"""The outbox — the owner-fired push queue and its state machine (V4 P2, §5.3).

Every push a brain sends is a row here, in the state home's SQLite database
(`$ALEXANDRIA_STATE_DIR/federation/outbox.sqlite`, WAL; override with
``FEDERATION_OUTBOX``). The queue enforces the two doctrine laws structurally:

- **Owner-fired.** A push is born ``staged`` and only the owner's :func:`fire`
  moves it to ``fired``; the worker deposits only ``fired`` rows. The one
  automatic-egress exception (§0 law 2) is a protocol receipt, staged already
  ``fired`` via :func:`enqueue_receipt`.
- **Seal at deposit, not stage.** The payload is stored as **plaintext** JSON so
  the staged manifest stays readable and a key rotation mid-queue still works;
  the worker seals it into an envelope at deposit time (P5), using the row ``id``
  as the envelope id.

State machine (``state`` column)::

    staged ──fire()──▶ fired ──deposit ok──▶ deposited ──receipt──▶ received│applied
                        │  ▲                     │
             deposit fail│  │backoff             └── (delivered; awaits receipt)
                        ▼  │
                    (retry, next_attempt)         403 unknown token ──▶ rejected
                        │
             14d after fired, still undelivered ──▶ expired

Terminal states: ``received``/``applied`` (success), ``expired``/``rejected``
(failure). **Expiry is 'gave up depositing'**: only an undelivered ``fired`` push
older than 14 days expires — a ``deposited`` push is delivered and never expires,
so a slow receipt (an offline peer that delivers late) is never mislabeled a
failure or lost. :func:`due` likewise refuses to re-deposit a push already past
its window, so the outbox is correct regardless of the worker tick's ordering.

**Delivery is at-least-once.** ``due`` hands a ``fired`` row to the worker, which
marks it ``deposited`` only *after* the network deposit; a crash in between (or a
second worker) re-deposits. That is safe because the receiver dedupes on the
authenticated envelope id (= the row id; bound into the sealed header in P1), so
a redelivery is idempotent at the peer. A ``depositing`` claim step would be
needed only if multiple workers ever ran (they don't — §5.6 is one worker).

Time is injectable everywhere (``now`` params) so tests drive the machine with a
synthetic clock — no sleeps. The DB path comes solely from the state home /
``FEDERATION_OUTBOX``; read-only calls never create it.
"""
import json
import os
import sqlite3
import time
import uuid
from contextlib import contextmanager
from pathlib import Path

from alexandria_vault.federation import identity

# State constants (also the set of legal `state` values).
STAGED, FIRED, DEPOSITED = "staged", "fired", "deposited"
RECEIVED, APPLIED, EXPIRED, REJECTED = "received", "applied", "expired", "rejected"
DECLINED = "declined"          # the peer's owner reviewed the push and chose not to file it
STATES = (STAGED, FIRED, DEPOSITED, RECEIVED, APPLIED, EXPIRED, REJECTED, DECLINED)

BACKOFF_BASE = 60          # first retry delay (s)
BACKOFF_CAP = 21600        # 6 h ceiling
EXPIRY_SECONDS = 14 * 24 * 3600   # an undelivered fired push expires 14 d after firing
BUSY_TIMEOUT_MS = 5000     # wait out a briefly-held write lock instead of raising

_SCHEMA = """
CREATE TABLE IF NOT EXISTS pushes (
    id           TEXT PRIMARY KEY,
    peer_fp      TEXT NOT NULL,
    kind         TEXT NOT NULL,
    payload      TEXT NOT NULL,          -- plaintext JSON body (sealed at deposit)
    state        TEXT NOT NULL,
    created      REAL NOT NULL,
    fired        REAL,
    attempts     INTEGER NOT NULL DEFAULT 0,
    next_attempt REAL,
    last_error   TEXT,
    receipt      TEXT                     -- JSON receipt body once delivered
);
"""


def _db_path(create=False):
    """Outbox DB path — under the federation state home, overridable via
    FEDERATION_OUTBOX (read lazily). ``create`` mkdirs the parent (0700 for the
    state home) for a write; a read passes ``create=False`` and never touches
    the filesystem, so introspection can't materialize the secrets directory."""
    override = os.environ.get("FEDERATION_OUTBOX")
    if override:
        p = Path(override).expanduser()
        if create:
            p.parent.mkdir(parents=True, exist_ok=True)
        return p
    return identity.federation_home(create=create) / "outbox.sqlite"


def _harden(path):
    """Owner-only (0600) on the outbox DB and its WAL/SHM sidecars. The outbox
    holds staged push bodies in PLAINTEXT (sealed only at deposit), so it is a
    secret like everything else in the state home — but sqlite3.connect creates
    the file under the process umask (0644 by default), unlike the 0600-from-
    creation writers in state/identity. Best-effort: a missing sidecar is fine."""
    for p in (path, path + "-wal", path + "-shm"):
        try:
            os.chmod(p, 0o600)
        except FileNotFoundError:
            pass


@contextmanager
def _conn(create=True):
    path = str(_db_path(create=create))
    conn = sqlite3.connect(path)
    try:
        conn.row_factory = sqlite3.Row
        conn.execute(f"PRAGMA busy_timeout={BUSY_TIMEOUT_MS}")
        if create:
            _harden(path)                          # the DB file exists after connect
            conn.execute("PRAGMA journal_mode=WAL")
            conn.execute(_SCHEMA)
            _harden(path)                          # WAL/SHM sidecars exist after the pragma
        yield conn
        conn.commit()
    finally:
        conn.close()


def _exists():
    return _db_path().exists()


def _backoff(attempts):
    """Delay before the next deposit attempt, given the failures so far."""
    return min(BACKOFF_BASE * (2 ** attempts), BACKOFF_CAP)


def _now(now):
    return time.time() if now is None else now


def _row_to_dict(row):
    return dict(row) if row is not None else None


# ---- enqueue ----------------------------------------------------------------
def stage(peer_fp, kind, body, *, now=None):
    """Store a new push in ``staged`` state; return its id (also the envelope id).

    ``body`` is the plaintext body dict, stored as JSON; nothing is sealed here.
    """
    if not peer_fp or not isinstance(kind, str) or not kind:
        raise ValueError("stage requires peer_fp and a non-empty kind")
    pid = uuid.uuid4().hex
    ts = _now(now)
    with _conn() as c:
        c.execute(
            "INSERT INTO pushes (id, peer_fp, kind, payload, state, created, attempts) "
            "VALUES (?,?,?,?,?,?,0)",
            (pid, peer_fp, kind, json.dumps(body, ensure_ascii=False), STAGED, ts))
    return pid


def enqueue_receipt(peer_fp, body, *, now=None):
    """Stage a receipt already ``fired`` — the one automatic egress (§0 law 2).
    Returns its id."""
    pid = uuid.uuid4().hex
    ts = _now(now)
    with _conn() as c:
        c.execute(
            "INSERT INTO pushes (id, peer_fp, kind, payload, state, created, fired, attempts) "
            "VALUES (?,?,?,?,?,?,?,0)",
            (pid, peer_fp, "receipt", json.dumps(body, ensure_ascii=False),
             FIRED, ts, ts))
    return pid


# ---- state transitions ------------------------------------------------------
def _transition(pid, to_state, from_states, **fields):
    """Move ``pid`` to ``to_state`` iff it is currently in one of ``from_states``,
    setting any extra columns atomically. Returns True if it transitioned. The
    single guarded-UPDATE shape every simple transition shares — a non-matching
    state no-ops (rowcount 0) rather than clobbering a row that moved on."""
    cols = ["state=?"]
    vals = [to_state]
    for k, v in fields.items():
        cols.append(f"{k}=?")
        vals.append(v)
    placeholders = ",".join("?" * len(from_states))
    vals += [pid, *from_states]
    with _conn() as c:
        cur = c.execute(
            f"UPDATE pushes SET {', '.join(cols)} WHERE id=? AND state IN ({placeholders})",
            vals)
        return cur.rowcount > 0


def fire(pid, *, now=None):
    """The owner's go: move a ``staged`` push to ``fired``. Returns True if it
    transitioned (a non-staged id returns False)."""
    return _transition(pid, FIRED, (STAGED,), fired=_now(now), next_attempt=None)


