65a96f3962
The client posted a body-only envelope to {DISPATCH_API_URL}/messages — an
endpoint that does not exist. Now: /dispatch/envelopes, mandatory DispatchCard
block per the platform pydantic model (int schema_version=1/state_code=0,
task_id=0, int4-safe epoch-second sequence_number), payload under card.event,
UUID card_uid (stable uuid5 for ULID-style asset ids). Inbox accepts both
card.event and legacy root body.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""Dispatch participant client (D4 / 08_DECISIONS D3-D4, INTER_MODULE_CONTRACT §2).
|
|
|
|
LegacyHUB joins the shared platform bus as participant ``legacyhub``. This module
|
|
publishes events (``LegacyhubDocumentIndexed``, ``AssetDerivativeReady``) to the
|
|
``dispatch-api`` over HTTP with an ``X-API-Key``, and handles inbound command
|
|
events delivered to the module's HTTP inbox.
|
|
|
|
Design notes:
|
|
- Publishing is best-effort and happens AFTER the DB commit; a dispatch outage
|
|
never fails document processing. Event ids are deterministic (uuid5 over the
|
|
business key) so re-emits are idempotent for downstream consumers.
|
|
- The body carries JSON only - asset ids and object-storage refs, never binary
|
|
payloads, presigned URLs, or local filesystem paths (10 §7).
|
|
- Internal Celery+Redis stays the intra-module queue; this is the inter-module
|
|
channel and the two are not conflated (D3).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app import __version__
|
|
from app.config import settings
|
|
from app.logging_config import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# Stable namespace for deterministic (idempotent) event ids.
|
|
_DISPATCH_NAMESPACE = uuid.UUID("6f1c8d2e-2a4b-5c6d-8e9f-0a1b2c3d4e5f")
|
|
|
|
EVENT_DOCUMENT_INDEXED = "LegacyhubDocumentIndexed"
|
|
EVENT_ASSET_DERIVATIVE_READY = "AssetDerivativeReady"
|
|
|
|
|
|
def deterministic_event_id(*parts: Any) -> str:
|
|
return str(uuid.uuid5(_DISPATCH_NAMESPACE, ":".join(str(p) for p in parts)))
|
|
|
|
|
|
def _card_uid(value: str) -> str:
|
|
"""Envelope card_uid must be a UUID (dispatch_api/models.py). Business ids
|
|
that are already UUIDs pass through; ULID-style asset ids map to a stable
|
|
uuid5 so re-emits keep the same card."""
|
|
try:
|
|
return str(uuid.UUID(str(value)))
|
|
except ValueError:
|
|
return str(uuid.uuid5(_DISPATCH_NAMESPACE, str(value)))
|
|
|
|
|
|
def build_envelope(
|
|
message_type: str,
|
|
card_uid: str,
|
|
body: dict[str, Any],
|
|
*,
|
|
event_id: str | None = None,
|
|
message_id: str | None = None,
|
|
created_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Build the dispatch JSON envelope (02 §2.4 + platform doc 23 profile).
|
|
|
|
The card block mirrors the mandatory DispatchCard fields of the platform
|
|
pydantic model (dispatch_api/models.py): int schema_version/state_code,
|
|
task_id 0 for out-of-ticket events, int4-safe epoch-second
|
|
sequence_number; the event payload nests under card.event untouched.
|
|
"""
|
|
now = created_at or datetime.now(tz=UTC).isoformat()
|
|
uid = _card_uid(card_uid)
|
|
return {
|
|
"message_id": message_id or str(uuid.uuid4()),
|
|
"event_id": event_id or str(uuid.uuid4()),
|
|
"card_uid": uid,
|
|
"task_id": 0,
|
|
"message_type": message_type,
|
|
"participant_code": settings.dispatch_participant_code,
|
|
"created_at": now,
|
|
"card": {
|
|
"schema_version": 1,
|
|
"card_uid": uid,
|
|
"task_id": 0,
|
|
"sequence_number": int(time.time()),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
"state_code": 0,
|
|
"specialist_status": "n/a",
|
|
"event": body,
|
|
},
|
|
"media_files": [],
|
|
"media_file_count": 0,
|
|
"local_delivery_status": {},
|
|
}
|
|
|
|
|
|
def publish(
|
|
message_type: str,
|
|
card_uid: str,
|
|
body: dict[str, Any],
|
|
*,
|
|
event_id: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""Publish one envelope to dispatch-api. Best-effort; returns the envelope on
|
|
success, ``None`` when disabled or on a (logged) transport failure."""
|
|
if not settings.dispatch_enabled:
|
|
logger.info("dispatch.disabled", message_type=message_type, card_uid=str(card_uid))
|
|
return None
|
|
if not settings.dispatch_api_url:
|
|
logger.warning("dispatch.no_url", message_type=message_type)
|
|
return None
|
|
|
|
envelope = build_envelope(message_type, card_uid, body, event_id=event_id)
|
|
try:
|
|
res = httpx.post(
|
|
f"{settings.dispatch_api_url.rstrip('/')}/dispatch/envelopes",
|
|
json=envelope,
|
|
headers={"X-API-Key": settings.dispatch_api_key},
|
|
timeout=settings.dispatch_timeout_seconds,
|
|
)
|
|
res.raise_for_status()
|
|
except Exception as exc: # noqa: BLE001 - dispatch must never break ingestion
|
|
logger.warning("dispatch.publish_failed", message_type=message_type, error=str(exc))
|
|
return None
|
|
|
|
logger.info("dispatch.published", message_type=message_type, event_id=envelope["event_id"])
|
|
return envelope
|
|
|
|
|
|
def publish_document_indexed(
|
|
document_id: uuid.UUID | str, *, asset_metadata: dict[str, Any] | None = None
|
|
) -> dict[str, Any] | None:
|
|
body: dict[str, Any] = {"document_id": str(document_id)}
|
|
card_uid = str(document_id)
|
|
if asset_metadata:
|
|
for key in ("asset_id", "owner_module", "owner_record_type", "owner_record_id", "manifest_version"):
|
|
if asset_metadata.get(key):
|
|
body[key] = asset_metadata[key]
|
|
if asset_metadata.get("asset_id"):
|
|
card_uid = str(asset_metadata["asset_id"])
|
|
event_id = deterministic_event_id(EVENT_DOCUMENT_INDEXED, card_uid)
|
|
return publish(EVENT_DOCUMENT_INDEXED, card_uid, body, event_id=event_id)
|
|
|
|
|
|
def publish_derivative_ready(
|
|
asset_metadata: dict[str, Any],
|
|
*,
|
|
derivative_type: str = "ocr_markdown",
|
|
content_type: str = "text/markdown",
|
|
) -> dict[str, Any] | None:
|
|
asset_id = asset_metadata.get("asset_id")
|
|
if not asset_id:
|
|
return None
|
|
body = {
|
|
"event_type": EVENT_ASSET_DERIVATIVE_READY, # canonical body-level type (10 §7)
|
|
"asset_id": asset_id,
|
|
"owner_module": asset_metadata.get("owner_module"),
|
|
"owner_record_type": asset_metadata.get("owner_record_type"),
|
|
"owner_record_id": asset_metadata.get("owner_record_id"),
|
|
"derivative_type": derivative_type,
|
|
"content_type": content_type,
|
|
"source_asset_sha256": asset_metadata.get("sha256"),
|
|
"generator": {"pipeline": "legacyhub", "version": __version__},
|
|
}
|
|
event_id = deterministic_event_id(EVENT_ASSET_DERIVATIVE_READY, asset_id, derivative_type)
|
|
return publish(EVENT_ASSET_DERIVATIVE_READY, str(asset_id), body, event_id=event_id)
|
|
|
|
|
|
def emit_after_indexing(document_id: uuid.UUID) -> None:
|
|
"""Emit indexing events after a document reaches INDEXING_COMPLETED.
|
|
|
|
Called at the tail of the pipeline (after commit). No-op when dispatch is
|
|
disabled. Always best-effort.
|
|
"""
|
|
if not settings.dispatch_enabled:
|
|
return
|
|
from app.ingestion.knowledge_ingest import ( # noqa: PLC0415
|
|
load_asset_ingest_options,
|
|
load_asset_metadata_for_document,
|
|
)
|
|
|
|
meta = load_asset_metadata_for_document(document_id) or {}
|
|
publish_document_indexed(document_id, asset_metadata=meta or None)
|
|
|
|
if meta.get("asset_id"):
|
|
options = load_asset_ingest_options(document_id)
|
|
if options.get("emit_events", True):
|
|
publish_derivative_ready(meta)
|
|
|
|
|
|
# ---- Inbound (HTTP inbox) ----
|
|
|
|
def handle_inbox_message(envelope: dict[str, Any]) -> dict[str, Any]:
|
|
"""Process one inbound dispatch envelope idempotently.
|
|
|
|
Idempotency: an ``audit_log`` row keyed on ``event_id`` marks a processed
|
|
event; a repeat delivery is acknowledged without reprocessing (at-least-once
|
|
delivery, 10 §7).
|
|
"""
|
|
from sqlalchemy import select # noqa: PLC0415
|
|
|
|
from app.db.audit import record_audit # noqa: PLC0415
|
|
from app.db.models import AuditLog # noqa: PLC0415
|
|
from app.db.session import session_scope # noqa: PLC0415
|
|
|
|
event_id = str(envelope.get("event_id") or "")
|
|
message_type = str(envelope.get("message_type") or "")
|
|
# Doc 23 profile nests the payload under card.event; legacy senders used a
|
|
# root-level body. Accept both at this external boundary.
|
|
card = envelope.get("card") or {}
|
|
body = card.get("event") or envelope.get("body") or {}
|
|
|
|
with session_scope() as db:
|
|
already = db.execute(
|
|
select(AuditLog).where(
|
|
AuditLog.action == "dispatch.inbox.processed",
|
|
AuditLog.entity_id == event_id,
|
|
)
|
|
).first()
|
|
if already is not None:
|
|
return {"status": "confirmed", "event_id": event_id, "duplicate": True}
|
|
|
|
handled = _handle_inbox(message_type, body)
|
|
record_audit(
|
|
db,
|
|
action="dispatch.inbox.processed",
|
|
actor=f"participant:{envelope.get('participant_code')}",
|
|
entity_type="dispatch_event",
|
|
entity_id=event_id,
|
|
details={"message_type": message_type, "handled": handled},
|
|
)
|
|
return {"status": "confirmed", "event_id": event_id, "handled": handled}
|
|
|
|
|
|
def _handle_inbox(message_type: str, body: dict[str, Any]) -> bool:
|
|
"""Return True if the message triggered an action."""
|
|
if message_type in ("ReindexRequested", "LegacyhubReindexRequested"):
|
|
return _enqueue_reindex(body)
|
|
if message_type in ("AssetTombstoned", "AssetDeleted"):
|
|
return _purge_projection(body)
|
|
logger.info("dispatch.inbox.ignored", message_type=message_type)
|
|
return False
|
|
|
|
|
|
def _purge_projection(body: dict[str, Any]) -> bool:
|
|
asset_id = body.get("asset_id")
|
|
if not asset_id:
|
|
logger.warning("dispatch.tombstone.no_asset_id")
|
|
return False
|
|
from app.indexing.projection import purge_projection_by_asset_id # noqa: PLC0415
|
|
|
|
result = purge_projection_by_asset_id(str(asset_id))
|
|
return bool(result.get("purged"))
|
|
|
|
|
|
def _enqueue_reindex(body: dict[str, Any]) -> bool:
|
|
from sqlalchemy import select # noqa: PLC0415
|
|
|
|
from app.db.models import AssetIngestJob # noqa: PLC0415
|
|
from app.db.session import session_scope # noqa: PLC0415
|
|
|
|
document_id = body.get("document_id")
|
|
asset_id = body.get("asset_id")
|
|
if document_id is None and asset_id is not None:
|
|
with session_scope() as db:
|
|
job = (
|
|
db.execute(
|
|
select(AssetIngestJob)
|
|
.where(AssetIngestJob.asset_id == str(asset_id))
|
|
.order_by(AssetIngestJob.created_at.desc())
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
document_id = str(job.document_id) if job is not None else None
|
|
if document_id is None:
|
|
logger.warning("dispatch.reindex.unresolved", asset_id=asset_id)
|
|
return False
|
|
|
|
from app.db.audit import record_audit # noqa: PLC0415
|
|
from app.db.session import session_scope # noqa: PLC0415
|
|
|
|
with session_scope() as db:
|
|
record_audit(
|
|
db,
|
|
action="asset.reindex",
|
|
actor="dispatch",
|
|
entity_type="document",
|
|
entity_id=str(document_id),
|
|
details={"asset_id": asset_id},
|
|
)
|
|
|
|
from app.workers.tasks import process_document # noqa: PLC0415
|
|
|
|
process_document.delay(str(document_id), str(uuid.uuid4()))
|
|
logger.info("dispatch.reindex.enqueued", document_id=str(document_id))
|
|
return True
|
|
|
|
|
|
__all__ = [
|
|
"EVENT_ASSET_DERIVATIVE_READY",
|
|
"EVENT_DOCUMENT_INDEXED",
|
|
"build_envelope",
|
|
"deterministic_event_id",
|
|
"emit_after_indexing",
|
|
"handle_inbox_message",
|
|
"publish",
|
|
"publish_derivative_ready",
|
|
"publish_document_indexed",
|
|
]
|