river-data-api

Time-series API for the RIVER platform: one store for sensor readings, grab samples and the values computed from them.

Readings arrive from sync services, batch posts, CSV imports and interactive entry, and land in the same tables whichever way they came. The dashboard on top of this API is river-data-ui, which is where the platform is described from a user's side.

(Not to be confused with river-api, the earlier Astrocast-era API this replaces.)

Table of contents

• Quick start
• Stack
• Data model
• How a reading arrives
• Measurement cadence
• Corrections
• Visits and calculations
• Background jobs
• Authentication and authorisation
• API surface
• Configuration
• Testing
• Deployment
• License

Quick start

The compose file in river-data-ui starts this API with its database, Keycloak, the R tool runner and the sync services, hot-reloading from this source tree. To run the API alone against a database you already have:

cp .env.example .env
cargo run

Migrations run on startup. The API is at http://localhost:3000 and its documentation at /docs.

Stack

• Axum 0.8 and SeaORM 2.0 on Rust 1.95, edition 2024
• PostgreSQL 18 with TimescaleDB 2.23 for hypertables, compression and continuous aggregates
• CrudCrate for the generated CRUD surface and its hooks
• utoipa and Scalar for the OpenAPI documentation
• Migrations are a workspace member (migration/), applied on startup and runnable by hand with cargo run -p migration -- up. The single baseline is migration/src/m20260924_000001_baseline.rs; the complete earlier chain is in commit 2d9500de

After a baseline flatten, rebuild an existing dev database. Back up anything to preserve for restore_cutover first, then run docker compose down -v && docker compose up -d from river-data-ui. This deletes the compose volumes; sync restores stream registrations and staged readings. Startup refuses recorded migrations absent from the checkout.

Data model

The path a measurement takes, generated from the migrated database:

Every table is in docs/schema.svg. scripts/generate-schema-docs.py also writes docs/schema.md, each subject area as an entity diagram with its columns listed underneath, which is generated locally and not committed.

Tables are grouped and shaded by what they are for: core (blue) is the measurement and what identifies it, catalog (green) the definitions the core points at, record (amber) the provenance and curation evidence about readings, and operations (grey) the machinery that runs the platform. The overview boxes each tier together, so the shape of the schema is visible without reading a single table name.

Both are regenerated by CI whenever a migration changes, from a database the migrations have actually been run against, so they cannot drift from the code. To regenerate them by hand, point DATABASE_URL at a migrated database, run scripts/generate-schema-docs.py, then dot -Tsvg docs/schema.dot -o docs/schema.svg and the same for schema-core.

In words:

Projects
  └── Subprojects
       └── Sites
            └── SiteParameters (this site measures this parameter, with these settings)
Parameters (the global catalog: a code, a name, default units and thresholds)
Sensors (instruments) → SensorDeployments, SensorCalibrations, StandardCurves
DataStreams (one incoming series from one source)
  └── Readings (hypertable), StatusEvents (hypertable)

A reading is keyed by (stream_id, time, replicate_index). It carries the site and parameter it was attributed to, the instrument, deployment and curves behind its value, and the visit it belongs to. Readings are compressed after 30 days and rolled up into hourly, six-hourly, twelve-hourly, daily, weekly and monthly continuous aggregates.

How a reading arrives

A stream is one series from one source, ie. a viewLinc channel or a portal column, unique on (source_system, source_key). A sync service registers its streams once and then pushes readings; the readings are stored immediately, whether or not anyone has said what they are.

A stream belongs to nothing until it is paired with a site parameter. Pairing is what attributes the readings, ie. it stamps site and parameter onto the stream's history and onto everything it writes afterwards. Attribution comes from the pairing and never from the request, so a caller cannot mint a site and parameter that no configuration backs.

Sources that are edited in place, ie. the portal databases, cannot be synced by a cursor. Those declare a completeness window with each push, and the request becomes a diff rather than an upsert: values that changed are corrected, values absent at source are stamped withdrawn (never deleted, and reversibly), and a pass that would change or withdraw an implausible share of the window applies only its new rows and raises a hold for a person to release. Every diffed pass commits a receipt, so what a cycle did is queryable afterwards.

The other write paths are POST /readings/batch, POST /readings/import_csv and POST /grab_samples. Grab values are screened against the site's seasonal history before they are stored: post them to POST /readings/seasonal_check first, read its warnings, and pass its check_id on the save (or on the commit of a spot or tool CSV, whose dry_run stores the check). The save is held to exactly the values the check screened, and a correction of a stored grab value through POST /readings/edits names one in its decision the same way.

Measurement cadence

Every reading is continuous, spot or derived, resolved at ingest from the reading, then the stream, then the instrument's declared frequency.

• Continuous is logger data, one value per instant.
• Spot is a grab sample: several replicates at one instant, whose mean, standard deviation and count are maintained by a database trigger over the unflagged replicates. The standard deviation is the sample sd (n-1).
• Derived is a formula over other parameters, recomputed when its inputs move.

Continuous aggregates exclude spot readings, so grabs are served from the replicate statistics instead. The public serving contract is versioned in routes/public/service.rs.

Corrections

A stored corrected value is a claim that some curve produced it, so nothing changes one half of that pair and leaves the other.

• A calibration is a slope and an intercept over [valid_from, valid_until) on an instrument. Editing either one enqueues a job that rewrites the values over that window, re-derives which calibration each reading belongs to, cascades to derived parameters and refreshes the rollups.
• A standard curve belongs to an instrument and is chosen by hand for one measurement, so no window query can resolve it. A curve a reading already uses cannot be edited; correcting one means a new curve.