def cancel(pid):
    """Cancel a push the owner staged/fired by mistake (F12). Deletes the row while it is
    still ``staged`` or ``fired`` — i.e. NOT yet marked deposited. A ``deposited`` (or
    later) push is already sealed into the peer's mailbox and cannot be recalled.

    BEST-EFFORT for a ``fired`` push: a tick deposits over the network and only THEN
    marks the row ``deposited``, so cancelling during that in-flight window removes the
    row but the blob may still land at the peer (the caller should not promise otherwise).
    A ``staged`` push has never been sent, so its cancel is definitive.

    Returns ``{cancelled: True, was}`` on success, else ``{cancelled: False, reason}``
    (``not-found`` | ``too-late`` (+state) | ``raced`` — a concurrent tick marked it
    deposited between the read and the delete)."""
    row = get(pid)
    if row is None:
        return {"cancelled": False, "reason": "not-found"}
    if row["state"] not in (STAGED, FIRED):
        return {"cancelled": False, "reason": "too-late", "state": row["state"]}
    with _conn() as c:
        cur = c.execute("DELETE FROM pushes WHERE id=? AND state IN (?,?)",
                        (pid, STAGED, FIRED))
    if cur.rowcount > 0:
        return {"cancelled": True, "was": row["state"]}
    return {"cancelled": False, "reason": "raced", "state": (get(pid) or {}).get("state")}


def mark_deposited(pid):
    """Deposit succeeded: ``fired`` → ``deposited`` (awaiting a receipt)."""
    return _transition(pid, DEPOSITED, (FIRED,), last_error=None, next_attempt=None)


def mark_rejected(pid, error):
    """A permanent deposit refusal (e.g. relay 403 unknown token — the peer
    revoked us): → ``rejected``, no retry."""
    return _transition(pid, REJECTED, (FIRED, DEPOSITED), last_error=str(error)[:500])


def mark_received(pid, receipt=None):
    """A ``received`` receipt arrived from the peer."""
    return _transition(pid, RECEIVED, (FIRED, DEPOSITED), receipt=_receipt_json(receipt))


def mark_applied(pid, receipt=None):
    """An ``applied`` receipt arrived — the peer filed the intents. Accepts a
    prior ``received`` too, so a two-phase ack upgrades cleanly to the stronger
    terminal (the protocol sends one receipt today; this just never traps)."""
    return _transition(pid, APPLIED, (FIRED, DEPOSITED, RECEIVED),
                       receipt=_receipt_json(receipt))


def mark_declined(pid, receipt=None):
    """A ``declined`` receipt arrived — the peer's owner reviewed the push in their
    inbox and chose not to file it. Terminal, like applied."""
    return _transition(pid, DECLINED, (FIRED, DEPOSITED, RECEIVED),
                       receipt=_receipt_json(receipt))


def _receipt_json(receipt):
    return json.dumps(receipt, ensure_ascii=False) if receipt is not None else None


def mark_deposit_failed(pid, error, *, now=None):
    """A transient deposit failure: bump attempts and schedule the next attempt
    with exponential backoff. The read-modify-write stays on one connection and
    the UPDATE is state-guarded, so a receipt that lands between the SELECT and
    the UPDATE (flipping the row out of ``fired``) makes this a no-op instead of
    clobbering the delivered row."""
    ts = _now(now)
    with _conn() as c:
        row = c.execute("SELECT attempts FROM pushes WHERE id=? AND state=?",
                        (pid, FIRED)).fetchone()
        if row is None:
            return False
        attempts = row["attempts"]
        cur = c.execute(
            "UPDATE pushes SET attempts=?, next_attempt=?, last_error=? "
            "WHERE id=? AND state=?",
            (attempts + 1, ts + _backoff(attempts), str(error)[:500], pid, FIRED))
        return cur.rowcount > 0


GC_AFTER = 30 * 24 * 3600   # delete a settled push this long after it settled (fired)


