#!/usr/bin/env python3
"""AICEP v2.3 transport-adapter interface (proposed Phase 2 extension).

The AICEP batch is the message. Adapters move opaque bytes and report delivery state; they never see
plaintext or keys. Their guarantees are hints only: the core still enforces seq, dedup, expiry, signatures.
"""
import json, os, shutil, time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from aicep_common import atomic_write, canonical, check_slug, load_json

READY = "READY"


@dataclass
class PublishReceipt:
    transport_ref: str
    durable: bool


@dataclass
class Delivery:
    batch_id: str
    path: str                 # local path of the fetched batch dir (opaque files)
    transport_ref: str
    delivery_count: int
    lease_until: int
    attrs: dict = field(default_factory=dict)


class TransportAdapter(ABC):
    @abstractmethod
    def publish(self, batch_dir: str, attrs: dict) -> PublishReceipt:
        """Durable, atomic-visibility, idempotent on batch_id."""

    @abstractmethod
    def fetch(self, compartment: str, max_items: int, lease_secs: int, consumer: str, folders=()) -> list:
        """Lease complete deliveries from the given sender folders only (v2.5); incomplete/staged
        batches and unlisted folders are never returned. Delivery.attrs['folder'] = source folder."""

    @abstractmethod
    def ack(self, delivery: Delivery) -> None:
        """Final transport settle after core commit."""

    @abstractmethod
    def nack(self, delivery: Delivery, retry_after_secs: int, reason: str) -> None:
        """Transient failure: release lease, re-visible after delay, delivery_count++."""

    @abstractmethod
    def dead_letter(self, delivery: Delivery, reason_record: dict) -> None:
        """Move to DLQ with signed reason record. Never deletes."""

    @abstractmethod
    def hold(self, delivery: Delivery) -> None:
        """Move an out-of-order batch to the holding area."""

    def lease_valid(self, delivery: Delivery, consumer: str) -> bool:
        """Re-check the lease right before commit (v2.4). Adapters without leases may return True."""
        return True

    @abstractmethod
    def list_pending(self, compartment: str) -> list:
        """Read-only view: [{batch_id, state, delivery_count}]."""

    @abstractmethod
    def capabilities(self) -> dict:
        ...


