NuklearRabbit
59d663a43a
Outbox dispatcher (background thread, FOR UPDATE SKIP LOCKED claim, exponential backoff, no transaction held during HTTP I/O). n8n callback endpoint with shared-secret auth and idempotency by event ID. Automation nav + UI with manual retry. 49 backend tests passing, ruff clean. Fixed a crash-on-redelivery bug in seeded outbox payloads and made the dispatcher defensive against malformed payloads. Verified the full live round trip against a real n8n instance: return -> outbox -> dispatcher -> n8n workflow -> callback -> succeeded, including the S5 failed-retry demo scenario.
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Header
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_db
|
|
from app.core.config import get_settings
|
|
from app.core.errors import AppError
|
|
from app.models.audit import AuditEvent
|
|
from app.models.outbox import OutboxEvent
|
|
from app.services.audit import record_audit_event
|
|
|
|
router = APIRouter(prefix="/api/v1/integrations/n8n", tags=["integrations"])
|
|
settings = get_settings()
|
|
|
|
|
|
@router.post("/return-callback")
|
|
def return_callback(
|
|
body: dict[str, Any],
|
|
idempotency_key: str = Header(..., alias="Idempotency-Key"),
|
|
service_token: str = Header(..., alias="X-Service-Token"),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
if service_token != settings.n8n_callback_token:
|
|
raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401)
|
|
|
|
try:
|
|
event_id = uuid.UUID(idempotency_key)
|
|
except ValueError as exc:
|
|
raise AppError(
|
|
"INVALID_IDEMPOTENCY_KEY", "Idempotency-Key must be the event's UUID.", status_code=422
|
|
) from exc
|
|
|
|
event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id))
|
|
if event is None:
|
|
raise AppError("EVENT_NOT_FOUND", "No outbox event matches this event ID.", status_code=404)
|
|
|
|
# Idempotent by event ID: n8n or our own dispatcher may redeliver the same event
|
|
# (e.g. a lost response after a timeout), so this callback must not double-record.
|
|
already_recorded = (
|
|
db.scalar(
|
|
select(AuditEvent.id).where(
|
|
AuditEvent.action == "n8n_return_followup_recorded",
|
|
AuditEvent.metadata_json["event_id"].astext == str(event_id),
|
|
)
|
|
)
|
|
is not None
|
|
)
|
|
if not already_recorded:
|
|
record_audit_event(
|
|
db,
|
|
actor_type="service",
|
|
actor_label="n8n",
|
|
action="n8n_return_followup_recorded",
|
|
entity_type="booking",
|
|
correlation_id=uuid.UUID(body.get("correlation_id"))
|
|
if body.get("correlation_id")
|
|
else None,
|
|
after={"follow_up": body.get("follow_up"), "summary": body.get("summary")},
|
|
metadata={"event_id": str(event_id)},
|
|
)
|
|
db.commit()
|
|
|
|
return {
|
|
"status": "recorded",
|
|
"event_id": str(event_id),
|
|
"occurred_at": datetime.now(UTC).isoformat(),
|
|
}
|