"""Typed customer projections for explicitly supported shared authority ledgers."""

from __future__ import annotations

import json
import os
import stat
from enum import StrEnum
from pathlib import Path
from typing import Any

from checkin_cli.customer_cleanup_fs import canonical, digest_bytes, safe_relative, write_private
from checkin_cli.customer_cleanup_models import CleanupBlockedError, CleanupIntegrityError


class SharedLedgerProjection(StrEnum):
    """Closed set of shared authorities with customer-safe projection logic."""

    DRAFT_GENERATIONS = "data/owner-actions/draft-generations.json"


_SELECTOR = "draft-generations-customer-v1"
_PROJECTION_SCHEMA = "customer-cleanup-draft-generations-projection-v1"
_RECORD_KEYS = frozenset({
    "schema_version", "generation", "state", "customer_key", "checkin_event_id",
    "checkin_revision", "draft_revision", "model_contract_version",
    "provider_contract_version", "delivery_provider_contract_version", "actor",
    "authority_digest", "predecessor_digest", "lineage_predecessor_digest",
    "lineage_parent_token", "created_at", "updated_at", "attempt", "max_attempts",
    "error", "generation_provider_receipt", "delivery_provider_receipt",
    "idempotency_key", "record_digest",
})
_KNOWN_STATES = frozenset({
    "generation_pending", "generating", "draft_created", "generation_failed", "held",
    "approved", "delivery_pending", "delivered", "sent_audited",
})
_TERMINAL_STATES = frozenset({"sent_audited"})


def _secure_read(path: Path, root: Path) -> tuple[bytes, os.stat_result]:
    relative = Path(safe_relative(path, root, ""))
    directory_flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW | os.O_DIRECTORY
    descriptor = -1
    directory = -1
    try:
        directory = os.open(root, directory_flags)
        for part in relative.parts[:-1]:
            child = os.open(part, directory_flags, dir_fd=directory)
            os.close(directory)
            directory = child
        descriptor = os.open(
            relative.name,
            os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW,
            dir_fd=directory,
        )
        before = os.fstat(descriptor)
        mode = stat.S_IMODE(before.st_mode)
        if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1:
            raise CleanupIntegrityError("shared ledger must be a regular no-hardlink file")
        if before.st_uid != os.geteuid() or mode & 0o077:
            raise CleanupIntegrityError("shared ledger must be owner-only")
        chunks: list[bytes] = []
        while chunk := os.read(descriptor, 1024 * 1024):
            chunks.append(chunk)
        after = os.fstat(descriptor)
        before_identity = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
        after_identity = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
        if before_identity != after_identity:
            raise CleanupIntegrityError("shared ledger changed while read")
        return b"".join(chunks), before
    except OSError as exc:
        raise CleanupIntegrityError(f"shared ledger cannot be securely opened: {path}") from exc
    finally:
        if descriptor >= 0:
            os.close(descriptor)
        if directory >= 0:
            os.close(directory)


def _parse_draft_generations(payload: bytes) -> dict[str, list[dict[str, Any]]]:
    try:
        document = json.loads(payload)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise CleanupIntegrityError("draft-generations shared ledger is invalid") from exc
    if type(document) is not dict:
        raise CleanupIntegrityError("draft-generations shared ledger schema is invalid")
    parsed: dict[str, list[dict[str, Any]]] = {}
    for token, history in document.items():
        if type(token) is not str or not token or type(history) is not list or not history:
            raise CleanupIntegrityError("draft-generations shared ledger schema is invalid")
        rows: list[dict[str, Any]] = []
        binding: str | None = None
        for row in history:
            if type(row) is not dict or set(row) != _RECORD_KEYS:
                raise CleanupIntegrityError("draft-generations customer binding is missing or ambiguous")
            customer = row.get("customer_key")
            if type(customer) is not str or not customer or (binding is not None and customer != binding):
                raise CleanupIntegrityError("draft-generations customer binding is missing or ambiguous")
            if row.get("schema_version") != "nutrition-draft-generation-v1":
                raise CleanupIntegrityError("draft-generations shared ledger schema is invalid")
            if type(row.get("state")) is not str:
                raise CleanupIntegrityError("draft-generations shared ledger state is invalid")
            binding = customer
            rows.append(row)
        parsed[token] = rows
    return parsed


