Stage 04.2 - journal and event log

app/src/storage/repositories/journal.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from src.storage.session import get_connection
+
+
+class JournalRepository:
+    def add_event(
+        self,
+        *,
+        level: str,
+        event_type: str,
+        message: str,
+        payload: dict[str, Any] | None = None,
+    ) -> None:
+        payload_json = json.dumps(payload, ensure_ascii=False) if payload is not None else None
+
+        with get_connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    '''
+                    INSERT INTO journal_events (level, event_type, message, payload_json)
+                    VALUES (%s, %s, %s, %s::jsonb)
+                    ''',
+                    (
+                        level.upper().strip(),
+                        event_type.strip(),
+                        message.strip(),
+                        payload_json,
+                    ),
+                )
+
+    def list_recent_events(self, limit: int = 10) -> list[dict[str, str]]:
+        with get_connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    '''
+                    SELECT id, created_at, level, event_type, message
+                    FROM journal_events
+                    ORDER BY created_at DESC, id DESC
+                    LIMIT %s
+                    ''',
+                    (limit,),
+                )
+                rows = cursor.fetchall()
+
+        items: list[dict[str, str]] = []
+        for row in rows:
+            items.append(
+                {
+                    "id": str(row[0]),
+                    "created_at": str(row[1]),
+                    "level": str(row[2]),
+                    "event_type": str(row[3]),
+                    "message": str(row[4]),
+                }
+            )
+        return items
+
+    def count_events(self) -> int:
+        with get_connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute("SELECT COUNT(*) FROM journal_events")
+                row = cursor.fetchone()
+
+        return int(row[0]) if row else 0
+
+    def list_recent_with_offset(self, limit: int, offset: int):
+        with get_connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    SELECT created_at, level, event_type, message
+                    FROM journal_events
+                    ORDER BY created_at DESC
+                    LIMIT %s OFFSET %s
+                    """,
+                    (limit, offset),
+                )
+                rows = cursor.fetchall()
+
+        return [
+            {
+                "created_at": str(r[0]),
+                "level": r[1],
+                "event_type": r[2],
+                "message": r[3],
+            }
+            for r in rows
+        ]

app/src/storage/schema.py

@@ -1,8 +1,5 @@
 from __future__ import annotations
-
 from src.storage.session import get_connection
-
-
 DDL = [
     '''
     CREATE TABLE IF NOT EXISTS balance_snapshots (
@@ -23,6 +20,14 @@ DDL = [
     )
     ''',
     '''
+    CREATE INDEX IF NOT EXISTS idx_journal_events_created_at
+    ON journal_events (created_at DESC)
+    ''',
+    '''
+    CREATE INDEX IF NOT EXISTS idx_journal_events_event_type
+    ON journal_events (event_type)
+    ''',
+    '''
     CREATE TABLE IF NOT EXISTS order_drafts (
         id BIGSERIAL PRIMARY KEY,
         created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
@@ -35,10 +40,8 @@ DDL = [
     )
     '''
 ]
-
-
 def init_schema() -> None:
     with get_connection() as connection:
         with connection.cursor() as cursor:
             for statement in DDL:
-                cursor.execute(statement)
+                cursor.execute(statement)