"""Receive: poll our own relay mailbox and process each sealed envelope (§5.5).

Push-only holds: this is the ONE place inbound federation is handled, and it only
ever acts on envelopes the brain *fetched itself* from its own mailbox — never a
listener. The gates run in V1 order, and a received envelope is DATA, never
instructions (the shared applier does the filing; nothing here executes message
content).

Gates, in order:
  1. allowlist — the envelope's ``from`` fingerprint must be a configured peer
  2. open — the sealed envelope must decrypt+authenticate under that peer's key
     (this also authenticates from/to/id/kind — the P1 header binding)
  4. validate_intents — only the fixed intent schema survives
  5. permissions/size — per-peer scope (handled inside apply_intents)
  6. dedupe — an already-processed envelope id is skipped

``receipt`` envelopes update the outbox (only for a push we actually sent that
peer); ``ping`` envelopes are acked with a receipt. A peer's first
``review_first_n`` *content-bearing* envelopes route wholesale to
``Agent/Federation Review.md`` regardless of permissions.

Robustness: every blob is processed in isolation — one malformed authenticated
body from an allowlisted peer is audited and dropped, never allowed to wedge the
queue. The envelope id is recorded as processed BEFORE the (non-idempotent) vault
write, via one atomic test-and-set under the cross-process state lock, so a
redelivered or concurrently-polled copy files at most once.
"""
import base64
import datetime as dt
import time
from pathlib import Path

from alexandria_vault.federation import (identity, envelope, outbox,
                                         apply as fed_apply, state as fed_statefiles)
from alexandria_vault.engine import federation_state as fstate

# Pages one tick may pull from the mailbox while paging PAST deferred blobs
# (RELAY-REVIEW F1). Each page is one poll against the relay's 600/hour bucket; at
# the 60 s default tick, 10 pages per tick IS that whole budget — so this cap is a
# ceiling for a pathological mailbox, not a steady state (a healthy tick is one page).
# Where a tick stops is remembered (state.poll_cursor_*) and the next tick resumes
# there, so a deferred head deeper than the cap still cannot starve what is behind it.
MAX_POLL_PAGES = 10


def peers_by_fingerprint(cfg):
    """The configured v2 (relay) peers, keyed by fingerprint. Shared by the receive
    and deposit paths so both agree on who is eligible."""
    return {p["fingerprint"]: p for p in (cfg.get("peers") or [])
            if p.get("fingerprint") and p.get("pubkey")}


def _relay_url(cfg):
    return (cfg.get("federation") or {}).get("relay_url")


def _fresh(ts_str, now):
    """True if an envelope's (authenticated) ts is within [now-14d, now+1d]. An
    unparseable, too-old, or far-future ts is refused — the replay-window bound."""
    try:
        t = dt.datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%SZ").replace(
            tzinfo=dt.timezone.utc).timestamp()
    except (TypeError, ValueError):
        return False
    n = time.time() if now is None else now
    return (n - fstate.MAX_ENVELOPE_AGE) <= t <= (n + 86400)


