fix: outbox consumer_inbox idempotency logic + tenant_plugin_activation per-tenant check

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:

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,

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",

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()
+    )