feat: save audit logs to postgres (M2-10911)

conftest.py

@@ -348,6 +348,20 @@ async def mock_reencrypt_kiq(mocker) -> AsyncGenerator[Any, Any]:
     yield mock
 
 
+@pytest.fixture(autouse=True)
+def mock_audit_event_kiq(mocker: MockerFixture):
+    """Prevent audit events from being persisted during tests.
+
+    ``audit.log()`` enqueues ``send_audit_event``, which the in-memory test
+    broker runs synchronously. That task opens its own DB session and commits
+    to ``audit_logs`` outside the test's transaction, leaking rows across tests.
+    Tests that assert audit behaviour mock ``log`` directly, and the audit
+    query/export tests seed ``audit_logs`` explicitly, so disabling the enqueue
+    here is safe.
+    """
+    return mocker.patch("apps.audit.service.send_audit_event.kiq")
+
+
 @pytest.fixture(scope="session")
 def uuid_zero() -> uuid.UUID:
     return uuid.UUID("00000000-0000-0000-0000-000000000000")

docker-compose.yaml

@@ -70,9 +70,6 @@ services:
       - mailhog
     ports:
       - "8000:80"
-    networks:
-      - opensearch-net
-      - default
     develop:
       watch:
         - action: sync
@@ -186,43 +183,7 @@ services:
       - './compose/minio:/etc/minio'
     entrypoint: /etc/minio/create_bucket.sh
 
-  opensearch:
-    image: opensearchproject/opensearch:latest
-    container_name: opensearch-node
-    environment:
-      discovery.type: single-node
-      OPENSEARCH_INITIAL_ADMIN_PASSWORD: "${OPENSEARCH__PASSWORD:-MyStrongPassword123!}"
-      node.name: opensearch
-    ports:
-      - "9200:9200"
-      - "9600:9600"
-    volumes:
-      - opensearch-data:/usr/share/opensearch/data
-    networks:
-      - opensearch-net
-
-  opensearch-dashboards:
-    image: opensearchproject/opensearch-dashboards:latest
-    container_name: opensearch-dashboards
-    environment:
-      OPENSEARCH_HOSTS: '["https://opensearch:9200"]'
-      OPENSEARCH_USERNAME: admin
-      OPENSEARCH_PASSWORD: "${OPENSEARCH__PASSWORD:-MyStrongPassword123!}"
-      OPENSEARCH_SSL_VERIFICATIONMODE: none
-      explore.enabled: true
-    depends_on:
-      - opensearch
-    ports:
-      - "5601:5601"
-    networks:
-      - opensearch-net
-
-
-networks:
-  opensearch-net:
-
 volumes:
   pg_data: {}
   pg_arb_data: {}
   datastore: {}
-  opensearch-data: {}

docs/audit-pipeline.md

@@ -33,14 +33,16 @@ API process                                 Worker process
       │
       └──────── RabbitMQ queue ──────────►  3. send_audit_event(payload)
                                                   │
-                                                  │  OpenSearchClient().index_document()
-                                                  │
+                                                  │  AuditLogCRUD(session).save(...)
+                                                  │  INSERT ... ON CONFLICT DO NOTHING
                                                   ▼
