"""Durable at-most-once customer notice after canonical activation.""" from __future__ import annotations import fcntl import hashlib import json import os from contextlib import contextmanager from dataclasses import asdict, dataclass, replace from pathlib import Path from typing import Iterator, Mapping from gateway.platforms.telegram_nutrition_addresses import ( AddressConfigurationError, parse_customer_private_delivery_address, ) ACTIVATION_COMPLETION_TEMPLATE = ( "고객 등록이 완료되었습니다.\n" "코칭 시작일: {starts_on}\n" "아침 체크인: 평일 {daily_time}\n" "시작일의 해당 시간 이후 첫 체크인 안내를 보내드릴게요." ) _TERMINAL_STATES = frozenset({"sent_audited", "unknown"}) class ActivationNoticeError(ValueError): """Activation notice state is malformed, stale, or unsafe.""" @dataclass(frozen=True, slots=True) class ActivationNoticeReceipt: reservation_id: str customer_key: str starts_on: str daily_time: str body: str destination: dict[str, str] authority_digest: str state: str provider_receipt: str | None = None reason: str | None = None append_sequence: int = 0 provider_authority: bool = False class ActivationNoticeStore: """Append-only reservation ledger for one completion notice per authority.""" def __init__(self, profile_root: Path | str) -> None: root = Path(profile_root) if root.is_symlink(): raise ActivationNoticeError("profile root symlinks are not allowed") root.mkdir(parents=True, exist_ok=True, mode=0o700) self.path = root / "data" / "activation-completion-notices.jsonl" self.lock_path = root / "data" / "activation-completion-notices.lock" @contextmanager def _locked(self) -> Iterator[None]: self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) self.path.parent.chmod(0o700) if self.path.is_symlink() or self.lock_path.is_symlink(): raise ActivationNoticeError("activation notice paths are unsafe") fd = os.open(self.lock_path, os.O_CREAT | os.O_RDWR, 0o600) try: os.chmod(self.lock_path, 0o600) with os.fdopen(fd, "a+", encoding="utf-8") as lock: fcntl.flock(lock.fileno(), fcntl.LOCK_EX) yield fcntl.flock(lock.fileno(), fcntl.LOCK_UN) except Exception: raise @staticmethod def _validate_destination(value: Mapping[str, object]) -> dict[str, str]: try: destination = parse_customer_private_delivery_address(value) except AddressConfigurationError as exc: raise ActivationNoticeError(str(exc)) from exc return { "user_id": str(destination.user_id), "chat_id": str(destination.chat_id), "topic_id": str(destination.topic_id), } def reserve( self, *, customer_key: str, starts_on: str, daily_time: str, destination: Mapping[str, object], authority_digest: str, ) -> ActivationNoticeReceipt: destination_pin = self._validate_destination(destination) if not customer_key or len(customer_key) > 128: raise ActivationNoticeError("activation notice customer is invalid") try: from datetime import date, time date.fromisoformat(starts_on) time.fromisoformat(daily_time) except (TypeError, ValueError) as exc: raise ActivationNoticeError("activation notice schedule is invalid") from exc if len(authority_digest) != 64 or any(c not in "0123456789abcdef" for c in authority_digest): raise ActivationNoticeError("activation notice authority digest is invalid") body = ACTIVATION_COMPLETION_TEMPLATE.format( starts_on=starts_on, daily_time=daily_time[:5], ) material = json.dumps( { "customer_key": customer_key, "starts_on": starts_on, "daily_time": daily_time[:5], "destination": destination_pin, "authority_digest": authority_digest, }, sort_keys=True, separators=(",", ":"), ) reservation_id = hashlib.sha256(material.encode()).hexdigest() with self._locked(): latest = self._latest_unlocked().get(reservation_id) if latest is not None: return latest receipt = ActivationNoticeReceipt( reservation_id=reservation_id, customer_key=customer_key, starts_on=starts_on, daily_time=daily_time[:5], body=body, destination=destination_pin, authority_digest=authority_digest, state="prepared", append_sequence=len(self._rows_unlocked()) + 1, ) self._append_unlocked(receipt) return receipt def mark_sending(self, receipt: ActivationNoticeReceipt) -> ActivationNoticeReceipt: return self._transition(receipt, "prepared", "sending", provider_authority=True) def mark_delivered( self, receipt: ActivationNoticeReceipt, *, provider_receipt: str ) -> ActivationNoticeReceipt: if not provider_receipt or len(provider_receipt) > 128: raise ActivationNoticeError("activation notice provider receipt is invalid") return self._transition( receipt, "sending", "delivered", provider_receipt=provider_receipt, ) def mark_audited(self, receipt: ActivationNoticeReceipt) -> ActivationNoticeReceipt: return self._transition(receipt, "delivered", "sent_audited") def mark_unknown( self, receipt: ActivationNoticeReceipt, *, reason: str ) -> ActivationNoticeReceipt: if receipt.state not in {"prepared", "sending"}: raise ActivationNoticeError("activation notice is not pending") return self._transition(receipt, receipt.state, "unknown", reason=reason[:128]) def _transition( self, receipt: ActivationNoticeReceipt, expected: str, target: str, **changes: object, ) -> ActivationNoticeReceipt: with self._locked(): current = self._latest_unlocked().get(receipt.reservation_id) if current is None or current.state != expected or current != replace(receipt, provider_authority=False): raise ActivationNoticeError(f"activation notice is not {expected}") updated = replace( current, state=target, append_sequence=len(self._rows_unlocked()) + 1, provider_authority=bool(changes.pop("provider_authority", False)), **changes, ) self._append_unlocked(updated) return updated def receipts(self) -> tuple[ActivationNoticeReceipt, ...]: with self._locked(): return tuple(self._rows_unlocked()) def latest(self) -> tuple[ActivationNoticeReceipt, ...]: with self._locked(): return tuple(self._latest_unlocked().values()) def _rows_unlocked(self) -> list[ActivationNoticeReceipt]: if not self.path.exists(): return [] if self.path.is_symlink() or not self.path.is_file(): raise ActivationNoticeError("activation notice ledger is unsafe") rows: list[ActivationNoticeReceipt] = [] try: for line in self.path.read_text(encoding="utf-8").splitlines(): raw = json.loads(line) if not isinstance(raw, dict): raise ValueError rows.append(ActivationNoticeReceipt(**raw)) except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: raise ActivationNoticeError("activation notice ledger is corrupt") from exc return rows def _latest_unlocked(self) -> dict[str, ActivationNoticeReceipt]: latest: dict[str, ActivationNoticeReceipt] = {} for row in self._rows_unlocked(): latest[row.reservation_id] = row return latest def _append_unlocked(self, receipt: ActivationNoticeReceipt) -> None: payload = asdict(replace(receipt, provider_authority=False)) with self.path.open("a", encoding="utf-8") as ledger: ledger.write(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n") ledger.flush() os.fsync(ledger.fileno()) self.path.chmod(0o600) __all__ = [ "ACTIVATION_COMPLETION_TEMPLATE", "ActivationNoticeError", "ActivationNoticeReceipt", "ActivationNoticeStore", ]