Database refactor to separate modules.

paskia/db/jsonl.py

@@ -0,0 +1,125 @@
+"""
+JSONL persistence layer for the database.
+
+Handles file I/O, JSON diffs, and persistence. Works with plain JSON/dict data.
+Uses aiofiles for async I/O operations.
+"""
+
+import logging
+from collections import deque
+from datetime import datetime, timezone
+from pathlib import Path
+
+import aiofiles
+import jsondiff
+import msgspec
+
+_logger = logging.getLogger(__name__)
+
+# Default database path
+DB_PATH_DEFAULT = "paskia.jsonl"
+
+
+class _ChangeRecord(msgspec.Struct):
+    """A single change record in the JSONL file."""
+
+    ts: datetime
+    actor: str
+    diff: dict
+
+
+# msgspec encoder for change records
+_change_encoder = msgspec.json.Encoder()
+
+
+async def load_jsonl(db_path: Path, empty_data: dict) -> dict:
+    """Load data from disk by applying change log.
+
+    Replays all changes from JSONL file using plain dicts (to handle
+    schema evolution).
+
+    Args:
+        db_path: Path to the JSONL database file
+        empty_data: Empty data structure to start with (as dict)
+
+    Returns:
+        The final state after applying all changes
+    """
+    data_dict = empty_data.copy()
+    if db_path.exists():
+        try:
+            # Read entire file at once and split into lines
+            async with aiofiles.open(db_path, "rb") as f:
+                content = await f.read()
+            for line_num, line in enumerate(content.split(b"\n"), 1):
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    change = msgspec.json.decode(line)
+                    # Apply the diff to current state (marshal=True for $-prefixed keys)
+                    data_dict = jsondiff.patch(data_dict, change["diff"], marshal=True)
+                except Exception as e:
+                    raise ValueError(f"Error parsing line {line_num}: {e}")
+        except (OSError, ValueError, msgspec.DecodeError) as e:
+            raise ValueError(f"Failed to load database: {e}")
+    return data_dict
+
+
+def compute_diff(previous: dict, current: dict) -> dict | None:
+    """Compute JSON diff between two states.
+
+    Args:
+        previous: Previous state (JSON-compatible dict)
+        current: Current state (JSON-compatible dict)
+
+    Returns:
+        The diff, or None if no changes
+    """
+    diff = jsondiff.diff(previous, current, marshal=True)
+    return diff if diff else None
+
+
+def create_change_record(actor: str, diff: dict) -> _ChangeRecord:
+    """Create a change record for persistence."""
+    return _ChangeRecord(
+        ts=datetime.now(timezone.utc),
+        actor=actor,
+        diff=diff,
+    )
+
+
+async def flush_changes(
+    db_path: Path,
+    pending_changes: deque[_ChangeRecord],
+) -> bool:
+    """Write all pending changes to disk.
+
+    Args:
+        db_path: Path to the JSONL database file
+        pending_changes: Queue of pending change records (will be cleared on success)
+
+    Returns:
+        True if flush succeeded, False otherwise
+    """
+    if not pending_changes:
+        return True
+
+    # Collect all pending changes
+    changes_to_write = list(pending_changes)
+    pending_changes.clear()
+
+    try:
+        # Build lines to append (keep as bytes, join with \n)
+        lines = [_change_encoder.encode(change) for change in changes_to_write]
+
+        # Append all lines in a single write (binary mode for Windows compatibility)
+        async with aiofiles.open(db_path, "ab") as f:
+            await f.write(b"\n".join(lines) + b"\n")
+        return True
+    except OSError:
+        _logger.exception("Failed to flush database changes")
+        # Re-queue the changes on failure
+        for change in reversed(changes_to_write):
+            pending_changes.appendleft(change)
+        return False