"""Crash-safe, forward-only customer authority archive and cleanup.""" from __future__ import annotations import json import uuid from collections.abc import Callable, Iterable from datetime import UTC, date, datetime from pathlib import Path from typing import Any from checkin_cli.customer_admin import ( _require_customer_key, _resolve_profile_root, _resolve_registry_path, profile_authority_lock, withdraw_customer, ) from checkin_cli.customer_cleanup_fs import ( PHASES, append_phase, canonical, copy_inventory, digest_bytes, freeze_archive, prune_sources, read_journal, source_inventory, validate_file, verify_archive, write_private, ) from checkin_cli.customer_cleanup_inventory import ( post_cleanup_authority_inventory, read_registry_state, relevant_rows, ) from checkin_cli.customer_cleanup_models import ( AuthorityInventory, CleanupBlockedError, CleanupError, CleanupIntegrityError, CleanupReceipt, ) from checkin_cli.customer_cleanup_shared import ( SharedLedgerProjection, copy_shared_projections, prepare_shared_projections, verify_shared_projections, ) __all__ = [ "AuthorityInventory", "CleanupBlockedError", "CleanupError", "CleanupIntegrityError", "CleanupReceipt", "SharedLedgerProjection", "archive_customer_cleanup", "post_cleanup_authority_inventory", "resume_customer_cleanup", ] def _paths(root: Path, key: str) -> tuple[Path, Path]: cleanup = root / "data" / "customer-cleanup" return cleanup / f"{key}.journal.jsonl", cleanup / "archives" def _bindings(root: Path, key: str, candidates: Iterable[Path]) -> dict[str, object]: registry = _resolve_registry_path(root) activation_paths = ( root / "data/customer-activation-journal.json", root / "data/customer-activation-audit.jsonl", ) sent_path = root / "data/scheduled-deliveries.jsonl" candidate_rows: list[dict[str, str]] = [] for requested in candidates: path = requested if requested.is_absolute() else root / requested candidate_rows.append({ "path": path.absolute().relative_to(root).as_posix(), "sha256": digest_bytes(path.read_bytes()), }) return { "registry": {"path": registry.relative_to(root).as_posix(), "sha256": digest_bytes(registry.read_bytes())}, "activation": [ {"path": path.relative_to(root).as_posix(), "rows_sha256": digest_bytes(canonical(relevant_rows(path, key)))} for path in activation_paths ], "sent_event": {"path": sent_path.relative_to(root).as_posix(), "rows_sha256": digest_bytes(canonical(relevant_rows(sent_path, key)))}, "candidate": sorted(candidate_rows, key=lambda row: row["path"]), } def _load_manifest(path: Path) -> dict[str, Any]: try: validate_file(path) value = json.loads(path.read_text(encoding="utf-8")) if ( not isinstance(value, dict) or not isinstance(value.get("inventory"), list) or not isinstance(value.get("shared_ledger_projections", []), list) ): raise TypeError except (OSError, TypeError, json.JSONDecodeError) as exc: raise CleanupIntegrityError("cleanup manifest is invalid") from exc return value def _receipt(root: Path, key: str, journal: Path, operation: str, phase: str) -> CleanupReceipt: archive = _paths(root, key)[1] / operation return CleanupReceipt(key, operation, phase, archive, archive / "manifest.json", journal, phase == "committed") def archive_customer_cleanup( profile_root: Path | str, customer_key: str, *, candidate_paths: Iterable[Path | str] = (), shared_ledger_projections: Iterable[SharedLedgerProjection] = (), kst_date: date | None = None, fault_injector: Callable[[str], None] | None = None, ) -> CleanupReceipt: """Archive, byte-verify, and prune one withdrawn customer's authorities.""" root = _resolve_profile_root(Path(profile_root)) key = _require_customer_key(customer_key) candidates = tuple(Path(path) for path in candidate_paths) shared = tuple(shared_ledger_projections) with profile_authority_lock(root): disabled, nonconsenting, _ = read_registry_state(root, key) if not disabled or not nonconsenting: withdrawal_day = kst_date or datetime.now(UTC).date() withdraw_customer(root, key, kst_date=withdrawal_day) journal, archives = _paths(root, key) rows = read_journal(journal) if rows: operation = str(rows[0]["operation_id"]) else: operation = uuid.uuid4().hex inventory = source_inventory(root, key, candidates) shared_bindings = prepare_shared_projections(root, key, shared) archive = archives / operation manifest = { "schema_version": "customer-cleanup-archive-v2", "customer_key": key, "operation_id": operation, "source_root": str(root), "destination": str(archive), "inputs": _bindings(root, key, candidates), "inventory": inventory, "shared_ledger_projections": shared_bindings, } manifest_payload = canonical(manifest) + b"\n" write_private(archive / "manifest.json", manifest_payload, exclusive=True) append_phase( journal, "prepared", operation, manifest_sha256=digest_bytes(manifest_payload), ) if fault_injector: fault_injector("prepared") return _resume_locked(root, key, operation, journal, fault_injector) def resume_customer_cleanup( profile_root: Path | str, customer_key: str, *, fault_injector: Callable[[str], None] | None = None, ) -> CleanupReceipt: """Resume the sole durable cleanup operation without moving backward.""" root = _resolve_profile_root(Path(profile_root)) key = _require_customer_key(customer_key) with profile_authority_lock(root): journal, _ = _paths(root, key) rows = read_journal(journal) if not rows: raise CleanupError("customer cleanup operation has not been prepared") return _resume_locked(root, key, str(rows[0]["operation_id"]), journal, fault_injector) def _resume_locked( root: Path, key: str, operation: str, journal: Path, fault: Callable[[str], None] | None, ) -> CleanupReceipt: archive = _paths(root, key)[1] / operation manifest_path = archive / "manifest.json" manifest = _load_manifest(manifest_path) rows = read_journal(journal) if digest_bytes(manifest_path.read_bytes()) != rows[0].get("manifest_sha256"): raise CleanupIntegrityError("cleanup manifest digest mismatch") if manifest.get("customer_key") != key or manifest.get("operation_id") != operation: raise CleanupIntegrityError("cleanup manifest authority mismatch") inventory = manifest["inventory"] shared = manifest.get("shared_ledger_projections", []) phase = str(rows[-1]["phase"]) if PHASES.index(phase) >= PHASES.index("copied_verified"): verify_archive(archive, inventory) verify_shared_projections(root, archive, key, shared) if phase == "prepared": copy_inventory(root, archive, inventory) copy_shared_projections(root, archive, key, shared) verify_archive(archive, inventory) verify_shared_projections(root, archive, key, shared) append_phase(journal, "copied_verified", operation) phase = "copied_verified" if fault: fault(phase) if phase == "copied_verified": authority = post_cleanup_authority_inventory(root, key) if authority.pending_count or authority.unknown_count: raise CleanupBlockedError("pending or unknown customer authority blocks cleanup") prune_sources(root, inventory, key) append_phase(journal, "source_pruned", operation) phase = "source_pruned" if fault: fault(phase) if phase == "source_pruned": verify_archive(archive, inventory) verify_shared_projections(root, archive, key, shared) authority = post_cleanup_authority_inventory(root, key) if authority.active_count or authority.pending_count or authority.unknown_count: raise CleanupBlockedError("customer authority remains after source pruning") freeze_archive(archive) append_phase(journal, "committed", operation) phase = "committed" if fault: fault(phase) return _receipt(root, key, journal, operation, phase)