feat(analytics): add permission-gated login-activity read api

CLAUDE.md

@@ -113,6 +113,7 @@ make frontend.build.api  # Regenerate frontend/lib/api/generated.ts from the bac
 
 # Database
 make migrations         # Apply Alembic migrations for both app and analytics databases (native)
+make analytics-backfill # Full-refresh TimescaleDB continuous aggregates (run once after an analytics migration adds one; no-op on SQLite). Pass -- --name <cagg> to target one
 make services.logs      # Tail logs for the service containers (make services.logs [service])
 make db-shell           # PostgreSQL shell
 make db-shell-analytics  # PostgreSQL shell on the analytics database
@@ -218,6 +219,15 @@ both. Generate an analytics migration with
 databases never share metadata: app models use `SQLModel.metadata`, analytics
 tables use `sparkth.core.analytics.models.analytics_metadata`.
 
+**Continuous aggregates need a one-off backfill after migrating.** A TimescaleDB
+continuous aggregate is created `WITH NO DATA` (creating it with data would backfill
+inside Alembic's transaction), and its refresh policy only covers a trailing window —
+so once the first policy run advances the materialization watermark, buckets older than
+that window vanish from the view and pre-migration history is lost. After applying an
+analytics migration that adds a continuous aggregate, run `make analytics-backfill` once
+on PostgreSQL to full-refresh it (`refresh_continuous_aggregate` over the whole range).
+It is idempotent and a no-op on SQLite.
+
 ### Preventing Split Heads
 
 Multiple Alembic heads occur when two branches each generate a migration from the same parent revision and merge independently. Before creating a new migration, always check for existing heads:

Makefile

@@ -58,6 +58,10 @@ migrations: ## Apply Alembic migrations for both the app and analytics databases
 	uv run alembic upgrade head
 	uv run alembic -c alembic_analytics.ini upgrade head
 
+.PHONY: analytics-backfill
+analytics-backfill: ## Backfill continuous aggregates full history (run once after migrating on Postgres; no-op on SQLite)
+	uv run python -m sparkth.cli.main analytics backfill-aggregates $(ARGS)
+
 .PHONY: rag-cleanup
 rag-cleanup: ## Run the RAG cleanup task (native)
 	uv run python -m sparkth.rag.cleanup

frontend/lib/api/generated.ts

@@ -4,6 +4,29 @@
  */
 
 export interface paths {
+    "/api/v1/analytics/login-activity": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /**
+         * Login Activity
+         * @description Return daily login counts (newest first) for the last ``days`` calendar days.
+         *
+         *     Days with no logins are omitted from the series (no zero-fill); consumers must
+         *     tolerate gaps.
+         */
+        get: operations["login_activity_api_v1_analytics_login_activity_get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/v1/auth/google/authorize": {
         parameters: {
             query?: never;
@@ -1591,6 +1614,16 @@ export interface components {
             /** Name */
             name?: string | null;
         };
+        /**
+         * LoginActivityPoint
+         * @description One day's login count. ``day`` is an ISO ``YYYY-MM-DD`` string.
+         */
+        LoginActivityPoint: {
+            /** Day */
+            day: string;
+            /** Login Count */
+            login_count: number;
+        };
         /** LogsResponse */
         LogsResponse: {
             /**
@@ -1944,6 +1977,37 @@ export interface components {
 }
 export type $defs = Record<string, never>;
 export interface operations {
+    login_activity_api_v1_analytics_login_activity_get: {
+        parameters: {
+            query?: {
+                days?: number;
+            };
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["LoginActivityPoint"][];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     google_authorize_api_v1_auth_google_authorize_get: {
         parameters: {
             query?: never;

sparkth/api/v1/analytics.py

@@ -0,0 +1,34 @@
+"""Analytics read API — dashboards call these endpoints; they read rollups only.
+
+Never touches the application database or raw events beyond the analytics read
+functions. Every endpoint is gated by the ``analytics.read`` permission. (Analytics
+is emitted server-side via ``ingest_event`` — validated against the schema on the
+``ANALYTICS_EVENTS`` hook; there is no HTTP emission endpoint — this read router is
+the analytics HTTP surface.)
+"""
+
+from fastapi import APIRouter, Depends, Query
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from sparkth.lib.analytics import LoginActivityPoint, get_login_activity
+from sparkth.lib.db import get_analytics_session
+from sparkth.lib.permissions import ANALYTICS_READ
+
+router = APIRouter()
+
+
+@router.get(
+    "/login-activity",
+    response_model=list[LoginActivityPoint],
+    dependencies=[Depends(ANALYTICS_READ.require_in_global_scope())],
+)
+async def login_activity(
+    days: int = Query(default=30, ge=1, le=365),
+    session: AsyncSession = Depends(get_analytics_session),
+) -> list[LoginActivityPoint]:
+    """Return daily login counts (newest first) for the last ``days`` calendar days.
+
+    Days with no logins are omitted from the series (no zero-fill); consumers must
+    tolerate gaps.
+    """
+    return await get_login_activity(session, days)

sparkth/api/v1/api.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 
-from sparkth.api.v1 import auth, file_parser, llm, permissions, user, user_plugins, whitelist
+from sparkth.api.v1 import analytics, auth, file_parser, llm, permissions, user, user_plugins, whitelist
 
 api_router = APIRouter()
 api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
@@ -10,3 +10,4 @@
 api_router.include_router(whitelist.router, prefix="/whitelist", tags=["Whitelist"])
 api_router.include_router(llm.router, prefix="/llm", tags=["LLM Configuration"])
 api_router.include_router(permissions.router, prefix="/permissions", tags=["Permissions"])
+api_router.include_router(analytics.router, prefix="/analytics", tags=["Analytics"])

sparkth/cli/analytics.py

@@ -0,0 +1,39 @@
+import asyncio
+
+import typer
+
+from sparkth.lib.analytics import ContinuousAggregateNotFound, backfill_continuous_aggregates
+
+app = typer.Typer(help="Analytics maintenance commands")
+
+
+@app.command("backfill-aggregates")
+def backfill_aggregates_command(
+    name: str | None = typer.Option(
+        None,
+        "--name",
+        help="Refresh only this continuous aggregate (default: refresh all).",
+    ),
+) -> None:
+    """Backfill the full history of TimescaleDB continuous aggregates.
+
+    Run once after an aggregate's migration is applied on PostgreSQL/TimescaleDB:
+    aggregates are created empty and their refresh policies only cover a trailing window,
+    so without this one-off full refresh any pre-migration history is lost. Idempotent.
+    On a non-PostgreSQL analytics database (SQLite in tests/e2e) it is a no-op.
+    """
+    try:
+        refreshed = asyncio.run(backfill_continuous_aggregates(name))
+    except ContinuousAggregateNotFound as exc:
+        typer.secho(str(exc), fg=typer.colors.RED)
+        raise typer.Exit(code=1) from None
+
+    if refreshed is None:
+        typer.secho(
+            "Skipped: analytics database is not PostgreSQL/TimescaleDB; nothing to refresh.",
+            fg=typer.colors.YELLOW,
+        )
+    elif not refreshed:
+        typer.secho("No continuous aggregates registered; nothing to refresh.", fg=typer.colors.YELLOW)
+    else:
+        typer.secho(f"Refreshed {len(refreshed)} aggregate(s): {', '.join(refreshed)}.", fg=typer.colors.GREEN)

sparkth/cli/main.py

@@ -1,12 +1,13 @@
 import typer
 
-from sparkth.cli import roles, users
+from sparkth.cli import analytics, roles, users
 from sparkth.lib.log import configure_logging
 
 app = typer.Typer(help="Root command for all CLI tools")
 
 app.add_typer(users.app, name="users")
 app.add_typer(roles.app, name="roles")
+app.add_typer(analytics.app, name="analytics")
 
 
 def main() -> None:

sparkth/core/analytics/exceptions.py

@@ -33,3 +33,11 @@ def __init__(self, plugin_name: str, event_type: str) -> None:
         super().__init__(
             f"Plugin '{plugin_name}' registered event '{event_type}', which is not namespaced under '{plugin_name}.'"
         )
+
+
+class ContinuousAggregateNotFound(Exception):
+    """Raised when a backfill targets a continuous aggregate that does not exist."""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+        super().__init__(f"No continuous aggregate named '{name}' exists in the analytics database")

sparkth/core/analytics/maintenance.py

@@ -0,0 +1,75 @@
+"""Administrative maintenance operations over the analytics database.
+
+Unlike the read path (:mod:`sparkth.core.analytics.reads`), which only issues
+``SELECT``s, this module runs Timescale-specific administrative procedures — currently
+the one-off backfill of continuous aggregates. Exposed to the CLI via
+``sparkth.lib.analytics``.
+"""
+
+from sqlalchemy import text
+
+from sparkth.core.analytics.db import get_analytics_engine
+from sparkth.core.analytics.exceptions import ContinuousAggregateNotFound
+from sparkth.lib.log import get_logger
+
+logger = get_logger(__name__)
+
+# Every continuous aggregate registered with TimescaleDB, so the backfill covers all of
+# them (present and future) rather than a hard-coded list that drifts as rollups are added.
+_LIST_CONTINUOUS_AGGREGATES = text(
+    "SELECT view_name FROM timescaledb_information.continuous_aggregates ORDER BY view_name"
+)
+
+# ``refresh_continuous_aggregate`` is a TimescaleDB stored procedure (invoked with CALL,
+# not SELECT) and cannot run inside a transaction block — so it is executed on an
+# AUTOCOMMIT connection below. ``NULL, NULL`` refreshes the entire range, materializing
+# all history rather than only the trailing window the aggregate's policy covers. The
+# target is bound and cast to ``regclass`` so a caller-supplied name is never interpolated
+# into SQL.
+_REFRESH_AGGREGATE = text("CALL refresh_continuous_aggregate(CAST(:name AS regclass), NULL, NULL)")
+
+
+async def backfill_continuous_aggregates(name: str | None = None) -> list[str] | None:
+    """Materialize the full history of continuous aggregates (all, or one by ``name``).
+
+    Continuous aggregates are created ``WITH NO DATA`` (so creation does not backfill
+    inside Alembic's transaction), and their refresh policies only cover a trailing
+    window (``start_offset``). Without a one-off full refresh, buckets older than that
+    window fall below the materialization watermark once the first policy run advances it
+    and disappear from the view — so any pre-migration history is silently lost. Run this
+    once after applying an aggregate's migration on PostgreSQL/TimescaleDB; it is
+    idempotent and safe to re-run.
+
+    Args:
+        name: Refresh only this aggregate. When ``None`` (default), refresh every
+            continuous aggregate discovered in the TimescaleDB catalog.
+
+    Returns:
+        The list of aggregate names refreshed (possibly empty if none are registered), or
+        ``None`` if skipped because the analytics database is not PostgreSQL/TimescaleDB
+        (e.g. SQLite in tests/e2e, where continuous aggregates do not exist and the read
+        path aggregates ``raw_events`` directly).
+
+    Raises:
+        ContinuousAggregateNotFound: if ``name`` is given but no such aggregate exists.
+    """
+    engine = get_analytics_engine()
+    if engine.dialect.name != "postgresql":
+        logger.info(
+            "Skipping continuous-aggregate backfill: analytics dialect is '%s', not 'postgresql'",
+            engine.dialect.name,
+        )
+        return None
+    # AUTOCOMMIT because refresh_continuous_aggregate cannot run inside a transaction block.
+    async with engine.execution_options(isolation_level="AUTOCOMMIT").connect() as conn:
+        available = [row.view_name for row in (await conn.execute(_LIST_CONTINUOUS_AGGREGATES)).all()]
+        if name is not None:
+            if name not in available:
+                raise ContinuousAggregateNotFound(name)
+            targets = [name]
+        else:
+            targets = available
+        for target in targets:
+            await conn.execute(_REFRESH_AGGREGATE, {"name": target})
+    logger.info("Refreshed %d continuous aggregate(s): %s", len(targets), ", ".join(targets) or "(none)")
+    return targets

sparkth/core/analytics/reads.py

@@ -0,0 +1,78 @@
+"""Read-side queries over analytics data.
+
+The read path never touches the application database — only analytics tables and
+rollups. Exposed to the API layer via ``sparkth.lib.analytics``.
+
+``get_login_activity`` is dialect-aware because the login-activity rollup only
+exists as a TimescaleDB continuous aggregate on PostgreSQL. On SQLite (the test /
+e2e database) the aggregate is not created (see the analytics migration), so we
+compute the same daily buckets directly from ``raw_events``. The two query variants
+below must stay semantically identical — same filter (``user.logged_in``), same
+day bucketing, same columns, and same date-floor window (last ``days`` calendar
+days).
+
+Day boundaries are UTC on both paths, and this is the premise that makes them
+identical: PG's ``time_bucket('1 day', occurred_at)`` buckets the timestamptz in
+UTC, while SQLite's ``date(occurred_at)`` buckets on the stored string's offset —
+they agree only because the ingest path stores/interprets events as UTC. If the PG
+side is ever parameterized with a bucketing timezone, the SQLite side must move in
+lockstep or the two will silently diverge.
+"""
+
+from pydantic import BaseModel
+from sqlalchemy import text
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+class LoginActivityPoint(BaseModel):
+    """One day's login count. ``day`` is an ISO ``YYYY-MM-DD`` string."""
+
+    day: str
+    login_count: int
+
+
+# Both queries are raw SQL rather than SQLAlchemy Core selects, for two reasons:
+#   1. The PostgreSQL query reads ``login_activity_daily`` — the TimescaleDB continuous
+#      aggregate. That relation is created by raw migration DDL and is deliberately NOT
+#      registered as a ``sa.Table`` on ``analytics_metadata`` (doing so would make Alembic
+#      autogenerate try to create/drop it, and it doesn't exist on SQLite). With no table
+#      object to select against, a Core query would need a throwaway Table declaration plus
+#      Postgres-only funcs (``to_char``) — i.e. the same SQL with more ceremony.
+#   2. The two dialects need genuinely different queries (different source relation, different
+#      day-bucketing), so Core could not unify them anyway. Writing them as two side-by-side
+#      SQL strings makes the "must stay semantically identical" invariant (see module docstring)
+#      easy to eyeball. ``raw_events`` IS a ``sa.Table``, so the SQLite branch could be Core, but
+#      mixing one Core branch with one raw branch would obscure that pairing.
+
+# PostgreSQL: read the pre-rolled continuous aggregate.
+_PG_SQL = text(
+    "SELECT to_char(day AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS day, login_count "
+    "FROM login_activity_daily "
+    "WHERE (day AT TIME ZONE 'UTC')::date >= (now() AT TIME ZONE 'UTC')::date - :days "
+    "ORDER BY day DESC "
+    "LIMIT :days"
+)
+
+# SQLite (tests/e2e): aggregate raw_events directly — same semantics as the cagg.
+_SQLITE_SQL = text(
+    "SELECT date(occurred_at) AS day, count(*) AS login_count "
+    "FROM raw_events "
+    "WHERE event_type = 'user.logged_in' "
+    "AND date(occurred_at) >= date('now', '-' || :days || ' days') "
+    "GROUP BY date(occurred_at) "
+    "ORDER BY day DESC "
+    "LIMIT :days"
+)
+
+
+async def get_login_activity(session: AsyncSession, days: int = 30) -> list[LoginActivityPoint]:
+    """Return daily login counts, newest first, for the last ``days`` calendar days.
+
+    The window is bounded by a date floor (``now - days``), so gap days never let
+    the series reach back beyond the requested window. Days with no logins are
+    omitted entirely — the series is not zero-filled; callers must tolerate gaps.
+    """
+    dialect = session.bind.dialect.name
+    sql = _PG_SQL if dialect == "postgresql" else _SQLITE_SQL
+    rows = (await session.execute(sql, {"days": days})).mappings().all()
+    return [LoginActivityPoint(day=str(row["day"]), login_count=row["login_count"]) for row in rows]

sparkth/core/permissions/__init__.py

@@ -141,6 +141,9 @@ def get_permission(name: str) -> Permission:
 # reading roles.
 PERMISSION_READ = Permission.create("permission.read")
 
+# The analytics read permission gates the in-app analytics dashboards / read API.
+ANALYTICS_READ = Permission.create("analytics.read")
+
 
 def _active_assignment_at_scope(
     user_id: int | None,