8 - API (refactor): Repository Pattern

api/docs/adr/000-vertical-slice-architecture.md

@@ -79,7 +79,7 @@ damnit_api/
 │
 ├── runs/            # the core domain: runs, variables, previews
 │   ├── models.py    #   plain dataclasses
-│   ├── repository.py
+│   ├── repository.py   #   `DamnitRepository` interface + backends; see ADR-005
 │   ├── serialization.py, preview.py
 │   └── gql.py       #   Strawberry types + Query/Subscription contributions
 │

api/docs/adr/001-error-classes.md

@@ -81,4 +81,4 @@ Subclasses are added per _caller-distinguishable situation_, not per call site.
    - Permission denials use the Strawberry permission mechanism (message on the permission class); `ForbiddenError` is the same concept for non-GraphQL paths.
 3. **Boundaries translate exceptions.** Infrastructure exceptions do not cross layer boundaries raw:
    - MyMdC client errors become `UpstreamServiceError`
-   - Locator/repository/filesystem failures become `ProposalNotFoundError`/`DataUnavailableError`, logged with context at the point of translation (log-then-raise).
+   - Locator/repository/filesystem failures become `ProposalNotFoundError`/`DataUnavailableError`, logged with context at the point of translation (log-then-raise). See [ADR-005](005-repository-pattern.md) for the repository boundary.

api/docs/adr/005-repository-pattern.md

@@ -0,0 +1,36 @@
+---
+date: 2026-07-07
+---
+
+# ADR-005 - Repository pattern for DAMNIT run data
+
+## Context and Problem Statement
+
+Run and variable data lives in per-proposal SQLite files (`runs.sqlite`) produced by DAMNIT itself. The schema is external: the API reads it but does not own it, must reflect it at runtime, and there is one database per proposal on a network filesystem.
+
+Reading that data directly with SQLAlchemy from the GraphQL resolvers welds the transport layer to the DAMNIT schema, makes resolver logic untestable without real SQLite fixtures, and scatters session handling and caching across call sites. The data-access layer needs a seam: a narrow interface the resolvers depend on, with the SQL, reflection, and caching behind it.
+
+## Considered Options
+
+- Direct SQLAlchemy access from the resolvers (the status quo).
+- A `DamnitRepository` interface the resolvers depend on, with SQL, reflection, and caching behind it.
+
+## Decision Outcome
+
+Chosen option: the repository interface, because it decouples resolvers from SQLAlchemy and the DAMNIT schema, lets them be tested against a lightweight backend, and gives caching and session handling a single home.
+
+### Consequences
+
+- Good: resolvers depend only on the interface and plain domain models, so they are testable against the CSV backend without SQLite fixtures.
+- Good: a new data source (a DAMNIT HTTP API, Parquet exports) is a new `runs/<backend>/` package implementing the same ABC.
+- Bad: the registry accumulates one repository, and one engine, per accessed proposal for the process lifetime; eviction is available but not automatic (acceptable at current scale).
+
+## Details
+
+### The rules
+
+1. **`DamnitRepository`** (`runs/repository.py`, ABC) is the only way application code reads DAMNIT run data. One instance per proposal. Contract: `get_runs`, `get_latest_runs`, `get_metadata`, `get_extracted_data` (must not block the event loop, see [ADR-004](004-proposal-path-locator.md)), and `invalidate_metadata_cache`.
+2. **Return types are plain domain dataclasses** (`runs/models.py`: `RunRecord`, `VariableValue`, `MetadataSnapshot`, `VariableInfo`, `TagInfo`, `KnownVariable`) - no framework or SQLAlchemy types. Serialisation to transport shapes happens outside the repository.
+3. **A `DamnitRepositoryRegistry`** (held on `AppState`, see [ADR-002](002-no-global-mutable-state.md)) lazily creates and caches one repository per `ProposalNumber` via an injected factory, and supports eviction (`pop`, `clear`).
+4. **Implementations.** `runs/sqlite/` is production: async SQLAlchemy over `runs.sqlite`, read-only sessions (`PRAGMA query_only = ON`), `NullPool`, connection timeouts, and a per-repo table-reflection cache plus a metadata TTL cache (the approved cache pattern of [ADR-002](002-no-global-mutable-state.md)). `runs/csv/` is dev/test: it reads `runs.csv` / `run_variables.csv` / `variables.csv` and exercises resolver and subscription logic without SQLite fixtures.
+5. Caching lives inside the implementations; callers never wrap repository calls in their own module-level caches.

api/docs/architecture.md

@@ -18,7 +18,7 @@ For more information, see [ADR-000](adr/000-vertical-slice-architecture.md).
 
 | Package                           | Capability                              | What belongs there                                                                                                        | Status  | Today                                                                     |
 | --------------------------------- | --------------------------------------- | ------------------------------------------------------------------------------------------------------------------------- | ------- | ------------------------------------------------------------------------- |