class LocalFolderAdapter(TransportAdapter):
    """Folder bus (v2.5): senders publish to <root>/<compartment>/pub/<folder>/<batch_id>/ where <folder> is the
    sender's registered slug; <root>/<compartment>/{_staging/<folder>,processing,holding,done,dlq}/<batch_id>/.
    The source folder is kept in _meta so hold/redrive return a batch to the same folder.
    Commit marker: READY written LAST (after manifest.json.sig); fetch skips dirs without it."""

    def __init__(self, root, now=time.time, crash_after_files=None):
        self.root, self.now = root, now
        self.crash_after_files = crash_after_files      # test hook: simulate half-upload

    def _d(self, comp, area, bid=""):
        return os.path.join(self.root, comp, area, bid)

    def _meta(self, comp, bid):
        return os.path.join(self.root, comp, "_meta", bid + ".json")

    def capabilities(self):
        return {"atomic_publish": "emulated(READY marker)", "native_dedup_window": 0, "native_ordering": False,
                "native_dlq": False, "native_lease": "emulated(lease file)", "max_blob_bytes": None}

    def publish(self, batch_dir, attrs):
        comp, bid = attrs["compartment"], attrs["batch_id"]
        folder = check_slug(attrs.get("folder"))                    # v2.5: per-agent folder required
        dest = os.path.join(self._d(comp, "pub"), folder, bid)
        if os.path.exists(os.path.join(dest, READY)):
            return PublishReceipt(dest, True)                       # idempotent
        stage = os.path.join(self._d(comp, "_staging"), folder, bid)
        if os.path.exists(stage):
            shutil.rmtree(stage)                                    # discard a half-upload
        os.makedirs(stage)
        names = sorted(n for n in os.listdir(batch_dir) if n != "manifest.json.sig") + ["manifest.json.sig"]
        for i, n in enumerate(names):                               # signature uploaded last
            if self.crash_after_files is not None and i >= self.crash_after_files:
                raise IOError("simulated crash mid-upload")
            shutil.copyfile(os.path.join(batch_dir, n), os.path.join(stage, n))
        open(os.path.join(stage, READY), "w").write("ok\n")          # commit marker last
        os.makedirs(os.path.dirname(dest), exist_ok=True)
        if os.path.exists(dest):
            shutil.rmtree(dest)
        os.rename(stage, dest)
        return PublishReceipt(dest, True)

    def place_raw(self, comp, bid, files: dict, ready=True, folder=None):
        """Test/ops helper: drop a (possibly incomplete) batch straight into pub/<folder>/ (no slug check)."""
        d = os.path.join(self._d(comp, "pub"), folder, bid) if folder else self._d(comp, "pub", bid)
        os.makedirs(d, exist_ok=True)
        for n, b in files.items():
            open(os.path.join(d, n), "wb").write(b)
        if ready:
            open(os.path.join(d, READY), "w").write("ok\n")

    def unknown_folders(self, comp, folders):
        """pub/ entries that are not a watched sender folder (ignored by fetch)."""
        base = self._d(comp, "pub")
        return sorted(n for n in os.listdir(base) if n not in set(folders)) if os.path.isdir(base) else []

    def _candidates(self, comp, folders):
        folders = set(folders or ())
        base = self._d(comp, "processing")
        if os.path.isdir(base):
            for bid in sorted(os.listdir(base)):
                yield "processing", None, bid, os.path.join(base, bid)
        for f in sorted(folders):
            fb = os.path.join(self._d(comp, "pub"), f)
            if os.path.isdir(fb) and not os.path.islink(fb):
                for bid in sorted(os.listdir(fb)):
                    yield "pub", f, bid, os.path.join(fb, bid)

    def fetch(self, comp, max_items=10, lease_secs=300, consumer="r1", folders=()):
        """folders: iterable of slugs, or {slug: sender_key_id} (v2.5.1: lets processing items that carry
        a verified skid in _meta resolve their CURRENT folder from the registry, not the stale one)."""
        out, now, taken = [], int(self.now()), set()
        by_skid = {v: k for k, v in folders.items()} if isinstance(folders, dict) else {}
        folders = set(folders or ())
        for area, folder, bid, d in self._candidates(comp, folders):
            if True:
                if bid in taken:
                    continue
                if not os.path.exists(os.path.join(d, READY)):
                    continue                                         # incomplete: skip
                meta = load_json(self._meta(comp, bid), default={"delivery_count": 0, "retry_after": 0})
                if os.path.islink(d):
                    continue                                         # v2.5.1: never follow a symlinked batch dir
                if area == "pub":
                    meta["folder"] = folder
                elif meta.get("skid") in by_skid:
                    meta["folder"] = by_skid[meta["skid"]]           # X3: resolve from registry by verified skid
                elif meta.get("folder") not in folders:
                    continue                                         # sender no longer watched (revoked)
                lease = meta.get("lease")
                if lease and lease["until"] > now and lease["consumer"] != consumer:
                    continue                                         # leased by someone else
                if meta.get("retry_after", 0) > now:
                    continue
                meta["delivery_count"] += 1
                meta["lease"] = {"consumer": consumer, "until": now + lease_secs}
                if area == "pub":
                    dst = self._d(comp, "processing", bid); os.makedirs(os.path.dirname(dst), exist_ok=True)
                    try:
                        os.rename(d, dst); d = dst
                    except OSError:
                        continue                                     # V2c: lost the race; skip, don't abort
                if not os.path.isdir(d):
                    continue
                atomic_write(self._meta(comp, bid), json.dumps(meta).encode())
                taken.add(bid)
                out.append(Delivery(bid, d, d, meta["delivery_count"], now + lease_secs,
                                    {"folder": meta.get("folder")}))
                if len(out) >= max_items:
                    return out
        return out

    def lease_valid(self, delivery, consumer):
        """V2a: True only if `consumer` still holds an unexpired lease on this delivery."""
        try:
            meta = load_json(self._meta(self._comp(delivery), delivery.batch_id), default={})
        except (OSError, ValueError):
            return False
        lease = meta.get("lease") or {}
        return lease.get("consumer") == consumer and int(lease.get("until", 0)) > int(self.now()) \
            and os.path.isdir(delivery.path)

    def batch_files(self, d):
        return {n for n in os.listdir(d) if n != READY}

    def _move(self, delivery, comp, area):
        dst = self._d(comp, area, delivery.batch_id); os.makedirs(os.path.dirname(dst), exist_ok=True)
        if os.path.exists(dst):
            shutil.rmtree(dst)
        os.rename(delivery.path, dst)
        return dst

    def _comp(self, delivery):
        return os.path.basename(os.path.dirname(os.path.dirname(delivery.path)))

    def ack(self, delivery):
        comp = self._comp(delivery); self._move(delivery, comp, "done")
        meta = load_json(self._meta(comp, delivery.batch_id), default={}); meta.pop("lease", None)
        atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())

    def nack(self, delivery, retry_after_secs, reason):
        comp = self._comp(delivery)
        meta = load_json(self._meta(comp, delivery.batch_id), default={"delivery_count": 0})
        meta.pop("lease", None); meta["retry_after"] = int(self.now()) + retry_after_secs; meta["last_reason"] = reason
        atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())

    def dead_letter(self, delivery, reason_record):
        comp = self._comp(delivery); dst = self._move(delivery, comp, "dlq")
        atomic_write(os.path.join(dst, "reason.json"), canonical(reason_record))

    def hold(self, delivery):
        comp = self._comp(delivery); self._move(delivery, comp, "holding")
        meta = load_json(self._meta(comp, delivery.batch_id), default={}); meta.pop("lease", None)
        meta["held_since"] = meta.get("held_since", int(self.now()))
        atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())

    def set_skid(self, delivery, skid):
        """X3: record the VERIFIED sender_key_id so later moves resolve the folder from the registry."""
        comp = self._comp(delivery)
        meta = load_json(self._meta(comp, delivery.batch_id), default={}); meta["skid"] = skid
        atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())

    def _pub_dst(self, comp, bid, folder=None):
        meta = load_json(self._meta(comp, bid), default={})
        if folder is not None:                                       # X3: caller resolved it from the registry
            meta["folder"] = check_slug(folder); atomic_write(self._meta(comp, bid), json.dumps(meta).encode())
        return os.path.join(self._d(comp, "pub"), check_slug(meta.get("folder")), bid)

    def release_held(self, comp, bid, folder=None):
        src, dst = self._d(comp, "holding", bid), self._pub_dst(comp, bid, folder)
        os.makedirs(os.path.dirname(dst), exist_ok=True); os.rename(src, dst)

    def redrive(self, comp, bid, operator: str, folder=None):
        """Human-only: move a dlq batch back to pub/<folder>/. Requires a named operator. Prefer
        Receiver.redrive, which resolves the folder from the registry by the verified signing key (X3)."""
        if not operator or not str(operator).strip():
            raise PermissionError("redrive requires a named human operator")
        src, dst = self._d(comp, "dlq", bid), self._pub_dst(comp, bid, folder)
        os.makedirs(os.path.dirname(dst), exist_ok=True)
        os.remove(os.path.join(src, "reason.json")); os.rename(src, dst)
        meta = load_json(self._meta(comp, bid), default={}); meta.update(delivery_count=0, redriven_by=operator)
        atomic_write(self._meta(comp, bid), json.dumps(meta).encode())

    def list_pending(self, comp):
        rows = []
        for area in ("_staging", "pub", "processing", "holding", "dlq", "done"):
            base = self._d(comp, area)
            if not os.path.isdir(base):
                continue
            items = []
            if area in ("_staging", "pub"):
                for f in sorted(os.listdir(base)):
                    if os.path.isdir(os.path.join(base, f)):
                        items += [(f, b, os.path.join(base, f, b)) for b in sorted(os.listdir(os.path.join(base, f)))]
            else:
                items = [(None, b, os.path.join(base, b)) for b in sorted(os.listdir(base))]
            for f, bid, p in items:
                meta = load_json(self._meta(comp, bid), default={})
                rows.append({"batch_id": bid, "state": area, "folder": f or meta.get("folder"),
                             "delivery_count": meta.get("delivery_count", 0),
                             "ready": os.path.exists(os.path.join(p, READY))})
        return rows