def poll_once(vault, cfg, relay_client, *, now=None, today=None, commit=None):
    """Fetch our mailbox and process every blob in isolation. Returns a list of
    per-envelope result dicts. Writes vault files directly; ``commit`` (if given) is
    called with a message after each envelope that wrote the vault — one commit per
    applied envelope (§5.6). The served worker passes the git-writer's committer
    (holding its write lock); the CLI passes its own path-scoped committer."""
    vault = Path(vault)
    today = today or dt.date.today().isoformat()
    priv, pub = identity.load_or_create()
    our_fp = identity.fingerprint(pub)
    mb = fed_statefiles.mailbox_for(_relay_url(cfg))
    bearer = mb.get("bearer")
    if not bearer:
        return []                                        # no mailbox yet — nothing to poll
    if mb.get("fingerprint") not in (None, our_fp):
        # The cached bearer was registered for a DIFFERENT identity (the brain.key was
        # rotated/restored). It would only 401 — don't poll; a re-pair re-registers.
        fstate.append_audit({"ts": today, "transport": "relay",
                             "skipped": "mailbox bearer is for a prior identity — re-pair"})
        return []
    peers = peers_by_fingerprint(cfg)
    results = []
    # Settle first: captures the owner filed or declined since the last pass get their
    # receipts (and their item files retired) before anything new arrives.
    settled = _settle_reviewed(vault, now)
    if settled:
        results.append({"status": "reconciled", "settled": settled,
                        "paths": [s["capture"] for s in settled if s.get("removed")]
                                 + [str(_rs().QUEUE_REL), str(_rs().NOTE_REL)]})
        if commit:
            commit(f"federation: receipts for {len(settled)} reviewed item(s)")

    # Paging (F1): a blob we DEFER (unknown sender, mid-pairing) stays unacked at the
    # head of the mailbox, and the relay's poll is oldest-first within a byte budget —
    # so without a cursor two deferred full-size blobs starved everything behind them
    # for their 14-day TTL. We remember what we deferred this tick, and when a page's
    # last blob is one of them we ask the relay for the page AFTER it; when the queue
    # moved (something was acked) we re-read from the head, skipping the deferred ids
    # we already saw. Deferred blobs are looked at again next tick (they may have
    # become a configured peer by then).
    seen_deferred = set()                    # deferred this tick: skip on a re-read
    handled = set()                          # processed this tick: skip on a re-read
    #                                          (an ack that failed would otherwise be
    #                                          re-read as a "duplicate" up to the cap — §10 #6)
    relay_url = _relay_url(cfg)
    # Resume where the previous tick ran out of pages (§10 #3), once: the tick after
    # a resumed one starts from the head again, so deferred blobs that became
    # processable are still revisited.
    stored = fed_statefiles.poll_cursor_get(relay_url)
    cursor = stored.get("cursor") if stored and not stored.get("used") else None
    if cursor:
        fed_statefiles.poll_cursor_set(relay_url, cursor, used=True)
    stopped_at = None                        # set when the cap is hit mid-walk
    pages = 0
    while pages < MAX_POLL_PAGES:
        pages += 1
        status, payload = relay_client.poll(our_fp, bearer, after=cursor)
        if status == 401:
            # Our cached bearer is for a mailbox the relay no longer has (its idle GC
            # reclaimed it — RELAY-REVIEW F4). Re-register now so the next tick polls a
            # live mailbox instead of 401ing forever; there is nothing to fetch yet.
            _reregister(cfg, relay_client, bearer, today, now)
            return results
        if status == 400 and cursor is not None:
            cursor = None                    # the resume point is gone (acked/purged/
            continue                         # swept): start this tick from the head
        if status != 200 or not payload:
            break
        blobs = payload.get("blobs", [])
        if not blobs:
            break
        if cursor is not None and any(b.get("id") == cursor for b in blobs):
            # A relay that honours `after` never returns the cursor blob itself: this
            # one ignored it (a pre-1.2.0 relay answers from the head). Looping would
            # only spend the poll budget (§10 #2) — the head is reachable next tick.
            break
        fresh = [b for b in blobs
                 if b.get("id") not in seen_deferred and b.get("id") not in handled]
        progressed = False
        for blob in fresh:
            result = _process_blob(blob, our_fp, bearer, priv, peers, relay_client,
                                   vault, cfg, today, now)
            results.append(result)
            if result.get("status") == "deferred":
                seen_deferred.add(blob.get("id"))
            else:
                handled.add(blob.get("id"))
                progressed = True
            wrote = (result.get("applied") or []) + (result.get("parked") or [])
            if commit and result.get("status") in ("applied", "received") and wrote:
                verb = "captured for review" if result["status"] == "received" else "applied"
                commit(f"federation: {len(wrote)} intent(s) from {result.get('from') or 'peer'} "
                       f"{verb}")
        last_id = blobs[-1].get("id")
        if last_id in seen_deferred:
            cursor = last_id                 # the head is stuck here: page past it
            stopped_at = last_id
        elif progressed:
            cursor = None                    # the queue moved: re-read from the head
            stopped_at = None
        else:
            break
    else:
        # Ran out of pages while still walking past deferred blobs: remember where,
        # so the next tick resumes instead of re-walking the same head.
        if stopped_at is not None:
            fed_statefiles.poll_cursor_set(relay_url, stopped_at, used=False)
            return results
    fed_statefiles.poll_cursor_set(relay_url, None)      # walked to the end: nothing to resume
    return results


def _reregister(cfg, relay_client, stale_bearer, today, now):
    from alexandria_vault.federation import pairing   # lazy: pairing imports sodium
    try:
        pairing.reregister_mailbox(cfg, relay_client, now=now, stale=stale_bearer)
        _audit({"ts": today, "transport": "relay",
                "reregistered": "relay answered 401 — mailbox was reclaimed; registered anew"})
    except Exception as exc:  # noqa: BLE001 — next tick tries again; the owner sees the audit
        _audit({"ts": today, "transport": "relay",
                "error": f"relay answered 401 and re-registration failed: {exc}"[:200]})


