fix(webhooks): JSONB-Containment statt .any() auf JSON-Spalte — behebt 158 fehlgeschlagene Outbox-Events

app/core/webhook_dispatcher.py

@@ -7,7 +7,8 @@ import logging
 import uuid
 from typing import Any
 
-from sqlalchemy import select
+from sqlalchemy import cast, select
+from sqlalchemy.dialects.postgresql import JSONB
 
 from app.core.db import get_session_factory
 from app.core.event_bus import EventBus, get_event_bus
@@ -48,7 +49,12 @@ async def _dispatch_event(payload: dict[str, Any]) -> None:
         stmt = select(Webhook).where(
             Webhook.tenant_id == tenant_id,
             Webhook.is_active == True,  # noqa: E712
-            Webhook.events.any(event_name),
+            # Webhook.events is a JSONB array column (NOT a relationship):
+            # events @> '["<event_name>"]' — JSONB containment instead of
+            # the invalid relationship .any() call that crashed every event
+            # with "Neither 'AnnotatedColumn' nor 'Comparator' object has an
+            # attribute 'any'" (158 failed outbox events in production).
+            cast(Webhook.events, JSONB).contains([event_name]),
         )
         result = await db.execute(stmt)
         webhooks = list(result.scalars().all())

app/services/webhook_service.py

@@ -14,7 +14,8 @@ from typing import Any
 from urllib.parse import urlparse
 
 import httpx
-from sqlalchemy import select
+from sqlalchemy import cast, select
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.visibility import apply_visibility_filter, check_single_entity_access
@@ -74,7 +75,10 @@ async def list_webhooks(
         Webhook.tenant_id == tenant_id,
     )
     if event:
