corvix.storage

Local persistence for polled notifications.

The JSON cache uses fcntl advisory locks and is therefore supported on Linux/POSIX platforms only.

Attributes

_NOTIFICATION_RECORD_COLUMNS
_DISMISSED_ROW_COLUMNS
_POLLER_STATUS_COLUMNS
SINGLE_USER_ID

Exceptions

StorageConfigError

Raised when required storage configuration (a database URL) is missing.

Classes

StorageBackend	Protocol for notification persistence backends.
NotificationCache	Read/write notification snapshots to a JSON file.
PostgresStorage	PostgreSQL-backed notification persistence implementing StorageBackend.

Functions

create_storage(→ PostgresStorage)	Return the configured PostgreSQL storage backend.
_fsync_directory(→ None)	Best-effort fsync of a directory after atomic replacement.
_coerce_context(→ dict[str, object])
_coerce_string_key_dict(→ dict[str, object] | None)
_require_str(→ str)
_optional_str(→ str | None)
_require_bool(→ bool)
_require_float(→ float)
_require_datetime(→ datetime.datetime)
_parse_account_errors(...)	Parse the account_errors JSONB value into a tuple of AccountError.
_coerce_str_list(→ list[str])

Module Contents

corvix.storage._NOTIFICATION_RECORD_COLUMNS = 18

corvix.storage._DISMISSED_ROW_COLUMNS = 2

corvix.storage._POLLER_STATUS_COLUMNS = 5

corvix.storage.SINGLE_USER_ID: uuid.UUID

exception corvix.storage.StorageConfigError

Bases: RuntimeError

Raised when required storage configuration (a database URL) is missing.

class corvix.storage.StorageBackend

Bases: Protocol

Protocol for notification persistence backends.

save_records(records: list[corvix.domain.NotificationRecord], generated_at: datetime.datetime) → None

load_records() → tuple[datetime.datetime | None, list[corvix.domain.NotificationRecord]]

save_status(status: corvix.domain.PollerStatus) → None

load_status() → corvix.domain.PollerStatus

dismiss_record(thread_id: str, account_id: str = 'primary') → None

mark_record_read(thread_id: str, account_id: str = 'primary') → None

get_dismissed_notification_keys() → list[str]

get_dismissed_thread_ids() → list[str]

close() → None

__enter__() → StorageBackend

__exit__(*args: object) → None

corvix.storage.create_storage(config: corvix.config.AppConfig) → PostgresStorage

Return the configured PostgreSQL storage backend.

PostgreSQL is required in all deployments; the JSON cache is no longer used as the shared store between the poller and the web service. Raises StorageConfigError when no database URL is configured.

class corvix.storage.NotificationCache

Read/write notification snapshots to a JSON file.

Implements StorageBackend using a single-user JSON file. The user_id parameter in protocol methods is ignored — the file is shared.

path: pathlib.Path

static _opt_str(value: object) → str | None

save(records: list[corvix.domain.NotificationRecord], generated_at: datetime.datetime | None = None, *, poller_status: corvix.domain.PollerStatus | None = None) → None

Persist records to disk.

load() → tuple[datetime.datetime | None, list[corvix.domain.NotificationRecord]]

Load snapshot from disk if available.

save_status(status: corvix.domain.PollerStatus) → None

Persist only the poller status without touching notifications.

load_status() → corvix.domain.PollerStatus

Load the poller status from the cache file.

_load_raw_generated_at() → str | None

_load_status_unlocked() → corvix.domain.PollerStatus

_load_unlocked() → tuple[datetime.datetime | None, list[corvix.domain.NotificationRecord]]

Load snapshot from disk without acquiring a file lock.

_save_unlocked(records: list[corvix.domain.NotificationRecord], generated_at: datetime.datetime, *, poller_status: corvix.domain.PollerStatus | None = None) → None

_exclusive_lock() → collections.abc.Iterator[None]

_shared_lock() → collections.abc.Iterator[None]