-| `runs/`                           | Run/variable data - the core domain     | Domain models, repository interface + implementations, serialisation, preview extraction, its GraphQL types and resolvers | Planned | `db.py` + `data.py`; resolvers in `graphql/queries.py`/`subscriptions.py` |
+| `runs/`                           | Run/variable data - the core domain     | Domain models, repository interface + implementations (see [ADR-005](adr/005-repository-pattern.md)), serialisation, preview extraction, its GraphQL types and resolvers | Partial | `runs/` (repository, models, sqlite + csv backends); resolvers still in `graphql/queries.py`/`subscriptions.py` |
 | `proposals/`                      | Proposal metadata and lookup            | Proposal models, MyMdC-backed metadata services, path locator (see [ADR-004](adr/004-proposal-path-locator.md))            | Planned | `metadata/`                                                               |
 | `auth/`                           | Authentication and authorisation        | OAuth flow, sessions, token store, `User`, permission classes, the membership policy                                      | Partial | Policy still in `metadata/services.py`                                    |
 | `contextfile/`                    | Context-file viewing                    | File reading, watching, its routes                                                                                        | Done    | As-is                                                                     |

api/src/damnit_api/auth/gql.py

@@ -35,7 +35,7 @@ async def proposals(
         if settings.is_local:
             from ..metadata.services import _local_proposal_meta, _local_proposal_number
 
-            proposal_number = await _local_proposal_number(info.context.damnit_registry)
+            proposal_number = await _local_proposal_number(info.context.repositories)
             if proposal_number is None:
                 return []
             return [ProposalMeta.from_pydantic(_local_proposal_meta(proposal_number))]

api/src/damnit_api/auth/routers.py

@@ -148,7 +148,7 @@ async def noauth_userinfo(request: Request):
 
     proposals = {}
     proposal_number = await _local_proposal_number(
-        get_app_state(request).damnit_registry
+        get_app_state(request).repositories
     )
     if proposal_number:
         proposals = {LOCAL_CYCLE: [proposal_number]}

api/src/damnit_api/graphql/queries.py

@@ -1,25 +1,19 @@
+import dataclasses
+
 import strawberry
-from sqlalchemy import and_, func, select
 from strawberry.scalars import JSON
 from strawberry.types import Info
 from strawberry.types.nodes import SelectedField
 
 from .. import get_logger
 from ..auth.permissions import PROPOSAL_PERMISSIONS
 from ..metadata.services import _get_proposal_meta, _update_proposal_meta
-from ..runs.preview import get_preview_data
-from ..runs.sqlite import async_table, get_session
-from ..runs.types import KNOWN_DTYPES, DamnitRun
+from ..runs.types import DamnitRun
 from ..shared.models import ProposalNumber
-from .metadata import fetch_metadata
-from .utils import DatabaseInput, fetch_info
+from .utils import DatabaseInput
 
 logger = get_logger()
 
-# Names that only `fetch_info` provides; `proposal` and `run` already come
-# from `fetch_variables`.
-RUN_INFO_NAMES = frozenset(KNOWN_DTYPES) - {"proposal", "run"}
-
 
 async def _ensure_damnit_path(info: Info, proposal: ProposalNumber) -> None:
     """Ensure the proposal has a DAMNIT path, refreshing from MyMdC if needed.
@@ -45,92 +39,6 @@ async def _ensure_damnit_path(info: Info, proposal: ProposalNumber) -> None:
             raise ValueError(msg)
 
 
-def group_by_run(record):
-    grouped = {}
-
-    for entry in record:
-        key = (entry["proposal"], entry["run"])
-        if key not in grouped:
-            grouped[key] = {
-                "proposal": {"value": entry["proposal"]},
-                "run": {"value": entry["run"]},
-            }
-        # Outer-join placeholder for a run with no matching variables.
-        if entry["name"] is None:
-            continue
-        grouped[key][entry["name"]] = {
-            "value": entry["value"],
-            "summary_type": entry["summary_type"],
-            "attributes": entry["attributes"],
-        }
-
-    return list(grouped.values())
-
-
-async def fetch_variables(registry, proposal, *, limit, offset, names=None):
-    table = await async_table(registry, proposal, name="run_variables")
-    if table is None:
-        return []
-
-    runs_subquery = (
-        select(table.c.proposal, table.c.run)
-        .distinct()
-        .order_by(table.c.run)
-        .limit(limit)
-        .offset(offset)
-        .subquery()
-    )
-
-    latest_timestamp_subquery = select(
-        table.c.proposal,
-        table.c.run,
-        table.c.name,
-        func.max(table.c.timestamp).label("latest_timestamp"),
-    ).where(table.c.run.in_(select(runs_subquery.c.run)))
-    if names is not None:
-        latest_timestamp_subquery = latest_timestamp_subquery.where(
-            table.c.name.in_(names)
-        )
-    latest_timestamp_subquery = latest_timestamp_subquery.group_by(
-        table.c.run, table.c.name
-    ).subquery()
-
-    # Outer-join from `runs_subquery` so a `names` filter that excludes every
-    # row of a paginated run still keeps the run in the page.
-    query = (
-        select(
-            runs_subquery.c.proposal,
-            runs_subquery.c.run,
-            latest_timestamp_subquery.c.name,
-            table.c.value,
-            table.c.summary_type,
-            table.c.attributes,
-        )
-        .select_from(runs_subquery)
-        .outerjoin(
-            latest_timestamp_subquery,
-            runs_subquery.c.run == latest_timestamp_subquery.c.run,
-        )
-        .outerjoin(
-            table,
-            and_(
-                table.c.proposal == latest_timestamp_subquery.c.proposal,
-                table.c.run == latest_timestamp_subquery.c.run,
-                table.c.name == latest_timestamp_subquery.c.name,
-                table.c.timestamp == latest_timestamp_subquery.c.latest_timestamp,
-            ),
-        )
-        .order_by(runs_subquery.c.run)
-    )
-
-    async with get_session(registry, proposal) as session:
-        result = await session.execute(query)
-        if not result:
-            raise ValueError  # TODO: Better error handling
-
-        return group_by_run(result.mappings().all())  # type: ignore[assignment]
-
-
 def _selected_variable_names(info: Info) -> list[str] | None:
     """Union the `names` arguments across every `variables` sub-selection.
     Returns None if any selection omits the argument (forces a full fetch).
@@ -149,12 +57,6 @@ def _selected_variable_names(info: Info) -> list[str] | None:
     return sorted(union) if found else None
 
 
-def _wants_run_info(names: list[str] | None) -> bool:
-    if names is None:
-        return True
-    return bool(set(names) & RUN_INFO_NAMES)
-
-
 @strawberry.type
 class Query:
     """
@@ -179,45 +81,37 @@ async def runs(
         await _ensure_damnit_path(info, proposal)
         names = _selected_variable_names(info)
 
-        variables = await fetch_variables(
-            info.context.damnit_registry,
-            proposal,
-            limit=per_page,
-            offset=(page - 1) * per_page,
-            names=names,
-        )
-
-        if not len(variables):
-            return []
-
-        if _wants_run_info(names):
-            info_rows = await fetch_info(
-                info.context.damnit_registry,
-                proposal,
-                runs=[v["run"]["value"] for v in variables],
+        try:
+            repo = info.context.repositories.get(proposal)
+            records = await repo.get_runs(
+                limit=per_page,
+                offset=(page - 1) * per_page,
+                variable_names=names,
             )
-        else:
-            info_rows = [{} for _ in variables]
-
-        return [
-            DamnitRun.from_db({**v, **i})
-            for v, i in zip(variables, info_rows, strict=True)
-        ]
+        except Exception:
+            logger.exception("Failed to get runs", proposal=proposal)
+            raise
+        return [DamnitRun.from_record(r) for r in records]
 
     @strawberry.field(permission_classes=PROPOSAL_PERMISSIONS)
     async def metadata(
         self,
         info: Info,
         database: DatabaseInput,
-    ) -> JSON:  # FIX: # pyright: ignore[reportInvalidTypeForm]
+    ) -> JSON:
         proposal = database.proposal
         await _ensure_damnit_path(info, proposal)
 
-        snapshot = await fetch_metadata(info.context.damnit_registry, proposal)
-        return {
-            **snapshot,
-            "timestamp": snapshot["timestamp"] * 1000,  # ms for JS
-        }  # pyright: ignore[reportReturnType]
+        try:
+            repo = info.context.repositories.get(proposal)
+            snapshot = await repo.get_metadata()
+        except Exception:
+            logger.exception("Failed to get metadata", proposal=proposal)
+            raise
+        result = dataclasses.asdict(snapshot)
+        result["runs"] = list(result["runs"])
+        result["timestamp"] = snapshot.timestamp * 1000  # ms for JS
+        return result
 
     @strawberry.field(permission_classes=PROPOSAL_PERMISSIONS)
     async def extracted_data(
@@ -226,12 +120,12 @@ async def extracted_data(
         database: DatabaseInput,
         run: int,
         variable: str,
-    ) -> JSON:  # FIX: # pyright: ignore[reportInvalidTypeForm]
-        await _ensure_damnit_path(info, database.proposal)
-        # TODO: Convert to Strawberry type
-        # and make it analogous to DamitVariable; e.g. `data`
-        return get_preview_data(  # FIX: # pyright: ignore[reportReturnType]
-            proposal_number=database.proposal,
-            run=run,
-            variable=variable,
-        )
+    ) -> JSON:
+        proposal = database.proposal
+        await _ensure_damnit_path(info, proposal)
+        try:
+            repo = info.context.repositories.get(proposal)
+            return await repo.get_extracted_data(run=run, variable=variable)
+        except Exception:
+            logger.exception("Failed to get extracted data", proposal=proposal)
+            raise