# Docs: /guides/send-cold-email-python
# LOCAL SIMULATION ONLY. SQLite transactions checkpoint events and attempts.
import argparse
import json
import sqlite3

FIXTURES = [
    ("ada@example.com", "Ada", "your webhook retry write-up"),
    ("bo@example.com", "Bo", "your queue debugging notes"),
    ("cy@example.com", "Cy", "your mail parsing experiment"),
    ("dev@example.com", "Dev", "your developer tooling post"),
]


def open_state(path: str) -> sqlite3.Connection:
    """Open or seed a persistent local simulation without resetting existing jobs."""
    db = sqlite3.connect(path)
    db.row_factory = sqlite3.Row
    db.executescript("""
        CREATE TABLE IF NOT EXISTS contacts(email TEXT PRIMARY KEY, name TEXT, context TEXT, stopped TEXT);
        CREATE TABLE IF NOT EXISTS jobs(key TEXT PRIMARY KEY, email TEXT, step INTEGER, due INTEGER, status TEXT);
        CREATE TABLE IF NOT EXISTS events(id TEXT PRIMARY KEY, email TEXT, kind TEXT);
        CREATE TABLE IF NOT EXISTS attempts(key TEXT PRIMARY KEY, email TEXT, step INTEGER, at INTEGER, text TEXT);
        CREATE TABLE IF NOT EXISTS pacing(id INTEGER PRIMARY KEY, next_slot INTEGER);
        INSERT OR IGNORE INTO pacing VALUES(1, 0);
    """)
    with db:
        for email, name, context in FIXTURES:
            db.execute("INSERT OR IGNORE INTO contacts VALUES(?,?,?,NULL)", (email, name, context))
            for step in (0, 1):
                db.execute("INSERT OR IGNORE INTO jobs VALUES(?,?,?,?, 'pending')", (f"wave1:{email}:{step}", email, step, 0 if step == 0 else None))
    return db


def apply_event(db: sqlite3.Connection, event_id: str, email: str, kind: str) -> bool:
    """Commit one internal stop event and cancel only its recipient's pending jobs."""
    if not event_id or kind not in {"reply", "opt-out", "bounce"}:
        raise ValueError("Invalid internal event")
    email = email.strip().lower()
    with db:
        db.execute("BEGIN IMMEDIATE")
        if not db.execute("SELECT 1 FROM contacts WHERE email=?", (email,)).fetchone():
            raise ValueError("Unknown recipient")
        inserted = db.execute("INSERT OR IGNORE INTO events VALUES(?,?,?)", (event_id, email, kind)).rowcount
        if not inserted:
            return False
        db.execute("UPDATE contacts SET stopped=COALESCE(stopped,?) WHERE email=?", (kind, email))
        db.execute("UPDATE jobs SET status='canceled' WHERE email=? AND status='pending'", (email,))
    return True


def tick(db: sqlite3.Connection, now: int) -> dict[str, object] | None:
    """Record at most one due simulated attempt within the global pacing limit."""
    if not isinstance(now, int) or now < 0:
        raise ValueError("Virtual clock must be nonnegative integer seconds")
    with db:
        db.execute("BEGIN IMMEDIATE")
        if now < db.execute("SELECT next_slot FROM pacing WHERE id=1").fetchone()[0]:
            return None
        job = db.execute("""SELECT j.*, c.name, c.context FROM jobs j JOIN contacts c ON c.email=j.email
            WHERE j.status='pending' AND j.due IS NOT NULL AND j.due<=? AND c.stopped IS NULL
            ORDER BY j.step, j.email LIMIT 1""", (now,)).fetchone()
        if not job:
            return None
        text = f'Hi {job["name"]}, I read {job["context"]}. ' + ("Following up once." if job["step"] else "Could we discuss it?")
        # Unique key and job transition commit together. This records simulation, not delivery.
        db.execute("INSERT INTO attempts VALUES(?,?,?,?,?)", (job["key"], job["email"], job["step"], now, text))
        db.execute("UPDATE jobs SET status='simulated' WHERE key=?", (job["key"],))
        db.execute("UPDATE pacing SET next_slot=? WHERE id=1", (now + 60,))
        if job["step"] == 0:
            db.execute("UPDATE jobs SET due=? WHERE email=? AND step=1", (now + 86400, job["email"]))
        return dict(job)


def scenario(db: sqlite3.Connection) -> dict[str, str | int]:
    """Run the deterministic four-recipient scenario and return persisted counts."""
    for now in (0, 60, 120, 180):
        tick(db, now)
    for event in (("reply-1", "ada@example.com", "reply"), ("optout-1", "bo@example.com", "opt-out"), ("bounce-1", "cy@example.com", "bounce"), ("reply-1", "ada@example.com", "reply")):
        apply_event(db, *event)
    tick(db, 86580)
    return {"mode": "LOCAL SIMULATION, no emails sent",
            "attempts": db.execute("SELECT COUNT(*) FROM attempts").fetchone()[0],
            "canceled": db.execute("SELECT COUNT(*) FROM jobs WHERE status='canceled'").fetchone()[0],
            "events": db.execute("SELECT COUNT(*) FROM events").fetchone()[0]}


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Local outreach simulation; no transport")
    parser.add_argument("--state", required=True)
    args = parser.parse_args()
    connection = open_state(args.state)
    try:
        print(json.dumps(scenario(connection), indent=2))
    finally:
        connection.close()