-                                            POST /audit-logs/_doc
-                                            → document stored in OpenSearch
+                                            row stored in the `audit_logs`
+                                            table (primary Postgres database)
 ```
 
-**The API never waits for OpenSearch.** As soon as the message is on the queue (step 2), the HTTP response can return. The write to OpenSearch happens in a separate worker process.
+**The API never waits for the audit write.** As soon as the message is on the queue (step 2), the HTTP response can return. The write to Postgres happens in a separate worker process.
+
+Audit events are stored in the primary Postgres database. A few fields are denormalized into typed columns for filtering/sorting; the full ECS document is kept in a JSONB `payload` column and is what the export endpoint returns.
 
 ---
 
@@ -51,11 +53,10 @@ API process                                 Worker process
 | `apps/audit/domain.py` | `AuditEvent` — the Pydantic model that defines every field an audit event can have |
 | `apps/audit/enums.py` | `EventAction`, `EventOutcome` — the vocabulary of valid actions and outcomes |
 | `apps/audit/service.py` | `audit.log()` — the one function callers use. Serializes and enqueues. |
-| `apps/audit/tasks.py` | `send_audit_event` — the Taskiq worker task. Writes to OpenSearch, handles retries. |
-| `apps/audit/index_mapping.py` | OpenSearch field type definitions for the `audit-logs` index |
-| `infrastructure/utility/opensearch_client.py` | Singleton `AsyncOpenSearch` wrapper used by the worker |
-| `infrastructure/lifespan.py` | Creates the `audit-logs` index on app startup (once, idempotent) |
-| `config/opensearch.py` | `OpenSearchSettings` — host, port, credentials, index name |
+| `apps/audit/tasks.py` | `send_audit_event` — the Taskiq worker task. Writes to Postgres, handles retries. |
+| `apps/audit/db/schemas.py` | `AuditLogSchema` — the `audit_logs` table (typed columns + JSONB `payload`) |
+| `apps/audit/crud.py` | `AuditLogCRUD` — idempotent insert and the applet-scoped export query |
+| `apps/audit/query_service.py` | `AuditQueryService` — rebuilds `AuditEvent`s from stored rows for the export endpoint |
 
 ---
 
@@ -83,39 +84,34 @@ Only `user_id` and `event_action` are required. Everything else is optional and
 
 ## Failure handling
 
-If OpenSearch is unavailable or returns an error:
+If the database write fails:
 
 1. The worker logs a warning and re-enqueues the task with a **5-second delay**.
 2. This repeats up to **3 times** (configurable via `retries`).
 3. After all retries are exhausted, the full event payload is logged at **ERROR level**, which captures it in Datadog. The event is not silently dropped.
 
+Each event carries a unique `event_id`, stored in a unique column. The insert uses `ON CONFLICT (event_id) DO NOTHING`, so a retried task that re-delivers the same event never creates a duplicate row.
+
 ---
 
-## OpenSearch index
+## Storage
 
-The `audit-logs` index is created once on app startup by `startup_opensearch()` in `lifespan.py`. The mapping is defined explicitly in `index_mapping.py` — field types are chosen for the queries we expect:
+Audit events live in the `audit_logs` table (`apps/audit/db/schemas.py`), created by an Alembic migration like any other table.
 
-- `keyword` — IDs, enums, statuses (exact match, aggregation)
-- `ip` — `client.ip` (CIDR range queries)
-- `date` — `@timestamp` (time-range queries)
-- `text` — URL paths, user agent strings (full-text search)
+- Typed columns — `event_id`, `event_timestamp`, `event_action`, `event_outcome`, `user_id`, `applet_ids` — back the export query's filter and sort.
+- `payload` (JSONB) holds the full event document; the export endpoint returns it verbatim.
 
-Adding new **event actions** (new values of `EventAction`) requires no mapping changes. Adding new **fields** to `AuditEvent` requires a corresponding entry in `index_mapping.py`.
+Indexes:
+- unique on `event_id` (idempotency)
+- `(event_timestamp, event_id)` — the date-range filter and sort order
+- GIN on `applet_ids` — `:applet_id = ANY(applet_ids)` membership lookups
 
----
+Adding new **event actions** (new values of `EventAction`) requires no schema change. Adding new **fields** to `AuditEvent` requires no schema change either — they are stored in the JSONB `payload` automatically; only promote a field to its own column (with a migration) if you need to filter or sort on it.
 
-## Configuration
+> Note: the legacy OpenSearch modules (`infrastructure/utility/opensearch_client.py`, `apps/audit/index_mapping.py`, `config/opensearch.py`) remain in the tree but are no longer wired into the audit pipeline.
 
-All OpenSearch settings are nested under `OPENSEARCH__` in the environment:
+---
 
-```bash
-OPENSEARCH__HOST=opensearch        # default: opensearch (Docker service name)
-OPENSEARCH__PORT=9200              # default: 9200
-OPENSEARCH__USER=admin             # default: admin
-OPENSEARCH__PASSWORD=admin         # default: admin
-OPENSEARCH__USE_SSL=True           # default: True (required by the Docker image)
-OPENSEARCH__VERIFY_CERTS=False     # default: False (self-signed cert in local/dev)
-OPENSEARCH__AUDIT_INDEX=audit-logs # default: audit-logs
-```
+## Retention
 
-For local development, if running the backend outside Docker, set `OPENSEARCH__HOST=localhost`.
+There is currently no retention policy — the `audit_logs` table grows unbounded (the previous OpenSearch setup had no retention either). When retention becomes necessary, the recommended approach is native Postgres range partitioning by `event_timestamp` (e.g. monthly) with a scheduled job that drops partitions older than a configurable window.

src/apps/audit/api.py

@@ -41,7 +41,7 @@ async def applet_audit_export(
         if from_datetime and to_datetime and from_datetime > to_datetime:
             raise InvalidAuditDateRangeError(path=["fromDatetime"])
 
-        events, total = await AuditQueryService().search_applet_events(
+        events, total = await AuditQueryService(session).search_applet_events(
             applet_id,
             from_datetime=from_datetime,
             to_datetime=to_datetime,

src/apps/audit/crud.py

@@ -0,0 +1,74 @@
+import datetime
+import uuid
+
+from sqlalchemy import and_, func, select
+from sqlalchemy.dialects.postgresql import insert
+
+from apps.audit.db.schemas import AuditLogSchema
+from infrastructure.database.crud import BaseCRUD
+
+DEFAULT_PAGE_SIZE = 1000
+
+
+def _to_naive_utc(value: datetime.datetime) -> datetime.datetime:
+    """Normalise a datetime to naive UTC to match the stored ``event_timestamp``."""
+    if value.tzinfo is not None:
+        value = value.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+    return value
+
+
+class AuditLogCRUD(BaseCRUD[AuditLogSchema]):
+    schema_class = AuditLogSchema
+
+    async def save(self, schema: AuditLogSchema) -> None:
+        """Insert an audit log row.
+
+        Uses ``ON CONFLICT (event_id) DO NOTHING`` so that a retried worker
+        task (which re-delivers the same event) does not create a duplicate
+        row — preserving the idempotency OpenSearch gave us via ``id=event.id``.
+        """
+        query = (
+            insert(AuditLogSchema)
+            .values(
+                event_id=schema.event_id,
+                event_timestamp=schema.event_timestamp,
+                event_action=schema.event_action,
+                event_outcome=schema.event_outcome,
+                user_id=schema.user_id,
+                applet_ids=schema.applet_ids,
+                payload=schema.payload,
+            )
+            .on_conflict_do_nothing(index_elements=[AuditLogSchema.event_id])
+        )
+        await self._execute(query)
+
+    async def search_applet_events(
+        self,
+        applet_id: uuid.UUID,
+        *,
+        from_datetime: datetime.datetime | None = None,
+        to_datetime: datetime.datetime | None = None,
+        page: int = 1,
+        limit: int = DEFAULT_PAGE_SIZE,
+    ) -> tuple[list[AuditLogSchema], int]:
+        conditions = [AuditLogSchema.applet_ids.contains([applet_id])]
+        if from_datetime is not None:
+            conditions.append(AuditLogSchema.event_timestamp >= _to_naive_utc(from_datetime))
+        if to_datetime is not None:
+            conditions.append(AuditLogSchema.event_timestamp < _to_naive_utc(to_datetime))
+
+        where = and_(*conditions)
+
+        query = (
+            select(AuditLogSchema)
+            .where(where)
+            .order_by(AuditLogSchema.event_timestamp.asc(), AuditLogSchema.event_id.asc())
+            .limit(limit)
+            .offset((page - 1) * limit)
+        )
+        rows = (await self._execute(query)).scalars().all()
+
+        count_query = select(func.count()).select_from(AuditLogSchema).where(where)
+        total = (await self._execute(count_query)).scalar() or 0
+
+        return list(rows), total

src/apps/audit/db/schemas.py

@@ -0,0 +1,30 @@
+from sqlalchemy import Column, DateTime, Index, String
+from sqlalchemy.dialects.postgresql import ARRAY, JSONB, UUID
+
+from infrastructure.database.base import Base
+
+
+class AuditLogSchema(Base):
+    """Audit event stored in Postgres.
+
+    Append-only. A few columns are denormalized from the ECS document for
+    filtering/sorting; the full event is kept in ``payload`` as the same
+    dotted-alias JSON that is produced by ``AuditEvent.model_dump(mode="json")``.
+    """
+
+    __tablename__ = "audit_logs"
+
+    event_id = Column(UUID(as_uuid=True), nullable=False, unique=True)
+    event_timestamp = Column(DateTime(), nullable=False)
+    event_action = Column(String(), nullable=False)
+    event_outcome = Column(String(), nullable=True)
+    user_id = Column(UUID(as_uuid=True), nullable=True, index=True)
+    applet_ids = Column(ARRAY(UUID(as_uuid=True)), nullable=True)
+    payload = Column(JSONB(), nullable=False)
+
+    __table_args__ = (
+        # Covers the date-range filter and the (timestamp, event_id) sort order.
+        Index("ix_audit_logs_event_timestamp_event_id", "event_timestamp", "event_id"),
+        # Covers `applet_ids @> ARRAY[:applet_id]` membership lookups.
+        Index("ix_audit_logs_applet_ids", "applet_ids", postgresql_using="gin"),
+    )

src/apps/audit/query_service.py

@@ -1,17 +1,15 @@
 import datetime
 import uuid
 
-from apps.audit.domain import AuditEvent
-from config import settings
-from infrastructure.utility.opensearch_client import DEFAULT_PAGE_SIZE, OpenSearchClient
+from sqlalchemy.ext.asyncio import AsyncSession
 
-SORT: list[dict] = [{"@timestamp": "asc"}, {"event.id": "asc"}]
+from apps.audit.crud import DEFAULT_PAGE_SIZE, AuditLogCRUD
+from apps.audit.domain import AuditEvent
 
 
 class AuditQueryService:
-    def __init__(self, client: OpenSearchClient | None = None) -> None:
-        self._client = client or OpenSearchClient()
-        self._index = settings.opensearch.audit_index
+    def __init__(self, session: AsyncSession) -> None:
+        self._crud = AuditLogCRUD(session)
 
     async def search_applet_events(
         self,
@@ -22,37 +20,12 @@ async def search_applet_events(
         page: int = 1,
         limit: int = DEFAULT_PAGE_SIZE,
     ) -> tuple[list[AuditEvent], int]:
-        query = self._build_query(applet_id, from_datetime, to_datetime)
-        response = await self._client.search(
-            self._index,
-            query=query,
-            sort=SORT,
-            size=limit,
-            from_=(page - 1) * limit,
+        rows, total = await self._crud.search_applet_events(
+            applet_id,
+            from_datetime=from_datetime,
+            to_datetime=to_datetime,
+            page=page,
+            limit=limit,
         )
-        hits = response.get("hits", {})
-        total = hits.get("total", {}).get("value", 0)
-        events = [AuditEvent.model_validate(hit["_source"]) for hit in hits.get("hits", [])]
+        events = [AuditEvent.model_validate(row.payload) for row in rows]
         return events, total
-
-    @staticmethod
-    def _build_query(
-        applet_id: uuid.UUID,
-        from_datetime: datetime.datetime | None,
-        to_datetime: datetime.datetime | None,
-    ) -> dict:
-        filters: list[dict] = [{"term": {"curious.applet_id": str(applet_id)}}]
-
-        timestamp_range: dict = {}
-        if from_datetime is not None:
-            if from_datetime.tzinfo is None:
-                from_datetime = from_datetime.replace(tzinfo=datetime.timezone.utc)
-            timestamp_range["gte"] = from_datetime.isoformat()
-        if to_datetime is not None:
-            if to_datetime.tzinfo is None:
-                to_datetime = to_datetime.replace(tzinfo=datetime.timezone.utc)
-            timestamp_range["lt"] = to_datetime.isoformat()
-        if timestamp_range:
-            filters.append({"range": {"@timestamp": timestamp_range}})
-
-        return {"bool": {"filter": filters}}