def _rs():
    from alexandria_vault.federation import review_store
    return review_store


def _process_blob(blob, our_fp, bearer, priv, peers, relay_client, vault, cfg,
                  today, now):
    blob_id = blob.get("id")
    # --- pre-crypto: parse (junk -> drop) ---
    try:
        raw = base64.b64decode(blob.get("body_b64", ""), validate=True)
        env = envelope.parse(raw)
    except Exception as exc:  # noqa: BLE001
        return _drop(our_fp, bearer, relay_client, blob_id, None, f"parse: {exc}", today)
    # --- gate 1: allowlist by from-fingerprint ---
    peer = peers.get(env["from"])
    if peer is None:
        # Could be a peer not yet in config (pairing race) — LEAVE it for a later
        # poll rather than delete; bounded by the relay's 14-day blob TTL.
        _audit({"ts": today, "transport": "relay", "from": env["from"],
                "deferred": "unknown sender"})
        return {"status": "deferred", "reason": "unknown sender"}
    # --- gate 2: open + authenticate (tamper/wrong-key -> drop) ---
    try:
        peer_pub = base64.b64decode(peer["pubkey"], validate=True)
        body = envelope.open_envelope(env, priv, peer_pub)
    except envelope.EnvelopeError as exc:
        return _drop(our_fp, bearer, relay_client, blob_id, env, f"open failed: {exc}",
                     today)
    # --- freshness: the ts is authenticated by open; a stale one is a malicious relay
    #     re-injecting a captured blob (a legit one can't outlive the 14-day blob TTL). ---
    if not _fresh(env.get("ts"), now):
        return _drop(our_fp, bearer, relay_client, blob_id, env,
                     "stale or future envelope (replay?)", today)
    eid = env["id"]
    peer_fp = env["from"]
    # --- gate 6: dedupe (authenticated id, per-peer + time-evicted) ---
    # ONE atomic test-and-set under the cross-process state lock, recorded BEFORE
    # acting: of N concurrent deliveries of the same id (served worker + CLI tick,
    # or a redelivered blob) exactly one caller sees "newly recorded" and files.
    # A separate is_processed_envelope() read here would reopen the race — two
    # pollers could both pass the stale check and double-file.
    if not fstate.mark_processed_envelope(peer_fp, eid, now=now):
        _ack(our_fp, bearer, relay_client, blob_id)
        return {"status": "duplicate", "from": peer.get("handle")}

    kind = env["kind"]
    try:
        if kind == "receipt":
            result = _handle_receipt(peer, body["receipt"])
        elif kind == "ping":
            result = _handle_ping(peer, eid, now)
        else:
            result = _handle_intents(vault, cfg, peer, body.get("intents"), today,
                                     eid, now)
    except Exception as exc:  # noqa: BLE001
        # A malformed authenticated body from an allowlisted peer must not wedge the
        # queue for everyone: audit, drop, move on (the id is already marked).
        _audit({"ts": today, "transport": "relay", "from": env["from"],
                "error": str(exc)})
        _ack(our_fp, bearer, relay_client, blob_id)
        return {"status": "error", "from": peer.get("handle"), "reason": str(exc)[:80]}
    _audit({"ts": today, "transport": "relay", "from": env["from"],
            "peer": peer.get("name"), "kind": kind, "result": result})
    _ack(our_fp, bearer, relay_client, blob_id)
    return result


def _handle_receipt(peer, receipt):
    """Resolve the outbox push a receipt acks — but ONLY if that push was actually
    sent to THIS peer. Envelope ids are cleartext transport headers, so without the
    binding a peer could forge an ack for a push we sent someone else."""
    of = receipt.get("of")
    push = outbox.get(of)
    if not push or push.get("peer_fp") != peer.get("fingerprint"):
        return {"status": "receipt-ignored", "from": peer.get("handle"), "of": of}
    st = receipt.get("status")
    if st == "applied":
        outbox.mark_applied(of, receipt)
    elif st == "received":
        outbox.mark_received(of, receipt)
    elif st == "declined":
        outbox.mark_declined(of, receipt)
    return {"status": "receipt", "from": peer.get("handle"), "of": of,
            "receipt_status": st}


