from __future__ import annotations import hashlib import importlib import json import os from dataclasses import replace from datetime import date, datetime, timezone import anyio from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from gateway.platforms.dualcoach_profile_package import ( DualCoachProfilePackage, ProfilePackageResolutionError, ) from gateway.platforms.nutrition_onboarding_reconciliation import ( NutritionOnboardingReconciler, ) from gateway.platforms.telegram_nutrition_onboarding import ( NUTRITION_ONBOARDING_API_VERSION, OPTIONAL_QUESTION_DEFAULTS, encode_callback_hash, ) from gateway.platforms.telegram_nutrition_onboarding_authority import ( feature_epoch_digest, ) from gateway.platforms.telegram_nutrition_onboarding_runtime_callback import ( TelegramNutritionOnboardingRuntimeCallbackMixin, ) from gateway.platforms.telegram_nutrition_onboarding_runtime_authority import ( TelegramNutritionOnboardingRuntimeAuthorityMixin, ) from gateway.platforms.telegram_nutrition_onboarding_runtime_collection import ( TelegramNutritionOnboardingRuntimeCollectionMixin, is_consent_withdrawal, ) from gateway.platforms.telegram_nutrition_onboarding_copy import ( parse_legacy_activity_answer, ) from gateway.platforms.telegram_nutrition_onboarding_runtime_errors import ( NutritionOnboardingRuntimeError, ) from gateway.platforms.telegram_nutrition_onboarding_runtime_publication import ( TelegramNutritionOnboardingRuntimePublicationMixin, publication_generation, ) from gateway.platforms.telegram_nutrition_onboarding_publication_outbox import ( GatewayOnboardingPublicationOutbox, ) from gateway.platforms.telegram_nutrition_onboarding_operator_ack import ( GatewayOperatorAttentionStore, ) from gateway.platforms.telegram_nutrition_onboarding_operator_recovery import ( OperatorNotificationRecoveryCapability, load_operator_notification_recovery, ) from gateway.platforms.telegram_nutrition_onboarding_owner_risk import ( OwnerRiskAcceptanceCapability, load_owner_risk_acceptance, ) from gateway.platforms.telegram_nutrition_addresses import StaffReviewRole from gateway.platforms.telegram_customer_bootstrap import ( BootstrapState, ConsentHandoff, Role, consent_handoff_digest, ) from gateway.platforms.telegram_polling_receipts import current_telegram_update_id class TelegramNutritionOnboardingRuntime( TelegramNutritionOnboardingRuntimeAuthorityMixin, TelegramNutritionOnboardingRuntimePublicationMixin, TelegramNutritionOnboardingRuntimeCollectionMixin, TelegramNutritionOnboardingRuntimeCallbackMixin, ): def __init__( self, *, adapter: Any, profile_root: Path, bootstrap_transport: Any, reconciler: NutritionOnboardingReconciler | None = None, ) -> None: self.adapter = adapter self.profile_root = Path(profile_root).resolve() self.bootstrap = bootstrap_transport configured_package = os.environ.get("DUALCOACH_PROFILE_PACKAGE") try: package = DualCoachProfilePackage.from_root( configured_package or self.profile_root / "workspace" / "checkin_cli" ) self.domain = package.resolve( "checkin_cli.nutrition_onboarding", importer=importlib.import_module, ) except ProfilePackageResolutionError as exc: raise NutritionOnboardingRuntimeError(str(exc)) from exc if ( getattr( self.domain, "NUTRITION_ONBOARDING_API_VERSION", None, ) != NUTRITION_ONBOARDING_API_VERSION ): raise NutritionOnboardingRuntimeError( "unsupported nutrition onboarding API version" ) self.profile_root.chmod(0o700) self.reconciler = reconciler or NutritionOnboardingReconciler() self._publication_outbox = GatewayOnboardingPublicationOutbox( self.profile_root ) self._operator_notification_recovery: ( OperatorNotificationRecoveryCapability | None ) = None self._owner_risk_acceptance: OwnerRiskAcceptanceCapability | None = None self._operator_attention_store: GatewayOperatorAttentionStore | None = None self._consent_recovery_lock = anyio.Lock() def configure_operator_notification_recovery( self, extra: dict[str, object], *, candidate_digest: str, ) -> None: owner = self._staff_address( StaffReviewRole.OWNER, self._registry_owner(), ) capability = load_operator_notification_recovery( extra, candidate_digest=candidate_digest, owner_user_id=owner.user_id, owner_chat_id=owner.chat_id, owner_topic_id=owner.topic_id, ) self._operator_notification_recovery = capability self._owner_risk_acceptance = load_owner_risk_acceptance( extra, candidate_digest=candidate_digest, owner_user_id=owner.user_id, owner_chat_id=owner.chat_id, owner_topic_id=owner.topic_id, ) self._operator_attention_store = ( GatewayOperatorAttentionStore(self.profile_root) if capability is not None else None ) def _acknowledge_operator_attention( self, *, session: Any, publication: Any, authority: Any, evidence: Any, callback_data: str, ) -> bool: capability = self._operator_notification_recovery store = self._operator_attention_store payload = getattr(publication, "payload", None) if ( capability is None or store is None or not isinstance(payload, dict) ): raise ValueError("operator attention acknowledgement is unavailable") attention_identity = payload.get("operator_attention_identity") candidate_epoch = payload.get("operator_delivery_epoch") raw_route = self._route(session, "owner") route = (str(raw_route[0]), str(raw_route[1])) records = tuple( record for record in self._publication_outbox.records() if record.session_id == str(session.session_id) and record.generation == publication.generation and record.payload == payload and record.route == route and record.role == "owner" and record.state == "COMMITTED" and record.message_id == publication.message_id ) if ( len(records) != 1 or attention_identity is None or candidate_epoch != capability.candidate_epoch ): raise ValueError("operator attention publication is unavailable") _, created = store.acknowledge( session_id=str(session.session_id), customer_key=str(session.customer_key), attention_identity=str(attention_identity), candidate_epoch=candidate_epoch, actor_user_id=evidence.actor_user_id, authority=( authority.owner_user_id, authority.owner_chat_id, authority.owner_topic_id, ), route=route, message_id=evidence.message_id, update_id=evidence.update_id, callback_data=callback_data, publication=records[0], ) return created async def start_after_consent( self, *, session: Any, query: Any, message: Any, ) -> None: current = self.bootstrap.store.get(session.session_id) if current.state is not BootstrapState.AWAITING_ACTIVATION: return if current.consent_handoff is None: try: current = self._bind_consent_handoff(current, query, message) except (OSError, TypeError, ValueError): await query.answer(text="동의 이벤트를 안전하게 확인하지 못했습니다.") return handoff = current.consent_handoff if handoff is None: return try: self.domain.record_current_registry_consent( self.profile_root, current.customer_key, recorded_on=handoff.recorded_at.astimezone( ZoneInfo("Asia/Seoul") ).date(), ) except (OSError, TypeError, ValueError): await query.answer( text="현재 동의를 안전하게 저장하지 못했습니다." ) return try: authority = self._current_authority(current) except (OSError, ValueError): await query.answer( text="온보딩 권한 또는 동의가 변경됐습니다." ) return evidence = self._handoff_evidence(handoff) service = self._service(current.customer_key) status = service.start_or_resume( authority=authority, evidence=evidence, consent_receipt_digest=handoff.provenance_digest, questionnaire_version="customer_v2", reference_date=handoff.recorded_at.astimezone( ZoneInfo("Asia/Seoul") ).date(), ) await self._publish(current, service, status) self._reconcile_consent_handoff(current) def _service(self, customer_key: str) -> Any: return self.domain.NutritionOnboardingService( profile_root=self.profile_root, customer_key=customer_key, enforce_reconciliation=True, ) def _finalize_owner_approval( self, *, session: Any, service: Any, authority: Any, ) -> Any: return service.finalize( starts_on=date.fromisoformat(session.customer_draft.starts_on), issued_at_kst=datetime.now(ZoneInfo("Asia/Seoul")), privacy_consent_digest=None, feature_epoch_digest=feature_epoch_digest( profile_root=self.profile_root, customer_key=session.customer_key, ), authority=authority, ) async def recover_waiting_session(self, session: Any) -> bool: """Restore one current customer card only from durable onboarding state.""" async with self._consent_recovery_lock: current = self.bootstrap.store.get(session.session_id) if current.state is not BootstrapState.AWAITING_ACTIVATION: return False handoff = current.consent_handoff if handoff is None: if self._operator_notification_recovery is None: return False try: authority = self._current_authority(current) service = self._service(current.customer_key) await self._recover_publication_receipts( service, session=current, ) status = service.status() if status.state.value == "finalizing": status = self._finalize_owner_approval( session=current, service=service, authority=authority, ) except (OSError, TypeError, ValueError): return False if status.state.value not in { "owner_review", "ready", "safety_hold", }: return False await self._publish(current, service, status) if status.state.value == "ready": self._reconcile_owner_callback_receipt( current, service, ) return True try: self.domain.record_current_registry_consent( self.profile_root, current.customer_key, recorded_on=handoff.recorded_at.astimezone( ZoneInfo("Asia/Seoul") ).date(), ) authority = self._current_authority(current) service = self._service(current.customer_key) await self._recover_publication_receipts(service, session=current) try: status = service.status() except FileNotFoundError: status = service.start_or_resume( authority=authority, evidence=self._handoff_evidence(handoff), consent_receipt_digest=handoff.provenance_digest, questionnaire_version="customer_v2", reference_date=handoff.recorded_at.astimezone( ZoneInfo("Asia/Seoul") ).date(), ) if status.state.value == "finalizing": status = self._finalize_owner_approval( session=current, service=service, authority=authority, ) except (OSError, TypeError, ValueError): return False await self._publish(current, service, status) self._reconcile_consent_handoff(current) if status.state.value == "ready": self._reconcile_owner_callback_receipt(current, service) return True def _bind_consent_handoff( self, session: Any, query: Any, message: Any, ) -> Any: recorded_at = datetime.now(timezone.utc) update_id = current_telegram_update_id() actor_id = getattr(getattr(query, "from_user", None), "id", None) chat_id = getattr(getattr(message, "chat", None), "id", None) topic_id = getattr(message, "message_thread_id", None) message_id = getattr(message, "message_id", None) callback_data = getattr(query, "data", None) if topic_id is None: topic_id = 0 if ( type(update_id) is not int or type(actor_id) is not int or type(chat_id) is not int or type(topic_id) is not int or type(message_id) is not int or not isinstance(callback_data, str) ): raise NutritionOnboardingRuntimeError( "consent handoff evidence is incomplete" ) customer = session.role_claim(Role.CUSTOMER) if customer is None: raise NutritionOnboardingRuntimeError( "consent handoff customer authority is unavailable" ) handoff = ConsentHandoff( update_id=update_id, actor_id=actor_id, chat_id=chat_id, topic_id=topic_id, message_id=message_id, callback_data=callback_data, bootstrap_generation=session.generation, recorded_at=recorded_at, provenance_digest="", ) handoff = replace( handoff, provenance_digest=consent_handoff_digest(session, handoff), ) return self.bootstrap.store.bind_consent_handoff( session.session_id, expected_generation=session.generation, handoff=handoff, ) def _handoff_evidence(self, handoff: ConsentHandoff) -> Any: return self.domain.MessageEvidence( actor_user_id=handoff.actor_id, chat_id=handoff.chat_id, topic_id=handoff.topic_id, message_id=handoff.message_id, update_id=handoff.update_id, ) def _reconcile_consent_handoff(self, session: Any) -> bool: current = self.bootstrap.store.get(session.session_id) handoff = current.consent_handoff if handoff is None or current.consent_recovery_reconciled: return False callback = getattr( self.adapter, "_reconcile_telegram_business_commit", None, ) if not callable(callback): return False callback( update_id=handoff.update_id, provenance_digest=handoff.provenance_digest, actor_id=handoff.actor_id, chat_id=handoff.chat_id, topic_id=handoff.topic_id, message_id=handoff.message_id, callback_data=handoff.callback_data, ) self.bootstrap.store.mark_consent_recovery_reconciled( current.session_id, expected_generation=current.generation, provenance_digest=handoff.provenance_digest, ) return True def _preserve_owner_callback_receipt( self, *, session: Any, service: Any, publication: Any, authority: Any, evidence: Any, callback_data: str, ) -> None: """Persist the authenticated Approve event before profile finalization.""" outbox = self._publication_outbox document = self.domain.read_private_json(service.session_path) consumed = document.get("consumed_updates") if not isinstance(consumed, list): raise ValueError("owner callback consumed-update authority is invalid") consumed_updates_before: list[int] = [] for item in consumed: if type(item) is not int or item < 0: raise ValueError("owner callback consumed-update authority is invalid") consumed_updates_before.append(item) if len(set(consumed_updates_before)) != len(consumed_updates_before): raise ValueError("owner callback consumed-update authority is invalid") route = self._route(session, "owner") actor_user_id = getattr(evidence, "actor_user_id", None) update_id = getattr(evidence, "update_id", None) message_id = getattr(evidence, "message_id", None) evidence_route = ( str(getattr(evidence, "chat_id", "")), str(getattr(evidence, "topic_id", "")), ) owner_user_id = getattr(authority, "owner_user_id", None) owner_chat_id = getattr(authority, "owner_chat_id", None) owner_topic_id = getattr(authority, "owner_topic_id", None) if ( type(actor_user_id) is not int or type(update_id) is not int or type(message_id) is not int or type(owner_user_id) is not int or type(owner_chat_id) is not int or type(owner_topic_id) is not int or actor_user_id != owner_user_id or evidence_route != route or message_id != getattr(publication, "message_id", None) ): raise ValueError("owner callback event authority does not match") authority_values = (owner_user_id, owner_chat_id, owner_topic_id) outbox.record_owner_callback( session_id=str(session.session_id), customer_key=str(session.customer_key), action="Approve", actor_user_id=actor_user_id, authority=authority_values, route=route, message_id=message_id, update_id=update_id, consumed_updates_before=tuple(consumed_updates_before), callback_data=callback_data, publication_generation=publication.generation, ) def _reconcile_owner_callback_receipt( self, session: Any, service: Any, ) -> bool: """Authenticate and receipt one owner commit whose UI recovered later.""" outbox = getattr(self, "_publication_outbox", None) ready_path = getattr(service, "ready_path", None) baseline_path = ready_path.parent / "baseline-v1.json" if isinstance(ready_path, Path) else None callback = getattr( getattr(self, "adapter", None), "_reconcile_telegram_business_commit", None, ) domain = getattr(self, "domain", None) read_private_json = getattr(domain, "read_private_json", None) if ( not isinstance(outbox, GatewayOnboardingPublicationOutbox) or not isinstance(ready_path, Path) or not isinstance(baseline_path, Path) or not callable(callback) or not callable(read_private_json) ): return False try: document = read_private_json(ready_path) baseline = read_private_json(baseline_path) consumed = document.get("consumed_updates") event = outbox.owner_callback(str(session.session_id)) if ( document.get("state") != "ready" or event is None or not isinstance(consumed, list) or tuple(consumed) != (*event.consumed_updates_before, event.update_id) ): return False update_id = event.update_id baseline_digest = document.get("baseline_digest") owner_review_receipt = baseline.get("owner_review_receipt") if ( not isinstance(baseline_digest, str) or len(baseline_digest) != 64 or baseline.get("digest") != baseline_digest or not isinstance(owner_review_receipt, str) or len(owner_review_receipt) != 64 ): return False authority = self._current_authority(session) current = service.store.load_session(str(session.session_id)) records = outbox.records() ready_matches = tuple( record for record in records if record.session_id == str(session.session_id) and record.role == "owner" and record.state == "COMMITTED" and record.payload.get("state") == "ready" and record.generation == getattr(current, "generation", None) and record.message_id == getattr(current, "message_id", None) and record.payload == getattr(current, "payload", None) and isinstance(record.receipt_integrity, str) and self._receipt_route_is_current(record, session) ) owner_matches = tuple( record for record in records if record.session_id == str(session.session_id) and record.role == "owner" and record.state == "COMMITTED" and record.payload.get("state") == "owner_review" and record.generation == event.publication_generation and record.message_id == event.message_id and record.route == event.route and record.dispatch_identity == event.publication_dispatch_identity and record.receipt_integrity == event.publication_receipt_integrity and self._receipt_route_is_current(record, session) ) authority_values = ( authority.owner_user_id, authority.owner_chat_id, authority.owner_topic_id, ) expected_callback_data = encode_callback_hash( action="owner_ok", generation=event.publication_generation, sid_hash=str(session.sid_hash), ) if ( len(ready_matches) != 1 or len(owner_matches) != 1 or event.customer_key != str(session.customer_key) or event.action != "Approve" or event.actor_user_id != authority.owner_user_id or event.authority != authority_values or event.route != tuple(str(item) for item in self._route(session, "owner")) or event.callback_data != expected_callback_data ): return False ready_receipt = ready_matches[0] owner_receipt = owner_matches[0] provenance_digest = hashlib.sha256( json.dumps( { "owner_review_receipt": owner_review_receipt, "owner_callback_event_integrity": event.event_integrity, "owner_publication_integrity": owner_receipt.receipt_integrity, "ready_publication_integrity": ready_receipt.receipt_integrity, "session_digest": hashlib.sha256( str(session.session_id).encode() ).hexdigest(), "update_id": update_id, }, sort_keys=True, separators=(",", ":"), ).encode() ).hexdigest() callback( update_id=update_id, provenance_digest=provenance_digest, actor_id=event.actor_user_id, chat_id=int(event.route[0]), topic_id=int(event.route[1]), message_id=event.message_id, callback_data=event.callback_data, ) except (AttributeError, OSError, TypeError, ValueError): return False return True def replayed_status_if_consumed( self, service: object, update_id: int | None, ) -> object | None: """Return durable status only when this exact update already committed.""" if type(update_id) is not int or update_id < 0: return None read_private_json = getattr( getattr(self, "domain", None), "read_private_json", None, ) status = getattr(service, "status", None) if not callable(read_private_json) or not callable(status): return None for path in ( getattr(service, "session_path", None), getattr(service, "ready_path", None), ): if not isinstance(path, Path) or not path.exists(): continue document = read_private_json(path) consumed_updates = document.get("consumed_updates") if isinstance(consumed_updates, list) and update_id in consumed_updates: return status() return None def _rewind_collection( self, *, service: Any, authority: Any, evidence: Any, target_cursor: int, ) -> Any: rewind = getattr(service, "rewind_collection", None) if callable(rewind): return rewind( authority=authority, evidence=evidence, target_cursor=target_cursor, ) fields = self.domain.QUESTION_FIELDS if target_cursor < 0 or target_cursor >= len(fields): raise NutritionOnboardingRuntimeError( "invalid onboarding navigation target" ) document = self.domain.load_mutable_session( session_path=service.session_path, customer_key=service.customer_key, authority=authority, evidence=evidence, role="customer", ) raw_answers = document.get("answers") if not isinstance(raw_answers, dict): raise NutritionOnboardingRuntimeError( "invalid onboarding answers" ) answers: dict[str, object] = {} for key, value in raw_answers.items(): if not isinstance(key, str): raise NutritionOnboardingRuntimeError( "invalid onboarding answers" ) answers[key] = value target_field = fields[target_cursor] if target_field == "schedule_constraints": recovered_activity = parse_legacy_activity_answer( answers.get("activity_category") ) if recovered_activity is not None: category, rationale = recovered_activity answers["activity_category"] = category answers["activity_rationale"] = rationale for field in fields[target_cursor:]: if not isinstance(field, str): raise NutritionOnboardingRuntimeError( "invalid onboarding navigation target" ) answers.pop(field, None) document["answers"] = answers document.update( state=self.domain.OnboardingState.COLLECTING.value, cursor=target_cursor, ) document.pop("reconciliation", None) self.domain.save_session( session_path=service.session_path, document=document, evidence=evidence, ) return self.domain.build_status( customer_key=service.customer_key, document=document, ) async def _continue_later( self, session: Any, service: Any, status: Any, ) -> None: await self._publish_paused(session, service, status) async def _resume(self, session: Any, service: Any) -> None: await self._publish(session, service, service.status())