From 07a99975ecf0f32b62d8a0f2ce1603a8b3b4ca5c Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Sun, 2 Aug 2026 23:25:54 +0200 Subject: [PATCH] Phase 5: Outbox DLQ, Monitoring, Consumer-Registry - Migration 0092: DLQ columns (error_message, failed_at) + consumer_inbox RLS fix - outbox.py: DLQ logic, replay functions, stats, consumer registry - app/routes/outbox.py: 5 API endpoints (stats, failed, replay, replay-all, consumer-registry) - outbox_deliveries tracking per consumer handler - 18/18 tests passing --- alembic/versions/0092_outbox_dlq.py | 77 +++++ app/core/outbox.py | 265 ++++++++++++++++- app/main.py | 2 + app/models/__init__.py | 1 + app/models/consumer_inbox.py | 2 +- app/models/outbox.py | 25 +- app/models/outbox_delivery.py | 57 ++++ app/routes/outbox.py | 130 +++++++++ test_report.md | 103 +++---- tests/conftest.py | 2 + tests/test_outbox.py | 10 +- tests/test_outbox_phase5.py | 422 ++++++++++++++++++++++++++++ 12 files changed, 1017 insertions(+), 79 deletions(-) create mode 100644 alembic/versions/0092_outbox_dlq.py create mode 100644 app/models/outbox_delivery.py create mode 100644 app/routes/outbox.py create mode 100644 tests/test_outbox_phase5.py diff --git a/alembic/versions/0092_outbox_dlq.py b/alembic/versions/0092_outbox_dlq.py new file mode 100644 index 0000000..bbc884b --- /dev/null +++ b/alembic/versions/0092_outbox_dlq.py @@ -0,0 +1,77 @@ +"""Add DLQ columns to event_outbox and fix consumer_inbox RLS policy. + +Phase 5: Dead-Letter-Queue support. +- Adds error_message TEXT and failed_at TIMESTAMPTZ to event_outbox +- Adds partial index for failed events +- Fixes consumer_inbox RLS policy (previous 0085 policy referenced + tenant_id column which does not exist on consumer_inbox; the + correct policy uses the event_id FK to event_outbox.tenant_id) + +Revision ID: 0092 +Revises: 0091 +""" + +from __future__ import annotations + +from alembic import op + +revision = "0092" +down_revision = "0091" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # 1. Add DLQ columns to event_outbox + op.execute( + "ALTER TABLE IF EXISTS event_outbox " + "ADD COLUMN IF NOT EXISTS error_message TEXT" + ) + op.execute( + "ALTER TABLE IF EXISTS event_outbox " + "ADD COLUMN IF NOT EXISTS failed_at TIMESTAMPTZ" + ) + + # 2. Partial index for efficient failed-event queries + op.execute( + "CREATE INDEX IF NOT EXISTS ix_outbox_failed " + "ON event_outbox (status, failed_at) WHERE status = 'failed'" + ) + + # 3. Fix consumer_inbox RLS policy + # Migration 0085 created a policy using tenant_id, but consumer_inbox + # has no tenant_id column. Drop the broken policy and create one + # that follows the same pattern as outbox_deliveries (0075): use the + # event_id FK to check event_outbox.tenant_id. + op.execute( + "DROP POLICY IF EXISTS consumer_inbox_tenant_isolation ON consumer_inbox" + ) + op.execute("DROP POLICY IF EXISTS tenant_isolation ON consumer_inbox") + op.execute("ALTER TABLE consumer_inbox ENABLE ROW LEVEL SECURITY") + op.execute("ALTER TABLE consumer_inbox FORCE ROW LEVEL SECURITY") + op.execute( + "CREATE POLICY consumer_inbox_tenant_isolation ON consumer_inbox " + "FOR ALL TO crm_api, crm_worker " + "USING (EXISTS (SELECT 1 FROM event_outbox " + "WHERE event_outbox.id = consumer_inbox.event_id " + "AND event_outbox.tenant_id = " + "NULLIF(current_setting('app.current_tenant_id', true), '')::uuid)) " + "WITH CHECK (EXISTS (SELECT 1 FROM event_outbox " + "WHERE event_outbox.id = consumer_inbox.event_id " + "AND event_outbox.tenant_id = " + "NULLIF(current_setting('app.current_tenant_id', true), '')::uuid))" + ) + # Ensure grants are in place + op.execute( + "GRANT SELECT, INSERT, UPDATE, DELETE ON consumer_inbox TO crm_api, crm_worker" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS ix_outbox_failed") + op.execute("ALTER TABLE event_outbox DROP COLUMN IF EXISTS failed_at") + op.execute("ALTER TABLE event_outbox DROP COLUMN IF EXISTS error_message") + # Restore the broken policy state (consumer_inbox RLS remains enabled) + op.execute( + "DROP POLICY IF EXISTS consumer_inbox_tenant_isolation ON consumer_inbox" + ) diff --git a/app/core/outbox.py b/app/core/outbox.py index 7cab41e..2aef2ce 100644 --- a/app/core/outbox.py +++ b/app/core/outbox.py @@ -15,6 +15,12 @@ Usage in services:: "tenant_id": str(tenant_id), }) # ... later, the transaction commits and the event is durable. + +Phase 5 additions: +- DLQ: ``error_message`` and ``failed_at`` columns on ``event_outbox`` +- Replay: ``replay_failed_event`` and ``replay_all_failed_events`` +- Monitoring: ``get_outbox_stats`` and ``get_failed_events`` +- Consumer registry: ``get_consumer_registry`` and ``outbox_deliveries`` """ from __future__ import annotations @@ -72,6 +78,8 @@ _FAIL_SQL = text( """ UPDATE event_outbox SET status = 'failed', + error_message = :error_message, + failed_at = now(), updated_at = now() WHERE id = :id """ @@ -88,6 +96,77 @@ _RETRY_SQL = text( """ ) +# ── Phase 5: outbox_deliveries SQL ────────────────────────────────────────── + +_INSERT_DELIVERY_SQL = text( + """ + INSERT INTO outbox_deliveries (event_id, consumer_name, status, attempt_count, last_error, processed_at) + VALUES (:event_id, :consumer_name, :status, :attempt_count, :last_error, :processed_at) + ON CONFLICT (event_id, consumer_name) DO UPDATE SET + status = EXCLUDED.status, + attempt_count = EXCLUDED.attempt_count, + last_error = EXCLUDED.last_error, + processed_at = EXCLUDED.processed_at, + updated_at = now() + """ +) + +# ── Phase 5: Replay SQL ───────────────────────────────────────────────────── + +_REPLAY_ONE_SQL = text( + """ + UPDATE event_outbox + SET status = 'pending', + attempts = 0, + error_message = NULL, + next_retry_at = NULL, + updated_at = now() + WHERE id = :event_id AND status = 'failed' + RETURNING id + """ +) + +_REPLAY_ALL_SQL = text( + """ + UPDATE event_outbox + SET status = 'pending', + attempts = 0, + error_message = NULL, + next_retry_at = NULL, + updated_at = now() + WHERE status = 'failed' AND tenant_id = :tenant_id + RETURNING id + """ +) + +# ── Phase 5: Stats SQL ────────────────────────────────────────────────────── + +_STATS_COUNT_SQL = text( + "SELECT status, COUNT(*) as count FROM event_outbox GROUP BY status" +) + +_STATS_OLDEST_PENDING_SQL = text( + """ + SELECT EXTRACT(EPOCH FROM (now() - created_at)) as age_seconds + FROM event_outbox + WHERE status = 'pending' + ORDER BY created_at ASC + LIMIT 1 + """ +) + +# ── Phase 5: Failed events SQL ────────────────────────────────────────────── + +_FAILED_EVENTS_SQL = text( + """ + SELECT id, tenant_id, event_name, error_message, failed_at, attempts, created_at + FROM event_outbox + WHERE status = 'failed' + ORDER BY failed_at DESC + LIMIT :limit OFFSET :offset + """ +) + def _json_payload(payload: dict[str, Any]) -> str: """Serialise payload to a JSON string suitable for JSONB cast.""" @@ -96,6 +175,17 @@ def _json_payload(payload: dict[str, Any]) -> str: return json.dumps(payload, default=str) +def _get_handler_name(handler: Any) -> str: + """Extract a human-readable name from a handler callable.""" + name = getattr(handler, "__name__", None) + if name: + return name + name = getattr(handler, "__qualname__", None) + if name: + return name + return str(handler) + + async def enqueue_outbox_event( db: AsyncSession, tenant_id: uuid.UUID, @@ -187,8 +277,43 @@ async def _process_single_outbox_event( logger.debug("Outbox event %s already processed, marking as published", event_id) return True + # Phase 5: Get handler names for consumer registry before publishing + handlers = list(event_bus._handlers.get(event_name, [])) + list(event_bus._handlers.get("*", [])) + handler_names = [_get_handler_name(h) for h in handlers] + results = await event_bus.publish_with_results(event_name, payload_dict) + # Phase 5: Write outbox_deliveries for each handler + current_attempt = attempts + 1 + for i, result in enumerate(results): + consumer_name = handler_names[i] if i < len(handler_names) else f"handler_{i}" + if result is None: + # Success + await db.execute( + _INSERT_DELIVERY_SQL, + { + "event_id": str(event_id), + "consumer_name": consumer_name, + "status": "delivered", + "attempt_count": current_attempt, + "last_error": None, + "processed_at": datetime.now(timezone.utc), + }, + ) + else: + # Failure + await db.execute( + _INSERT_DELIVERY_SQL, + { + "event_id": str(event_id), + "consumer_name": consumer_name, + "status": "failed", + "attempt_count": current_attempt, + "last_error": str(result), + "processed_at": None, + }, + ) + # Check if any handlers were registered at all handler_count = len(results) # If any handler raised, treat as failure @@ -219,10 +344,13 @@ async def _process_single_outbox_event( ) new_attempts = attempts + 1 if new_attempts >= max_attempts: - await db.execute(_FAIL_SQL, {"id": str(event_id)}) + await db.execute( + _FAIL_SQL, + {"id": str(event_id), "error_message": str(exc)}, + ) logger.warning( - "Outbox event %s marked as failed after %d attempts", - event_id, new_attempts, + "Outbox event %s marked as failed after %d attempts: %s", + event_id, new_attempts, exc, ) else: backoff = timedelta(seconds=(2 ** new_attempts) * 10) @@ -294,3 +422,134 @@ async def process_outbox_batch( await db.commit() return published_count + + +# ── Phase 5: Replay functions ──────────────────────────────────────────────── + +async def replay_failed_event(db: AsyncSession, event_id: uuid.UUID) -> bool: + """Replay a single failed outbox event. + + Resets the event to ``pending`` status with attempts=0, + error_message=NULL, next_retry_at=NULL. + + Args: + db: Active async SQLAlchemy session with tenant context set. + event_id: UUID of the event to replay. + + Returns: + True if the event was replayed, False if not found or not in 'failed' status. + """ + result = await db.execute( + _REPLAY_ONE_SQL, + {"event_id": str(event_id)}, + ) + row = result.first() + return row is not None + + +async def replay_all_failed_events(db: AsyncSession, tenant_id: uuid.UUID) -> int: + """Replay all failed outbox events for a tenant. + + Resets all failed events to ``pending`` status with attempts=0, + error_message=NULL, next_retry_at=NULL. + + Args: + db: Active async SQLAlchemy session with tenant context set. + tenant_id: Tenant scope for replay. + + Returns: + Number of events replayed. + """ + result = await db.execute( + _REPLAY_ALL_SQL, + {"tenant_id": str(tenant_id)}, + ) + return len(result.fetchall()) + + +# ── Phase 5: Monitoring functions ─────────────────────────────────────────── + +async def get_outbox_stats(db: AsyncSession) -> dict[str, Any]: + """Get outbox statistics for the current tenant. + + Requires tenant context to be set (RLS filters automatically). + + Returns: + Dict with: + - ``counts``: dict mapping each status to its count. + - ``total``: total number of events. + - ``oldest_pending_age_seconds``: age of oldest pending event in seconds, + or None if no pending events. + """ + result = await db.execute(_STATS_COUNT_SQL) + counts = {row[0]: row[1] for row in result.fetchall()} + + total = sum(counts.values()) + + oldest_result = await db.execute(_STATS_OLDEST_PENDING_SQL) + oldest_row = oldest_result.first() + oldest_pending_age_seconds = float(oldest_row[0]) if oldest_row else None + + return { + "counts": counts, + "total": total, + "oldest_pending_age_seconds": oldest_pending_age_seconds, + } + + +async def get_failed_events( + db: AsyncSession, + limit: int = 50, + offset: int = 0, +) -> list[dict[str, Any]]: + """Get failed outbox events for the current tenant. + + Requires tenant context to be set (RLS filters automatically). + + Args: + db: Active async SQLAlchemy session with tenant context set. + limit: Maximum number of events to return. + offset: Number of events to skip. + + Returns: + List of dicts with: id, tenant_id, event_name, error_message, + failed_at, attempts, created_at, status. + """ + result = await db.execute( + _FAILED_EVENTS_SQL, + {"limit": limit, "offset": offset}, + ) + return [ + { + "id": str(row[0]), + "tenant_id": str(row[1]), + "event_name": row[2], + "error_message": row[3], + "failed_at": row[4].isoformat() if row[4] else None, + "attempts": row[5], + "created_at": row[6].isoformat() if row[6] else None, + "status": "failed", + } + for row in result.fetchall() + ] + + +# ── Phase 5: Consumer registry ────────────────────────────────────────────── + +def get_consumer_registry() -> dict[str, list[str]]: + """Get the registered event handler registry from the in-process event bus. + + Returns a mapping of ``event_name`` to a list of consumer (handler) names. + This is read from ``event_bus._handlers`` at call time. + + Returns: + Dict mapping event names to lists of handler names. + """ + from app.core.event_bus import get_event_bus + + event_bus = get_event_bus() + registry: dict[str, list[str]] = {} + for event_name, handlers in event_bus._handlers.items(): + if handlers: + registry[event_name] = [_get_handler_name(h) for h in handlers] + return registry diff --git a/app/main.py b/app/main.py index 3e64345..43d1485 100644 --- a/app/main.py +++ b/app/main.py @@ -69,6 +69,7 @@ from app.routes import ( policies, guest_auth, guests, + outbox, ) @@ -443,6 +444,7 @@ def create_app() -> FastAPI: app.include_router(guest_auth.router) app.include_router(guests.router) app.include_router(workspaces.router) + app.include_router(outbox.router) # ── Register plugin routes for all built-in plugins ── # Routes are registered at app creation time so OpenAPI docs are complete. diff --git a/app/models/__init__.py b/app/models/__init__.py index b6adff6..3dc9888 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -13,6 +13,7 @@ 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.outbox_delivery import OutboxDelivery from app.models.guest_invitation import GuestInvitation from app.models.entity_policy import EntityPolicy from app.models.permission_template import PermissionTemplate diff --git a/app/models/consumer_inbox.py b/app/models/consumer_inbox.py index 93bd68c..54d1125 100644 --- a/app/models/consumer_inbox.py +++ b/app/models/consumer_inbox.py @@ -25,7 +25,7 @@ class ConsumerInbox(Base): ) id: Mapped[uuid.UUID] = mapped_column( - PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + PGUUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid(), ) event_id: Mapped[uuid.UUID] = mapped_column( PGUUID(as_uuid=True), diff --git a/app/models/outbox.py b/app/models/outbox.py index c1faf35..dd7fe35 100644 --- a/app/models/outbox.py +++ b/app/models/outbox.py @@ -5,7 +5,7 @@ from __future__ import annotations import uuid from datetime import datetime -from sqlalchemy import DateTime, Integer, String, func +from sqlalchemy import DateTime, Integer, String, Text, func from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import UUID as PGUUID from sqlalchemy.orm import Mapped, mapped_column @@ -54,3 +54,26 @@ class EventOutbox(Base): published_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) + # Envelope columns (migration 0075) + aggregate_type: Mapped[str | None] = mapped_column( + String(100), nullable=True, + ) + aggregate_id: Mapped[uuid.UUID | None] = mapped_column( + PGUUID(as_uuid=True), nullable=True, + ) + occurred_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), + ) + correlation_id: Mapped[uuid.UUID | None] = mapped_column( + PGUUID(as_uuid=True), nullable=True, + ) + schema_version: Mapped[int] = mapped_column( + Integer, nullable=False, server_default="1", + ) + # Phase 5: DLQ columns + error_message: Mapped[str | None] = mapped_column( + Text, nullable=True, + ) + failed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, + ) diff --git a/app/models/outbox_delivery.py b/app/models/outbox_delivery.py new file mode 100644 index 0000000..24a9de7 --- /dev/null +++ b/app/models/outbox_delivery.py @@ -0,0 +1,57 @@ +"""Outbox delivery model for per-consumer delivery tracking.""" + +from __future__ import annotations + +import uuid +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Integer, 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 OutboxDelivery(Base): + """Tracks per-consumer delivery status for outbox events. + + Each row represents one consumer (event handler) processing one outbox + event. An event is only fully 'published' when all mandatory deliveries + succeed. + """ + + __tablename__ = "outbox_deliveries" + __table_args__ = ( + UniqueConstraint("event_id", "consumer_name", name="uq_outbox_deliveries_event_consumer"), + ) + + id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), + primary_key=True, + server_default=func.gen_random_uuid(), + ) + event_id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), + ForeignKey("event_outbox.id", ondelete="CASCADE"), + nullable=False, + ) + consumer_name: Mapped[str] = mapped_column(String(150), nullable=False) + status: Mapped[str] = mapped_column( + String(30), nullable=False, server_default="pending", + ) + attempt_count: Mapped[int] = mapped_column( + Integer, nullable=False, server_default="0", + ) + next_attempt_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, + ) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + processed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), + ) diff --git a/app/routes/outbox.py b/app/routes/outbox.py new file mode 100644 index 0000000..35405cc --- /dev/null +++ b/app/routes/outbox.py @@ -0,0 +1,130 @@ +"""Outbox monitoring and management endpoints — DLQ, stats, replay, consumer registry. + +All endpoints require authentication and admin role. +Tenant context is set automatically via the ``get_current_user`` dependency +(used internally by ``require_admin``), which calls ``set_tenant_context`` +on the shared database session. RLS policies then filter all outbox queries +automatically. +""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.db import get_db +from app.core.outbox import ( + get_consumer_registry, + get_failed_events, + get_outbox_stats, + replay_all_failed_events, + replay_failed_event, +) +from app.deps import require_admin + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/v1/outbox", tags=["outbox"]) + + +@router.get("/stats") +async def outbox_stats( + db: AsyncSession = Depends(get_db), + current_user: dict[str, Any] = Depends(require_admin), +): + """Outbox statistics: counts per status, oldest pending age, total events. + + Returns a JSON object with: + - ``counts``: dict mapping each status to its count (pending, processing, + published, failed, no_handlers). + - ``total``: total number of events for the current tenant. + - ``oldest_pending_age_seconds``: age in seconds of the oldest pending + event, or ``null`` if none pending. + """ + return await get_outbox_stats(db) + + +@router.get("/failed") +async def outbox_failed( + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), + db: AsyncSession = Depends(get_db), + current_user: dict[str, Any] = Depends(require_admin), +): + """List failed outbox events with error details (paginated). + + Each event includes: id, tenant_id, event_name, error_message, + failed_at, attempts, created_at, status. + """ + events = await get_failed_events(db, limit=limit, offset=offset) + return { + "events": events, + "limit": limit, + "offset": offset, + "count": len(events), + } + + +@router.post("/replay/{event_id}") +async def outbox_replay_single( + event_id: str, + db: AsyncSession = Depends(get_db), + current_user: dict[str, Any] = Depends(require_admin), +): + """Replay a single failed outbox event. + + Resets the event to ``pending`` status with attempts=0, + error_message=NULL, next_retry_at=NULL. + """ + try: + eid = uuid.UUID(event_id) + except ValueError: + raise HTTPException( + status_code=400, + detail={ + "detail": "Invalid event ID format", + "code": "invalid_uuid", + }, + ) + + replayed = await replay_failed_event(db, eid) + if not replayed: + raise HTTPException( + status_code=404, + detail={ + "detail": "Failed event not found or not in 'failed' status", + "code": "not_found", + }, + ) + return {"replayed": True, "event_id": event_id} + + +@router.post("/replay-all") +async def outbox_replay_all( + db: AsyncSession = Depends(get_db), + current_user: dict[str, Any] = Depends(require_admin), +): + """Replay all failed outbox events for the current tenant. + + Resets all failed events to ``pending`` status. + Returns the number of events replayed. + """ + tenant_id = uuid.UUID(current_user["tenant_id"]) + count = await replay_all_failed_events(db, tenant_id) + return {"replayed_count": count, "tenant_id": str(tenant_id)} + + +@router.get("/consumer-registry") +async def outbox_consumer_registry( + current_user: dict[str, Any] = Depends(require_admin), +): + """List registered event handlers from the in-process event bus. + + Returns a mapping of ``event_name`` to a list of consumer (handler) names. + This is read from ``event_bus._handlers`` at request time. + """ + return {"registry": get_consumer_registry()} diff --git a/test_report.md b/test_report.md index c924ab9..8f70468 100644 --- a/test_report.md +++ b/test_report.md @@ -1,82 +1,47 @@ -# Test Report — LeoCRM Fix Branch +# Test Report — Phase 5: Outbox DLQ, Monitoring, Consumer-Registry -**Date**: 2026-07-27 -**Branch**: main (leocrm-fix) +## Date: 2026-08-02 -## Test Results - -### Backend: AI Copilot Tests (tests/test_ai_copilot.py) +## Test Execution ``` -76 passed, 2 warnings in 63.31s +cd /a0/usr/workdir/leocrm-fix && python -m pytest tests/test_outbox.py tests/test_outbox_phase5.py -v ``` -**AC Tests (all pass):** -- AC1: test_ac1_copilot_query_returns_proposed_actions ✅ -- AC2: test_ac2_copilot_execute_action_success ✅ -- AC3: test_ac3_copilot_execute_blocked_by_rbac ✅ -- AC4: test_ac4_copilot_history_paginated ✅ -- AC5: test_ac5_copilot_action_logged_in_audit ✅ -- AC6: test_ac6_copilot_tenant_isolation ✅ -- AC7: test_ac7_copilot_field_level_permissions ✅ +## Results: 18 passed, 0 failed -**Other tests fixed:** -- test_copilot_unauthenticated: Fixed 401→403 for POST (CSRF middleware returns 403) -- test_route_copilot_history_unauthenticated: GET returns 401 (no CSRF needed) -- test_route_copilot_execute_unauthenticated: Fixed 401→403 for POST -- action_mapper tests: Updated /api/v1/companies → /api/v1/contacts (unified contact model) -- llm_client tests: Fixed ai_client → client variable, api_base default '' -- service tests: Updated /api/v1/companies → /api/v1/contacts, PATCH/DELETE return 400 (unsupported) +### Existing Tests (test_outbox.py) — 6/6 passed +- test_enqueue_outbox_event_inserts_pending_row ✅ +- test_process_outbox_batch_publishes_events ✅ +- test_process_outbox_batch_empty_returns_zero ✅ +- test_process_outbox_batch_retry_on_failure ✅ +- test_process_outbox_batch_max_attempts_marks_failed ✅ +- test_enqueue_multiple_events_and_batch_size ✅ -### Frontend: TypeScript Type Check +### Phase 5 Tests (test_outbox_phase5.py) — 12/12 passed +- test_failed_event_has_error_message ✅ (DLQ: error_message + failed_at set) +- test_replay_failed_event ✅ (single replay: failed→pending) +- test_replay_failed_event_not_found ✅ (404 case) +- test_replay_all_failed_events ✅ (bulk replay: 3 events reset) +- test_get_outbox_stats ✅ (counts per status, total, oldest pending age) +- test_get_outbox_stats_empty ✅ (empty tenant returns zeros) +- test_get_failed_events ✅ (failed events with error details) +- test_get_failed_events_pagination ✅ (limit/offset pagination) +- test_get_consumer_registry ✅ (event_name→handler_names mapping) +- test_outbox_deliveries_written_on_success ✅ (status='delivered') +- test_outbox_deliveries_written_on_failure ✅ (status='failed', last_error set) +- test_route_import ✅ (all 5 endpoints registered) +## Syntax Check ``` -cd frontend && npx tsc --noEmit -# Exit code 0 — no errors +python -c 'import app.core.outbox; import app.routes.outbox; import app.models.outbox; import app.models.consumer_inbox; import app.models.outbox_delivery' +→ All imports OK ``` -### Event Loop Fix - -Added `asyncio_default_fixture_loop_scope = "session"` and `asyncio_default_test_loop_scope = "session"` to pyproject.toml to fix 'Event loop is closed' error when running multiple AI copilot tests in sequence. - -## Changes Summary - -### 1. Backend Security Fixes -- **RCE Dead Code repariert** (`app/routes/plugins.py`): Security-Check (`_check_dangerous_imports`) wurde VOR `exec_module()` verschoben. Zuvor war exec_module vor dem Security-Check, was eine RCE-Lücke war (auch wenn alle Upload-Endpoints deaktiviert waren). -- **verify_ws_origin verschärft** (`app/core/auth.py`): Leerer Origin-Header wird jetzt abgelehnt (return False) wenn CORS konfiguriert ist, statt automatisch akzeptiert zu werden. - -### 2. Test Infrastructure Fixes (conftest.py) -- Neuer `ai_app` und `ai_client` Fixture mit `init_permission_registry(active_plugin_names={'ai_assistant'})` -- `login_client` setzt jetzt CSRF-Token und Origin als Client-Default-Header -- `SESSION_COOKIE_SECURE=false` und `SESSION_COOKIE_SAMESITE=lax` werden vor allen Imports gesetzt -- `get_settings.cache_clear()` nach env-Override -- `pyproject.toml`: `asyncio_default_fixture_loop_scope = "session"` und `asyncio_default_test_loop_scope = "session"` hinzugefügt -- `tests/test_ai_copilot.py`: `/api/v1/companies` → `/api/v1/contacts` (Companies sind Contacts mit type='company'). 15 weitere Test-Fixes (action_mapper paths, llm_client variables, service test paths, unauthenticated test assertions). - -### 3. Event Bus Lücken geschlossen -- system_notif/plugin.py: Added conversation.created, participant.joined, participant.left, reaction.added to manifest events list -- Added handler methods: on_conversation_created, on_participant_joined, on_participant_left, on_reaction_added -- Added event titles for new events in _create_system_notification - -### 4. Frontend Integration: SavedFilterBar -- ContactsList.tsx: Added SavedFilterBar with entityType="contacts" in middle pane -- Mail.tsx: Added SavedFilterBar with entityType="mail" in mail list pane -- Calendar.tsx: Added SavedFilterBar with entityType="calendar" in calendar view pane - -### 5. Frontend Integration: TagSelector -- ContactsList.tsx: Added TagSelector with entityType="contact" in middle pane -- Mail.tsx: Added TagSelector with entityType="file" in mail list pane -- Calendar.tsx: Added TagSelector with entityType="calendar_entry" in calendar view pane - -### 6. Event Loop Fix -- pyproject.toml: Added asyncio_default_fixture_loop_scope and asyncio_default_test_loop_scope = "session" -- Fixed 15 pre-existing test failures (action_mapper, llm_client, service tests) caused by unified contact model migration - ## Smoke Test -- Backend: All 76 AI copilot tests pass including AC1-AC7 -- Frontend: TypeScript compilation passes with 0 errors -- Event bus: system_notif plugin now subscribes to conversation.created, participant.joined/left, reaction.added -- RCE Dead Code: Security-Check (_check_dangerous_imports) wird VOR exec_module() ausgeführt -- verify_ws_origin: Leerer Origin-Header wird abgelehnt bei konfiguriertem CORS -- conftest.py: ai_app/ai_client Fixtures mit ai_assistant Plugin-Aktivierung, CSRF-Token, Origin-Header -- Frontend-Integration: SavedFilterBar und TagSelector in ContactsList, Mail, Calendar integriert +- All 5 API endpoints registered under `/api/v1/outbox/` +- DLQ columns (error_message, failed_at) functional in event_outbox +- Replay functions reset failed events to pending correctly +- outbox_deliveries entries written per-consumer during processing +- Consumer registry reads from event_bus._handlers at runtime +- RLS: tenant context required for all monitoring queries diff --git a/tests/conftest.py b/tests/conftest.py index 42fea15..8a5c0ba 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -91,6 +91,8 @@ from app.plugins.builtins.tags.models import Tag, TagAssignment # noqa: F401 from app.plugins.builtins.tasks import TasksPlugin # noqa: F401 from app.plugins.builtins.tasks.models import Task # noqa: F401 from app.models.outbox import EventOutbox # noqa: F401 +from app.models.consumer_inbox import ConsumerInbox # noqa: F401 +from app.models.outbox_delivery import OutboxDelivery # noqa: F401 from app.models.saved_filter import SavedFilter # noqa: F401 from app.plugins.registry import reset_registry_for_testing # noqa: F401 from app.services.plugin_service import reset_plugin_service_for_testing # noqa: F401 diff --git a/tests/test_outbox.py b/tests/test_outbox.py index 6df40eb..61717b1 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -67,7 +67,7 @@ async def test_process_outbox_batch_publishes_events( # Use a separate session to simulate the worker async with session_factory() as worker_session: - count = await process_outbox_batch(worker_session, batch_size=10) + count = await process_outbox_batch(worker_session, batch_size=10, tenant_ids=[tenant_id]) assert count == 1 assert len(received_events) == 1 @@ -116,7 +116,7 @@ async def test_process_outbox_batch_retry_on_failure( await db_session.commit() async with session_factory() as worker_session: - count = await process_outbox_batch(worker_session, batch_size=10) + count = await process_outbox_batch(worker_session, batch_size=10, tenant_ids=[tenant_id]) assert count == 0 # nothing was successfully published @@ -162,7 +162,7 @@ async def test_process_outbox_batch_max_attempts_marks_failed( await db_session.commit() async with session_factory() as worker_session: - count = await process_outbox_batch(worker_session, batch_size=10) + count = await process_outbox_batch(worker_session, batch_size=10, tenant_ids=[tenant_id]) assert count == 0 @@ -200,14 +200,14 @@ async def test_enqueue_multiple_events_and_batch_size( await db_session.commit() async with session_factory() as worker_session: - count = await process_outbox_batch(worker_session, batch_size=3) + count = await process_outbox_batch(worker_session, batch_size=3, tenant_ids=[tenant_id]) assert count == 3 assert len(received) == 3 # Process the remaining 2 async with session_factory() as worker_session: - count2 = await process_outbox_batch(worker_session, batch_size=3) + count2 = await process_outbox_batch(worker_session, batch_size=3, tenant_ids=[tenant_id]) assert count2 == 2 assert len(received) == 5 finally: diff --git a/tests/test_outbox_phase5.py b/tests/test_outbox_phase5.py new file mode 100644 index 0000000..3cb9052 --- /dev/null +++ b/tests/test_outbox_phase5.py @@ -0,0 +1,422 @@ +"""Tests for Phase 5: Outbox DLQ, Monitoring, Consumer-Registry. + +Covers: +- DLQ: failed events have error_message and failed_at set +- Replay: single and bulk replay resets failed events to pending +- Stats: get_outbox_stats returns correct counts per status +- Failed events listing: get_failed_events returns error details +- Consumer registry: get_consumer_registry returns event→handlers mapping +- outbox_deliveries: entries are written during event processing +""" + +from __future__ import annotations + +import uuid + +import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from app.core.db import set_tenant_context +from app.core.event_bus import get_event_bus +from app.core.outbox import ( + enqueue_outbox_event, + get_consumer_registry, + get_failed_events, + get_outbox_stats, + process_outbox_batch, + replay_all_failed_events, + replay_failed_event, +) + + +# ── DLQ: error_message and failed_at ────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_failed_event_has_error_message( + db_session: AsyncSession, + session_factory: async_sessionmaker[AsyncSession], +): + """When a handler fails permanently, error_message and failed_at are set.""" + tenant_id = uuid.uuid4() + + async def _always_fails(payload: dict) -> None: + raise RuntimeError("Permanent failure for DLQ test") + + bus = get_event_bus() + bus.subscribe("test.dlq.fail", _always_fails) + try: + # Insert event with attempts = max_attempts - 1 so next failure marks it failed + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, attempts, max_attempts) " + "VALUES (:tid, 'test.dlq.fail', CAST(:payload AS JSONB), 'pending', 4, 5)" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + await db_session.flush() + await db_session.commit() + + async with session_factory() as worker_session: + count = await process_outbox_batch( + worker_session, batch_size=10, tenant_ids=[tenant_id] + ) + + assert count == 0 # nothing was successfully published + + await set_tenant_context(db_session, tenant_id) + rows = ( + await db_session.execute( + text( + "SELECT status, error_message, failed_at " + "FROM event_outbox WHERE tenant_id = :tid" + ), + {"tid": str(tenant_id)}, + ) + ).fetchall() + assert rows[0][0] == "failed" + assert rows[0][1] is not None + assert "Permanent failure" in rows[0][1] + assert rows[0][2] is not None # failed_at + finally: + bus.unsubscribe("test.dlq.fail", _always_fails) + + +# ── Replay single failed event ──────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_replay_failed_event(db_session: AsyncSession): + """replay_failed_event resets a failed event to pending.""" + tenant_id = uuid.uuid4() + + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, attempts, max_attempts, error_message, failed_at) " + "VALUES (:tid, 'test.replay', CAST(:payload AS JSONB), 'failed', 5, 5, 'some error', now())" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + await db_session.flush() + await db_session.commit() + + row = ( + await db_session.execute( + text("SELECT id FROM event_outbox WHERE tenant_id = :tid"), + {"tid": str(tenant_id)}, + ) + ).first() + event_id = row[0] + + await set_tenant_context(db_session, tenant_id) + replayed = await replay_failed_event(db_session, event_id) + await db_session.commit() + + assert replayed is True + + rows = ( + await db_session.execute( + text( + "SELECT status, attempts, error_message, next_retry_at " + "FROM event_outbox WHERE id = :eid" + ), + {"eid": str(event_id)}, + ) + ).fetchall() + assert rows[0][0] == "pending" + assert rows[0][1] == 0 + assert rows[0][2] is None + assert rows[0][3] is None + + +@pytest.mark.asyncio +async def test_replay_failed_event_not_found(db_session: AsyncSession): + """replay_failed_event returns False for non-existent or non-failed events.""" + tenant_id = uuid.uuid4() + fake_id = uuid.uuid4() + + await set_tenant_context(db_session, tenant_id) + replayed = await replay_failed_event(db_session, fake_id) + await db_session.commit() + + assert replayed is False + + +# ── Replay all failed events for a tenant ───────────────────────────────────── + +@pytest.mark.asyncio +async def test_replay_all_failed_events(db_session: AsyncSession): + """replay_all_failed_events resets all failed events for a tenant.""" + tenant_id = uuid.uuid4() + + for i in range(3): + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, attempts, max_attempts, error_message, failed_at) " + "VALUES (:tid, :event_name, CAST(:payload AS JSONB), 'failed', 5, 5, :error, now())" + ), + { + "tid": str(tenant_id), + "event_name": f"test.replay_all.{i}", + "payload": '{"k": "v"}', + "error": f"error {i}", + }, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + count = await replay_all_failed_events(db_session, tenant_id) + await db_session.commit() + + assert count == 3 + + rows = ( + await db_session.execute( + text( + "SELECT status, attempts, error_message " + "FROM event_outbox WHERE tenant_id = :tid" + ), + {"tid": str(tenant_id)}, + ) + ).fetchall() + for row in rows: + assert row[0] == "pending" + assert row[1] == 0 + assert row[2] is None + + +# ── Outbox stats ─────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_get_outbox_stats(db_session: AsyncSession): + """get_outbox_stats returns correct counts per status.""" + tenant_id = uuid.uuid4() + + for status in ["pending", "pending", "published", "failed"]: + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status) " + "VALUES (:tid, 'test.stats', CAST(:payload AS JSONB), :status)" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}', "status": status}, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + stats = await get_outbox_stats(db_session) + + assert stats["counts"]["pending"] == 2 + assert stats["counts"]["published"] == 1 + assert stats["counts"]["failed"] == 1 + assert stats["total"] == 4 + assert stats["oldest_pending_age_seconds"] is not None + + +@pytest.mark.asyncio +async def test_get_outbox_stats_empty(db_session: AsyncSession): + """get_outbox_stats returns zeros when no events exist.""" + tenant_id = uuid.uuid4() + await set_tenant_context(db_session, tenant_id) + + stats = await get_outbox_stats(db_session) + + assert stats["total"] == 0 + assert stats["oldest_pending_age_seconds"] is None + + +# ── Failed events listing ───────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_get_failed_events(db_session: AsyncSession): + """get_failed_events returns failed events with error details.""" + tenant_id = uuid.uuid4() + + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, attempts, max_attempts, error_message, failed_at) " + "VALUES (:tid, 'test.failed_list', CAST(:payload AS JSONB), 'failed', 5, 5, 'test error', now())" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + events = await get_failed_events(db_session, limit=10, offset=0) + + assert len(events) == 1 + assert events[0]["event_name"] == "test.failed_list" + assert events[0]["error_message"] == "test error" + assert events[0]["status"] == "failed" + assert events[0]["attempts"] == 5 + + +@pytest.mark.asyncio +async def test_get_failed_events_pagination(db_session: AsyncSession): + """get_failed_events respects limit and offset.""" + tenant_id = uuid.uuid4() + + for i in range(5): + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, attempts, max_attempts, error_message, failed_at) " + "VALUES (:tid, :name, CAST(:payload AS JSONB), 'failed', 5, 5, :error, now())" + ), + { + "tid": str(tenant_id), + "name": f"test.pag.{i}", + "payload": '{"k": "v"}', + "error": f"err {i}", + }, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + + page1 = await get_failed_events(db_session, limit=2, offset=0) + page2 = await get_failed_events(db_session, limit=2, offset=2) + + assert len(page1) == 2 + assert len(page2) == 2 + + +# ── Consumer registry ────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_get_consumer_registry(): + """get_consumer_registry returns event_name → consumer_names mapping.""" + bus = get_event_bus() + + async def handler_alpha(payload: dict) -> None: + pass + + async def handler_beta(payload: dict) -> None: + pass + + bus.subscribe("test.registry.event", handler_alpha) + bus.subscribe("test.registry.event", handler_beta) + try: + registry = get_consumer_registry() + assert "test.registry.event" in registry + assert len(registry["test.registry.event"]) == 2 + assert "handler_alpha" in registry["test.registry.event"] + assert "handler_beta" in registry["test.registry.event"] + finally: + bus.unsubscribe("test.registry.event", handler_alpha) + bus.unsubscribe("test.registry.event", handler_beta) + + +# ── outbox_deliveries written during processing ─────────────────────────────── + +@pytest.mark.asyncio +async def test_outbox_deliveries_written_on_success( + db_session: AsyncSession, + session_factory: async_sessionmaker[AsyncSession], +): + """outbox_deliveries entries are written with status='delivered' on success.""" + tenant_id = uuid.uuid4() + + async def _success_handler(payload: dict) -> None: + pass + + bus = get_event_bus() + bus.subscribe("test.deliveries.ok", _success_handler) + try: + await enqueue_outbox_event( + db_session, tenant_id, "test.deliveries.ok", {"key": "value"} + ) + await db_session.flush() + await db_session.commit() + + async with session_factory() as worker_session: + count = await process_outbox_batch( + worker_session, batch_size=10, tenant_ids=[tenant_id] + ) + + assert count == 1 + + await set_tenant_context(db_session, tenant_id) + rows = ( + await db_session.execute( + text( + "SELECT d.consumer_name, d.status, d.last_error " + "FROM outbox_deliveries d " + "JOIN event_outbox e ON d.event_id = e.id " + "WHERE e.tenant_id = :tid" + ), + {"tid": str(tenant_id)}, + ) + ).fetchall() + assert len(rows) == 1 + assert rows[0][0] == "_success_handler" + assert rows[0][1] == "delivered" + assert rows[0][2] is None # no error + finally: + bus.unsubscribe("test.deliveries.ok", _success_handler) + + +@pytest.mark.asyncio +async def test_outbox_deliveries_written_on_failure( + db_session: AsyncSession, + session_factory: async_sessionmaker[AsyncSession], +): + """outbox_deliveries entries are written with status='failed' on handler error.""" + tenant_id = uuid.uuid4() + + async def _failing_handler(payload: dict) -> None: + raise RuntimeError("Delivery failure") + + bus = get_event_bus() + bus.subscribe("test.deliveries.fail", _failing_handler) + try: + await enqueue_outbox_event( + db_session, tenant_id, "test.deliveries.fail", {"key": "value"} + ) + await db_session.flush() + await db_session.commit() + + async with session_factory() as worker_session: + count = await process_outbox_batch( + worker_session, batch_size=10, tenant_ids=[tenant_id] + ) + + assert count == 0 # not published + + await set_tenant_context(db_session, tenant_id) + rows = ( + await db_session.execute( + text( + "SELECT d.consumer_name, d.status, d.last_error " + "FROM outbox_deliveries d " + "JOIN event_outbox e ON d.event_id = e.id " + "WHERE e.tenant_id = :tid" + ), + {"tid": str(tenant_id)}, + ) + ).fetchall() + assert len(rows) == 1 + assert rows[0][0] == "_failing_handler" + assert rows[0][1] == "failed" + assert rows[0][2] is not None + assert "Delivery failure" in rows[0][2] + finally: + bus.unsubscribe("test.deliveries.fail", _failing_handler) + + +# ── Route import smoke test ──────────────────────────────────────────────────── + +def test_route_import(): + """Verify that the outbox route module imports cleanly.""" + from app.routes.outbox import router + + assert router is not None + assert router.prefix == "/api/v1/outbox" + # Check all 5 endpoints are registered + paths = {route.path for route in router.routes} + assert "/api/v1/outbox/stats" in paths + assert "/api/v1/outbox/failed" in paths + assert "/api/v1/outbox/replay/{event_id}" in paths + assert "/api/v1/outbox/replay-all" in paths + assert "/api/v1/outbox/consumer-registry" in paths