"""Python 3.13 receiver helpers. Keep the SQLite inbox on durable storage.

Pass the exact request bytes before JSON parsing. Return 204 after accept() commits,
including for duplicates. A background job retries pending inbox rows until the
authenticated status/result GET succeeds; a webhook is not an instruction to rerun.
"""
import hashlib
import hmac
import json
import re
import sqlite3
import time
from urllib.parse import urlsplit
from uuid import UUID


def verify(body: bytes, header: str, secrets: list[str], *, now=None) -> dict:
    if len(body) > 4096:
        raise ValueError("Webhook body too large")
    match = re.fullmatch(r"t=([0-9]{1,12}),v1=([0-9a-f]{64})", header)
    if not match:
        raise ValueError("Invalid signature header")
    timestamp = int(match[1])
    if abs((time.time() if now is None else now) - timestamp) > 300:
        raise ValueError("Expired signature")
    signed = match[1].encode("ascii") + b"." + body
    if not any(hmac.compare_digest(hmac.new(secret.encode("ascii"), signed, hashlib.sha256).hexdigest(), match[2])
               for secret in secrets):
        raise ValueError("Invalid signature")
    event = json.loads(body)
    UUID(event["event_id"])
    if type(event["sequence"]) is not int or event["sequence"] < 1:
        raise ValueError("Invalid sequence")
    result_path(event)  # Validates execution IDs. Never follow an arbitrary payload URL.
    return event


def result_path(event: dict) -> str:
    if event["aggregate_kind"] == "SESSION":
        return f"/v1/sessions/{UUID(event['session_id'])}/turns/{UUID(event['turn_id'])}"
    if event["aggregate_kind"] == "RUN":
        return f"/v1/runs/{UUID(event['run_id'])}"
    raise ValueError("Unsupported aggregate")


def result_url(api_origin: str, event: dict) -> str:
    origin = urlsplit(api_origin)
    if (origin.scheme != "https" or not origin.netloc or origin.username or origin.password
            or origin.path not in ("", "/") or origin.query or origin.fragment):
        raise ValueError("Configure the API HTTPS origin, not a URL from the webhook")
    return api_origin.rstrip("/") + result_path(event)


def open_inbox(path: str) -> sqlite3.Connection:
    db = sqlite3.connect(path)
    db.execute("CREATE TABLE IF NOT EXISTS webhook_inbox (event_id TEXT PRIMARY KEY, body BLOB NOT NULL, done INTEGER NOT NULL DEFAULT 0)")
    db.commit()
    return db


def accept(db: sqlite3.Connection, body: bytes, header: str, secrets: list[str], *, now=None) -> bool:
    event = verify(body, header, secrets, now=now)
    with db:
        inserted = db.execute("INSERT INTO webhook_inbox(event_id,body) VALUES (?,?) ON CONFLICT(event_id) DO NOTHING",
                              (event["event_id"], body)).rowcount
        if db.execute("SELECT body FROM webhook_inbox WHERE event_id=?", (event["event_id"],)).fetchone()[0] != body:
            raise ValueError("Event identity changed")
    return bool(inserted)


async def refresh_pending(db, api_origin, fetch_current, apply_current, *, limit=100):
    """Single inbox consumer; callbacks fetch with the API key/no redirects and apply idempotently.

    An exception leaves the row pending for the next pass (including after restart).
    Both the final notification and a later-arriving old progress event fetch CURRENT
    API state. The notification's status/sequence never replace that authoritative state.
    """
    pending = db.execute("SELECT event_id,body FROM webhook_inbox WHERE done=0 ORDER BY rowid LIMIT ?", (limit,)).fetchall()
    for event_id, raw in pending:
        current = await fetch_current(result_url(api_origin, json.loads(raw)))
        await apply_current(current)
        with db:
            db.execute("UPDATE webhook_inbox SET done=1 WHERE event_id=?", (event_id,))
    return len(pending)


# Keep event IDs for at least the event's 30-day replay horizon. A shared production
# receiver uses a shared transactional inbox, not one SQLite file per server.