Merge remote-tracking branch 'gitea/main' into codex/iwooos-wazuh-boundary-guard-20260624

apps/api/src/services/channel_event_dossier_service.py

@@ -2052,7 +2052,25 @@ async def _fetch_source_review_decisions_by_work_item(
 
     result = await db.execute(
         text("""
-            SELECT DISTINCT ON (context->>'work_item_id')
+            WITH ranked AS (
+                SELECT
+                    id,
+                    incident_id,
+                    actor,
+                    action_detail,
+                    success,
+                    context,
+                    created_at,
+                    ROW_NUMBER() OVER (
+                        PARTITION BY context->>'work_item_id'
+                        ORDER BY created_at DESC
+                    ) AS rn
+                FROM alert_operation_log
+                WHERE actor = :actor
+                  AND context->>'schema_version' = :schema_version
+                  AND COALESCE(context->>'project_id', :project_id) = :project_id
+            )
+            SELECT
                 id,
                 incident_id,
                 actor,
@@ -2060,11 +2078,9 @@ async def _fetch_source_review_decisions_by_work_item(
                 success,
                 context,
                 created_at
-            FROM alert_operation_log
-            WHERE actor = :actor
-              AND context->>'schema_version' = :schema_version
-              AND COALESCE(context->>'project_id', :project_id) = :project_id
-            ORDER BY context->>'work_item_id', created_at DESC
+            FROM ranked
+            WHERE rn = 1
+            ORDER BY created_at DESC
             LIMIT :limit
         """),
         {
@@ -2094,7 +2110,25 @@ async def _fetch_source_applies_by_work_item(
 
     result = await db.execute(
         text("""
-            SELECT DISTINCT ON (context->>'work_item_id')
+            WITH ranked AS (
+                SELECT
+                    id,
+                    incident_id,
+                    actor,
+                    action_detail,
+                    success,
+                    context,
+                    created_at,
+                    ROW_NUMBER() OVER (
+                        PARTITION BY context->>'work_item_id'
+                        ORDER BY created_at DESC
+                    ) AS rn
+                FROM alert_operation_log
+                WHERE actor = :actor
+                  AND context->>'schema_version' = :schema_version
+                  AND COALESCE(context->>'project_id', :project_id) = :project_id
+            )
+            SELECT
                 id,
                 incident_id,
                 actor,
@@ -2102,11 +2136,9 @@ async def _fetch_source_applies_by_work_item(
                 success,
                 context,
                 created_at
-            FROM alert_operation_log
-            WHERE actor = :actor
-              AND context->>'schema_version' = :schema_version
-              AND COALESCE(context->>'project_id', :project_id) = :project_id
-            ORDER BY context->>'work_item_id', created_at DESC
+            FROM ranked
+            WHERE rn = 1
+            ORDER BY created_at DESC
             LIMIT :limit
         """),
         {

apps/api/tests/test_channel_event_dossier_service.py

@@ -23,6 +23,49 @@ from src.services.channel_event_dossier_service import (
 )
 
 
+class _FakeMappings:
+    def all(self) -> list[dict]:
+        return []
+
+
+class _FakeResult:
+    def mappings(self) -> _FakeMappings:
+        return _FakeMappings()
+
+
+class _CaptureSqlDb:
+    def __init__(self) -> None:
+        self.statements: list[str] = []
+
+    async def execute(self, statement, params):  # noqa: ANN001, ANN201
+        self.statements.append(str(statement))
+        return _FakeResult()
+
+
+@pytest.mark.asyncio
+async def test_source_correlation_read_model_ranks_latest_rows_before_limit() -> None:
+    db = _CaptureSqlDb()
+
+    await channel_event_dossier_service._fetch_source_review_decisions_by_work_item(
+        db,
+        project_id="awoooi",
+        limit=300,
+    )
+    await channel_event_dossier_service._fetch_source_applies_by_work_item(
+        db,
+        project_id="awoooi",
+        limit=300,
+    )
+
+    assert len(db.statements) == 2
+    for sql in db.statements:
+        assert "ROW_NUMBER() OVER" in sql
+        assert "PARTITION BY context->>'work_item_id'" in sql
+        assert "WHERE rn = 1" in sql
+        assert "ORDER BY created_at DESC" in sql
+        assert "SELECT DISTINCT ON" not in sql
+
+
 def test_build_dossier_event_summarizes_source_envelope() -> None:
     event = build_dossier_event(
         {