def gc(now=None):
    """Delete settled pushes older than GC_AFTER so the queue can't grow without
    bound: terminal rows (applied/received/expired/rejected) AND `deposited` rows
    (delivered; a `deposited` receipt never gets a receipt of its own, so it would
    otherwise live forever). Live rows (staged/fired) are never touched. Returns the
    count removed.

    Age is measured from ``fired``, NOT ``created``: a push can sit ``staged`` for
    any length of time before the owner fires it, and aging from stage time would
    reclaim a freshly-``deposited`` row the moment it was delivered — leaving a late
    receipt nothing to land on. Every GC-eligible state was reached through firing
    (receipts are born fired), so ``fired`` is set on all of them; delivery and
    settlement then happen within the 14-day deposit window after it, so a
    ``deposited`` row always survives well past its delivery. ``COALESCE`` is
    belt-and-braces for a hand-edited row."""
    if not _exists():
        return 0
    cutoff = _now(now) - GC_AFTER
    settled = (RECEIVED, APPLIED, EXPIRED, REJECTED, DECLINED, DEPOSITED)
    with _conn() as c:
        placeholders = ",".join("?" * len(settled))
        cur = c.execute(
            f"DELETE FROM pushes WHERE COALESCE(fired, created)<? "
            f"AND state IN ({placeholders})",
            (cutoff, *settled))
        return cur.rowcount


def expire_due(now=None):
    """Expire only *undelivered* pushes: a ``fired`` row still not deposited more
    than 14 days after firing → ``expired``. A ``deposited`` push is delivered and
    is never swept (its receipt may arrive late). Returns the count expired."""
    if not _exists():
        return 0
    cutoff = _now(now) - EXPIRY_SECONDS
    with _conn() as c:
        cur = c.execute(
            "UPDATE pushes SET state=? WHERE state=? AND fired IS NOT NULL AND fired<?",
            (EXPIRED, FIRED, cutoff))
        return cur.rowcount


# ---- worker read ------------------------------------------------------------
def due(now=None):
    """Fired pushes ready for a deposit attempt: ``next_attempt`` has arrived AND
    the 14-day window has not elapsed (an over-age push is left for
    :func:`expire_due`, never re-deposited). Oldest first, as dicts carrying
    id/peer_fp/kind/payload/body for sealing."""
    if not _exists():
        return []
    ts = _now(now)
    cutoff = ts - EXPIRY_SECONDS
    with _conn(create=False) as c:
        rows = c.execute(
            "SELECT * FROM pushes WHERE state=? AND (next_attempt IS NULL OR next_attempt<=?) "
            "AND fired IS NOT NULL AND fired>=? ORDER BY created ASC",
            (FIRED, ts, cutoff)).fetchall()
    out = []
    for r in rows:
        d = _row_to_dict(r)
        d["body"] = json.loads(d["payload"])
        out.append(d)
    return out


# ---- introspection (read-only: never creates the DB or state home) ----------
def get(pid):
    if not _exists():
        return None
    with _conn(create=False) as c:
        return _row_to_dict(c.execute("SELECT * FROM pushes WHERE id=?", (pid,)).fetchone())


def list_for(peer_fp=None, *, limit=50):
    """Recent pushes (optionally for one peer), newest first — for `peer_status`."""
    if not _exists():
        return []
    with _conn(create=False) as c:
        if peer_fp is None:
            rows = c.execute("SELECT * FROM pushes ORDER BY created DESC LIMIT ?",
                             (int(limit),)).fetchall()
        else:
            rows = c.execute("SELECT * FROM pushes WHERE peer_fp=? ORDER BY created DESC "
                             "LIMIT ?", (peer_fp, int(limit))).fetchall()
    return [_row_to_dict(r) for r in rows]


def status(peer_fp=None):
    """Counts by state (optionally for one peer) — the digest/status summary."""
    if not _exists():
        return {}
    with _conn(create=False) as c:
        if peer_fp is None:
            rows = c.execute("SELECT state, COUNT(*) n FROM pushes GROUP BY state").fetchall()
        else:
            rows = c.execute("SELECT state, COUNT(*) n FROM pushes WHERE peer_fp=? "
                             "GROUP BY state", (peer_fp,)).fetchall()
    return {r["state"]: r["n"] for r in rows}
