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

2026-06-12 19:39:30 +00:00
11 changed files with 121 additions and 742 deletions
@@ -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)
+            await asyncio.sleep(CLEANUP_INTERVAL)
-            # Flush pending changes to disk
+            cleanup_expired()
            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()
        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:
+                if loop is task_loop:
                    _logger.debug("Background task in different event loop, restarting")
                    _background_task = None
                else:
                    # 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
@@ -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()}")
@@ -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.migrations import MigrationCtx
-from paskia.db.logging import log_change
+from paskia.db.structs import DB, Config
 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
 _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()))
+        rr = replay_jsonl(content)
-        data_dict = r.state
+        data_dict = rr.state
-        version = r.v
+        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)
            return msgspec.json.decode(msgspec.json.encode(data_dict), type=DB)
        except msgspec.ValidationError as e:
            raise DatabaseError(f"{path.resolve()}: {e}") from None
    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()
@@ -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."""
+    """Load database from JSONL file using kanta."""
-    if _ops._db._store:
+    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))
+    db = DB()
-    await store.load(str(db_path), rp_id=rp_id)
+    kanta = Kanta(
-    _ops._db = store.db
+        str(db_path),
-    _ops._db._store = store
+        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:
@@ -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()
@@ -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)
@@ -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:
@@ -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
@@ -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."""
+        """Wrap writes in transaction. Delegates to Kanta."""
-        return self._store.transaction(action, ctx, user=user)
+        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
@@ -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 = [
@@ -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"))
+        db = DB()
-        store = JsonlStore(db, f.name)
+        kanta = Kanta(
-        db._store = store
+            f.name,
-        await store.load()
+            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