def _project(
    payload: bytes, customer_key: str
) -> tuple[bytes, list[str], int, int, int, int]:
    ledger = _parse_draft_generations(payload)
    selected = {
        token: history
        for token, history in sorted(ledger.items())
        if history[0]["customer_key"] == customer_key
    }
    record_digests = [
        digest_bytes(canonical(row))
        for history in selected.values()
        for row in history
    ]
    projection = canonical({
        "schema_version": _PROJECTION_SCHEMA,
        "customer_key": customer_key,
        "histories": selected,
    }) + b"\n"
    active = pending = unknown = terminal = 0
    pending_states = {"generation_pending", "generating", "delivery_pending", "delivered"}
    for history in selected.values():
        state = history[-1]["state"]
        if state in _TERMINAL_STATES:
            terminal += 1
        elif state in pending_states:
            pending += 1
        elif state in _KNOWN_STATES:
            active += 1
        else:
            unknown += 1
    return projection, record_digests, active, pending, unknown, terminal


def prepare_shared_projections(
    root: Path, customer_key: str, requested: tuple[SharedLedgerProjection, ...]
) -> list[dict[str, object]]:
    seen: set[SharedLedgerProjection] = set()
    bindings: list[dict[str, object]] = []
    for authority in requested:
        if type(authority) is not SharedLedgerProjection or authority in seen:
            raise CleanupIntegrityError("unsupported shared ledger projection")
        seen.add(authority)
        relative = authority.value
        path = root / relative
        payload, info = _secure_read(path, root)
        projection, record_digests, _, _, _, _ = _project(payload, customer_key)
        bindings.append({
            "source_relative_path": relative,
            "source_sha256": digest_bytes(payload),
            "source_size": len(payload),
            "source_device": info.st_dev,
            "source_inode": info.st_ino,
            "source_mode": stat.S_IMODE(info.st_mode),
            "selector_version": _SELECTOR,
            "selected_record_count": len(record_digests),
            "ordered_record_digests": record_digests,
            "projection_sha256": digest_bytes(projection),
            "projection_archive_path": f"projections/{relative}",
        })
    return bindings


def _recompute(
    root: Path,
    customer_key: str,
    binding: dict[str, object],
    *,
    require_terminal: bool,
) -> bytes:
    if binding.get("source_relative_path") != SharedLedgerProjection.DRAFT_GENERATIONS.value:
        raise CleanupIntegrityError("unsupported shared ledger projection in manifest")
    path = root / str(binding["source_relative_path"])
    payload, info = _secure_read(path, root)
    projection, digests, active, pending, unknown, _ = _project(payload, customer_key)
    expected = {
        "source_sha256": digest_bytes(payload),
        "source_size": len(payload),
        "source_device": info.st_dev,
        "source_inode": info.st_ino,
        "source_mode": stat.S_IMODE(info.st_mode),
        "selector_version": _SELECTOR,
        "selected_record_count": len(digests),
        "ordered_record_digests": digests,
        "projection_sha256": digest_bytes(projection),
        "projection_archive_path": f"projections/{binding['source_relative_path']}",
    }
    if any(binding.get(key) != value for key, value in expected.items()):
        raise CleanupIntegrityError("shared ledger changed after cleanup preparation")
    if require_terminal and (active or pending or unknown):
        raise CleanupBlockedError("pending or unknown customer authority blocks cleanup")
    return projection


def copy_shared_projections(
    root: Path, archive: Path, customer_key: str, bindings: list[dict[str, object]]
) -> None:
    for binding in bindings:
        projection = _recompute(root, customer_key, binding, require_terminal=True)
        destination = archive / str(binding["projection_archive_path"])
        write_private(destination, projection, exclusive=not destination.exists())


def verify_shared_projections(
    root: Path, archive: Path, customer_key: str, bindings: list[dict[str, object]]
) -> None:
    for binding in bindings:
        expected = _recompute(root, customer_key, binding, require_terminal=False)
        path = archive / str(binding["projection_archive_path"])
        archived, _ = _secure_read(path, archive)
        if archived != expected or digest_bytes(archived) != binding.get("projection_sha256"):
            raise CleanupIntegrityError("shared ledger archive verification failed")
        try:
            projected = json.loads(archived)
            histories = projected["histories"]
        except (KeyError, TypeError, json.JSONDecodeError) as exc:
            raise CleanupIntegrityError("shared ledger archive verification failed") from exc
        if (
            projected.get("schema_version") != _PROJECTION_SCHEMA
            or projected.get("customer_key") != customer_key
            or type(histories) is not dict
            or any(
                row.get("customer_key") != customer_key
                for history in histories.values()
                for row in history
            )
        ):
            raise CleanupIntegrityError("shared ledger archive contains foreign customer authority")


def shared_authority_counts(root: Path, customer_key: str) -> tuple[int, int, int, int]:
    path = root / SharedLedgerProjection.DRAFT_GENERATIONS.value
    if not path.exists() and not path.is_symlink():
        return 0, 0, 0, 0
    payload, _ = _secure_read(path, root)
    _, _, active, pending, unknown, terminal = _project(payload, customer_key)
    return active, pending, unknown, terminal