corvix.web.app

Litestar app serving Corvix dashboard data and UI.

Attributes

logger
THEMES
_STATIC_ROOT
_STATIC_ASSETS_DIR
_ASSET_FILENAMES
_ASSET_CACHE_CONTROL
_INDEX_HTML_TEMPLATE
INDEX_HTML
_MEDIA_TYPE_HTML
_LOGIN_HTML
_DEPRECATION_HEADER
_DEPRECATION_HEADER_VALUE
_SSE_DEFAULT_POLL_INTERVAL_SECONDS
_SSE_KEEPALIVE_SECONDS
_snapshot_body_cache
_snapshot_body_cache_lock
_DEPRECATED_HEADERS
_config_cache
_storage_state
app

Classes

_ConfigCache	Mutable container for the module-level AppConfig cache.
_StorageState	Mutable container for the module-level storage backend.

Functions

_asset_version_token(→ str)
index(→ litestar.Response[str])	Serve the dashboard single-page UI.
dashboard_index(→ litestar.Response[str])	Serve the dashboard SPA for bookmarkable dashboard URLs.
_get_auth_secret(→ str)	Return the configured secret, delegating to middleware._get_secret().
login_page(→ litestar.Response[Any])	Serve the login form, or redirect to / when auth is not configured.
login(→ litestar.Response[None])	Validate the submitted token and issue a session cookie on success.
logout(→ litestar.Response[None])	Clear the session cookie and redirect to the login page.
_health_error(→ dict[str, object])
_health_check_staleness(→ dict[str, object])
_health_response(→ litestar.Response[dict[str, object]])
_read_health_poller_status(...)	Resolve the poller status for the health check, or a failure payload.
_health_impl(→ litestar.Response[dict[str, object]])	Compute and return the health check response.
_snapshot_impl(→ corvix.web.schemas.SnapshotResponse)	Compute and return the typed snapshot payload.
_notification_rule_snippets_impl(...)	Compute and return the typed rule-snippets payload.
health(→ litestar.Response[dict[str, object]])	Health endpoint for container checks.
metrics_endpoint(→ litestar.Response[bytes])	Expose Prometheus metrics in text exposition format for scraping.
api_themes(→ dict[str, object])	Return available theme presets.
dashboards(→ dict[str, object])	List configured dashboard names.
snapshot(→ corvix.web.schemas.SnapshotResponse)	Return the selected dashboard data from storage.
_sse_poll_interval(→ float)	Return the server-side SSE poll interval in seconds.
_snapshot_event_body(→ str)	Build the snapshot payload and serialize it to a compact JSON string.
_cached_snapshot_event_body(→ str)	Return the snapshot body for dashboard, reusing a build newer than ttl.
_snapshot_error_payload(→ str)	Serialize an SSE snapshot-error payload for error.
_snapshot_event_generator(...)	Yield SSE messages for dashboard, pushing only on change.
events(→ litestar.response.ServerSentEvent)	Stream dashboard snapshots as Server-Sent Events.
notification_rule_snippets(...)	Return prefilled ignore-rule snippets for a notification.
dismiss_notification(→ litestar.Response[None])	Dismiss a notification thread (removes it from the GitHub inbox).
mark_notification_read(→ litestar.Response[None])	Mark a notification thread as read in GitHub and local storage.
health_deprecated(→ litestar.Response[dict[str, object]])	Deprecated: use /api/v1/health.
api_themes_deprecated(→ litestar.Response[dict[str, ...)	Deprecated: use /api/v1/themes.
dashboards_deprecated(→ litestar.Response[dict[str, ...)	Deprecated: use /api/v1/dashboards.
snapshot_deprecated(...)	Deprecated: use /api/v1/snapshot.
notification_rule_snippets_deprecated(...)	Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/rule-snippets.
dismiss_notification_deprecated(→ litestar.Response[None])	Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/dismiss.
dismiss_notification_default_account(...)	Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/dismiss.
mark_notification_read_deprecated(...)	Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/mark-read.
mark_notification_read_default_account(...)	Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/mark-read.
_dismiss_notification_impl(→ litestar.Response[None])
_mark_notification_read_impl(→ litestar.Response[None])
_require_account(→ corvix.config.GitHubAccountConfig)
_build_github_client(...)
_default_account_id(→ str)
_find_record(→ corvix.domain.NotificationRecord | None)
_yaml_quoted(→ str)
_yaml_scalar(→ str)
_slug_token(→ str)
_rule_name_for_record(→ str)
_rule_match_lines(…)
_context_predicate_lines(→ list[str])
_anchored_title_regex(→ str)
_context_path_value(→ tuple[bool, object | None])
_dashboard_ignore_rule_snippet(→ str)
_global_exclude_rule_snippet(→ str)
set_storage_backend(→ None)	Inject the storage backend used by route handlers.
_get_storage(→ corvix.storage.StorageBackend)	Return the injected backend, or lazily build PostgreSQL storage from config.
_clear_config_cache(→ None)	Discard the cached AppConfig so the next request reloads from disk.
_load_runtime_config(→ corvix.config.AppConfig)	Return the cached AppConfig, re-parsing from disk only when the file changes.
_install_sighup_handler(→ None)	Register a SIGHUP handler that clears the config cache.
_select_dashboard(→ corvix.config.DashboardSpec)
_dashboard_names(→ list[str])
_configure_observability(→ None)	Configure structured logging and optional tracing at app startup.
run(→ None)	Run app with uvicorn.

Module Contents

corvix.web.app.logger

corvix.web.app.THEMES: dict[str, dict[str, str]]

corvix.web.app._STATIC_ROOT

corvix.web.app._STATIC_ASSETS_DIR = ''

corvix.web.app._ASSET_FILENAMES = ('app.js', 'index.css', 'favicon.svg')

corvix.web.app._ASSET_CACHE_CONTROL

corvix.web.app._asset_version_token() → str

corvix.web.app._INDEX_HTML_TEMPLATE

corvix.web.app.INDEX_HTML

corvix.web.app._MEDIA_TYPE_HTML = 'text/html'

corvix.web.app._LOGIN_HTML = Multiline-String

Show Value

"""<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Corvix — Sign in</title>
  <style>
    *, *::before, *::after { box-sizing: border-box; margin: 0; padding: 0; }
    body {
      font-family: system-ui, sans-serif;
      display: flex; height: 100vh;
      align-items: center; justify-content: center;
      background: #07111f; color: #edf3ff;
    }
    form {
      display: flex; flex-direction: column; gap: 1rem;
      min-width: 300px; padding: 2rem;
      background: #0e1a2b; border-radius: 8px;
    }
    h2 { font-size: 1.4rem; color: #74c0fc; }
    input[type=password] {
      padding: .65rem .9rem; border-radius: 6px;
      border: 1px solid #223753; background: #132238;
      color: #edf3ff; font-size: 1rem;
    }
    input[type=password]:focus { outline: 2px solid #74c0fc; border-color: transparent; }
    button {
      padding: .65rem .9rem; border-radius: 6px; border: none;
      background: #74c0fc; color: #07111f;
      font-size: 1rem; font-weight: 600; cursor: pointer;
    }
    button:hover { background: #a5d4ff; }
  </style>
</head>
<body>
  <form method="post" action="/login">
    <h2>Corvix</h2>
    <input type="password" name="token" placeholder="Secret token" required autofocus>
    <button type="submit">Sign in</button>
  </form>
</body>
</html>"""

corvix.web.app.index() → litestar.Response[str]

Serve the dashboard single-page UI.

corvix.web.app.dashboard_index(dashboard_name: str) → litestar.Response[str]

Serve the dashboard SPA for bookmarkable dashboard URLs.

corvix.web.app._get_auth_secret() → str

Return the configured secret, delegating to middleware._get_secret().

Using the shared implementation ensures consistent TTL caching, memoized misconfiguration logging, and _FILE support in one place.

corvix.web.app.login_page() → litestar.Response[Any]

Serve the login form, or redirect to / when auth is not configured.

async corvix.web.app.login(request: litestar.Request) → litestar.Response[None]

Validate the submitted token and issue a session cookie on success.

The Secure attribute is set when the request arrived over HTTPS. The scheme is read from request.url.scheme which reflects the real protocol when uvicorn is started with --proxy-headers (trusting X-Forwarded-Proto from a reverse proxy). This avoids raw header inspection from application code, which can be spoofed by untrusted clients not behind a proxy.

corvix.web.app.logout() → litestar.Response[None]

Clear the session cookie and redirect to the login page.

corvix.web.app._health_error(poller_status: corvix.domain.PollerStatus) → dict[str, object]

corvix.web.app._health_check_staleness(last_poll_str: str) → dict[str, object]

corvix.web.app._health_response(payload: dict[str, object]) → litestar.Response[dict[str, object]]

corvix.web.app._read_health_poller_status() → corvix.domain.PollerStatus | dict[str, object]

Resolve the poller status for the health check, or a failure payload.

corvix.web.app._DEPRECATION_HEADER = 'Deprecation'

corvix.web.app._DEPRECATION_HEADER_VALUE = 'true'

corvix.web.app._health_impl(extra_headers: dict[str, str] | None = None) → litestar.Response[dict[str, object]]

Compute and return the health check response.

corvix.web.app._snapshot_impl(dashboard: str | None = None) → corvix.web.schemas.SnapshotResponse

Compute and return the typed snapshot payload.

corvix.web.app._notification_rule_snippets_impl(account_id: str, thread_id: str, dashboard: str | None = None) → corvix.web.schemas.RuleSnippetsResponse

Compute and return the typed rule-snippets payload.

corvix.web.app.health() → litestar.Response[dict[str, object]]

Health endpoint for container checks.

Returns 200 with {“status”: “ok”} when config and storage are readable, the poller is running, and the poller’s last poll time is not stale.

Returns 503 with {“status”: “unhealthy”} and one of these reasons: “config_unavailable”, “storage_unavailable”, “invalid_cache”, “poller_not_running”, “poller_error”, “invalid_poll_time”, or “stale”.

corvix.web.app.metrics_endpoint() → litestar.Response[bytes]

Expose Prometheus metrics in text exposition format for scraping.

corvix.web.app.api_themes() → dict[str, object]

Return available theme presets.

corvix.web.app.dashboards() → dict[str, object]

List configured dashboard names.

corvix.web.app.snapshot(dashboard: str | None = None) → corvix.web.schemas.SnapshotResponse

Return the selected dashboard data from storage.

corvix.web.app._SSE_DEFAULT_POLL_INTERVAL_SECONDS = 3.0

corvix.web.app._SSE_KEEPALIVE_SECONDS = 20.0

corvix.web.app._sse_poll_interval() → float

Return the server-side SSE poll interval in seconds.

Read from CORVIX_SSE_POLL_INTERVAL_SECONDS; falls back to the default when unset, non-numeric, or non-positive.

corvix.web.app._snapshot_event_body(dashboard: str | None) → str

Build the snapshot payload and serialize it to a compact JSON string.

Uses msgspec (the encoder Litestar uses for the equivalent HTTP route) so the SSE body and the GET /api/v1/snapshot response share one serialization path and stay byte-for-byte consistent.

corvix.web.app._snapshot_body_cache: dict[str | None, tuple[float, str]]

corvix.web.app._snapshot_body_cache_lock

corvix.web.app._cached_snapshot_event_body(dashboard: str | None, ttl: float) → str

Return the snapshot body for dashboard, reusing a build newer than ttl.

Within a ttl-second window (one poll interval) concurrent SSE connections watching the same dashboard share a single storage read and serialization. A strict age < ttl comparison means a lone connection, whose ticks are spaced one interval apart, still rebuilds every tick — so the cache adds no latency in the common single-client case.

corvix.web.app._snapshot_error_payload(error: Exception) → str

Serialize an SSE snapshot-error payload for error.

HTTPException carries a client-safe detail and status code; any other exception is reported generically (its message is not leaked to the client).

async corvix.web.app._snapshot_event_generator(dashboard: str | None) → collections.abc.AsyncIterator[litestar.response.ServerSentEventMessage]

Yield SSE messages for dashboard, pushing only on change.

Emits a snapshot event whenever the serialized payload differs from the last one sent, a snapshot-error event when the payload cannot be produced, and a comment-only keep-alive when nothing has changed for a while (so proxies do not drop an idle connection). The blocking storage read runs in a worker thread to avoid stalling the event loop.

Any error building the snapshot is reported to the client and the stream keeps running, recovering on a later tick; this avoids tearing down the connection (and triggering a client reconnection storm) on a transient storage or serialization failure.

async corvix.web.app.events(dashboard: str | None = None) → litestar.response.ServerSentEvent

Stream dashboard snapshots as Server-Sent Events.

Replaces fixed-interval client polling: the connection stays open and the server pushes a snapshot event only when the data changes, cutting both latency and per-cycle overhead when nothing has happened.

corvix.web.app.notification_rule_snippets(account_id: str, thread_id: str, dashboard: str | None = None) → corvix.web.schemas.RuleSnippetsResponse

Return prefilled ignore-rule snippets for a notification.

corvix.web.app.dismiss_notification(account_id: str, thread_id: str) → litestar.Response[None]

Dismiss a notification thread (removes it from the GitHub inbox).

Calls DELETE /notifications/threads/{id} on GitHub, then marks the record as dismissed in local storage. Returns 204 No Content on success.

corvix.web.app.mark_notification_read(account_id: str, thread_id: str) → litestar.Response[None]

Mark a notification thread as read in GitHub and local storage.

corvix.web.app._DEPRECATED_HEADERS

corvix.web.app.health_deprecated() → litestar.Response[dict[str, object]]

Deprecated: use /api/v1/health.

corvix.web.app.api_themes_deprecated() → litestar.Response[dict[str, object]]

Deprecated: use /api/v1/themes.

corvix.web.app.dashboards_deprecated() → litestar.Response[dict[str, object]]

Deprecated: use /api/v1/dashboards.

corvix.web.app.snapshot_deprecated(dashboard: str | None = None) → litestar.Response[corvix.web.schemas.SnapshotResponse]

Deprecated: use /api/v1/snapshot.

corvix.web.app.notification_rule_snippets_deprecated(account_id: str, thread_id: str, dashboard: str | None = None) → litestar.Response[corvix.web.schemas.RuleSnippetsResponse]

Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/rule-snippets.

corvix.web.app.dismiss_notification_deprecated(account_id: str, thread_id: str) → litestar.Response[None]

Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/dismiss.

corvix.web.app.dismiss_notification_default_account(thread_id: str) → litestar.Response[None]

Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/dismiss.

corvix.web.app.mark_notification_read_deprecated(account_id: str, thread_id: str) → litestar.Response[None]

Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/mark-read.

corvix.web.app.mark_notification_read_default_account(thread_id: str) → litestar.Response[None]

Deprecated: use /api/v1/notifications/{account_id}/{thread_id}/mark-read.

corvix.web.app._dismiss_notification_impl(account_id: str, thread_id: str) → litestar.Response[None]

corvix.web.app._mark_notification_read_impl(account_id: str, thread_id: str) → litestar.Response[None]

corvix.web.app._require_account(config: corvix.config.AppConfig, account_id: str) → corvix.config.GitHubAccountConfig

corvix.web.app._build_github_client(config: corvix.config.AppConfig, account: corvix.config.GitHubAccountConfig, token: str) → corvix.ingestion.GitHubNotificationsClient

corvix.web.app._default_account_id(config: corvix.config.AppConfig) → str

corvix.web.app._find_record(*, records: list[corvix.domain.NotificationRecord], account_id: str, thread_id: str) → corvix.domain.NotificationRecord | None

corvix.web.app._yaml_quoted(value: str) → str

corvix.web.app._yaml_scalar(value: object) → str

corvix.web.app._slug_token(value: str) → str

corvix.web.app._rule_name_for_record(record: corvix.domain.NotificationRecord) → str

corvix.web.app._rule_match_lines(*, record: corvix.domain.NotificationRecord, include_context: Literal[False]) → list[str]

corvix.web.app._rule_match_lines(*, record: corvix.domain.NotificationRecord, include_context: Literal[True]) → list[str] | None

corvix.web.app._context_predicate_lines(*, record: corvix.domain.NotificationRecord) → list[str]

corvix.web.app._anchored_title_regex(title: str) → str

corvix.web.app._context_path_value(*, context: collections.abc.Mapping[str, object], path: str) → tuple[bool, object | None]

corvix.web.app._dashboard_ignore_rule_snippet(match_lines: list[str]) → str

corvix.web.app._global_exclude_rule_snippet(*, record: corvix.domain.NotificationRecord, match_lines: list[str]) → str

class corvix.web.app._ConfigCache

Mutable container for the module-level AppConfig cache.

config: corvix.config.AppConfig | None = None

path: str | None = None

mtime: float | None = None

corvix.web.app._config_cache

class corvix.web.app._StorageState

Mutable container for the module-level storage backend.

backend: corvix.storage.StorageBackend | None = None

lock: threading.Lock

corvix.web.app._storage_state

corvix.web.app.set_storage_backend(backend: corvix.storage.StorageBackend | None) → None

Inject the storage backend used by route handlers.

Production wiring leaves this unset and the backend is built lazily from config (PostgreSQL is required). Tests inject a backend directly and reset it to None afterwards.

corvix.web.app._get_storage() → corvix.storage.StorageBackend

Return the injected backend, or lazily build PostgreSQL storage from config.

The built backend is cached so its connection pool is reused across requests. Building is guarded by a lock so concurrent first requests don’t each create (and leak) a connection pool. Raises HTTPException (500) when no database is configured.

corvix.web.app._clear_config_cache() → None

Discard the cached AppConfig so the next request reloads from disk.

corvix.web.app._load_runtime_config() → corvix.config.AppConfig

Return the cached AppConfig, re-parsing from disk only when the file changes.

Config is read from the path in the CORVIX_CONFIG environment variable (default: corvix.yaml). The file’s mtime is checked on every call; the YAML is only re-parsed when either the path or the mtime differs from the last successful load, eliminating redundant I/O on every request.

corvix.web.app._install_sighup_handler() → None

Register a SIGHUP handler that clears the config cache.

Sending SIGHUP to the server process forces the config to be reloaded from disk on the next request without restarting the process:

kill -HUP <pid>

The handler is a no-op on platforms that do not support SIGHUP (e.g. Windows).

corvix.web.app._select_dashboard(dashboards: list[corvix.config.DashboardSpec], selected_name: str | None) → corvix.config.DashboardSpec

corvix.web.app._dashboard_names(dashboards: list[corvix.config.DashboardSpec]) → list[str]

corvix.web.app._configure_observability() → None

Configure structured logging and optional tracing at app startup.

Runs as a Litestar startup hook so it applies whether the app is launched via corvix serve or directly through uvicorn corvix.web.app:app.

corvix.web.app.app

corvix.web.app.run() → None

Run app with uvicorn.