diff --git a/paskia/db/background.py b/paskia/db/background.py index ef03c6c..bb75215 100644 --- a/paskia/db/background.py +++ b/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 diff --git a/paskia/db/filelock.py b/paskia/db/filelock.py deleted file mode 100644 index 874f1f9..0000000 --- a/paskia/db/filelock.py +++ /dev/null @@ -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()}") diff --git a/paskia/db/jsonl.py b/paskia/db/jsonl.py index 95b862e..a51e755 100644 --- a/paskia/db/jsonl.py +++ b/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() diff --git a/paskia/db/lifecycle.py b/paskia/db/lifecycle.py index e7607f8..c1474e6 100644 --- a/paskia/db/lifecycle.py +++ b/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: diff --git a/paskia/db/logging.py b/paskia/db/logging.py index f5fb4b0..a48a54d 100644 --- a/paskia/db/logging.py +++ b/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() diff --git a/paskia/db/migrations.py b/paskia/db/migrations.py index 4b2b9a4..8b1ca61 100644 --- a/paskia/db/migrations.py +++ b/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) diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 08a50d9..69fd38e 100644 --- a/paskia/db/operations.py +++ b/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: diff --git a/paskia/db/snapshot.py b/paskia/db/snapshot.py deleted file mode 100644 index 8a14f16..0000000 --- a/paskia/db/snapshot.py +++ /dev/null @@ -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 diff --git a/paskia/db/structs.py b/paskia/db/structs.py index fb176e1..03f4833 100644 --- a/paskia/db/structs.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 522630a..6ddc3d1 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/tests/conftest.py b/tests/conftest.py index 019f68c..01bf2a7 100644 --- a/tests/conftest.py +++ b/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