#!/usr/bin/env python3
"""
Inbound-mail ledger (off-repo, leaf module)
===========================================
Message-ID idempotency for the webhook path, and the thread map that lets a reply
land back on the conversation it belongs to.

**Why this has to exist.** Postmark redelivers on any non-2xx, and one inbound
message costs a model routing pass plus a git commit. Without a claim taken
*before* the model call, a single retry files the same mail twice — two notes, two
commits, and an owner who now distrusts the filing. The claim is therefore
pessimistic: it is written first and completed after, so a retry that arrives
mid-ingest is told ``in_flight`` rather than being allowed to start a second pass.

Lives OUTSIDE the vault checkout (``ALEXANDRIA_STATE_DIR``, default
``~/.alexandria``; a container sets it to the volume) because it holds sender
Message-IDs — correspondence metadata that must never be committed or synced.

Deliberately a **separate file** from ``engine/team_state.py`` despite sitting in
the same directory: that store is driven by an IMAP cron, this one by a threaded
server, and sharing a JSON file between them would be a data race.

Degradation is asymmetric on purpose. An unreadable/unwritable ledger returns
"new" rather than raising, because dropping the owner's mail is a worse failure
than filing it twice — and the duplicate is git-revertible while the drop is
invisible.
"""
import hashlib
import json
import os
import threading
import time
from pathlib import Path


def _state_dir():
    d = os.environ.get("ALEXANDRIA_STATE_DIR")
    return Path(d).expanduser() if d else (Path.home() / ".alexandria")


def _path():
    # Resolved per call, not at import: the tests (and a re-exec'd container)
    # set ALEXANDRIA_STATE_DIR after this module is already loaded.
    return Path(os.environ.get("MAIL_STATE",
                               str(_state_dir() / "mail-inbound-state.json")))


INFLIGHT_TTL = 900        # a claim older than this is assumed dead and reclaimable
_KEEP = 500               # ring cap, so the ledger can't grow without bound
_LOCK = threading.Lock()  # the server is multi-threaded; the file is one resource


def key_for(tenant, message_id=None, provider_message_id=None) -> str:
    """The dedup key: tenant + the most durable id the message carries.

    Prefers the RFC ``Message-ID`` (stable across redeliveries) over the provider's
    own id. Hashed so the ledger never stores a raw correspondent id in the clear.
    Returns ``""`` when the message has no id at all — the caller must then skip
    dedup rather than collapse every anonymous message onto one key."""
    ident = str(message_id or provider_message_id or "").strip().lower()
    if not ident:
        return ""
    return hashlib.sha256(f"{tenant or ''}\0{ident}".encode("utf-8")).hexdigest()


def _load():
    try:
        data = json.loads(_path().read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except Exception:  # noqa: BLE001 — absent/corrupt ledger behaves as empty
        return {}


def _save(data):
    """Atomic replace, best-effort. A failed write must not fail an ingest that
    has already landed in git."""
    order = data.get("order") or []
    if len(order) > _KEEP:
        for stale in order[:-_KEEP]:
            (data.get("seen") or {}).pop(stale, None)
        data["order"] = order[-_KEEP:]
    try:
        p = _path()
        p.parent.mkdir(parents=True, exist_ok=True)
        tmp = p.with_suffix(".tmp")
        tmp.write_text(json.dumps(data), encoding="utf-8")
        os.replace(tmp, p)
        return True
    except Exception:  # noqa: BLE001
        return False


def claim(key, subject=""):
    """Stake a claim on one message before doing any expensive or committing work.

    Returns ``"new"`` (proceed), ``"in_flight"`` (a retry arrived while the first
    pass is still running — answer 200 and do nothing), or ``{"done": result}``
    (already handled; replay the recorded answer)."""
    if not key:
        return "new"                      # no id to dedup on — never collapse these
    with _LOCK:
        data = _load()
        seen = data.setdefault("seen", {})
        entry = seen.get(key)
        now = time.time()
        if entry:
            if entry.get("state") == "done":
                return {"done": entry.get("result") or {}}
            if now - float(entry.get("at") or 0) < INFLIGHT_TTL:
                return "in_flight"
            # A stale claim means the process died mid-ingest. Retrying is the
            # right call: a lost message is worse than a possible duplicate.
        seen[key] = {"state": "in_flight", "at": now, "subject": str(subject or "")[:120]}
        data.setdefault("order", []).append(key)
        _save(data)
        return "new"


def complete(key, result=None):
    """Record the outcome so a later redelivery replays it instead of re-ingesting."""
    if not key:
        return
    with _LOCK:
        data = _load()
        entry = (data.setdefault("seen", {})).setdefault(key, {})
        entry.update(state="done", at=time.time(), result=result or {})
        _save(data)


def release(key):
    """Drop a claim that never completed, so an immediate retry can proceed rather
    than waiting out the TTL. Used when the ingest failed before writing anything."""
    if not key:
        return
    with _LOCK:
        data = _load()
        if (data.get("seen") or {}).pop(key, None) is not None:
            data["order"] = [k for k in (data.get("order") or []) if k != key]
            _save(data)


# ---- thread map -------------------------------------------------------------------
# Which capture note a conversation belongs to, keyed by every Message-ID seen in it.
# Reconstructible from the notes' `email_thread_root` frontmatter if this is ever
# lost, so it is a cache with a long memory rather than a source of truth.

def remember_thread(tenant, ids, note, root=None):
    """Map each of ``ids`` to ``note`` so a reply threading off any of them finds it."""
    ids = [str(i).strip() for i in (ids or []) if str(i or "").strip()]
    if not ids or not note:
        return
    with _LOCK:
        data = _load()
        threads = data.setdefault("threads", {})
        for i in ids:
            k = key_for(tenant, i)
            if k:
                threads[k] = {"note": note, "root": root or ids[0], "at": time.time()}
        order = data.setdefault("thread_order", [])
        order.extend(key_for(tenant, i) for i in ids)
        if len(order) > _KEEP:
            for stale in order[:-_KEEP]:
                threads.pop(stale, None)
            data["thread_order"] = order[-_KEEP:]
        _save(data)


def thread_for(tenant, ids):
    """The recorded thread for the first of ``ids`` we know, else ``None``.

    Callers pass References newest-last, so this walks in reverse: the nearest
    ancestor is a better answer than the thread root when a conversation forked."""
    data = _load()
    threads = data.get("threads") or {}
    for i in reversed([str(x).strip() for x in (ids or []) if str(x or "").strip()]):
        hit = threads.get(key_for(tenant, i))
        if hit:
            return hit
    return None
