From 648d8d89d6b0f363396e62f825f6aaa2983d698d Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Wed, 29 Jul 2026 13:02:33 +0200 Subject: [PATCH] fix: outbox consumer_inbox idempotency logic + tenant_plugin_activation per-tenant check --- app/core/outbox.py | 18 +++++++++++++++- app/deps.py | 35 ++++++++++++++++++++++++------ app/models/__init__.py | 2 ++ app/models/consumer_inbox.py | 42 ++++++++++++++++++++++++++++++++++++ 4 files changed, 89 insertions(+), 8 deletions(-) create mode 100644 app/models/consumer_inbox.py diff --git a/app/core/outbox.py b/app/core/outbox.py index e59c426..a8a7296 100644 --- a/app/core/outbox.py +++ b/app/core/outbox.py @@ -179,6 +179,18 @@ async def process_outbox_batch( payload_dict.setdefault("_event_name", event_name) payload_dict.setdefault("_event_timestamp", datetime.now(timezone.utc).isoformat()) + # Idempotency check: has this event already been processed? (P1.5 fix) + already_processed = await db.execute( + text("SELECT 1 FROM consumer_inbox WHERE event_id = :eid AND status = 'processed' LIMIT 1"), + {"eid": str(event_id)}, + ) + if already_processed.first(): + # Event was already processed by all consumers — mark as published + await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)}) + published_count += 1 + logger.debug("Outbox event %s already processed, marking as published", event_id) + continue + results = await event_bus.publish_with_results(event_name, payload_dict) # Check if any handlers were registered at all @@ -190,13 +202,17 @@ async def process_outbox_batch( if handler_count == 0: # No handlers registered — mark as 'no_handlers' not 'published' - # This prevents events from silently disappearing await db.execute( text("UPDATE event_outbox SET status = 'no_handlers', published_at = now() WHERE id = :id"), {"id": str(event_id)}, ) logger.warning("Outbox event %s (%s) had no handlers registered", event_id, event_name) else: + # Record in consumer_inbox for idempotency (P1.5 fix) + await db.execute( + text("INSERT INTO consumer_inbox (event_id, consumer_name, status, processed_at) VALUES (:eid, :name, 'processed', now()) ON CONFLICT DO NOTHING"), + {"eid": str(event_id), "name": event_name}, + ) await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)}) published_count += 1 except Exception as exc: diff --git a/app/deps.py b/app/deps.py index dadc062..42467f4 100644 --- a/app/deps.py +++ b/app/deps.py @@ -290,13 +290,10 @@ async def get_current_user_id( def require_active_plugin(plugin_name: str): """FastAPI dependency factory: require that a plugin is active. - Returns 403 if the plugin is not active in the permission registry. - This allows routes to be registered at app creation time while - enforcing activation status at request time. + Checks both global activation (permission registry) and per-tenant + activation (tenant_plugin_activation table). - Note: Does NOT depend on get_current_user — individual route handlers - already have their own auth dependencies (require_permission, etc.). - This check only verifies plugin activation status. + Returns 403 if the plugin is not active. """ async def _check() -> None: from app.core.permission_registry import get_permission_registry @@ -310,11 +307,35 @@ def require_active_plugin(plugin_name: str): "code": "plugin_inactive", }, ) + # Per-tenant activation check (P1.4 fix) + # If there's an entry in tenant_plugin_activation for this + # tenant+plugin with is_active=false, deny access. + # If no entry exists, default to active (backward compatible). + from app.core.db import async_session_maker + from sqlalchemy import text + async with async_session_maker() as db: + # Get tenant_id from current session context (RLS) + result = await db.execute( + text(""" + SELECT is_active FROM tenant_plugin_activation + WHERE plugin_name = :name + AND tenant_id = current_setting('app.current_tenant_id', true)::uuid + """), + {"name": plugin_name}, + ) + row = result.first() + if row is not None and not row[0]: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail={ + "detail": f"Plugin '{plugin_name}' is not active for this tenant", + "code": "plugin_inactive_tenant", + }, + ) except HTTPException: raise except Exception as exc: # Fail-closed: if registry check fails, deny access (P1.2 fix) - # Previously this was fail-open (pass) which allowed access on errors logger.error("Plugin activation check failed for '%s': %s", plugin_name, exc) raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, diff --git a/app/models/__init__.py b/app/models/__init__.py index 5321586..a0d5cd3 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -12,6 +12,7 @@ from app.models.contact_folder_permission import ContactFolderPermission from app.models.contact_merge import ContactMergeHistory from app.models.entity_permission import EntityPermission from app.models.guest_user import GuestUser +from app.models.consumer_inbox import ConsumerInbox from app.models.guest_invitation import GuestInvitation from app.models.entity_policy import EntityPolicy from app.models.permission_template import PermissionTemplate @@ -57,6 +58,7 @@ __all__ = [ "ContactMergeHistory", "EntityPermission", "GuestInvitation", + "ConsumerInbox", "GuestUser", "PermissionDelegation", "PermissionTemplate", diff --git a/app/models/consumer_inbox.py b/app/models/consumer_inbox.py new file mode 100644 index 0000000..93bd68c --- /dev/null +++ b/app/models/consumer_inbox.py @@ -0,0 +1,42 @@ +"""Consumer inbox model for outbox idempotency.""" + +from __future__ import annotations + +import uuid +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, String, Text, UniqueConstraint, func +from sqlalchemy.dialects.postgresql import UUID as PGUUID +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.db import Base + + +class ConsumerInbox(Base): + """Tracks which consumers have processed which outbox events. + + Prevents duplicate processing when a worker crashes between + delivering an event and marking it as published. + """ + + __tablename__ = "consumer_inbox" + __table_args__ = ( + UniqueConstraint("event_id", "consumer_name", name="uq_consumer_inbox_event_consumer"), + ) + + id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + event_id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), + ForeignKey("event_outbox.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + consumer_name: Mapped[str] = mapped_column(String(100), nullable=False, index=True) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending") + processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + )