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

Classes

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

Functions

_fsync_directory(→ None)	Best-effort fsync of a directory after atomic replacement.
_coerce_context(→ dict[str, object])
_require_str(→ str)
_optional_str(→ str | None)
_require_bool(→ bool)
_require_float(→ float)
_require_datetime(→ datetime.datetime)
_coerce_str_list(→ list[str])

Module Contents

corvix.storage._NOTIFICATION_RECORD_COLUMNS = 18

corvix.storage._DISMISSED_ROW_COLUMNS = 2

class corvix.storage.StorageBackend

Bases: Protocol

Protocol for notification persistence backends.

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

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

dismiss_record(user_id: corvix.types.UserId, thread_id: str, account_id: str = 'primary') → None

mark_record_read(user_id: corvix.types.UserId, thread_id: str, account_id: str = 'primary') → None

get_dismissed_notification_keys(user_id: corvix.types.UserId) → list[str]

get_dismissed_thread_ids(user_id: corvix.types.UserId) → list[str]

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

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

Persist records to disk.

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

Load snapshot from disk if available.

_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) → None

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

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

Save records; user_id ignored in single-user mode.

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

Load records; user_id ignored in single-user mode.

dismiss_record(user_id: corvix.types.UserId, thread_id: str, account_id: str = 'primary') → None

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

mark_record_read(user_id: corvix.types.UserId, 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(user_id: corvix.types.UserId) → list[str]

Return account-scoped keys of dismissed records.

get_dismissed_thread_ids(user_id: corvix.types.UserId) → list[str]

Backward-compatible API returning only thread IDs.

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).

connection_string: str

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

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

Upsert records for user_id. Preserves dismissed flag on conflict.

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

Load all records for user_id ordered by snapshot_at descending.

dismiss_record(user_id: corvix.types.UserId, thread_id: str, account_id: str = 'primary') → None

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

mark_record_read(user_id: corvix.types.UserId, thread_id: str, account_id: str = 'primary') → None

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

get_dismissed_notification_keys(user_id: corvix.types.UserId) → list[str]

Return account-scoped keys where dismissed=true for user_id.

get_dismissed_thread_ids(user_id: corvix.types.UserId) → list[str]

Backward-compatible API returning only thread IDs.

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._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._coerce_str_list(value: object, field: str) → list[str]