"""Independent post-cleanup customer authority inventory.""" from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any from checkin_cli.customer_admin import ( CustomerAdminError, _require_customer_key, _resolve_profile_root, _resolve_registry_path, ) from checkin_cli.customer_cleanup_fs import read_journal, validate_file, verify_archive from checkin_cli.customer_cleanup_models import ( AuthorityInventory, CleanupIntegrityError, ) from checkin_cli.customer_cleanup_shared import ( shared_authority_counts, verify_shared_projections, ) _CATEGORIES = ( "registry", "bootstrap", "onboarding", "outbox", "key", "owner_action", "service_state", "event", "customer_data", "activation", "sent_event", "candidate", ) _PENDING_STATES = {"prepared", "sending", "delivered", "audit_pending"} _UNKNOWN_STATES = {"unknown", "delivery_unknown"} _JSONL_AUTHORITIES = { "customer-activation-audit.jsonl", "scheduled-deliveries.jsonl", } _ACTIVATION_JOURNAL = "customer-activation-journal.json" _ACTIVATION_JOURNAL_REQUIRED_STRINGS = ( "transaction_id", "customer_id", "registry_path", "data_root", "checklist_evidence_path", "audit_path", "registry_sha256", "previous_registry_sha256", "audit_record_sha256", "created_at", "prepared_at", ) def read_registry_state(root: Path, key: str) -> tuple[bool, bool, bytes]: path = _resolve_registry_path(root) validate_file(path) payload = path.read_bytes() try: document = json.loads(payload) customer = next(row for row in document["customers"] if row["customer_key"] == key) except (KeyError, TypeError, StopIteration, json.JSONDecodeError) as exc: raise CustomerAdminError(f"unknown customer: {key}") from exc consent = customer.get("ai_processing_consent") granted = consent.get("granted") if isinstance(consent, dict) else None return customer.get("enabled") is False, granted is False, payload def _customer_binding(row: dict[str, Any]) -> str: customer_key = row.get("customer_key") customer_id = row.get("customer_id") bindings = [ value for value in (customer_key, customer_id) if isinstance(value, str) and value ] if not bindings or len(set(bindings)) != 1: raise TypeError("authority customer binding is invalid") if customer_key is not None and customer_key != bindings[0]: raise TypeError("authority customer_key is invalid") if customer_id is not None and customer_id != bindings[0]: raise TypeError("authority customer_id is invalid") return bindings[0] def _validate_activation_journal(row: object) -> dict[str, Any]: if not isinstance(row, dict): raise TypeError("activation journal must be an object") if type(row.get("version")) is not int or row["version"] not in {1, 2, 3}: raise TypeError("activation journal version is invalid") if any(not isinstance(row.get(field), str) or not row[field] for field in _ACTIVATION_JOURNAL_REQUIRED_STRINGS): raise TypeError("activation journal required field is invalid") if not isinstance(row.get("state"), str) or not row["state"]: raise TypeError("activation journal state is invalid") if not isinstance(row.get("recovery_required"), bool): raise TypeError("activation journal recovery binding is invalid") if not isinstance(row.get("previous_registry"), dict): raise TypeError("activation journal rollback registry is invalid") previous_audit = row.get("previous_audit_sha256") if previous_audit is not None and ( not isinstance(previous_audit, str) or not previous_audit ): raise TypeError("activation journal previous audit binding is invalid") if row["state"] == "committed" and ( not isinstance(row.get("committed_at"), str) or not row["committed_at"] ): raise TypeError("activation journal commit binding is invalid") if row["state"] == "abandoned" and ( not isinstance(row.get("abandoned_at"), str) or not row["abandoned_at"] ): raise TypeError("activation journal abandonment binding is invalid") if row["version"] in {2, 3} and not isinstance( row.get("nutrition_activation_receipt"), dict ): raise TypeError("activation journal nutrition receipt is invalid") if row["version"] == 3 and any( not isinstance(row.get(field), str) or not row[field] for field in ( "staff_membership_evidence_path", "staff_membership_evidence_sha256", "staff_chat_inventory_sha256", "membership_subscription_epoch_id", ) ): raise TypeError("activation journal membership binding is invalid") _customer_binding(row) return row def relevant_rows(path: Path, key: str) -> list[dict[str, Any]]: if path.name != _ACTIVATION_JOURNAL and path.name not in _JSONL_AUTHORITIES: raise CleanupIntegrityError(f"unsupported authority file: {path}") if not path.exists(): return [] validate_file(path) try: payload = path.read_text(encoding="utf-8") if path.name == _ACTIVATION_JOURNAL: row = _validate_activation_journal(json.loads(payload)) return [row] if _customer_binding(row) == key else [] rows: list[dict[str, Any]] = [] for line in payload.splitlines(): value = json.loads(line) if not isinstance(value, dict): raise TypeError("authority row must be an object") if _customer_binding(value) == key: rows.append(value) return rows except (OSError, TypeError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise CleanupIntegrityError(f"authority file is invalid: {path}") from exc def _load_inventory_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): raise TypeError return value except (OSError, TypeError, json.JSONDecodeError) as exc: raise CleanupIntegrityError("cleanup manifest is invalid") from exc def post_cleanup_authority_inventory( profile_root: Path | str, customer_key: str ) -> AuthorityInventory: """Independently enumerate every supported customer authority namespace.""" root = _resolve_profile_root(Path(profile_root)) key = _require_customer_key(customer_key) disabled, nonconsenting, _ = read_registry_state(root, key) categories = dict.fromkeys(_CATEGORIES, 0) active = 0 if not disabled or not nonconsenting: categories["registry"] = active = 1 customer_root = root / "data/customers" / key customer_files = list(customer_root.rglob("*")) if customer_root.exists() else [] for path in customer_files: if not path.is_file(): continue label = path.relative_to(customer_root).as_posix().lower().replace("-", "_") category = next( (name for name in ( "bootstrap", "onboarding", "outbox", "key", "owner_action", "service_state", "event", ) if name in label), "customer_data", ) categories[category] += 1 active += 1 shared_active, pending, unknown, terminal_owner_actions = shared_authority_counts(root, key) active += shared_active categories["owner_action"] += shared_active + terminal_owner_actions schedule_path = root / "data/scheduled-deliveries.jsonl" for row in relevant_rows(schedule_path, key): state = str(row.get("state", row.get("status", ""))) if state in _PENDING_STATES: pending += 1 elif state in _UNKNOWN_STATES or not state: unknown += 1 else: categories["sent_event"] += 1 activation_paths = ( root / "data/customer-activation-journal.json", root / "data/customer-activation-audit.jsonl", ) for path in activation_paths: for row in relevant_rows(path, key): categories["activation"] += 1 if path.name == _ACTIVATION_JOURNAL and ( row.get("state") not in {"committed", "abandoned"} or row.get("recovery_required") is not False ): unknown += 1 cleanup = root / "data/customer-cleanup" excluded = {schedule_path, *activation_paths} for path in (root / "data").rglob("*"): if not path.is_file() or path in excluded or cleanup in path.parents: continue if customer_root in path.parents: continue relative = path.relative_to(root).as_posix().lower().replace("-", "_") if key not in relative: continue category = next( (name for name in ( "bootstrap", "onboarding", "outbox", "key", "owner_action", "service_state", "event", ) if name in relative), "candidate", ) categories[category] += 1 active += 1 journal = cleanup / f"{key}.journal.jsonl" journal_rows = read_journal(journal) committed = bool(journal_rows and journal_rows[-1].get("phase") == "committed") verified = False if journal_rows: operation = str(journal_rows[0]["operation_id"]) archive = cleanup / "archives" / operation try: manifest_path = archive / "manifest.json" manifest = _load_inventory_manifest(manifest_path) manifest_digest = hashlib.sha256(manifest_path.read_bytes()).hexdigest() if manifest_digest != journal_rows[0].get("manifest_sha256"): raise CleanupIntegrityError("cleanup manifest digest mismatch") verify_archive(archive, manifest["inventory"]) verify_shared_projections( root, archive, key, manifest.get("shared_ledger_projections", []), ) verified = True except CleanupIntegrityError: pass orphan = active if committed else 0 return AuthorityInventory.build( customer_key=key, disabled=disabled, nonconsenting=nonconsenting, categories=categories, active_count=active, pending_count=pending, unknown_count=unknown, orphan_count=orphan, archive_verified=verified, journal_committed=committed, )