62 lines
3.1 KiB
Python
62 lines
3.1 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import CheckConstraint, DateTime, Index, Integer, String, Text
|
|
from sqlalchemy.dialects.postgresql import JSONB, UUID
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from app.core.db import Base
|
|
from app.models.mixins import TimestampMixin
|
|
|
|
DELIVERY_STATUSES = ("pending", "delivering", "succeeded", "failed")
|
|
|
|
# The one delivery failure the demo seed deliberately plants (BK-H-0020, see
|
|
# seed/workflow_runs.csv). It exists to show retry and audit working, so it must never
|
|
# be read as an integration-health problem: it is a scripted prop, not evidence that
|
|
# n8n is unhealthy. A dedicated error code -- rather than the generic
|
|
# "connectionError" a real timeout produces -- is what lets every reader tell the two
|
|
# apart without guessing from the message text.
|
|
#
|
|
# It is deliberately a `last_error_code` value and not a new column: the code is
|
|
# already persisted, already surfaced to the UI, and already localizable, so no schema
|
|
# change or migration is needed. A genuine later failure of this same event overwrites
|
|
# the code with the real one, which is exactly right -- from that moment it *is* a real
|
|
# failure.
|
|
DEMO_SCENARIO_ERROR_CODE = "demoScenarioTimeout"
|
|
|
|
|
|
def is_demo_scenario_failure(event: "OutboxEvent") -> bool:
|
|
"""True for the prepared demo failure, false for every real one."""
|
|
return event.delivery_status == "failed" and event.last_error_code == DEMO_SCENARIO_ERROR_CODE
|
|
|
|
|
|
class OutboxEvent(TimestampMixin, Base):
|
|
__tablename__ = "outbox_events"
|
|
__table_args__ = (
|
|
CheckConstraint(
|
|
"delivery_status IN ('pending','delivering','succeeded','failed')",
|
|
name="ck_outbox_delivery_status",
|
|
),
|
|
CheckConstraint("attempts >= 0", name="ck_outbox_attempts"),
|
|
Index("ix_outbox_delivery_next_attempt", "delivery_status", "next_attempt_at"),
|
|
)
|
|
|
|
event_id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
|
|
)
|
|
event_type: Mapped[str] = mapped_column(String(60), nullable=False)
|
|
aggregate_type: Mapped[str] = mapped_column(String(30), nullable=False)
|
|
aggregate_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
|
|
payload_json: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
|
|
occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
|
|
delivery_status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
|
|
attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
|
last_error: Mapped[str | None] = mapped_column(Text)
|
|
# Stable, localizable classification of last_error -- the frontend renders a
|
|
# localized summary from this code as the primary text and shows last_error itself
|
|
# only under "Technical details" (section 10 of docs/fleet-ops-correction/
|
|
# current-gap-audit.md). Kept alongside the raw message for backward compatibility.
|
|
last_error_code: Mapped[str | None] = mapped_column(String(60))
|
|
external_run_id: Mapped[str | None] = mapped_column(String(120))
|