400 lines
12 KiB
Python
400 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Iterator
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from services.shared.core import new_id, utc_now_iso
|
|
from services.shared.models import EventEnvelope, EventOutboxItem
|
|
from services.shared.sql_models import EventInboxRow, EventOutboxRow
|
|
|
|
EXCHANGE_NAME = "mvpcc.domain.events"
|
|
AUDIT_QUEUE = "mvpcc.audit.events"
|
|
REPORTING_QUEUE = "mvpcc.reporting.events"
|
|
PREFETCH_COUNT = 20
|
|
RETRY_SCHEDULE = (5, 15, 60, 300)
|
|
|
|
|
|
def _bool_env(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def event_bus_enabled() -> bool:
|
|
return _bool_env("EVENT_BUS_ENABLED", False)
|
|
|
|
|
|
def consumer_enabled() -> bool:
|
|
return _bool_env("EVENT_BUS_CONSUMER_ENABLED", True)
|
|
|
|
|
|
def event_bus_url() -> str:
|
|
return os.getenv("EVENT_BUS_URL", "amqp://guest:guest@localhost:5672/").strip()
|
|
|
|
|
|
def event_bus_exchange() -> str:
|
|
return os.getenv("EVENT_BUS_EXCHANGE", EXCHANGE_NAME).strip() or EXCHANGE_NAME
|
|
|
|
|
|
def event_bus_audit_queue() -> str:
|
|
return os.getenv("EVENT_BUS_AUDIT_QUEUE", AUDIT_QUEUE).strip() or AUDIT_QUEUE
|
|
|
|
|
|
def event_bus_reporting_queue() -> str:
|
|
return os.getenv("EVENT_BUS_REPORTING_QUEUE", REPORTING_QUEUE).strip() or REPORTING_QUEUE
|
|
|
|
|
|
def event_bus_dispatch_batch_size() -> int:
|
|
try:
|
|
return max(1, int(os.getenv("EVENT_BUS_DISPATCH_BATCH_SIZE", "50")))
|
|
except ValueError:
|
|
return 50
|
|
|
|
|
|
def event_bus_max_attempts() -> int:
|
|
try:
|
|
return max(1, int(os.getenv("EVENT_BUS_MAX_ATTEMPTS", "5")))
|
|
except ValueError:
|
|
return 5
|
|
|
|
|
|
def event_bus_poll_interval_seconds() -> float:
|
|
try:
|
|
return max(0.25, float(os.getenv("EVENT_BUS_POLL_INTERVAL_SECONDS", "2")))
|
|
except ValueError:
|
|
return 2.0
|
|
|
|
|
|
def _lazy_import_pika():
|
|
try:
|
|
import pika # type: ignore
|
|
except ImportError as exc:
|
|
raise RuntimeError("pika is required for EVENT_BUS_ENABLED=1") from exc
|
|
return pika
|
|
|
|
|
|
def _iso_in(seconds: int) -> str:
|
|
dt = datetime.now(timezone.utc).timestamp() + seconds
|
|
return datetime.fromtimestamp(dt, tz=timezone.utc).replace(microsecond=0).isoformat()
|
|
|
|
|
|
def retry_delay_seconds(attempt_count: int) -> int:
|
|
if attempt_count <= 1:
|
|
return RETRY_SCHEDULE[0]
|
|
if attempt_count == 2:
|
|
return RETRY_SCHEDULE[1]
|
|
if attempt_count == 3:
|
|
return RETRY_SCHEDULE[2]
|
|
return RETRY_SCHEDULE[3]
|
|
|
|
|
|
def routing_key_for(event_type: str) -> str:
|
|
return event_type.strip()
|
|
|
|
|
|
def build_envelope(
|
|
*,
|
|
event_type: str,
|
|
producer: str,
|
|
entity_type: str,
|
|
entity_id: str,
|
|
payload: dict[str, Any],
|
|
correlation_id: str | None = None,
|
|
event_id: str | None = None,
|
|
event_version: int = 1,
|
|
) -> EventEnvelope:
|
|
return EventEnvelope(
|
|
event_id=event_id or new_id("evt"),
|
|
event_type=event_type,
|
|
event_version=event_version,
|
|
occurred_at=utc_now_iso(),
|
|
producer=producer,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
correlation_id=correlation_id,
|
|
routing_key=routing_key_for(event_type),
|
|
payload=payload,
|
|
)
|
|
|
|
|
|
def outbox_item_from_row(row: EventOutboxRow) -> EventOutboxItem:
|
|
envelope = json.loads(row.payload_json or "{}")
|
|
return EventOutboxItem(
|
|
event_id=row.event_id,
|
|
event_type=row.event_type,
|
|
event_version=row.event_version,
|
|
producer_service=row.producer_service,
|
|
entity_type=row.entity_type,
|
|
entity_id=row.entity_id,
|
|
correlation_id=row.correlation_id,
|
|
routing_key=row.routing_key,
|
|
payload=envelope.get("payload") if isinstance(envelope.get("payload"), dict) else {},
|
|
status=row.status,
|
|
attempt_count=row.attempt_count,
|
|
last_error=row.last_error,
|
|
available_at=row.available_at,
|
|
published_at=row.published_at,
|
|
created_at=row.created_at,
|
|
updated_at=row.updated_at,
|
|
)
|
|
|
|
|
|
def append_outbox_event(
|
|
session: Session,
|
|
*,
|
|
event_type: str,
|
|
producer_service: str,
|
|
entity_type: str,
|
|
entity_id: str,
|
|
payload: dict[str, Any],
|
|
correlation_id: str | None = None,
|
|
) -> EventOutboxRow:
|
|
envelope = build_envelope(
|
|
event_type=event_type,
|
|
producer=producer_service,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
payload=payload,
|
|
correlation_id=correlation_id,
|
|
)
|
|
now = utc_now_iso()
|
|
row = EventOutboxRow(
|
|
event_id=envelope.event_id,
|
|
event_type=envelope.event_type,
|
|
event_version=envelope.event_version,
|
|
producer_service=envelope.producer,
|
|
entity_type=envelope.entity_type,
|
|
entity_id=envelope.entity_id,
|
|
correlation_id=envelope.correlation_id,
|
|
routing_key=envelope.routing_key,
|
|
payload_json=json.dumps(envelope.model_dump(), ensure_ascii=False),
|
|
status="pending",
|
|
attempt_count=0,
|
|
last_error=None,
|
|
available_at=now,
|
|
published_at=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
return row
|
|
|
|
|
|
def list_publishable_outbox(session: Session, limit: int | None = None) -> list[EventOutboxRow]:
|
|
now = utc_now_iso()
|
|
stmt = (
|
|
select(EventOutboxRow)
|
|
.where(
|
|
EventOutboxRow.available_at <= now,
|
|
EventOutboxRow.status.in_(["pending", "failed"]),
|
|
)
|
|
.order_by(EventOutboxRow.id.asc())
|
|
)
|
|
rows = session.execute(stmt).scalars().all()
|
|
if limit is None:
|
|
limit = event_bus_dispatch_batch_size()
|
|
return rows[:limit]
|
|
|
|
|
|
def mark_outbox_published(session: Session, row: EventOutboxRow) -> None:
|
|
now = utc_now_iso()
|
|
row.status = "published"
|
|
row.published_at = now
|
|
row.updated_at = now
|
|
row.last_error = None
|
|
|
|
|
|
def mark_outbox_failed(session: Session, row: EventOutboxRow, error: str) -> None:
|
|
row.attempt_count += 1
|
|
row.status = "failed"
|
|
row.last_error = error[:2000]
|
|
row.available_at = _iso_in(retry_delay_seconds(row.attempt_count))
|
|
row.updated_at = utc_now_iso()
|
|
|
|
|
|
def retry_outbox_event(session: Session, row: EventOutboxRow) -> None:
|
|
now = utc_now_iso()
|
|
row.status = "pending"
|
|
row.last_error = None
|
|
row.available_at = now
|
|
row.updated_at = now
|
|
|
|
|
|
def inbox_seen(session: Session, consumer_name: str, event_id: str) -> bool:
|
|
row = session.execute(
|
|
select(EventInboxRow).where(
|
|
EventInboxRow.consumer_name == consumer_name,
|
|
EventInboxRow.event_id == event_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
return row is not None
|
|
|
|
|
|
def record_inbox(
|
|
session: Session,
|
|
*,
|
|
consumer_name: str,
|
|
event_id: str,
|
|
event_type: str,
|
|
status: str,
|
|
notes: str | None = None,
|
|
) -> EventInboxRow:
|
|
row = EventInboxRow(
|
|
consumer_name=consumer_name,
|
|
event_id=event_id,
|
|
event_type=event_type,
|
|
processed_at=utc_now_iso(),
|
|
status=status,
|
|
notes=notes,
|
|
)
|
|
session.add(row)
|
|
return row
|
|
|
|
|
|
@contextmanager
|
|
def rabbitmq_connection() -> Iterator[tuple[Any, Any]]:
|
|
pika = _lazy_import_pika()
|
|
params = pika.URLParameters(event_bus_url())
|
|
connection = pika.BlockingConnection(params)
|
|
channel = connection.channel()
|
|
try:
|
|
yield connection, channel
|
|
finally:
|
|
try:
|
|
connection.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def ensure_rabbitmq_topology(channel: Any) -> None:
|
|
exchange = event_bus_exchange()
|
|
audit_queue = event_bus_audit_queue()
|
|
reporting_queue = event_bus_reporting_queue()
|
|
channel.exchange_declare(exchange=exchange, exchange_type="topic", durable=True)
|
|
for queue_name, bindings in (
|
|
(audit_queue, ["interaction.*", "agent.state.changed", "call.recording.ready", "ivr.completed"]),
|
|
(
|
|
reporting_queue,
|
|
["interaction.created", "interaction.closed", "agent.state.changed", "ivr.completed", "call.recording.ready"],
|
|
),
|
|
):
|
|
dlq_name = f"{queue_name}.dlq"
|
|
channel.queue_declare(queue=dlq_name, durable=True)
|
|
channel.queue_declare(
|
|
queue=queue_name,
|
|
durable=True,
|
|
arguments={
|
|
"x-dead-letter-exchange": "",
|
|
"x-dead-letter-routing-key": dlq_name,
|
|
},
|
|
)
|
|
for binding in bindings:
|
|
channel.queue_bind(queue=queue_name, exchange=exchange, routing_key=binding)
|
|
|
|
|
|
def publish_envelope(channel: Any, envelope: dict[str, Any], headers: dict[str, Any] | None = None) -> None:
|
|
pika = _lazy_import_pika()
|
|
body = json.dumps(envelope, ensure_ascii=False).encode("utf-8")
|
|
properties = pika.BasicProperties(
|
|
content_type="application/json",
|
|
delivery_mode=2,
|
|
headers=headers or {},
|
|
)
|
|
channel.basic_publish(
|
|
exchange=event_bus_exchange(),
|
|
routing_key=envelope["routing_key"],
|
|
body=body,
|
|
properties=properties,
|
|
)
|
|
|
|
|
|
def dispatch_outbox_batch(session: Session, limit: int | None = None) -> dict[str, int]:
|
|
rows = list_publishable_outbox(session, limit=limit)
|
|
if not rows:
|
|
return {"claimed": 0, "published": 0, "failed": 0}
|
|
|
|
published = 0
|
|
failed = 0
|
|
try:
|
|
with rabbitmq_connection() as (_, channel):
|
|
ensure_rabbitmq_topology(channel)
|
|
for row in rows:
|
|
try:
|
|
publish_envelope(channel, json.loads(row.payload_json or "{}"))
|
|
mark_outbox_published(session, row)
|
|
published += 1
|
|
except Exception as exc: # noqa: BLE001
|
|
mark_outbox_failed(session, row, str(exc))
|
|
failed += 1
|
|
session.commit()
|
|
except Exception as exc: # noqa: BLE001
|
|
for row in rows:
|
|
mark_outbox_failed(session, row, str(exc))
|
|
session.commit()
|
|
failed = len(rows)
|
|
published = 0
|
|
return {"claimed": len(rows), "published": published, "failed": failed}
|
|
|
|
|
|
def consume_one_message(
|
|
queue_name: str,
|
|
handler,
|
|
*,
|
|
max_attempts: int | None = None,
|
|
) -> bool:
|
|
if max_attempts is None:
|
|
max_attempts = event_bus_max_attempts()
|
|
with rabbitmq_connection() as (_, channel):
|
|
ensure_rabbitmq_topology(channel)
|
|
channel.basic_qos(prefetch_count=PREFETCH_COUNT)
|
|
method, properties, body = channel.basic_get(queue=queue_name, auto_ack=False)
|
|
if not method:
|
|
return False
|
|
|
|
headers = dict((properties.headers or {}))
|
|
try:
|
|
envelope = json.loads(body.decode("utf-8"))
|
|
handler(envelope)
|
|
channel.basic_ack(delivery_tag=method.delivery_tag)
|
|
return True
|
|
except Exception as exc: # noqa: BLE001
|
|
retry_count = int(headers.get("x-retry-count", 0))
|
|
if retry_count + 1 < max_attempts:
|
|
next_headers = dict(headers)
|
|
next_headers["x-retry-count"] = retry_count + 1
|
|
publish_envelope(channel, json.loads(body.decode("utf-8")), headers=next_headers)
|
|
channel.basic_ack(delivery_tag=method.delivery_tag)
|
|
else:
|
|
dlq_name = f"{queue_name}.dlq"
|
|
pika = _lazy_import_pika()
|
|
channel.basic_publish(
|
|
exchange="",
|
|
routing_key=dlq_name,
|
|
body=body,
|
|
properties=pika.BasicProperties(
|
|
content_type="application/json",
|
|
delivery_mode=2,
|
|
headers={**headers, "error": str(exc)[:2000]},
|
|
),
|
|
)
|
|
channel.basic_ack(delivery_tag=method.delivery_tag)
|
|
return False
|
|
|
|
|
|
def poll_forever(loop_fn) -> None:
|
|
while True:
|
|
try:
|
|
loop_fn()
|
|
except Exception:
|
|
time.sleep(event_bus_poll_interval_seconds())
|
|
else:
|
|
time.sleep(event_bus_poll_interval_seconds())
|