An instrument with no calibration stores no corrected value at all, rather than an identity correction that reads as one. docs/internals.md has the full rules, alongside the database triggers, CRUD hooks and background loops that run without a caller.

Visits and calculations

A visit (collection_events) is one site at one instant: the wide row the R/Shiny portals stored, as an entity. Every path that lands attributed grab readings attaches them to one.

The lab calculations the portals carried are R scripts stored in the database and executed by an OpenCPU sidecar (tools-runner/) that has no database, no credentials and no route out. Each script is one tool <- function(inputs, constants, curves) and is run verbatim: fidelity to the portal maths is structural, not tested for. A run is recorded server-side with its inputs, constants, curves and outputs, and a save cites that run; the client never authors provenance and cannot store a value the run did not produce.

Writing a value at a visit re-runs the calculations that read it, in dependency order, as a job. There is no timed global recompute: values that are missing or stale are found by an audit that only reports, and repaired by a scoped run.

Background jobs

Anything slower than a request is a tracked row in reprocessing_jobs: reprocessing, aggregate refreshes, derived computation, pairing plan apply and revert, retags, audits, recomputes and the scheduled sweeps. Each job has a status, a log timeline, a retry with backoff, a rerun and a cancel, and each reports what it did in one shape (detail.counts), so "how much is each job actually doing" is a query rather than log parsing.

Jobs are claimed, not assigned (FOR UPDATE SKIP LOCKED with a lease and a reaper), and enqueueing is idempotent on a dedupe key, so any replica can run any job and a restart loses no pending retry. Recurring work is rows in schedules with a cron expression, editable and runnable on demand.

Authentication and authorisation

Keycloak (realm river-data) authenticates people; API tokens and sync session tokens cover machines. A login gets in only with one of four ordered realm roles:

Role	Level	Grants
riverdata-intern	Intern	Read data and metadata
riverdata-river	River	Intern, plus write data and field metadata
riverdata-manager	Manager	River, plus manage instruments and the catalog
riverdata-admin	Administrator	Everything: users, tokens, onboarding, all projects

Non-administrators see only the projects granted to them (user_project_grants). Every route is gated by a capability check in common/middleware.rs; a token carries read_metadata, read_data, write_metadata or write_data, may be scoped to one project, and can never reach the routes that mint tokens or sync credentials, which are administrator-only.

The development realm is keycloak-realm-dev.json in the UI repository, with one fixture user per level.

API surface

Prefix	Auth
/api/	Keycloak JWT or API token, per-route capability
/api/public/	None. Read-only, per project, for external partners
/api/sync/	Sync service credentials, then a session token
/docs	The OpenAPI documentation for the authenticated API

Readings and aggregates are served as JSON, NDJSON or CSV, with start and end, optional site and parameter filters, and annotations for measurement type, sample statistics, curves and origin. A public project's own documentation is at /api/public/{code}/docs.

Configuration

Variable	Description	Default
DATABASE_URL	PostgreSQL connection string	required
API_PORT	Server port	3000
DEPLOYMENT	Environment name (local, dev, stage, prod)	local
KEYCLOAK_URL, KEYCLOAK_REALM, KEYCLOAK_CLIENT_ID	Token validation	required
KEYCLOAK_ADMIN_CLIENT_ID, KEYCLOAK_ADMIN_CLIENT_SECRET	User management	required
TOOLS_RUNNER_URL	OpenCPU endpoint for tool scripts	none, tools disabled
METEOSWISS_BASE_URL, METEOSWISS_INTERVAL_SECONDS	SMN pressure feed and its cadence	OGD SMN, 3600
CACHE_TTL_SECONDS, CACHE_MAX_BYTES	Response cache	300, 209715200
DISABLE_RATE_LIMITING	Turn off the public rate limiter	false
ALARM_SWEEP_INTERVAL_SECONDS	Alarm reconciliation cadence	60
JOB_MAX_RETRIES, JOB_RETRY_BACKOFF_SECONDS	Job retry policy	3, 60
VAPID_PRIVATE_KEY_PEM, VAPID_PUBLIC_KEY, VAPID_SUBJECT	Web Push	unset, push disabled

Startup validates the configuration and fails rather than starting half-configured.

Testing

Tests are written at the lowest level that can fail for the right reason: an inline #[cfg(test)] module for anything that is a decision, tests/<theme>/ for what only real SQL can show, and tests/e2e/ for a user story crossing several domains.

cargo test --lib            # no database needed

# Integration tests, against a database of their own
DATABASE_URL=postgresql://postgres:psql@localhost:5444/river_test \
  RUST_MIN_STACK=16777216 cargo test --tests -- --test-threads=1

docker compose --profile test up -d river-db-test-db river-db-test-watcher in the UI repository starts that database and a watcher that reruns the suite on every change. CI runs the same suite against TimescaleDB, Keycloak and a tool runner built from the commit.

Deployment

The baseline requires an empty database. Back up the existing database before replacing it, then start the API against the fresh database to apply the baseline. It seeds the twelve portal constants; projects, sites and calculation definitions are supplied through the application. Use restore_cutover to carry curated state from the backup after rebuilding the metadata.

Multi-stage Docker build with cargo-chef. The image and the R runner image are published together and pinned by digest in the ArgoCD manifests; the runner is a sidecar in the API pod, reachable only over loopback. One replica, deliberately: latency lives in the database queries, not in the API process.

License

MIT