Remove built-in database, replace with kanta package. No disk format changes.

paskia/db/background.py

@@ -1,67 +1,61 @@
 """
 Background task for database maintenance.
 
-Periodically flushes pending changes to disk and cleans up expired items.
+Kanta handles periodic flushing to disk.  This module keeps a small
+companion task that periodically cleans up expired sessions/tokens.
 """
 
 import asyncio
 import logging
-from datetime import UTC, datetime
+import os
+import signal
+
+from kanta.exceptions import DatabaseError
 
 import paskia.db.operations as _ops
 from paskia.db.lifecycle import cleanup_expired
 
-FLUSH_INTERVAL = 0.1  # Flush to disk
 CLEANUP_INTERVAL = 1  # Expired item cleanup
 
-
 _logger = logging.getLogger(__name__)
 _background_task: asyncio.Task | None = None
 
 
+def _sigterm_on_error(error: DatabaseError) -> None:
+    """Exit the server when a database write fails."""
+    _logger.error("Fatal database error: %s", error)
+    os.kill(os.getpid(), signal.SIGTERM)
+
+
 async def flush() -> None:
     """Write all pending database changes to disk."""
-    store = _ops._db._store
+    store = _ops._store
     if store is None:
         _logger.warning("flush() called but _store is None")
         return
-    await store.flush()
+    try:
+        await store.flush()
+    except DatabaseError as e:
+        _sigterm_on_error(e)
 
 
 async def _background_loop():
-    """Background task that periodically flushes changes and cleans up."""
+    """Background task that periodically cleans up expired items."""
     # Run cleanup immediately on startup to clear old expired items
     cleanup_expired()
-    await flush()
-
-    last_cleanup = datetime.now(UTC)
 
     while True:
         try:
-            await asyncio.sleep(FLUSH_INTERVAL)
-            # Flush pending changes to disk
-            await flush()
-
-            # Run cleanup periodically
-            now = datetime.now(UTC)
-            if (now - last_cleanup).total_seconds() >= CLEANUP_INTERVAL:
-                cleanup_expired()
-                await flush()  # Flush cleanup changes
-                last_cleanup = now
-
-            # Conditionally write a snapshot to speed up future startups
-            if _ops._db._store is not None:
-                _ops._db._store.maybe_snapshot()
+            await asyncio.sleep(CLEANUP_INTERVAL)
+            cleanup_expired()
         except asyncio.CancelledError:
-            # Final flush before exit
-            await flush()
             break
         except Exception:
             _logger.debug("Error in database background loop", exc_info=True)
 
 
 async def start_background():
-    """Start the background flush/cleanup task."""
+    """Start the background cleanup task."""
     global _background_task
 
     # Check if task exists but is no longer running (e.g., after uvicorn reload)
@@ -75,16 +69,15 @@ async def start_background():
                 # Check if task is in current event loop
                 loop = asyncio.get_running_loop()
                 task_loop = _background_task.get_loop()
-                if loop is not task_loop:
-                    _logger.debug("Background task in different event loop, restarting")
-                    _background_task = None
-                else:
+                if loop is task_loop:
                     # Task is already running in same loop - idempotent, just return
                     # This happens with dual IPv4+IPv6 endpoints sharing the same process
                     _logger.debug(
                         "Background task already running in same loop, skipping"
                     )
                     return
+                _logger.debug("Background task in different event loop, restarting")
+                _background_task = None
             except Exception as e:
                 _logger.debug("Error checking background task loop: %s, restarting", e)
                 _background_task = None
@@ -94,7 +87,7 @@ async def start_background():
 
 
 async def stop_background():
-    """Stop the background task, flush pending changes, and release the file lock."""
+    """Stop the background cleanup task and close kanta."""
     global _background_task
     if _background_task:
         _background_task.cancel()
@@ -103,7 +96,12 @@ async def stop_background():
         except asyncio.CancelledError:
             pass
         _background_task = None
-    _ops._db._store.close()
+    store = _ops._store
+    if store is not None:
+        try:
+            await store.close()
+        except DatabaseError as e:
+            _sigterm_on_error(e)
 
 
 # Aliases for backwards compatibility

paskia/db/filelock.py

