"""Consume QMS §7 asset events from the dispatch service inbox (C2). LegacyHUB is subscribed to ``AssetRegistered``/``AssetApproved`` emitted by QMS for release/evidence objects in the shared MinIO. This consumer polls ``GET /dispatch/inbox/service/{participant}`` (the inbox returns the same head message until it is confirmed), maps the §7 payload to an ``AssetManifestEnvelope`` and runs the local knowledge-ingest, then confirms the delivery. Outcome policy: - ``AssetRegistered`` → ingest; accepted/duplicate/rejected all confirm (rejected is a permanent verdict, retrying cannot change it). - ``AssetApproved`` / unknown types → confirm without work (no head-of-line blocking on messages that carry nothing to do). - Malformed payload → confirm as poison (permanently unmappable). - Transient infrastructure failure (object storage / DB down) → NO confirm and stop the run; the next scheduled run retries the same message. Secrets (X-API-Key) are never logged. """ from __future__ import annotations from collections.abc import Callable from typing import Any import httpx from app.config import settings from app.logging_config import get_logger logger = get_logger(__name__) MANIFEST_VERSION = "1.0" CONSUMED_EVENT_TYPES = {"AssetRegistered", "AssetApproved"} class TransientIngestError(Exception): """Ingest failed for a reason a later retry can fix (storage/DB outage).""" def manifest_envelope_from_event(event: dict[str, Any]) -> dict[str, Any]: """Map a §7 asset-event payload to an AssetManifestEnvelope dict (§8).""" ref = event.get("manifest_ref") if not isinstance(ref, dict): raise ValueError("event has no manifest_ref") for required in ("asset_id", "owner_module", "owner_record_type", "owner_record_id", "content_type", "size_bytes", "sha256"): if event.get(required) in (None, ""): raise ValueError(f"event misses required field: {required}") object_key = str(ref.get("object_key") or "") filename = object_key.rsplit("/", 1)[-1] or None return { "manifest": { "manifest_version": MANIFEST_VERSION, "asset": { "asset_id": str(event["asset_id"]), "owner_module": str(event["owner_module"]), "owner_record_type": str(event["owner_record_type"]), "owner_record_id": str(event["owner_record_id"]), "asset_kind": str(event.get("asset_kind") or "document"), "title": None, "original_filename": filename, "content_type": str(event["content_type"]), "size_bytes": int(event["size_bytes"]), "sha256": str(event["sha256"]), "created_at": None, "created_by": event.get("actor"), }, "storage": { "provider": str(ref.get("provider") or "s3"), "bucket": str(ref.get("bucket") or ""), "object_key": object_key, "version_id": None, }, "security": { "gate_status": str((event.get("security") or {}).get("gate_status") or ""), }, "retention": { "policy_id": f"{event['owner_module']}-{event['owner_record_type']}", "retain_until": None, "legal_hold": False, }, "derivatives": [], "links": [], "source": {"source_system": str(event["owner_module"])}, } } def _event_from_message(message: dict[str, Any]) -> dict[str, Any]: envelope = message.get("envelope") or {} card = envelope.get("card") or {} event = card.get("event") return event if isinstance(event, dict) else {} def consume_inbox( *, fetch: Callable[[], dict[str, Any] | None], ingest: Callable[[dict[str, Any]], str], confirm: Callable[[str], Any], limit: int = 25, ) -> dict[str, Any]: """Drain up to ``limit`` messages; injected callables keep this pure.""" summary: dict[str, Any] = {"ingested": 0, "confirmed": 0, "skipped": 0} for _ in range(limit): message = fetch() if not message: break message_id = str(message.get("message_id") or "") message_type = str(message.get("message_type") or "") if message_type == "AssetRegistered": try: envelope = manifest_envelope_from_event(_event_from_message(message)) except ValueError as exc: logger.warning("dispatch_consume.poison", message_id=message_id, error=str(exc)) confirm(message_id) summary["confirmed"] += 1 summary["skipped"] += 1 continue try: status = ingest(envelope) except TransientIngestError as exc: logger.warning("dispatch_consume.transient", message_id=message_id, error=str(exc)) summary["stopped"] = "transient_error" break summary["ingested"] += 1 logger.info("dispatch_consume.ingested", message_id=message_id, status=str(status)) else: if message_type not in CONSUMED_EVENT_TYPES: logger.info("dispatch_consume.unhandled_type", message_id=message_id, message_type=message_type) summary["skipped"] += 1 confirm(message_id) summary["confirmed"] += 1 return summary # ---------------- live wiring (HTTP fetch/confirm + local ingest) ---------------- def _http_fetch(client: httpx.Client) -> dict[str, Any] | None: response = client.get( f"{settings.dispatch_api_url.rstrip('/')}/dispatch/inbox/service/" f"{settings.dispatch_participant_code}", headers={"X-API-Key": settings.dispatch_api_key}, ) response.raise_for_status() payload = response.json() return payload if payload.get("envelope") else None def _http_confirm(client: httpx.Client, message_id: str) -> None: response = client.post( f"{settings.dispatch_api_url.rstrip('/')}/dispatch/deliveries/confirm", headers={"X-API-Key": settings.dispatch_api_key}, json={"message_id": message_id, "participant_code": settings.dispatch_participant_code}, ) response.raise_for_status() def _local_ingest(envelope: dict[str, Any]) -> str: from app.api.schemas import AssetManifestEnvelope from app.ingestion.knowledge_ingest import KnowledgeIngestError, accept_knowledge_ingest request = AssetManifestEnvelope.model_validate(envelope) idempotency_key = ( f"dispatch-consume-{request.manifest.asset.asset_id}-{request.manifest.asset.sha256}" ) try: response = accept_knowledge_ingest(request, idempotency_key=idempotency_key) except KnowledgeIngestError as exc: if exc.reason_code == "object_read_failed": raise TransientIngestError(exc.message) from exc logger.warning("dispatch_consume.ingest_rejected", reason_code=exc.reason_code) return f"rejected:{exc.reason_code}" return str(response.status) def consume_once(limit: int = 25) -> dict[str, Any]: if not (settings.dispatch_enabled and settings.dispatch_api_url and settings.dispatch_api_key): return {"skipped_run": "dispatch_disabled"} with httpx.Client(timeout=settings.dispatch_timeout_seconds) as client: return consume_inbox( fetch=lambda: _http_fetch(client), ingest=_local_ingest, confirm=lambda message_id: _http_confirm(client, message_id), limit=limit, )