"""Disabled-customer registry projection and activation authority receipts.""" from __future__ import annotations import fcntl import hmac import json import os import tempfile from copy import deepcopy from pathlib import Path from typing import Any, cast from checkin_cli.nutrition_onboarding_contract import canonical_digest def _require_digest(value: str, *, field: str) -> None: if len(value) != 64 or any(character not in "0123456789abcdef" for character in value): raise ValueError(f"{field} must be a lowercase SHA-256 digest") def project_nutrition_document( document: dict[str, object], *, customer_key: str, nutrition_profile: dict[str, object], plan: dict[str, object], ) -> dict[str, object]: projected = deepcopy(document) customers = projected.get("customers") if not isinstance(customers, list): raise ValueError("registry customers must be a list") matches = [ customer for customer in customers if isinstance(customer, dict) and customer.get("customer_key") == customer_key ] if len(matches) != 1: raise ValueError(f"unknown customer: {customer_key}") target = cast(dict[str, object], matches[0]) if target.get("enabled") is not False: raise ValueError("nutrition onboarding projection requires a disabled customer") target["profile"] = deepcopy(nutrition_profile) target.pop("nutrition_profile", None) target["plan"] = deepcopy(plan) return projected def apply_nutrition_onboarding_projection( profile_root: Path, *, customer_key: str, nutrition_profile: dict[str, object], plan: dict[str, object], artifact_bundle_digest: str, readiness_receipt_digest: str, ) -> dict[str, object]: _require_digest(artifact_bundle_digest, field="artifact_bundle_digest") _require_digest(readiness_receipt_digest, field="readiness_receipt_digest") root = Path(profile_root).resolve() registry_path = root / "customers" / "registry.json" data_root = root / "data" data_root.mkdir(mode=0o700, parents=True, exist_ok=True) lock_path = data_root / ".nutrition-onboarding-projection.lock" journal_path = data_root / "nutrition-onboarding-projection-journal.jsonl" descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) with os.fdopen(descriptor, "r+b") as lock: fcntl.flock(lock.fileno(), fcntl.LOCK_EX) document = json.loads(registry_path.read_text(encoding="utf-8")) if not isinstance(document, dict): raise ValueError("registry document must be an object") projected = project_nutrition_document( document, customer_key=customer_key, nutrition_profile=nutrition_profile, plan=plan, ) projection_digest = canonical_digest(projected) prior = _projection_rows(journal_path) for row in prior: if row.get("customer_key") != customer_key or row.get("state") != "committed": continue if ( row.get("artifact_bundle_digest") == artifact_bundle_digest and row.get("readiness_receipt_digest") == readiness_receipt_digest and row.get("customer_projection_digest") == projection_digest ): return dict(row) raise ValueError("conflicting nutrition onboarding projection replay") transaction_id = canonical_digest( { "customer_key": customer_key, "artifact_bundle_digest": artifact_bundle_digest, "readiness_receipt_digest": readiness_receipt_digest, "customer_projection_digest": projection_digest, } ) prepared: dict[str, object] = { "schema_version": "nutrition_projection_v1", "transaction_id": transaction_id, "state": "prepared", "customer_key": customer_key, "artifact_bundle_digest": artifact_bundle_digest, "readiness_receipt_digest": readiness_receipt_digest, "customer_projection_digest": projection_digest, } _append_journal(journal_path, prepared) _atomic_registry_write(registry_path, projected) committed: dict[str, object] = {**prepared, "state": "committed"} _append_journal(journal_path, committed) return committed def build_nutrition_activation_receipt_v2( *, customer_key: str, readiness_receipt_digest: str, readiness_bundle_digest: str, customer_projection_digest: str, input_reconciliation_digest: str, ) -> dict[str, str]: for field, value in ( ("readiness_receipt_digest", readiness_receipt_digest), ("readiness_bundle_digest", readiness_bundle_digest), ("customer_projection_digest", customer_projection_digest), ("input_reconciliation_digest", input_reconciliation_digest), ): _require_digest(value, field=field) receipt = { "schema_version": "nutrition_activation_v2", "customer_key": customer_key, "readiness_receipt_digest": readiness_receipt_digest, "readiness_bundle_digest": readiness_bundle_digest, "customer_projection_digest": customer_projection_digest, "input_reconciliation_digest": input_reconciliation_digest, } return {**receipt, "digest": canonical_digest(receipt)} def customer_nutrition_projection_digest(customer: dict[str, object]) -> str: return canonical_digest( { "customer_key": customer.get("customer_key"), "nutrition_profile": ( customer.get("profile") if "profile" in customer else customer.get("nutrition_profile") ), "plan": customer.get("plan"), } ) def build_legacy_activation_authority( *, customer_key: str, activation_receipt_digest: str, registry_projection_digest: str, owner_digest: str, ) -> dict[str, str]: for field, value in ( ("activation_receipt_digest", activation_receipt_digest), ("registry_projection_digest", registry_projection_digest), ("owner_digest", owner_digest), ): _require_digest(value, field=field) manifest = { "schema_version": "nutrition_legacy_activation_authority_v1", "customer_key": customer_key, "activation_receipt_digest": activation_receipt_digest, "registry_projection_digest": registry_projection_digest, "owner_digest": owner_digest, } return {**manifest, "digest": canonical_digest(manifest)} def validate_legacy_activation_authority( manifest: dict[str, str], *, customer_key: str, activation_receipt_digest: str, registry_projection_digest: str, owner_digest: str, ) -> bool: expected = build_legacy_activation_authority( customer_key=customer_key, activation_receipt_digest=activation_receipt_digest, registry_projection_digest=registry_projection_digest, owner_digest=owner_digest, ) return all( isinstance(manifest.get(key), str) and hmac.compare_digest(manifest[key], value) for key, value in expected.items() ) def _projection_rows(path: Path) -> list[dict[str, object]]: if not path.exists(): return [] rows: list[dict[str, object]] = [] for line in path.read_text(encoding="utf-8").splitlines(): value: Any = json.loads(line) if not isinstance(value, dict): raise ValueError("invalid nutrition projection journal") rows.append(value) return rows def _append_journal(path: Path, row: dict[str, object]) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) descriptor = os.open(path, os.O_CREAT | os.O_APPEND | os.O_WRONLY, 0o600) with os.fdopen(descriptor, "ab") as handle: handle.write( json.dumps(row, sort_keys=True, separators=(",", ":")).encode() + b"\n" ) handle.flush() os.fsync(handle.fileno()) def _atomic_registry_write(path: Path, document: dict[str, object]) -> None: original_mode = path.stat().st_mode & 0o777 descriptor, temporary = tempfile.mkstemp(prefix=".registry.", dir=path.parent) temp_path = Path(temporary) try: os.fchmod(descriptor, original_mode) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(document, handle, ensure_ascii=False, indent=2) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temp_path, path) directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(directory) finally: os.close(directory) finally: temp_path.unlink(missing_ok=True)