@@ -1,247 +0,0 @@
-"""Cross-platform locked file for the database (no separate .lock files).
-
-Unix:    open() + fcntl.flock (advisory, cooperative among processes that flock).
-Windows: CreateFileW with FILE_SHARE_READ (OS-enforced, allows readers, blocks writers).
-
-A single file descriptor is opened once for both reading and writing.
-The lock is acquired atomically (on Windows) or immediately after open (on Unix),
-and the same descriptor is used for the lifetime of the process: first to read
-the existing content, then to append new writes.
-"""
-
-import logging
-import os
-import sys
-from pathlib import Path
-
-_logger = logging.getLogger(__name__)
-
-
-def _fatal(msg: str) -> None:
-    """Log a fatal error and exit immediately, bypassing exception handlers."""
-    _logger.critical(msg)
-    os._exit(1)
-
-
-if sys.platform == "win32":
-    import ctypes
-    from ctypes import wintypes
-
-    _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
-
-    _GENERIC_READ = 0x80000000
-    _GENERIC_WRITE = 0x40000000
-    _FILE_SHARE_READ = 0x00000001
-    _OPEN_EXISTING = 3
-    _OPEN_ALWAYS = 4
-    _FILE_ATTRIBUTE_NORMAL = 0x80
-    _FILE_BEGIN = 0
-    _FILE_END = 2
-    _ERROR_SHARING_VIOLATION = 32
-    _INVALID_FILE_SIZE = 0xFFFFFFFF
-
-    _kernel32.CreateFileW.restype = wintypes.HANDLE
-    _kernel32.CreateFileW.argtypes = [
-        wintypes.LPCWSTR,
-        wintypes.DWORD,
-        wintypes.DWORD,
-        ctypes.c_void_p,
-        wintypes.DWORD,
-        wintypes.DWORD,
-        wintypes.HANDLE,
-    ]
-    _kernel32.ReadFile.restype = wintypes.BOOL
-    _kernel32.ReadFile.argtypes = [
-        wintypes.HANDLE,
-        ctypes.c_void_p,
-        wintypes.DWORD,
-        ctypes.POINTER(wintypes.DWORD),
-        ctypes.c_void_p,
-    ]
-    _kernel32.WriteFile.restype = wintypes.BOOL
-    _kernel32.WriteFile.argtypes = [
-        wintypes.HANDLE,
-        ctypes.c_void_p,
-        wintypes.DWORD,
-        ctypes.POINTER(wintypes.DWORD),
-        ctypes.c_void_p,
-    ]
-    _kernel32.GetFileSize.restype = wintypes.DWORD
-    _kernel32.GetFileSize.argtypes = [
-        wintypes.HANDLE,
-        ctypes.POINTER(wintypes.DWORD),
-    ]
-    _kernel32.SetFilePointer.restype = wintypes.DWORD
-    _kernel32.SetFilePointer.argtypes = [
-        wintypes.HANDLE,
-        wintypes.LONG,
-        ctypes.POINTER(wintypes.LONG),
-        wintypes.DWORD,
-    ]
-    _kernel32.CloseHandle.restype = wintypes.BOOL
-    _kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
-
-    def _is_invalid_handle(handle) -> bool:
-        return ctypes.c_void_p(handle).value == ctypes.c_void_p(-1).value
-
-else:
-    import fcntl
-
-
-class LockedFile:
-    """A file opened with an exclusive write lock.
-
-    Usage::
-
-        f = LockedFile()
-        f.open(path)          # open + lock (read+write)
-        content = f.read()    # read entire content
-        f.write(data)         # append data (seeks to end first)
-        f.close()             # release lock + close fd
-
-    Unix:    fcntl.flock (advisory) — read-only callers that don't flock are unaffected.
-    Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers.
-    """
-
-    def __init__(self) -> None:
-        self._fd: int | None = None  # Unix fd or Windows HANDLE
-
-    def open(self, path: Path, *, create: bool = False) -> None:
-        """Open *path* for read+write with an exclusive lock.
-
-        Args:
-            path:   File to open and lock.
-            create: If True, create the file if it doesn't exist (bootstrap).
-
-        Raises:
-            SystemExit: If the file is locked by another process or not found.
-        """
-        if self._fd is not None:
-            return  # Already open (idempotent)
-
-        if sys.platform == "win32":
-            self._open_win32(path, create)
-        else:
-            self._open_unix(path, create)
-
-    def open_and_read(self, path: Path) -> bytes:
-        """Open *path* with exclusive lock and read all content.
-
-        Combined operation for efficient use with asyncio.to_thread().
-        """
-        self.open(path)
-        return self.read()
-
-    def read(self) -> bytes:
-        """Read the entire file content from the beginning."""
-        if self._fd is None:
-            raise RuntimeError("LockedFile.read() called on a closed file")
-
-        if sys.platform == "win32":
-            return self._read_win32()
-        else:
-            return self._read_unix()
-
-    def write(self, data: bytes) -> None:
-        """Append *data* to the end of the file."""
-        if self._fd is None:
-            raise RuntimeError("LockedFile.write() called on a closed file")
-
-        if sys.platform == "win32":
-            self._write_win32(data)
-        else:
-            self._write_unix(data)
-
-    def close(self) -> None:
-        """Release the lock and close the file."""
-        if self._fd is None:
-            return
-        if sys.platform == "win32":
-            _kernel32.CloseHandle(self._fd)
-        else:
-            os.close(self._fd)
-        self._fd = None
-
-    @property
-    def is_open(self) -> bool:
-        return self._fd is not None
-
-    # -- Unix ----------------------------------------------------------------
-
-    def _open_unix(self, path: Path, create: bool) -> None:
-        flags = os.O_RDWR | (os.O_CREAT if create else 0)
-        try:
-            fd = os.open(path, flags, 0o666)
-        except FileNotFoundError:
-            _fatal(f"Database file not found: {path.resolve()}")
-        try:
-            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except OSError:
-            os.close(fd)
-            _fatal(f"🛑 {path.resolve()}: database already locked by another instance")
-        self._fd = fd
-
-    def _read_unix(self) -> bytes:
-        os.lseek(self._fd, 0, os.SEEK_SET)
-        chunks = []
-        while True:
-            chunk = os.read(self._fd, 1 << 20)  # 1 MiB
-            if not chunk:
-                break
-            chunks.append(chunk)
-        return b"".join(chunks)
-
-    def _write_unix(self, data: bytes) -> None:
-        os.lseek(self._fd, 0, os.SEEK_END)
-        os.write(self._fd, data)
-
-    # -- Windows -------------------------------------------------------------
-
-    def _open_win32(self, path: Path, create: bool) -> None:
-        disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
-        handle = _kernel32.CreateFileW(
-            str(path),
-            _GENERIC_READ | _GENERIC_WRITE,
-            _FILE_SHARE_READ,
-            None,
-            disposition,
-            _FILE_ATTRIBUTE_NORMAL,
-            None,
-        )
-        if _is_invalid_handle(handle):
-            err = ctypes.get_last_error()
-            if err == _ERROR_SHARING_VIOLATION:
-                _fatal(
-                    f"🛑 {path.resolve()}: database already locked by another instance"
-                )
-            _fatal(f"Failed to open database {path.resolve()}: Windows error {err}")
-        self._fd = handle
-
-    def _read_win32(self) -> bytes:
-        _kernel32.SetFilePointer(self._fd, 0, None, _FILE_BEGIN)
-        size = _kernel32.GetFileSize(self._fd, None)
-        if size == _INVALID_FILE_SIZE:
-            raise OSError(
-                f"GetFileSize failed: Windows error {ctypes.get_last_error()}"
-            )
-        if size == 0:
-            return b""
-        buf = ctypes.create_string_buffer(size)
-        bytes_read = wintypes.DWORD()
-        ok = _kernel32.ReadFile(self._fd, buf, size, ctypes.byref(bytes_read), None)
-        if not ok:
-            raise OSError(f"ReadFile failed: Windows error {ctypes.get_last_error()}")
-        return buf.raw[: bytes_read.value]
-
-    def _write_win32(self, data: bytes) -> None:
-        _kernel32.SetFilePointer(self._fd, 0, None, _FILE_END)
-        written = wintypes.DWORD()
-        ok = _kernel32.WriteFile(
-            self._fd,
-            data,
-            len(data),
-            ctypes.byref(written),
-            None,
-        )
-        if not ok:
-            raise OSError(f"WriteFile failed: Windows error {ctypes.get_last_error()}")

