diff --git a/app/core/webhook_dispatcher.py b/app/core/webhook_dispatcher.py index c8fced2..b9432db 100644 --- a/app/core/webhook_dispatcher.py +++ b/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 @> '[""]' — 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()) diff --git a/app/services/webhook_service.py b/app/services/webhook_service.py index 054a280..daab7f6 100644 --- a/app/services/webhook_service.py +++ b/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 @> '[""]' — 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 diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py new file mode 100644 index 0000000..6e54181 --- /dev/null +++ b/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 @> '[""]' 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()