"""Docs: docs/website/content/bulk-email-language-series.md"""
import argparse
import json
import os
import re
from pathlib import Path

parser = argparse.ArgumentParser(description="Private bulk messages; MOCK by default")
parser.add_argument("--live", action="store_true")
parser.add_argument("--retry-failed", action="store_true")
parser.add_argument("--ledger")
parser.add_argument("--scenario", default="mixed", choices=["mixed", "timeout", "short", "reordered", "badstatus", "crash", "scheduled"])
args = parser.parse_args()
if args.live and args.scenario != "mixed":
    parser.error("Fault scenarios are mock-only")
sender = os.environ.get("MAILKITE_FROM") if args.live else "mock@demo.invalid"
if args.live and (not sender or not os.environ.get("MAILKITE_API_KEY")):
    parser.error("Set MAILKITE_FROM and MAILKITE_API_KEY")
path = Path(args.ledger or ("python-live.json" if args.live else "python-mock.json"))
fixture = ([{"to": "test-bulk-guide@mailkite.dev", "name": "Controlled test"}] if args.live else
           [{"to": f"person{i}@example.com", "name": f"Reader {i}"} for i in range(53)])
seen = set()
recipients = []
for row in fixture:
    if not re.fullmatch(r"[^\s@<>]+@[^\s@<>]+\.[^\s@<>]+", row["to"]) or not row["name"] or re.search(r"[\r\n]", row["name"]):
        raise ValueError("Invalid fixture")
    if row["to"].lower() not in seen:
        seen.add(row["to"].lower())
        recipients.append(row)
job = {"version": 1, "mode": "LIVE" if args.live else "MOCK", "from": sender, "recipients": recipients}
ledger = json.loads(path.read_text()) if path.exists() else {
    "job": job, "rows": [dict(r, state="pending", attempts=0) for r in recipients], "requests": []}
if ledger["job"] != job:
    raise ValueError("Ledger belongs to a different job or mode")


def save() -> None:
    """Replace the private single-worker ledger after writing a complete snapshot."""
    temporary = Path(str(path) + ".tmp")
    with os.fdopen(os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as file:
        json.dump(ledger, file, indent=2)
    os.replace(temporary, path)


for row in ledger["rows"]:
    if row["state"] == "inflight":
        row["state"] = "unknown"
save()


class MockMailKite:
    def sendBatch(self, batch: dict) -> dict:
        """Return ordered mock outcomes or simulate a lost/malformed response."""
        if args.scenario == "timeout":
            raise TimeoutError("Simulated lost response after request")
        results = []
        for recipient in batch["recipients"]:
            to = recipient["to"]
            row = next(r for r in ledger["rows"] if r["to"] == to)
            if to == "person1@example.com" and row["attempts"] == 1:
                result = {"to": to, "status": "failed", "code": "mock_transient", "error": "MOCK temporary refusal"}
            elif to == "person2@example.com":
                result = {"to": to, "status": "failed", "code": "recipient_suppressed", "error": "MOCK suppression"}
            else:
                status = "scheduled" if args.scenario == "scheduled" else "sent"
                prefix = "ssnd" if status == "scheduled" else "msg"
                result = {"to": to, "status": status, "id": f"{prefix}_mock_{to.split('@')[0]}"}
            results.append(result)
        if args.scenario == "short":
            results = results[1:]
        if args.scenario == "reordered":
            results.reverse()
        if args.scenario == "badstatus":
            results[0]["status"] = "queued"
        return dict(results=results, **{s: sum(r["status"] == s for r in results) for s in ("sent", "scheduled", "failed")})


if args.live:
    from mailkite import MailKite
    mk = MailKite(os.environ["MAILKITE_API_KEY"])
else:
    mk = MockMailKite()
eligible = [r for r in ledger["rows"] if r["state"] == "pending" or
            (args.retry_failed and r["state"] == "failed" and r.get("code") == "mock_transient")]
for offset in range(0, len(eligible), 50):
    chunk = eligible[offset:offset + 50]
    for row in chunk:
        row["state"] = "inflight"
        row["attempts"] += 1
    ledger["requests"].append([r["to"] for r in chunk])
    save()  # Persist intent before the SDK request.
    if args.scenario == "crash":
        raise SystemExit(75)
    try:
        response = mk.sendBatch({
            "from": sender,
            "subject": "Your update, {{name}}",
            "text": "Hi {{name}}, your requested product update is ready.",
            "recipients": [{"to": r["to"], "templateData": {"name": r["name"]}} for r in chunk],
        })
        results = response["results"]
        if not isinstance(results, list) or len(results) != len(chunk):
            raise ValueError("Result cardinality mismatch")
        for row, result in zip(chunk, results):
            if result["to"] != row["to"] or result["status"] not in ("sent", "scheduled", "failed") or (result["status"] != "failed" and not result.get("id")):
                raise ValueError("Invalid ordered result")
        for status in ("sent", "scheduled", "failed"):
            if response[status] != sum(r["status"] == status for r in results):
                raise ValueError("Count mismatch")
    except Exception:
        for row in chunk:
            row["state"] = "unknown"
        ledger["problem"] = "Request or response ambiguous; reconcile before resending"
        save()
        break
    for row, result in zip(chunk, results):
        row.update(state=result["status"], id=result.get("id"), code=result.get("code"))
    save()
print(json.dumps({"mode": job["mode"], "ledger": str(path), "counts": {
    s: sum(r["state"] == s for r in ledger["rows"]) for s in ("pending", "sent", "scheduled", "failed", "unknown")}}))