paskia/db/jsonl.py

@@ -1,82 +1,20 @@
 """
-JSONL persistence layer for the database.
+JSONL read-only loader using kanta.
 """
 
-import asyncio
-import copy
 import logging
-import os
-import signal
-from collections import deque
-from contextlib import contextmanager
-from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any
-from uuid import UUID
 
-import jsondiff
 import msgspec
+from kanta import replay as replay_jsonl
+from kanta.migrate import MigrationRegistry
 
-from paskia.db.filelock import LockedFile
-from paskia.db.logging import log_change
-from paskia.db.migrations import (
-    DBVER,
-    MigrationCtx,
-    apply_all_migrations,
-    apply_migrations_readonly,
-)
-from paskia.db.snapshot import SnapshotState
-from paskia.db.structs import DB, Config, SessionContext
+from paskia.db.migrations import MigrationCtx
+from paskia.db.structs import DB, Config
 
 _logger = logging.getLogger(__name__)
 
 
-class ReplayResult(msgspec.Struct, frozen=False):
-    """Return value of _replay_from_data"""
-
-    state: dict = {}
-    v: int = 0
-    ts: datetime | None = None
-    snapts: datetime | None = None
-    changes: int = 0
-
-
-class DatabaseError(ValueError):
-    """Exception raised for database loading errors."""
-
-
-def _replay_from_data(data: bytes, db_path: str) -> ReplayResult:
-    """Replay database state from file data, using the last snapshot if available."""
-    resolved_path = str(Path(db_path).resolve())
-    result = ReplayResult()
-
-    # Find and apply the last snapshot
-    snap, start_offset = SnapshotState.load(data)
-    if snap:
-        result.state = snap.state
-        result.v = snap.v
-        result.snapts = snap.ts
-
-    # Replay change records after the snapshot
-    lines = data[start_offset:].split(b"\n")
-    for raw in lines:
-        line = raw.strip()
-        if not line:
-            continue
-        try:
-            change = msgspec.json.decode(line, type=ChangeRecord)
-        except msgspec.DecodeError as e:
-            raise DatabaseError(
-                f"{resolved_path}: {e}\n{line.decode(errors='replace')}"
-            )
-        result.state = jsondiff.patch(result.state, change.diff, marshal=True)
-        result.v = change.v
-        result.ts = change.ts
-        result.changes += 1
-
-    return result
-
-
 def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB:
     """Replay JSONL and apply migrations to produce a DB, without writing anything.
 
@@ -89,21 +27,21 @@ def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB:
 
     try:
         content = path.read_bytes()
-        r = _replay_from_data(content, str(path.resolve()))
-        data_dict = r.state
-        version = r.v
+        rr = replay_jsonl(content)
+        data_dict = rr.state
+        version = rr.version
 
         if not data_dict:
             return DB(config=Config(rp_id=rp_id))
 
         # Apply migrations in-memory (no persistence)
-        apply_migrations_readonly(data_dict, version, MigrationCtx(rp_id=rp_id))
+        registry = MigrationRegistry.from_module("paskia.db.migrations")
+        version = registry.apply(
+            data_dict, version, MigrationCtx(rp_id=rp_id), silent=True
+        )
 
         # Decode to msgspec struct
-        try:
-            return msgspec.json.decode(msgspec.json.encode(data_dict), type=DB)
-        except msgspec.ValidationError as e:
-            raise DatabaseError(f"{path.resolve()}: {e}") from None
+        return msgspec.json.decode(msgspec.json.encode(data_dict), type=DB)
     except OSError as e:
         _logger.exception("Failed to load database")
         raise SystemExit(f"{e}")
@@ -112,241 +50,3 @@ def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB:
     except Exception as e:
         _logger.exception("Unexpected error loading database")
         raise SystemExit(f"{e}")
-
-
-class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
-    ts: datetime = msgspec.field(default_factory=lambda: datetime.now(UTC))
-    a: str = ""  # action (e.g., "migrate", "login", "create_user")
-    v: int = 0  # schema version after this change
-    u: str | None = None  # user UUID who performed the action (None for system)
-    diff: dict
-
-
-def compute_diff(previous: dict, current: dict) -> dict | None:
-    return jsondiff.diff(previous, current, marshal=True) or None
-
-
-# Actions that are allowed to create a new database file
-_BOOTSTRAP_ACTIONS = frozenset({"bootstrap"})
-
-
-class JsonlStore:
-    """JSONL persistence layer for a DB instance."""
-
-    def __init__(self, db: DB, db_path: str):
-        self.db: DB = db
-        self.db_path = Path(db_path)
-        self._file = LockedFile()
-        self._flush_failed = False
-        self._statedict: dict[str, Any] = {}
-        self._pending_changes: deque[ChangeRecord] = deque()
-        self._current_action: str = "system"
-        self._current_user: str | None = None
-        self._in_transaction: bool = False
-        self._transaction_snapshot: dict[str, Any] | None = None
-        self._v: int = DBVER  # Schema version for new databases
-        self._snapshot = SnapshotState()
-
-    async def load(
-        self, db_path: str | None = None, *, rp_id: str = "localhost"
-    ) -> None:
-        """Load data from JSONL change log."""
-        if db_path is not None:
-            self.db_path = Path(db_path)
-        self._rp_id = rp_id
-        if not self.db_path.exists():
-            return
-
-        # Open with exclusive write lock and read contents — single threadpool call
-        content = await asyncio.to_thread(self._file.open_and_read, self.db_path)
-
-        # Replay change log to reconstruct state (snapshot-accelerated)
-        try:
-            r = _replay_from_data(content, str(self.db_path.resolve()))
-            statedict = r.state
-            self._v = r.v
-            self._snapshot.ts = r.snapts
-            self._snapshot.changes = r.changes
-        except (OSError, ValueError, msgspec.DecodeError, DatabaseError) as e:
-            raise SystemExit(f"{e}")
-        except Exception as e:
-            _logger.exception("Unexpected error loading database")
-            raise SystemExit(f"{e}")
-
-        if not statedict:
-            return
-
-        # Set previous state for diffing (will be updated by _queue_change)
-        self._statedict = copy.deepcopy(statedict)
-
-        # Callback to persist each migration
-        async def persist_migration(
-            action: str, new_version: int, current: dict
-        ) -> None:
-            self._v = new_version
-            self._queue_change(action, new_version, current)
-
-        # Apply schema migrations one at a time
-        await apply_all_migrations(
-            statedict,
-            self._v,
-            persist_migration,
-            MigrationCtx(rp_id=rp_id),
-        )
-
-        # Decode to msgspec struct
-        decoder = msgspec.json.Decoder(DB)
-        self.db = decoder.decode(msgspec.json.encode(statedict))
-        self.db._store = self
-
-        # Normalize via msgspec round-trip (handles omit_defaults etc.)
-        # This ensures _previous_builtins matches what msgspec would produce
-        normalized_dict = msgspec.to_builtins(self.db)
-        await persist_migration("migrate:msgspec", self._v, normalized_dict)
-
-    def _queue_change(
-        self, action: str, version: int, current: dict, user: str | None = None
-    ) -> None:
-        """Queue a change record and log it.
-
-        Args:
-            action: The action name for the change record
-            version: The schema version for the change record
-            current: The current state as a plain dict
-            user: Optional user UUID who performed the action
-        """
-        diff = compute_diff(self._statedict, current)
-        if not diff:
-            return
-        self._pending_changes.append(
-            ChangeRecord(
-                a=action,
-                v=version,
-                u=user,
-                diff=diff,
-            )
-        )
-
-        # Log the change with user display name if available
-        user_display = None
-        if user:
-            try:
-                user_uuid = UUID(user)
-                if user_uuid in self.db.users:
-                    user_display = self.db.users[user_uuid].display_name
-            except (ValueError, KeyError):
-                user_display = user
-
-        log_change(action, diff, user_display, self._statedict, self.db)
-        self._statedict = copy.deepcopy(current)
-
-    @contextmanager
-    def transaction(
-        self,
-        action: str,
-        ctx: SessionContext | None = None,
-        *,
-        user: str | None = None,
-    ):
-        """Wrap writes in transaction. Queues change on successful exit.
-
-        Args:
-            action: Describes the operation (e.g., "Created user", "Login")
-            ctx: Session context of user performing the action (None for system operations)
-            user: User UUID string (alternative to ctx when full context unavailable)
-        """
-        if self._in_transaction:
-            raise RuntimeError("Nested transactions are not supported")
-
-        # Check for out-of-transaction modifications
-        current_state = msgspec.to_builtins(self.db)
-        if current_state != self._statedict:
-            # Allow bootstrap to create a new database from empty state
-            is_bootstrap = action in _BOOTSTRAP_ACTIONS
-            if is_bootstrap and not self._statedict:
-                pass  # Expected: creating database from scratch
-            else:
-                diff = compute_diff(self._statedict, current_state)
-                diff_json = msgspec.json.encode(diff).decode()
-                _logger.critical(
-                    "Database state modified outside of transaction! "
-                    "This indicates a bug where DB changes occurred without a transaction wrapper.\n"
-                    f"Changes detected:\n{diff_json}"
-                )
-                raise SystemExit(1)
-
-        old_action = self._current_action
-        old_user = self._current_user
-        self._current_action = action
-        # Prefer ctx.user.uuid if ctx provided, otherwise use user param
-        self._current_user = str(ctx.user.uuid) if ctx else user
-        self._in_transaction = True
-        self._transaction_snapshot = current_state
-
-        try:
-            yield
-            current = msgspec.to_builtins(self.db)
-            self._queue_change(
-                self._current_action, self._v, current, self._current_user
-            )
-        except Exception:
-            # Rollback on error: restore from snapshot
-            _logger.warning("Transaction '%s' failed, rolling back changes", action)
-            if self._transaction_snapshot is not None:
-                decoder = msgspec.json.Decoder(DB)
-                self.db = decoder.decode(
-                    msgspec.json.encode(self._transaction_snapshot)
-                )
-                self.db._store = self
-            raise
-        finally:
-            self._current_action = old_action
-            self._current_user = old_user
-            self._in_transaction = False
-            self._transaction_snapshot = None
-
-    async def flush(self) -> None:
-        """Write all pending changes to disk.
-
-        On failure, logs an error and sends SIGTERM to trigger graceful shutdown.
-        """
-        if self._flush_failed or not self._pending_changes:
-            return
-
-        if not self._file.is_open:
-            first_action = self._pending_changes[0].a
-            if first_action not in _BOOTSTRAP_ACTIONS:
-                _logger.error(
-                    "Refusing to create database file with action '%s' - "
-                    "only bootstrap can create a new database",
-                    first_action,
-                )
-                self._flush_failed = True
-                os.kill(os.getpid(), signal.SIGTERM)
-                return
-            # Bootstrap: create and open the file with lock
-            await asyncio.to_thread(self._file.open, self.db_path, create=True)
-
-        changes_to_write = list(self._pending_changes)
-
-        try:
-            lines = [msgspec.json.encode(change) for change in changes_to_write]
-            if not lines:
-                self._pending_changes.clear()
-                return
-
-            await asyncio.to_thread(self._file.write, b"\n".join(lines) + b"\n")
-            self._snapshot.record_lines(len(lines))
-            self._pending_changes.clear()
-        except OSError as e:
-            _logger.error("Failed to flush database: %s", e)
-            self._flush_failed = True
-            os.kill(os.getpid(), signal.SIGTERM)
-
-    def maybe_snapshot(self) -> None:
-        """Write a snapshot if conditions are met."""
-        self._snapshot.maybe_write(self._file, self._v, self._statedict)
-
-    def close(self) -> None:
-        """Release the file lock and close the file."""
-        self._file.close()

paskia/db/lifecycle.py

@@ -3,29 +3,52 @@ Database lifecycle: initialization and maintenance.
 """
 
 import logging