def _handle_ping(peer, eid, now):
    # Ack a ping with a receipt so the sender's outbound ping resolves (round-trip).
    outbox.enqueue_receipt(peer["fingerprint"],
                           envelope.receipt_body(eid, "received"), now=now)
    return {"status": "ping", "from": peer.get("handle")}


def _handle_intents(vault, cfg, peer, intents, today, eid, now):
    """Content-bearing envelope → CAPTURE (default) or apply (a peer the owner opted in
    with ``auto_apply: true``). Capture is an ingestion: each intent becomes a pending
    item in ``Agent/inbox/`` with the sender's destination as a suggestion, plus a ledger
    entry carrying the envelope id; the peer gets a ``received`` receipt now and an
    ``applied``/``declined`` one after the owner acts (see ``review_store.reconcile``).
    Nothing peer-supplied is written into the brain's notes on this path."""
    valid = fed_apply.validate_intents({"intents": intents or []})   # gate 4
    if not valid:
        # Nothing to file: just ack with an (empty) receipt so the sender's push resolves.
        outbox.enqueue_receipt(peer["fingerprint"],
                               envelope.receipt_body(eid, "applied", [], []), now=now)
        return {"status": "applied", "from": peer.get("handle"),
                "applied": [], "parked": [], "paths": []}
    fstate.bump_peer_intent_count(peer["fingerprint"])           # only content counts
    if peer.get("auto_apply") is True:
        # The owner's explicit per-peer opt-in (a trusted automated feed): the fixed
        # applier writes in-scope intents; out-of-scope ones still park.
        applied, parked = fed_apply.apply_intents(vault, peer, valid, cfg, today)  # gate 5
        outbox.enqueue_receipt(
            peer["fingerprint"],
            envelope.receipt_body(eid, "applied", applied, parked), now=now)
        return {"status": "applied", "from": peer.get("handle"),
                "applied": applied, "parked": parked, "paths": []}
    parked, paths = _capture_all(vault, peer, valid, today, eid)
    outbox.enqueue_receipt(
        peer["fingerprint"],
        envelope.receipt_body(eid, "received", [], parked), now=now)
    return {"status": "received", "from": peer.get("handle"),
            "applied": [], "parked": parked, "paths": paths}


def _capture_all(vault, peer, intents, today, eid):
    """Capture each intent as an inbox item + ledger entry. Returns (labels, paths)."""
    from alexandria_vault.federation import review_store
    labels, paths = [], []
    for it in intents:
        t = (it.get("type") or "?").strip().lower()
        area_raw = (it.get("area") or "").strip() or None
        area = fed_apply._safe_component(area_raw) if area_raw else None
        in_scope = not (area_raw and area is None) and fed_apply._permits(peer, t, area)
        reason = "held for your review" if in_scope else "outside this peer's grant"
        iid, rel = review_store.capture(vault, peer, it, reason, today, eid=eid)
        labels.append(f"{t} -> inbox ({iid})")
        paths.append(rel)
    paths += [str(review_store.QUEUE_REL), str(review_store.NOTE_REL)]
    return labels, paths


def _settle_reviewed(vault, now):
    """Send the receipts for captures the owner has acted on since the last pass."""
    from alexandria_vault.federation import review_store
    settled = review_store.reconcile(vault)
    for s in settled:
        if s.get("eid") and s.get("peer_fp"):
            outbox.enqueue_receipt(
                s["peer_fp"],
                envelope.receipt_body(s["eid"], s["outcome"],
                                      [s["label"]] if s["outcome"] == "applied" else [],
                                      [] if s["outcome"] == "applied" else [s["label"]]),
                now=now)
    return settled


def _ack(our_fp, bearer, relay_client, blob_id):
    """Delete the blob from the mailbox — the pickup ack. Best-effort."""
    try:
        relay_client.delete_blob(our_fp, bearer, blob_id)
    except Exception:  # noqa: BLE001
        pass


def _drop(our_fp, bearer, relay_client, blob_id, env, reason, today):
    """Audit + delete a blob we will never process (junk/tamper)."""
    _audit({"ts": today, "transport": "relay", "from": (env or {}).get("from"),
            "rejected": reason})
    _ack(our_fp, bearer, relay_client, blob_id)
    return {"status": "rejected", "reason": reason[:80]}


def _audit(record):
    fstate.append_audit(record)