Acquire a shared read lock when possible; fall through unlocked otherwise.

When the lock file cannot be created (read-only filesystem, missing permissions), no concurrent writer can be holding the exclusive lock from this process either, so reads proceed without it. Torn reads remain prevented under normal operation where both reader and writer can open the lock file.

save_records(records: list[corvix.domain.NotificationRecord], generated_at: datetime.datetime) → None

Save records to the JSON cache.

load_records() → tuple[datetime.datetime | None, list[corvix.domain.NotificationRecord]]

Load records from the JSON cache.

dismiss_record(thread_id: str, account_id: str = 'primary') → None

Mark a record as dismissed by account/thread id in the JSON file.

mark_record_read(thread_id: str, account_id: str = 'primary') → None

Mark a record as read by account/thread id in the JSON file.

get_dismissed_notification_keys() → list[str]

Return account-scoped keys of dismissed records.

get_dismissed_thread_ids() → list[str]

Return thread IDs of dismissed records.

close() → None

No-op; the JSON backend holds no long-lived resources.

__enter__() → NotificationCache

__exit__(*_args: object) → None

class corvix.storage.PostgresStorage

PostgreSQL-backed notification persistence implementing StorageBackend.

Uses psycopg (sync) so it is safe to use from CLI commands and the synchronous Litestar route handlers (sync_to_thread=False is not used with this backend — callers should run in a thread pool if needed).

A psycopg_pool.ConnectionPool is created at construction time so that TCP connections are reused across method calls rather than being opened and torn down per operation. Call close() when the storage is no longer needed, or use it as a context manager:

with PostgresStorage(connection_string=url) as storage:
    storage.save_records(...)

connection_string: str

min_pool_size: int = 1

max_pool_size: int = 10

_pool: psycopg_pool.ConnectionPool[psycopg.Connection[tuple[object, Ellipsis]]] | None = None

__post_init__() → None

close() → None

Close all pooled connections and release resources.

__enter__() → PostgresStorage

__exit__(*_args: object) → None

_connect() → contextlib.AbstractContextManager[psycopg.Connection[tuple[object, Ellipsis]]]

Return a pooled connection context-manager.

Usage is identical to the previous psycopg.connect() call:

with self._connect() as conn:
    ...

save_records(records: list[corvix.domain.NotificationRecord], generated_at: datetime.datetime) → None

Upsert records. Preserves dismissed flag on conflict.

load_records() → tuple[datetime.datetime | None, list[corvix.domain.NotificationRecord]]

Load all records ordered by snapshot_at descending.

save_status(status: corvix.domain.PollerStatus) → None

Upsert the poller status row.

load_status() → corvix.domain.PollerStatus

Load the poller status, defaulting to unknown.

dismiss_record(thread_id: str, account_id: str = 'primary') → None

Set dismissed=true for a specific account/thread id.

mark_record_read(thread_id: str, account_id: str = 'primary') → None

Set unread=false for a specific account/thread id.

get_dismissed_notification_keys() → list[str]

Return account-scoped keys where dismissed=true.

get_dismissed_thread_ids() → list[str]

Return thread IDs of dismissed records.

corvix.storage._fsync_directory(path: pathlib.Path) → None

Best-effort fsync of a directory after atomic replacement.

corvix.storage._coerce_context(value: object) → dict[str, object]

corvix.storage._coerce_string_key_dict(value: object) → dict[str, object] | None

corvix.storage._require_str(value: object, field: str) → str

corvix.storage._optional_str(value: object, field: str) → str | None

corvix.storage._require_bool(value: object, field: str) → bool

corvix.storage._require_float(value: object, field: str) → float

corvix.storage._require_datetime(value: object, field: str) → datetime.datetime

corvix.storage._parse_account_errors(value: object) → tuple[corvix.domain.AccountError, Ellipsis]

Parse the account_errors JSONB value into a tuple of AccountError.

corvix.storage._coerce_str_list(value: object, field: str) → list[str]