+import os
+import signal
 from datetime import UTC, datetime
 
+from kanta import Kanta
+from kanta.exceptions import DatabaseError
+
 import paskia.db.operations as _ops
 from paskia import oidc_notify
 from paskia.authsession import EXPIRES
-from paskia.db.jsonl import JsonlStore
+from paskia.db.migrations import MigrationCtx
 from paskia.db.paths import db_file_path
+from paskia.db.structs import DB
 
 _logger = logging.getLogger(__name__)
 
 
+def _fatal_error(error: DatabaseError) -> None:
+    """Fatal error callback: terminate the process on background write failures."""
+    _logger.error("Fatal database error: %s", error)
+    os.kill(os.getpid(), signal.SIGTERM)
+
+
 async def init(rp_id: str, *args, **kwargs):
-    """Load database from JSONL file."""
-    if _ops._db._store:
+    """Load database from JSONL file using kanta."""
+    if _ops._store is not None:
         _logger.debug("Database already initialized, skipping reload")
         return
     db_path = db_file_path(rp_id=rp_id, create_root=True)
-    store = JsonlStore(_ops._db, str(db_path))
-    await store.load(str(db_path), rp_id=rp_id)
-    _ops._db = store.db
-    _ops._db._store = store
+    db = DB()
+    kanta = Kanta(
+        str(db_path),
+        db,
+        migrations="paskia.db.migrations",
+        migration_ctx=MigrationCtx(rp_id=rp_id),
+        fatal_error=_fatal_error,
+    )
+    try:
+        await kanta.open()
+    except DatabaseError as e:
+        raise SystemExit(f"{e}") from e
+    _ops._store = kanta
+    _ops._db = db
+    _ops._db._store = kanta
     # Request a snapshot after successful startup