-        stmt = stmt.where(Webhook.events.any(event))
+        # Webhook.events is a JSONB array column (NOT a relationship):
+        # events @> '["<event>"]' — JSONB containment instead of the invalid
+        # relationship .any() call (same fix as webhook_dispatcher).
+        stmt = stmt.where(cast(Webhook.events, JSONB).contains([event]))
     if user_id and not is_system_admin:
         stmt = await apply_visibility_filter(
             db, stmt, "webhook", Webhook, user_id, tenant_id, is_system_admin

tests/test_webhooks.py

@@ -0,0 +1,173 @@
+"""Webhook regression tests — JSONB containment on Webhook.events.
+
+Regression (2026-09-14, 158 failed outbox events in production):
+Webhook.events is a JSONB array column, NOT a relationship. Both
+webhook_service.list_webhooks (event filter) and
+webhook_dispatcher._dispatch_event called Webhook.events.any(event),
+which only exists on relationship attributes. Every event publish crashed
+with "Neither 'AnnotatedColumn' nor 'Comparator' object has an attribute
+'any'" and the outbox marked the events as failed after retries.
+
+Fix: PostgreSQL JSONB containment events @> '["<event>"]' via
+cast(Webhook.events, JSONB).contains([event]).
+
+These tests prove both fixed code paths against a real PostgreSQL test DB
+(the JSONB cast only compiles correctly on PostgreSQL, not SQLite).
+"""
+
+from __future__ import annotations
+
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.core.webhook_dispatcher import _dispatch_event
+from app.models.webhook import Webhook
+from app.services.webhook_service import list_webhooks
+
+
+@pytest_asyncio.fixture
+def test_session_factory(session_factory: async_sessionmaker[AsyncSession]) -> async_sessionmaker[AsyncSession]:
+    """The test session factory — patches the dispatcher's global factory.
+
+    ``_dispatch_event`` opens its own session via ``get_session_factory()``
+    (the app engine). In tests the global engine points at the app DB unless
+    ``reset_engine_for_testing`` is active, so we patch the symbol the
+    dispatcher resolves at call time to use the test engine instead.
+    """
+    return session_factory
+
+
+def _make_webhook(
+    tenant_id: uuid.UUID,
+    events: list[str],
+    *,
+    url: str = "https://example.com/hook",
+    is_active: bool = True,
+) -> Webhook:
+    return Webhook(
+        tenant_id=tenant_id,
+        url=url,
+        events=events,
+        is_active=is_active,
+        retry_count=3,
+        timeout_seconds=30,
+    )
+
+
+@pytest_asyncio.fixture
+async def tenant_id(db_session: AsyncSession) -> uuid.UUID:
+    """A fresh tenant id; Webhook rows are tenant-scoped."""
+    return uuid.uuid4()
+
+
+@pytest.mark.asyncio
+class TestListWebhooksEventFilter:
+    """webhook_service.list_webhooks — event filter uses JSONB containment."""
+
+    async def test_event_filter_returns_matching_webhook(
+        self, db_session: AsyncSession, tenant_id: uuid.UUID
+    ) -> None:
+        """Regression: .any() raised AttributeError before the JSONB fix."""
+        matching = _make_webhook(tenant_id, ["file.deleted", "contact.created"])
+        other = _make_webhook(tenant_id, ["contact.created"])
+        db_session.add_all([matching, other])
+        await db_session.flush()
+
+        # Before the fix this raised:
+        # Neither 'AnnotatedColumn' object nor 'Comparator' object
+        # has an attribute 'any'
+        result = await list_webhooks(db_session, tenant_id, event="file.deleted")
+
+        assert len(result) == 1, "event filter must select only matching webhooks"
+        assert result[0].id == matching.id
+        assert result[0].url == matching.url
+
+    async def test_event_filter_empty_when_no_match(
+        self, db_session: AsyncSession, tenant_id: uuid.UUID
+    ) -> None:
+        db_session.add(_make_webhook(tenant_id, ["contact.created"]))
+        await db_session.flush()
+
+        result = await list_webhooks(db_session, tenant_id, event="file.deleted")
+        assert result == []
+
+    async def test_no_event_filter_returns_all(
+        self, db_session: AsyncSession, tenant_id: uuid.UUID
+    ) -> None:
+        db_session.add_all([
+            _make_webhook(tenant_id, ["file.deleted"]),
+            _make_webhook(tenant_id, ["contact.created"]),
+        ])
+        await db_session.flush()
+
+        result = await list_webhooks(db_session, tenant_id)
+        assert len(result) == 2
+
+
+@pytest.mark.asyncio
+class TestDispatcherEventMatching:
+    """webhook_dispatcher._dispatch_event — wildcard handler query."""
+
+    async def test_dispatch_matches_webhook_by_event(
+        self, db_session: AsyncSession, tenant_id: uuid.UUID, test_session_factory
+    ) -> None:
+        """Regression: the dispatcher query crashed with .any() and every
+        outbox event failed (158 production events 2026-08-27)."""
+        hook = _make_webhook(tenant_id, ["file.deleted", "contact.created"])
+        db_session.add(hook)
+        await db_session.commit()  # dispatcher opens its own session
+
+        sent = AsyncMock(return_value={"success": True, "status_code": 200})
+        with patch("app.core.webhook_dispatcher.send_webhook", sent), \
+             patch("app.core.webhook_dispatcher.get_session_factory", return_value=test_session_factory):
+            # Before the fix this raised AttributeError and the outbox
+            # marked the event failed after retries.
+            await _dispatch_event({
+                "event_name": "file.deleted",
+                "tenant_id": str(tenant_id),
+                "data": {"file_id": "abc"},
+            })
+
+        sent.assert_awaited_once()
+        dispatched_hook = sent.await_args.args[0]
+        assert dispatched_hook.id == hook.id
+        assert sent.await_args.args[1] == "file.deleted"
+        assert sent.await_args.args[2] == {"file_id": "abc"}
+
+    async def test_dispatch_skips_webhook_with_other_events(
+        self, db_session: AsyncSession, tenant_id: uuid.UUID, test_session_factory
+    ) -> None:
+        db_session.add(_make_webhook(tenant_id, ["contact.created"]))
+        await db_session.commit()
+
+        sent = AsyncMock(return_value={"success": True, "status_code": 200})
+        with patch("app.core.webhook_dispatcher.send_webhook", sent), \
+             patch("app.core.webhook_dispatcher.get_session_factory", return_value=test_session_factory):
+            await _dispatch_event({
+                "event_name": "file.deleted",
+                "tenant_id": str(tenant_id),
+                "data": {},
+            })
+
+        sent.assert_not_awaited()
+
+    async def test_dispatch_skips_inactive_webhook(
+        self, db_session: AsyncSession, tenant_id: uuid.UUID, test_session_factory
+    ) -> None:
+        db_session.add(_make_webhook(tenant_id, ["file.deleted"], is_active=False))
+        await db_session.commit()
+
+        sent = AsyncMock(return_value={"success": True, "status_code": 200})
+        with patch("app.core.webhook_dispatcher.send_webhook", sent), \
+             patch("app.core.webhook_dispatcher.get_session_factory", return_value=test_session_factory):
+            await _dispatch_event({
+                "event_name": "file.deleted",
+                "tenant_id": str(tenant_id),
+                "data": {},
+            })
+
+        sent.assert_not_awaited()