-    store._snapshot.request_force()
+    kanta.request_snapshot()
 
 
 def cleanup_expired() -> int:

paskia/db/logging.py

@@ -14,6 +14,8 @@ import sys
 from typing import TYPE_CHECKING, Any
 from uuid import UUID
 
+from kanta.logging import configure_logging as configure_kanta_logging
+
 if TYPE_CHECKING:
     from paskia.db.structs import DB
 
@@ -464,3 +466,5 @@ def configure_db_logging() -> None:
     logger.addHandler(handler)
     logger.setLevel(logging.INFO)
     logger.propagate = False
+    # Kanta logs changes through its own logger; wire it to the same output.
+    configure_kanta_logging()

paskia/db/migrations.py

@@ -6,17 +6,15 @@ Each migration should be idempotent and only run when needed.
 """
 
 import base64
-from collections.abc import Awaitable, Callable
-
-import msgspec
 
 from paskia.util.crypto import secret_key
 
 
-class MigrationCtx(msgspec.Struct):
+class MigrationCtx:
     """Context passed to each migration function."""
 
-    rp_id: str
+    def __init__(self, rp_id: str):
+        self.rp_id = rp_id
 
 
 def migrate_v1(d: dict, ctx: MigrationCtx) -> None:
@@ -42,7 +40,10 @@ def migrate_v4(d: dict, ctx: MigrationCtx) -> None:
     # Session keys changed to hashes, drop old sessions
     d["sessions"] = {}
     # Create OIDC structure with a generated new key
-    d["oidc"] = {"clients": {}, "key": base64.standard_b64encode(secret_key()).decode()}
+    d["oidc"] = {
+        "clients": {},
+        "key": base64.standard_b64encode(secret_key()).decode(),
+    }
 
 
 def migrate_v5(d: dict, ctx: MigrationCtx) -> None:
@@ -50,38 +51,3 @@ def migrate_v5(d: dict, ctx: MigrationCtx) -> None:
     listen = d["config"].get("listen")
     if listen and isinstance(listen, str):
         d["config"]["listen"] = [listen]
-
-
-migrations = sorted(
-    [f for n, f in globals().items() if n.startswith("migrate_v")],
-    key=lambda f: int(f.__name__.removeprefix("migrate_v")),
-)
-
-DBVER = len(migrations)  # Used by bootstrap to set initial version
-
-
-def apply_migrations_readonly(
-    data_dict: dict,
-    current_version: int,
-    ctx: MigrationCtx,
-) -> int:
-    """Apply migration functions in-place without persistence.
-
-    Returns the new version after all migrations.
-    """
-    while current_version < DBVER:
-        migrations[current_version](data_dict, ctx)
-        current_version += 1
-    return current_version
-
-
-async def apply_all_migrations(
-    data_dict: dict,
-    current_version: int,
-    persist: Callable[[str, int, dict], Awaitable[None]],
-    ctx: MigrationCtx,
-) -> None:
-    while current_version < DBVER:
-        migrations[current_version](data_dict, ctx)
-        current_version += 1
-        await persist(f"migrate:v{current_version}", current_version, data_dict)

paskia/db/operations.py

@@ -12,6 +12,7 @@ from datetime import UTC, datetime, timedelta
 from uuid import UUID
 
 import uuid7
+from kanta import Kanta
 
 from paskia import oidc_notify
 from paskia.config import SESSION_LIFETIME
@@ -38,6 +39,7 @@ _UNSET = object()
 
 # Global database instance (empty until init() loads data)
 _db = DB(config=Config(rp_id="uninitialized.invalid"))
+_store: Kanta[DB] | None = None
 
 
 def is_username_taken(username: str, exclude_uuid: UUID | None = None) -> bool:

paskia/db/snapshot.py

@@ -1,88 +0,0 @@
-"""
-Snapshot handling for JSONL database persistence.
-"""
-
-import logging
-from datetime import UTC, datetime
-from typing import Any
-
-import msgspec
-
-_logger = logging.getLogger(__name__)
-
-LINEPREFIX = b"SNAPSHOT "
-MINDIFFS = 100
-
-
-class Snapshot(msgspec.Struct):
-    """Snapshot data structure for database persistence."""
-
-    ts: datetime
-    v: int
-    state: dict[str, Any]
-
-
-class SnapshotState:
-    """Tracks snapshot timing and line counts for a database file."""
-
-    def __init__(self) -> None:
-        self.ts: datetime | None = None
-        self.changes: int = 0
-        self._force_pending: bool = False
-
-    def request_force(self) -> None:
-        """Request a forced snapshot on the next maybe_write call."""
-        self._force_pending = True
-
-    def record_lines(self, count: int) -> None:
-        self.changes += count
-
-    def maybe_write(self, file, version: int, state: dict) -> None:
-        """Write a snapshot if conditions are met (enough changes, and Sunday UTC or forced)."""
-        if self.changes < MINDIFFS:
-            return
-        force = self._force_pending
-        now = datetime.now(UTC)
-        if not force and now.weekday() != 6:  # 6 = Sunday
-            return
-        sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
-        if not force and self.ts is not None and self.ts >= sunday_midnight:
-            return
-        if not file.is_open:
-            return
-        try:
-            self._write(file, version, state, now)
-            self._force_pending = False
-        except Exception as exc:
-            _logger.error("snapshot: failed to write snapshot: %r", exc)
-
-    def _write(self, file, version: int, state: dict, now: datetime) -> None:
-        """Write a snapshot and update internal state."""
-        data = msgspec.json.encode(Snapshot(ts=now, v=version, state=state))
-        file.write(LINEPREFIX + data + b"\n")
-        self.changes = 0
-        self.ts = now
-
-    @staticmethod
-    def load(data: bytes) -> tuple[Snapshot | None, int]:
-        """Find and parse the last snapshot in file data.
-
-        Returns (snapshot, replay_offset) where replay_offset is the byte
-        position to start replaying change records from. If no valid snapshot
-        is found, returns (None, 0).
-        """
-        marker = b"\n" + LINEPREFIX
-        pos = data.rfind(marker)
-        if pos != -1:
-            pos += 1  # skip the newline
-        elif data.startswith(LINEPREFIX):
-            pos = 0
-        else:
-            return None, 0
-
-        end = data.find(b"\n", pos)
-        if end == -1:
-            raise ValueError("Incomplete snapshot line at end of file")
-
-        snap = msgspec.json.decode(data[pos + len(LINEPREFIX) : end], type=Snapshot)
-        return snap, end + 1

paskia/db/structs.py

@@ -9,6 +9,7 @@ import msgspec
 import uuid7
 
 from paskia import db
+from paskia.db.logging import UuidResolver
 from paskia.util import passphrase as passphrase_util
 from paskia.util.crypto import hash_secret
 
@@ -618,7 +619,7 @@ class Config(msgspec.Struct, omit_defaults=True):
 class DB(msgspec.Struct, dict=True, omit_defaults=False):
     """In-memory database. Access fields directly for reads."""
 
-    config: Config
+    config: Config = msgspec.field(default_factory=lambda: Config(rp_id="localhost"))
     permissions: dict[UUID, Permission] = {}
     orgs: dict[UUID, Org] = {}
     roles: dict[UUID, Role] = {}
@@ -652,8 +653,21 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
             client.uuid = uuid
 
     def transaction(self, action, ctx=None, *, user=None):
-        """Wrap writes in transaction. Delegates to JsonlStore."""
-        return self._store.transaction(action, ctx, user=user)
+        """Wrap writes in transaction. Delegates to Kanta."""
+        user_id = str(ctx.user.uuid) if ctx else user
+        user_display = None
+        if user_id:
+            try:
+                user_uuid = UUID(user_id)
+                if user_uuid in self.users:
+                    user_display = self.users[user_uuid].display_name
+            except (ValueError, KeyError):
+                user_display = user_id
+        previous_state = msgspec.to_builtins(self)
+        resolver = UuidResolver(self, previous_state).resolve
+        return self._store.transaction(
+            action, user=user_id, user_display=user_display, resolver=resolver
+        )
 
     def session_ctx(
         self, session_secret: str, host: str | None = None

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "msgspec>=0.20.0",
     "fastapi-vue>=1.1.0",
     "ua-parser[regex]>=1.0.1",
+    "kanta>=0.1.1",
 ]
 [dependency-groups]
 dev = [

tests/conftest.py

@@ -22,13 +22,13 @@ from uuid import UUID
 import httpx
 import pytest
 import pytest_asyncio
+from kanta import Kanta
 
 import paskia.db.operations as ops_db
 from paskia import globals as paskia_globals
 from paskia.authsession import reset_expires
 from paskia.config import SESSION_LIFETIME
 from paskia.db import (
-    Config,
     Credential,
     Org,
     Permission,
@@ -40,7 +40,7 @@ from paskia.db import (
     create_role,
     create_user,
 )
-from paskia.db.jsonl import JsonlStore
+from paskia.db.migrations import MigrationCtx
 from paskia.db.operations import DB
 from paskia.db.structs import Session
 from paskia.fastapi.mainapp import app
@@ -59,7 +59,7 @@ def event_loop():
 
 @pytest_asyncio.fixture(scope="function")
 async def test_db() -> AsyncGenerator[DB, None]:
-    """Create an in-memory JSON database for testing.
+    """Create a temporary JSONL database for testing using kanta.
 
     Uses bootstrap() to properly initialize the database with:
     - auth:admin and auth:org:admin permissions
@@ -67,18 +67,24 @@ async def test_db() -> AsyncGenerator[DB, None]:
     - An admin user with the Administration role
     """
     with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=True) as f:
-        db = DB(config=Config(rp_id="test.example.com"))
-        store = JsonlStore(db, f.name)
-        db._store = store
-        await store.load()
+        db = DB()
+        kanta = Kanta(
+            f.name,
+            db,
+            migrations="paskia.db.migrations",
+            migration_ctx=MigrationCtx(rp_id="test.example.com"),
+        )
+        await kanta.open()
+        ops_db._store = kanta
         ops_db._db = db
-        ops_db._store = store
+        ops_db._db._store = kanta
         # Bootstrap creates the initial permissions, org, role, and admin user
         bootstrap(
             org_name="Test Organization",
             admin_name="Test Admin",
         )
-        yield db
+        yield ops_db._db
+        await kanta.close()
         ops_db._db = None
         ops_db._store = None