diff --git a/Cargo.lock b/Cargo.lock index 81ecee513..829f1c6bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3477,7 +3477,7 @@ dependencies = [ [[package]] name = "helios-observability" -version = "0.1.47" +version = "0.2.1" dependencies = [ "axum", "chrono", @@ -4760,15 +4760,6 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" -[[package]] -name = "memoffset" -version = "0.9.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" -dependencies = [ - "autocfg", -] - [[package]] name = "metrics" version = "0.24.6" @@ -6012,7 +6003,7 @@ dependencies = [ "libc", "once_cell", "raw-cpuid", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "web-sys", "winapi", ] diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index 30e0068d4..019f26ce1 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -25,18 +25,32 @@ use helios_audit::{ }; use helios_auth::{AuthConfig, InMemoryJtiCache, JtiCache, JwksBearerAuthProvider, JwksCache}; use helios_persistence::{BackendKind, ResourceStorage, TenantContext}; -use helios_rest::{AuthMiddlewareState, ServerConfig, StorageBackendMode, create_app_with_auth}; +use helios_rest::{AuthMiddlewareState, ServerConfig, StorageBackendMode}; use tracing::info; use helios_persistence::backends::local_fs::LocalFsOutputStore; +#[cfg(any(feature = "sqlite", feature = "postgres"))] +use helios_persistence::core::SettingsStore; use helios_persistence::core::{ BulkExportJobStore, BulkSubmitJobStore, DefaultExportWorker, DefaultSubmitWorker, ExportOutputStore, SubmitInputFetcher, WorkerId, }; #[cfg(any(feature = "sqlite", feature = "postgres"))] use helios_rest::bulk_export_auth::BearerScopeAuth; -#[cfg(any(feature = "sqlite", feature = "postgres"))] +// The S3+Elasticsearch composite is the only remaining caller of the plain +// bulk builder: S3 has no settings store, and its bulk-export job state rides +// on an embedded SQLite sidecar. +#[cfg(all(feature = "s3", feature = "elasticsearch", feature = "sqlite"))] use helios_rest::create_app_with_auth_and_bulk; +// Settings-capable standalone/composite backends (SQLite, PostgreSQL, MongoDB) +// host the per-user settings store, wired alongside bulk export/submit. +#[cfg(any(feature = "sqlite", feature = "postgres"))] +use helios_rest::create_app_with_auth_bulk_and_settings; +// S3 does not host a settings store (tracked follow-up #199), so its startup +// paths use the plain app builder. Every other standalone/composite backend now +// wires the per-user settings store via `create_app_with_auth_bulk_and_settings`. +#[cfg(feature = "s3")] +use helios_rest::create_app_with_auth; #[cfg(feature = "sqlite")] use helios_persistence::backends::sqlite::{SqliteBackend, SqliteBackendConfig}; @@ -502,32 +516,33 @@ async fn start_mongodb( let backend = Arc::new(backend); let serve_audit_state = audit_state.clone(); - // MongoDB primary; embedded SQLite sidecar for job state. - #[cfg(feature = "sqlite")] - { - let jobs = build_embedded_job_store(&config)?; - if let Some(bundle) = build_bulk_export(&config, backend.clone(), jobs).await? { - let app = create_app_with_auth_and_bulk( - backend, - config.clone(), - auth_config, - auth_state, - audit_state, - Some(bundle), - None, - ); - return serve(app, &config, serve_audit_state).await; + // MongoDB is a full standalone primary, so it also hosts the per-user + // settings store: it keeps ownership of the backend Arc and wires the + // settings-capable builder (like the SQLite/Postgres backends). + let settings_store: Option> = Some(backend.clone()); + + // MongoDB primary; embedded SQLite sidecar for bulk-export job state. + let export_bundle = { + #[cfg(feature = "sqlite")] + { + let jobs = build_embedded_job_store(&config)?; + build_bulk_export(&config, backend.clone(), jobs).await? } - } + #[cfg(not(feature = "sqlite"))] + { + None + } + }; - let app = create_app_with_auth( - Arc::try_unwrap(backend).unwrap_or_else(|_| { - unreachable!("backend Arc is uniquely owned when bulk export is disabled") - }), + let app = create_app_with_auth_bulk_and_settings( + backend, config.clone(), auth_config, auth_state, audit_state, + export_bundle, + None, + settings_store, ); serve(app, &config, serve_audit_state).await } @@ -863,29 +878,20 @@ async fn start_sqlite( let serve_audit_state = audit_state.clone(); let backend = Arc::new(create_sqlite_backend(&config)?); + // The SQLite backend also hosts the per-user settings store, so it always + // keeps ownership of the backend Arc and uses the settings-capable builder. + let settings_store: Option> = Some(backend.clone()); let export_bundle = build_bulk_export(&config, backend.clone(), backend.clone()).await?; let submit_bundle = build_bulk_submit(&config, backend.clone()).await?; - if export_bundle.is_some() || submit_bundle.is_some() { - let app = create_app_with_auth_and_bulk( - backend, - config.clone(), - auth_config, - auth_state, - audit_state, - export_bundle, - submit_bundle, - ); - return serve(app, &config, serve_audit_state).await; - } - - let app = create_app_with_auth( - Arc::try_unwrap(backend).unwrap_or_else(|_| { - unreachable!("backend Arc is uniquely owned when bulk export is disabled") - }), + let app = create_app_with_auth_bulk_and_settings( + backend, config.clone(), auth_config, auth_state, audit_state, + export_bundle, + submit_bundle, + settings_store, ); serve(app, &config, serve_audit_state).await } @@ -1416,29 +1422,22 @@ async fn start_sqlite_elasticsearch( let serve_audit_state = audit_state.clone(); let composite = Arc::new(composite); + // The per-user settings store lives on the SQLite primary (Elasticsearch is + // search-only), so it is wired from the underlying `sqlite` backend even + // though the app is served over the composite storage. + let settings_store: Option> = Some(sqlite.clone()); + let export_bundle = build_bulk_export(&config, sqlite.clone(), sqlite.clone()).await?; let submit_bundle = build_bulk_submit(&config, sqlite.clone()).await?; - if export_bundle.is_some() || submit_bundle.is_some() { - let app = create_app_with_auth_and_bulk( - composite, - config.clone(), - auth_config, - auth_state, - audit_state, - export_bundle, - submit_bundle, - ); - return serve(app, &config, serve_audit_state).await; - } - - let app = create_app_with_auth( - Arc::try_unwrap(composite).unwrap_or_else(|_| { - unreachable!("composite Arc is uniquely owned when bulk export is disabled") - }), + let app = create_app_with_auth_bulk_and_settings( + composite, config.clone(), auth_config, auth_state, audit_state, + export_bundle, + submit_bundle, + settings_store, ); serve(app, &config, serve_audit_state).await } @@ -1484,29 +1483,20 @@ async fn start_postgres( let backend = Arc::new(backend); let serve_audit_state = audit_state.clone(); + // The PostgreSQL backend also hosts the per-user settings store, so it always + // keeps ownership of the backend Arc and uses the settings-capable builder. + let settings_store: Option> = Some(backend.clone()); let export_bundle = build_bulk_export(&config, backend.clone(), backend.clone()).await?; let submit_bundle = build_bulk_submit(&config, backend.clone()).await?; - if export_bundle.is_some() || submit_bundle.is_some() { - let app = create_app_with_auth_and_bulk( - backend, - config.clone(), - auth_config, - auth_state, - audit_state, - export_bundle, - submit_bundle, - ); - return serve(app, &config, serve_audit_state).await; - } - - let app = create_app_with_auth( - Arc::try_unwrap(backend).unwrap_or_else(|_| { - unreachable!("backend Arc is uniquely owned when bulk export is disabled") - }), + let app = create_app_with_auth_bulk_and_settings( + backend, config.clone(), auth_config, auth_state, audit_state, + export_bundle, + submit_bundle, + settings_store, ); serve(app, &config, serve_audit_state).await } @@ -1650,29 +1640,22 @@ async fn start_postgres_elasticsearch( let serve_audit_state = audit_state.clone(); let composite = Arc::new(composite); + // The per-user settings store lives on the PostgreSQL primary (Elasticsearch + // is search-only), so it is wired from the underlying `pg` backend even + // though the app is served over the composite storage. + let settings_store: Option> = Some(pg.clone()); + let export_bundle = build_bulk_export(&config, pg.clone(), pg.clone()).await?; let submit_bundle = build_bulk_submit(&config, pg.clone()).await?; - if export_bundle.is_some() || submit_bundle.is_some() { - let app = create_app_with_auth_and_bulk( - composite, - config.clone(), - auth_config, - auth_state, - audit_state, - export_bundle, - submit_bundle, - ); - return serve(app, &config, serve_audit_state).await; - } - - let app = create_app_with_auth( - Arc::try_unwrap(composite).unwrap_or_else(|_| { - unreachable!("composite Arc is uniquely owned when bulk export is disabled") - }), + let app = create_app_with_auth_bulk_and_settings( + composite, config.clone(), auth_config, auth_state, audit_state, + export_bundle, + submit_bundle, + settings_store, ); serve(app, &config, serve_audit_state).await } @@ -1808,32 +1791,33 @@ async fn start_mongodb_elasticsearch( let serve_audit_state = audit_state.clone(); let composite = Arc::new(composite); - // MongoDB primary; embedded SQLite sidecar for job state. - #[cfg(feature = "sqlite")] - { - let jobs = build_embedded_job_store(&config)?; - if let Some(bundle) = build_bulk_export(&config, mongo.clone(), jobs).await? { - let app = create_app_with_auth_and_bulk( - composite, - config.clone(), - auth_config, - auth_state, - audit_state, - Some(bundle), - None, - ); - return serve(app, &config, serve_audit_state).await; + // The per-user settings store lives on the MongoDB primary (Elasticsearch is + // search-only), so it is wired from the underlying `mongo` backend even + // though the app is served over the composite storage. + let settings_store: Option> = Some(mongo.clone()); + + // MongoDB primary; embedded SQLite sidecar for bulk-export job state. + let export_bundle = { + #[cfg(feature = "sqlite")] + { + let jobs = build_embedded_job_store(&config)?; + build_bulk_export(&config, mongo.clone(), jobs).await? } - } + #[cfg(not(feature = "sqlite"))] + { + None + } + }; - let app = create_app_with_auth( - Arc::try_unwrap(composite).unwrap_or_else(|_| { - unreachable!("composite Arc is uniquely owned when bulk export is disabled") - }), + let app = create_app_with_auth_bulk_and_settings( + composite, config.clone(), auth_config, auth_state, audit_state, + export_bundle, + None, + settings_store, ); serve(app, &config, serve_audit_state).await } diff --git a/crates/hts/README.md b/crates/hts/README.md index 8e0fc5bc5..0e63f0e07 100644 --- a/crates/hts/README.md +++ b/crates/hts/README.md @@ -10,7 +10,7 @@ It can also be used standalone as a general-purpose FHIR terminology service, in An open test server will soon be available at https://hts.heliossoftware.com/ for experimentation and evaluation. -HTS currently uses SQLite as its database backend. PostgreSQL support is planned for a future release - see [Storage Backends](#storage-backends) for details. +HTS supports both SQLite (the zero-config default) and PostgreSQL database backends, with full feature parity between them - see [Storage Backends](#storage-backends) for details. ### Terminology Data @@ -422,10 +422,17 @@ On top of the SQL layer, the SQLite backend keeps small bounded in-memory caches ### PostgreSQL -PostgreSQL backend support is planned for a future release. The schema, query patterns, and persistence trait surface have been designed with multi-backend portability in mind, and the integration is being staged behind feature work tracked separately. Until it lands, all production deployments should use the SQLite backend documented above. +HTS fully supports PostgreSQL as an alternative backend, with feature parity with SQLite across all terminology operations, CRUD, search, and import formats. The schema mirrors the SQLite layout and is applied automatically on first connection, so no manual migration step is required. + +Build with the `postgres` feature, then select the backend with `--storage-backend postgres` (or `HTS_STORAGE_BACKEND=postgres`) and a `postgresql://` connection string: ```bash -# Coming soon +# Build with PostgreSQL support +cargo build --release -p helios-hts --features postgres + +# Run against PostgreSQL +hts run --storage-backend postgres \ + --database-url "postgresql://user:pass@localhost/hts" ``` ## API Endpoints diff --git a/crates/persistence/Cargo.toml b/crates/persistence/Cargo.toml index 41e2f396d..6a88dc8f9 100644 --- a/crates/persistence/Cargo.toml +++ b/crates/persistence/Cargo.toml @@ -114,7 +114,7 @@ tokio = { version = "1", features = ["full", "test-util"] } tokio-test = "0.4" tempfile = "3" criterion = { version = "0.5", features = ["async_tokio"] } -testcontainers-modules = { version = "0.15", features = ["postgres", "elastic_search", "minio"] } +testcontainers-modules = { version = "0.15", features = ["postgres", "elastic_search", "minio", "mongo"] } paste = "1.0" # `watchdog` pulls in `signal_hook::iterator`/`SIGQUIT`, which only compile on Unix. diff --git a/crates/persistence/src/backends/mongodb/mod.rs b/crates/persistence/src/backends/mongodb/mod.rs index 1c68e3f24..683a5dd13 100644 --- a/crates/persistence/src/backends/mongodb/mod.rs +++ b/crates/persistence/src/backends/mongodb/mod.rs @@ -20,5 +20,6 @@ mod bulk_export; pub(crate) mod schema; mod search_impl; mod storage; +mod user_settings; pub use backend::{MongoBackend, MongoBackendConfig}; diff --git a/crates/persistence/src/backends/mongodb/schema.rs b/crates/persistence/src/backends/mongodb/schema.rs index 230448b09..8566b5376 100644 --- a/crates/persistence/src/backends/mongodb/schema.rs +++ b/crates/persistence/src/backends/mongodb/schema.rs @@ -12,7 +12,7 @@ use crate::error::{BackendError, StorageError, StorageResult}; use super::backend::MongoBackendConfig; /// Current MongoDB schema version. -pub const SCHEMA_VERSION: i32 = 4; +pub const SCHEMA_VERSION: i32 = 5; /// Initialize MongoDB collections/indexes required by the backend. /// @@ -43,6 +43,7 @@ pub async fn initialize_schema_async(database: &Database) -> StorageResult<()> { ensure_resources_indexes(database).await?; ensure_history_indexes(database).await?; ensure_search_indexes(database).await?; + ensure_user_settings_indexes(database).await?; set_schema_version(database, SCHEMA_VERSION).await?; Ok(()) } @@ -54,6 +55,7 @@ pub async fn migrate_schema_async(database: &Database) -> StorageResult<()> { ensure_resources_indexes(database).await?; ensure_history_indexes(database).await?; ensure_search_indexes(database).await?; + ensure_user_settings_indexes(database).await?; set_schema_version(database, SCHEMA_VERSION).await?; } Ok(()) @@ -250,6 +252,27 @@ async fn ensure_search_indexes(database: &Database) -> StorageResult<()> { Ok(()) } +/// Index for the per-user UI settings store. One document per user, keyed by a +/// unique `user_key`, so the settings store can rely on the index to serialize +/// concurrent first-writes (a lost insert race surfaces as a duplicate-key +/// error). Kept separate from the FHIR `resources` collection so UI preferences +/// never surface in FHIR machinery (`CapabilityStatement`, search, history, +/// export). +async fn ensure_user_settings_indexes(database: &Database) -> StorageResult<()> { + let user_settings = + database.collection::(super::user_settings::USER_SETTINGS_COLLECTION); + + create_index( + &user_settings, + doc! { "user_key": 1_i32 }, + "idx_user_settings_key", + true, + ) + .await?; + + Ok(()) +} + async fn create_index( collection: &Collection, keys: Document, diff --git a/crates/persistence/src/backends/mongodb/user_settings.rs b/crates/persistence/src/backends/mongodb/user_settings.rs new file mode 100644 index 000000000..02ea02cc8 --- /dev/null +++ b/crates/persistence/src/backends/mongodb/user_settings.rs @@ -0,0 +1,266 @@ +//! MongoDB implementation of the per-user [`SettingsStore`]. +//! +//! Each user owns a single document in the `user_settings` collection holding an +//! opaque JSON settings document (stored as a JSON string, mirroring the SQLite +//! backend's JSON blob) plus a monotonic `version` used for optimistic locking. +//! +//! MongoDB standalone deployments do not support multi-document transactions, so +//! rather than a `SELECT … FOR UPDATE` read-modify-write (as SQLite/PostgreSQL +//! use) writes here are made atomic by a **version-conditioned update**: the +//! current document is read, the new document computed, then persisted with a +//! filter pinned to the version that was read. If a concurrent writer bumped the +//! version in between, the update matches nothing and the write is retried +//! (unconditional writes) or surfaced as an optimistic-lock failure (conditional +//! `If-Match` writes). A `Some(0)` precondition asserts the document does not yet +//! exist; the unique index on `user_key` turns a lost insert race into the same +//! lock failure. This yields the same read-modify-write + monotonic-version +//! semantics as the relational stores without requiring a replica set. + +use async_trait::async_trait; +use chrono::Utc; +use mongodb::bson::{Document, doc}; +use mongodb::error::Error as MongoError; +use serde_json::Value; + +use crate::core::user_settings::{SettingsStore, StoredUserSettings, apply_merge_patch}; +use crate::error::{BackendError, ConcurrencyError, StorageError, StorageResult}; + +use super::MongoBackend; + +/// Name of the collection backing the per-user settings store. +pub(crate) const USER_SETTINGS_COLLECTION: &str = "user_settings"; + +/// Bound on retries when an *unconditional* write loses the version-compare race +/// with a concurrent writer. A single unconditional writer can be beaten at most +/// once per other concurrent writer before it wins, so this caps the tolerated +/// concurrent-writer count. Contention on a single user's document is expected to +/// be near-zero (a user updates their own settings), so this generous bound is +/// never approached in practice; exceeding it surfaces as a concurrency error +/// rather than looping forever. +const MAX_WRITE_RETRIES: usize = 32; + +impl MongoBackend { + /// Read-modify-write a user's settings document using a version-conditioned + /// update so concurrent writers cannot lose data or skip the `If-Match` + /// precondition check. + /// + /// `compute` receives the currently stored document (or `None` when the user + /// has no settings yet) and returns the document to persist. It may be called + /// more than once when an unconditional write retries after losing a race. + async fn write_settings( + &self, + user_key: &str, + if_match_version: Option, + compute: impl Fn(Option) -> Value + Send, + ) -> StorageResult { + let db = self.get_database().await?; + let collection = db.collection::(USER_SETTINGS_COLLECTION); + + for _ in 0..=MAX_WRITE_RETRIES { + let existing = collection + .find_one(doc! { "user_key": user_key }) + .await + .map_err(|e| backend_err(format!("read user_settings: {e}")))?; + + let current_version = existing + .as_ref() + .and_then(|d| d.get_i64("version").ok()) + .unwrap_or(0); + + if let Some(expected) = if_match_version + && expected != current_version + { + return Err(lock_failure(user_key, expected, current_version)); + } + + let current_doc = match &existing { + Some(d) => Some(decode_document(d)?), + None => None, + }; + + let new_doc = compute(current_doc); + let new_version = current_version + 1; + let now = Utc::now(); + let data = serde_json::to_string(&new_doc) + .map_err(|e| backend_err(format!("encode user_settings: {e}")))?; + let now_bson = mongodb::bson::DateTime::from_millis(now.timestamp_millis()); + + if current_version == 0 { + // No document yet: insert. A unique index on `user_key` makes a + // concurrent insert fail with a duplicate-key error, which we + // treat as a lost race (retry) or a lock failure (conditional). + match collection + .insert_one(doc! { + "user_key": user_key, + "data": &data, + "version": new_version, + "updated_at": now_bson, + }) + .await + { + Ok(_) => { + return Ok(StoredUserSettings { + user_key: user_key.to_string(), + document: new_doc, + version: new_version, + updated_at: now, + }); + } + Err(e) if is_duplicate_key_error(&e) => { + if let Some(expected) = if_match_version { + let actual = reload_version(&collection, user_key).await?; + return Err(lock_failure(user_key, expected, actual)); + } + continue; + } + Err(e) => { + return Err(backend_err(format!("insert user_settings: {e}"))); + } + } + } + + // A document exists: update only if its version is unchanged since we + // read it, so a concurrent writer's change is never clobbered. + let result = collection + .update_one( + doc! { "user_key": user_key, "version": current_version }, + doc! { "$set": { + "data": &data, + "version": new_version, + "updated_at": now_bson, + }}, + ) + .await + .map_err(|e| backend_err(format!("write user_settings: {e}")))?; + + if result.matched_count == 1 { + return Ok(StoredUserSettings { + user_key: user_key.to_string(), + document: new_doc, + version: new_version, + updated_at: now, + }); + } + + // Version moved under us. A conditional write fails the precondition; + // an unconditional write retries against the new state. + if let Some(expected) = if_match_version { + let actual = reload_version(&collection, user_key).await?; + return Err(lock_failure(user_key, expected, actual)); + } + } + + Err(StorageError::Concurrency( + ConcurrencyError::OptimisticLockFailure { + resource_type: "UserSettings".to_string(), + id: user_key.to_string(), + expected_etag: "W/\"*\"".to_string(), + actual_etag: None, + }, + )) + } +} + +#[async_trait] +impl SettingsStore for MongoBackend { + async fn get_settings(&self, user_key: &str) -> StorageResult> { + let db = self.get_database().await?; + let collection = db.collection::(USER_SETTINGS_COLLECTION); + let row = collection + .find_one(doc! { "user_key": user_key }) + .await + .map_err(|e| backend_err(format!("read user_settings: {e}")))?; + + match row { + None => Ok(None), + Some(d) => { + let document = decode_document(&d)?; + let version = d + .get_i64("version") + .map_err(|e| backend_err(format!("read user_settings version: {e}")))?; + let updated_at = d + .get_datetime("updated_at") + .map(|dt| { + chrono::DateTime::::from_timestamp_millis(dt.timestamp_millis()) + .unwrap_or_else(Utc::now) + }) + .unwrap_or_else(|_| Utc::now()); + Ok(Some(StoredUserSettings { + user_key: user_key.to_string(), + document, + version, + updated_at, + })) + } + } + } + + async fn put_settings( + &self, + user_key: &str, + document: Value, + if_match_version: Option, + ) -> StorageResult { + self.write_settings(user_key, if_match_version, move |_current| document.clone()) + .await + } + + async fn patch_settings( + &self, + user_key: &str, + merge_patch: Value, + if_match_version: Option, + ) -> StorageResult { + self.write_settings(user_key, if_match_version, move |current| { + apply_merge_patch( + current.unwrap_or_else(|| Value::Object(Default::default())), + &merge_patch, + ) + }) + .await + } +} + +/// Decodes the JSON `data` field of a stored `user_settings` document. +fn decode_document(doc: &Document) -> StorageResult { + let data = doc + .get_str("data") + .map_err(|e| backend_err(format!("read user_settings data: {e}")))?; + serde_json::from_str(data).map_err(|e| backend_err(format!("decode stored user_settings: {e}"))) +} + +/// Reloads the live version for a user, used to report the actual version in an +/// optimistic-lock failure. A vanished document reads back as version 0. +async fn reload_version( + collection: &mongodb::Collection, + user_key: &str, +) -> StorageResult { + let row = collection + .find_one(doc! { "user_key": user_key }) + .await + .map_err(|e| backend_err(format!("reload user_settings version: {e}")))?; + Ok(row.and_then(|d| d.get_i64("version").ok()).unwrap_or(0)) +} + +fn is_duplicate_key_error(err: &MongoError) -> bool { + err.to_string().contains("E11000") +} + +/// Builds an `OptimisticLockFailure` for a `user_settings` write whose +/// `If-Match` precondition did not match the live version. +fn lock_failure(user_key: &str, expected: i64, actual: i64) -> StorageError { + StorageError::Concurrency(ConcurrencyError::OptimisticLockFailure { + resource_type: "UserSettings".to_string(), + id: user_key.to_string(), + expected_etag: format!("W/\"{expected}\""), + actual_etag: Some(format!("W/\"{actual}\"")), + }) +} + +fn backend_err(message: String) -> StorageError { + StorageError::Backend(BackendError::Internal { + backend_name: "mongodb".to_string(), + message, + source: None, + }) +} diff --git a/crates/persistence/src/backends/postgres/mod.rs b/crates/persistence/src/backends/postgres/mod.rs index b1808bef1..7c8bfe66c 100644 --- a/crates/persistence/src/backends/postgres/mod.rs +++ b/crates/persistence/src/backends/postgres/mod.rs @@ -80,5 +80,6 @@ pub mod search; mod search_impl; mod storage; mod transaction; +mod user_settings; pub use backend::{PostgresBackend, PostgresConfig}; diff --git a/crates/persistence/src/backends/postgres/schema.rs b/crates/persistence/src/backends/postgres/schema.rs index 536482f7b..d1d928099 100644 --- a/crates/persistence/src/backends/postgres/schema.rs +++ b/crates/persistence/src/backends/postgres/schema.rs @@ -3,7 +3,7 @@ use crate::error::{BackendError, StorageResult}; /// Current schema version. -pub const SCHEMA_VERSION: i32 = 12; +pub const SCHEMA_VERSION: i32 = 13; /// Initialize the database schema. pub async fn initialize_schema(client: &deadpool_postgres::Client) -> StorageResult<()> { @@ -275,6 +275,7 @@ async fn migrate_schema( 9 => migrate_v9_to_v10(client).await?, 10 => migrate_v10_to_v11(client).await?, 11 => migrate_v11_to_v12(client).await?, + 12 => migrate_v12_to_v13(client).await?, _ => { return Err(pg_error(format!("Unknown schema version: {}", version))); } @@ -767,6 +768,28 @@ async fn migrate_v11_to_v12(client: &deadpool_postgres::Client) -> StorageResult add_bulk_submit_worker_schema(client, "Migration v11->v12").await } +/// v12 -> v13: Add the `user_settings` table backing the per-user UI settings +/// store (theme, default tenant, active FHIR version, recent queries, …). +/// +/// One opaque JSONB document is stored per user, keyed by `user_key`, with a +/// monotonic `version` for optimistic locking. This table is independent of the +/// FHIR `resources` table so UI preferences never leak into FHIR machinery. +async fn migrate_v12_to_v13(client: &deadpool_postgres::Client) -> StorageResult<()> { + client + .execute( + "CREATE TABLE IF NOT EXISTS user_settings ( + user_key TEXT PRIMARY KEY, + data JSONB NOT NULL, + version BIGINT NOT NULL DEFAULT 1, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + )", + &[], + ) + .await + .map_err(|e| pg_error(format!("Migration v12->v13 failed: {}", e)))?; + Ok(()) +} + fn pg_error(message: String) -> crate::error::StorageError { crate::error::StorageError::Backend(BackendError::Internal { backend_name: "postgres".to_string(), diff --git a/crates/persistence/src/backends/postgres/user_settings.rs b/crates/persistence/src/backends/postgres/user_settings.rs new file mode 100644 index 000000000..aa57447ab --- /dev/null +++ b/crates/persistence/src/backends/postgres/user_settings.rs @@ -0,0 +1,156 @@ +//! PostgreSQL implementation of the per-user [`SettingsStore`]. +//! +//! Each user owns a single row in the `user_settings` table holding an opaque +//! JSONB document plus a monotonic `version` used for optimistic locking. Writes +//! run a `SELECT … FOR UPDATE` read-modify-write inside a transaction so +//! concurrent updates to the same user serialize correctly and the `If-Match` +//! precondition is checked against the live row. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use serde_json::Value; + +use crate::core::user_settings::{SettingsStore, StoredUserSettings, apply_merge_patch}; +use crate::error::{BackendError, ConcurrencyError, StorageError, StorageResult}; + +use super::PostgresBackend; + +impl PostgresBackend { + /// Read-modify-write a user's settings document inside a single transaction, + /// locking the row with `SELECT … FOR UPDATE`. + /// + /// `compute` receives the currently stored document (or `None` when the user + /// has no settings yet) and returns the document to persist. The optimistic + /// `if_match_version` precondition — where `Some(0)` asserts "does not yet + /// exist" — is checked against the locked row before `compute` runs. + async fn write_settings( + &self, + user_key: &str, + if_match_version: Option, + compute: impl FnOnce(Option) -> Value + Send, + ) -> StorageResult { + let mut client = self.get_client().await?; + let txn = client + .transaction() + .await + .map_err(|e| backend_err(format!("begin user_settings transaction: {e}")))?; + + let current = txn + .query_opt( + "SELECT version, data FROM user_settings WHERE user_key = $1 FOR UPDATE", + &[&user_key], + ) + .await + .map_err(|e| backend_err(format!("read user_settings: {e}")))?; + + let (current_version, current_doc) = match ¤t { + Some(row) => { + let version: i64 = row.get(0); + let doc: Value = row.get(1); + (version, Some(doc)) + } + None => (0, None), + }; + + if let Some(expected) = if_match_version + && expected != current_version + { + return Err(lock_failure(user_key, expected, current_version)); + } + + let new_doc = compute(current_doc); + let new_version = current_version + 1; + let now = Utc::now(); + + txn.execute( + "INSERT INTO user_settings (user_key, data, version, updated_at) + VALUES ($1, $2, $3, $4) + ON CONFLICT (user_key) + DO UPDATE SET data = $2, version = $3, updated_at = $4", + &[&user_key, &new_doc, &new_version, &now], + ) + .await + .map_err(|e| backend_err(format!("write user_settings: {e}")))?; + + txn.commit() + .await + .map_err(|e| backend_err(format!("commit user_settings: {e}")))?; + + Ok(StoredUserSettings { + user_key: user_key.to_string(), + document: new_doc, + version: new_version, + updated_at: now, + }) + } +} + +#[async_trait] +impl SettingsStore for PostgresBackend { + async fn get_settings(&self, user_key: &str) -> StorageResult> { + let client = self.get_client().await?; + let row = client + .query_opt( + "SELECT data, version, updated_at FROM user_settings WHERE user_key = $1", + &[&user_key], + ) + .await + .map_err(|e| backend_err(format!("read user_settings: {e}")))?; + + Ok(row.map(|row| { + let document: Value = row.get(0); + let version: i64 = row.get(1); + let updated_at: DateTime = row.get(2); + StoredUserSettings { + user_key: user_key.to_string(), + document, + version, + updated_at, + } + })) + } + + async fn put_settings( + &self, + user_key: &str, + document: Value, + if_match_version: Option, + ) -> StorageResult { + self.write_settings(user_key, if_match_version, move |_current| document) + .await + } + + async fn patch_settings( + &self, + user_key: &str, + merge_patch: Value, + if_match_version: Option, + ) -> StorageResult { + self.write_settings(user_key, if_match_version, move |current| { + apply_merge_patch( + current.unwrap_or_else(|| Value::Object(Default::default())), + &merge_patch, + ) + }) + .await + } +} + +/// Builds an `OptimisticLockFailure` for a `user_settings` write whose +/// `If-Match` precondition did not match the live version. +fn lock_failure(user_key: &str, expected: i64, actual: i64) -> StorageError { + StorageError::Concurrency(ConcurrencyError::OptimisticLockFailure { + resource_type: "UserSettings".to_string(), + id: user_key.to_string(), + expected_etag: format!("W/\"{expected}\""), + actual_etag: Some(format!("W/\"{actual}\"")), + }) +} + +fn backend_err(message: String) -> StorageError { + StorageError::Backend(BackendError::Internal { + backend_name: "postgres".to_string(), + message, + source: None, + }) +} diff --git a/crates/persistence/src/backends/sqlite/mod.rs b/crates/persistence/src/backends/sqlite/mod.rs index c84226e5b..4131e6c18 100644 --- a/crates/persistence/src/backends/sqlite/mod.rs +++ b/crates/persistence/src/backends/sqlite/mod.rs @@ -75,5 +75,6 @@ pub mod search; mod search_impl; mod storage; mod transaction; +mod user_settings; pub use backend::{SqliteBackend, SqliteBackendConfig}; diff --git a/crates/persistence/src/backends/sqlite/schema.rs b/crates/persistence/src/backends/sqlite/schema.rs index 171436cd4..4a6082f3a 100644 --- a/crates/persistence/src/backends/sqlite/schema.rs +++ b/crates/persistence/src/backends/sqlite/schema.rs @@ -5,7 +5,7 @@ use rusqlite::Connection; use crate::error::StorageResult; /// Current schema version. -pub const SCHEMA_VERSION: i32 = 12; +pub const SCHEMA_VERSION: i32 = 13; /// Initialize the database schema. pub fn initialize_schema(conn: &Connection) -> StorageResult<()> { @@ -269,6 +269,7 @@ fn migrate_schema(conn: &Connection, from_version: i32) -> StorageResult<()> { 9 => migrate_v9_to_v10(conn)?, 10 => migrate_v10_to_v11(conn)?, 11 => migrate_v11_to_v12(conn)?, + 12 => migrate_v12_to_v13(conn)?, _ => { return Err(crate::error::StorageError::Backend( crate::error::BackendError::Internal { @@ -1190,6 +1191,27 @@ fn migrate_v11_to_v12(conn: &Connection) -> StorageResult<()> { add_bulk_submit_worker_schema(conn) } +/// Migrate from schema version 12 to version 13. +/// +/// Adds the `user_settings` table that backs the per-user UI settings store +/// (theme, default tenant, active FHIR version, recent queries, …). One opaque +/// JSON document is stored per user, keyed by `user_key`, with a monotonic +/// `version` for optimistic locking. This table is intentionally independent of +/// the FHIR `resources` table so UI preferences never leak into FHIR machinery. +fn migrate_v12_to_v13(conn: &Connection) -> StorageResult<()> { + conn.execute( + "CREATE TABLE IF NOT EXISTS user_settings ( + user_key TEXT NOT NULL PRIMARY KEY, + data BLOB NOT NULL, + version INTEGER NOT NULL DEFAULT 1, + updated_at TEXT NOT NULL + )", + [], + ) + .map_err(|e| migration_err(format!("create user_settings table: {e}")))?; + Ok(()) +} + fn migration_err(message: String) -> crate::error::StorageError { crate::error::StorageError::Backend(crate::error::BackendError::Internal { backend_name: "sqlite".to_string(), diff --git a/crates/persistence/src/backends/sqlite/user_settings.rs b/crates/persistence/src/backends/sqlite/user_settings.rs new file mode 100644 index 000000000..f88bc0f79 --- /dev/null +++ b/crates/persistence/src/backends/sqlite/user_settings.rs @@ -0,0 +1,305 @@ +//! SQLite implementation of the per-user [`SettingsStore`]. +//! +//! Each user owns a single row in the `user_settings` table holding an opaque +//! JSON document plus a monotonic `version` used for optimistic locking. Writes +//! perform a read-modify-write inside a transaction so concurrent updates to the +//! same user cannot lose data or skip the `If-Match` precondition check. + +use async_trait::async_trait; +use chrono::Utc; +use rusqlite::{OptionalExtension, params}; +use serde_json::Value; + +use crate::core::user_settings::{SettingsStore, StoredUserSettings, apply_merge_patch}; +use crate::error::{BackendError, ConcurrencyError, StorageError, StorageResult}; + +use super::SqliteBackend; + +impl SqliteBackend { + /// Read-modify-write a user's settings document inside a single transaction. + /// + /// `compute` receives the currently stored document (or `None` when the user + /// has no settings yet) and returns the document to persist. The optimistic + /// `if_match_version` precondition — where `Some(0)` asserts "does not yet + /// exist" — is checked against the live row before `compute` runs. + fn write_settings( + &self, + user_key: &str, + if_match_version: Option, + compute: impl FnOnce(Option) -> Value, + ) -> StorageResult { + let mut conn = self.get_connection()?; + let txn = conn + .transaction() + .map_err(|e| backend_err(format!("begin user_settings transaction: {e}")))?; + + let current: Option<(i64, Vec)> = txn + .query_row( + "SELECT version, data FROM user_settings WHERE user_key = ?1", + [user_key], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional() + .map_err(|e| backend_err(format!("read user_settings: {e}")))?; + + let current_version = current.as_ref().map(|(v, _)| *v).unwrap_or(0); + if let Some(expected) = if_match_version + && expected != current_version + { + return Err(lock_failure(user_key, expected, current_version)); + } + + let current_doc = match ¤t { + Some((_, data)) => Some( + serde_json::from_slice(data) + .map_err(|e| backend_err(format!("decode stored user_settings: {e}")))?, + ), + None => None, + }; + + let new_doc = compute(current_doc); + let new_version = current_version + 1; + let now = Utc::now(); + let updated_at = now.to_rfc3339(); + let data = serde_json::to_vec(&new_doc) + .map_err(|e| backend_err(format!("encode user_settings: {e}")))?; + + txn.execute( + "INSERT INTO user_settings (user_key, data, version, updated_at) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(user_key) DO UPDATE SET data = ?2, version = ?3, updated_at = ?4", + params![user_key, data, new_version, updated_at], + ) + .map_err(|e| backend_err(format!("write user_settings: {e}")))?; + + txn.commit() + .map_err(|e| backend_err(format!("commit user_settings: {e}")))?; + + Ok(StoredUserSettings { + user_key: user_key.to_string(), + document: new_doc, + version: new_version, + updated_at: now, + }) + } +} + +#[async_trait] +impl SettingsStore for SqliteBackend { + async fn get_settings(&self, user_key: &str) -> StorageResult> { + let conn = self.get_connection()?; + let row: Option<(Vec, i64, String)> = conn + .query_row( + "SELECT data, version, updated_at FROM user_settings WHERE user_key = ?1", + [user_key], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .optional() + .map_err(|e| backend_err(format!("read user_settings: {e}")))?; + + match row { + None => Ok(None), + Some((data, version, updated_at)) => { + let document = serde_json::from_slice(&data) + .map_err(|e| backend_err(format!("decode stored user_settings: {e}")))?; + let updated_at = chrono::DateTime::parse_from_rfc3339(&updated_at) + .map(|dt| dt.with_timezone(&Utc)) + .unwrap_or_else(|_| Utc::now()); + Ok(Some(StoredUserSettings { + user_key: user_key.to_string(), + document, + version, + updated_at, + })) + } + } + } + + async fn put_settings( + &self, + user_key: &str, + document: Value, + if_match_version: Option, + ) -> StorageResult { + self.write_settings(user_key, if_match_version, move |_current| document) + } + + async fn patch_settings( + &self, + user_key: &str, + merge_patch: Value, + if_match_version: Option, + ) -> StorageResult { + self.write_settings(user_key, if_match_version, move |current| { + apply_merge_patch( + current.unwrap_or_else(|| Value::Object(Default::default())), + &merge_patch, + ) + }) + } +} + +/// Builds an `OptimisticLockFailure` for a `user_settings` write whose +/// `If-Match` precondition did not match the live version. +fn lock_failure(user_key: &str, expected: i64, actual: i64) -> StorageError { + StorageError::Concurrency(ConcurrencyError::OptimisticLockFailure { + resource_type: "UserSettings".to_string(), + id: user_key.to_string(), + expected_etag: format!("W/\"{expected}\""), + actual_etag: Some(format!("W/\"{actual}\"")), + }) +} + +fn backend_err(message: String) -> StorageError { + StorageError::Backend(BackendError::Internal { + backend_name: "sqlite".to_string(), + message, + source: None, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn backend() -> SqliteBackend { + let backend = SqliteBackend::in_memory().expect("in-memory backend"); + backend.init_schema().expect("init schema"); + backend + } + + #[tokio::test] + async fn get_returns_none_for_unknown_user() { + let backend = backend(); + assert!(backend.get_settings("ghost").await.unwrap().is_none()); + } + + #[tokio::test] + async fn put_then_get_round_trips_document() { + let backend = backend(); + let doc = json!({"theme": "dark", "recentQueries": {"Patient": ["name=smith"]}}); + let stored = backend.put_settings("u1", doc.clone(), None).await.unwrap(); + assert_eq!(stored.version, 1); + + let fetched = backend.get_settings("u1").await.unwrap().unwrap(); + assert_eq!(fetched.document, doc); + assert_eq!(fetched.version, 1); + } + + #[tokio::test] + async fn version_increments_on_each_write() { + let backend = backend(); + backend + .put_settings("u1", json!({"a": 1}), None) + .await + .unwrap(); + let second = backend + .put_settings("u1", json!({"a": 2}), None) + .await + .unwrap(); + assert_eq!(second.version, 2); + } + + #[tokio::test] + async fn patch_merges_and_deletes_keys() { + let backend = backend(); + backend + .put_settings( + "u1", + json!({"theme": "dark", "defaultTenant": "acme"}), + None, + ) + .await + .unwrap(); + let patched = backend + .patch_settings("u1", json!({"theme": "light", "defaultTenant": null}), None) + .await + .unwrap(); + assert_eq!(patched.document, json!({"theme": "light"})); + assert_eq!(patched.version, 2); + } + + #[tokio::test] + async fn patch_on_missing_user_creates_document() { + let backend = backend(); + let patched = backend + .patch_settings("u1", json!({"theme": "dark"}), None) + .await + .unwrap(); + assert_eq!(patched.document, json!({"theme": "dark"})); + assert_eq!(patched.version, 1); + } + + #[tokio::test] + async fn stale_if_match_is_rejected() { + let backend = backend(); + backend + .put_settings("u1", json!({"a": 1}), None) + .await + .unwrap(); // version 1 + let err = backend + .put_settings("u1", json!({"a": 2}), Some(0)) + .await + .unwrap_err(); + assert!(matches!( + err, + StorageError::Concurrency(ConcurrencyError::OptimisticLockFailure { .. }) + )); + } + + #[tokio::test] + async fn matching_if_match_succeeds() { + let backend = backend(); + // Some(0) asserts "does not exist yet" for the first write. + backend + .put_settings("u1", json!({"a": 1}), Some(0)) + .await + .unwrap(); + let updated = backend + .put_settings("u1", json!({"a": 2}), Some(1)) + .await + .unwrap(); + assert_eq!(updated.version, 2); + } + + #[tokio::test] + async fn get_settings_surfaces_decode_error_for_corrupt_row() { + let backend = backend(); + // Write a row whose `data` blob is not valid JSON, bypassing the store. + { + let conn = backend.get_connection().unwrap(); + conn.execute( + "INSERT INTO user_settings (user_key, data, version, updated_at) \ + VALUES (?1, ?2, ?3, ?4)", + rusqlite::params![ + "corrupt", + b"not json".as_slice(), + 1_i64, + "2020-01-01T00:00:00Z" + ], + ) + .unwrap(); + } + let err = backend.get_settings("corrupt").await.unwrap_err(); + assert!(matches!(err, StorageError::Backend(_))); + } + + #[tokio::test] + async fn get_settings_tolerates_unparseable_timestamp() { + let backend = backend(); + // Valid JSON document but a malformed `updated_at`: read must not fail. + { + let conn = backend.get_connection().unwrap(); + conn.execute( + "INSERT INTO user_settings (user_key, data, version, updated_at) \ + VALUES (?1, ?2, ?3, ?4)", + rusqlite::params!["odd-ts", b"{}".as_slice(), 3_i64, "not-a-timestamp"], + ) + .unwrap(); + } + let stored = backend.get_settings("odd-ts").await.unwrap().unwrap(); + assert_eq!(stored.version, 3); + assert_eq!(stored.document, json!({})); + } +} diff --git a/crates/persistence/src/core/mod.rs b/crates/persistence/src/core/mod.rs index cfd0c5751..054b72e52 100644 --- a/crates/persistence/src/core/mod.rs +++ b/crates/persistence/src/core/mod.rs @@ -102,6 +102,7 @@ pub mod search; pub mod sof_runner; pub mod storage; pub mod transaction; +pub mod user_settings; pub mod versioned; // Re-export main types @@ -156,4 +157,5 @@ pub use transaction::{ BundleEntry, BundleEntryResult, BundleMethod, BundleProvider, BundleResult, BundleType, IsolationLevel, LockingStrategy, Transaction, TransactionOptions, TransactionProvider, }; +pub use user_settings::{SettingsStore, StoredUserSettings, apply_merge_patch}; pub use versioned::{VersionConflictInfo, VersionedStorage, check_version_match, normalize_etag}; diff --git a/crates/persistence/src/core/user_settings.rs b/crates/persistence/src/core/user_settings.rs new file mode 100644 index 000000000..734a0f04e --- /dev/null +++ b/crates/persistence/src/core/user_settings.rs @@ -0,0 +1,163 @@ +//! Per-user UI settings storage. +//! +//! This module defines [`SettingsStore`], a small storage abstraction for an +//! opaque, per-user JSON settings document. It is deliberately *separate* from +//! the FHIR [`ResourceStorage`](crate::core::ResourceStorage) hierarchy: UI +//! preferences such as theme, default tenant, active FHIR version, or recent +//! queries are private user state, not FHIR resources, and should not surface +//! in `CapabilityStatement`, `_history`, `_search`, or `$export`. +//! +//! # Design +//! +//! - **One document per user.** Each user owns a single JSON *object* keyed by a +//! caller-supplied `user_key` (e.g. `"{issuer}|{subject}"` from an +//! authenticated principal, or a fixed local key when auth is disabled). +//! - **Opaque and extensible.** The document is stored as an arbitrary +//! [`serde_json::Value`] object, so new settings keys require no schema or +//! code changes — the frontend owns the document shape. +//! - **User-global.** Settings are keyed by user only, not by tenant: a +//! preference like "default tenant" is inherently cross-tenant. Settings that +//! genuinely need per-tenant scoping should nest inside the document (e.g. +//! `{"perTenant": {"": {...}}}`) rather than changing the key. +//! - **Optimistically lockable.** Each document carries a monotonic `version` +//! that increments on every write and is surfaced to clients as a weak ETag +//! (`W/"{version}"`). Callers may pass `if_match_version` to make a write +//! conditional and avoid lost updates. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use serde_json::Value; + +use crate::error::StorageResult; + +/// A stored per-user settings document together with its optimistic-lock +/// version and last-modified timestamp. +#[derive(Debug, Clone)] +pub struct StoredUserSettings { + /// The opaque key identifying the owning user. + pub user_key: String, + /// The settings document. Always a JSON object. + pub document: Value, + /// Monotonic version, bumped on every write. Surfaced as the `W/"{version}"` + /// ETag so clients can perform conditional (`If-Match`) updates. + pub version: i64, + /// Timestamp of the most recent write. + pub updated_at: DateTime, +} + +/// Storage abstraction for opaque, per-user JSON settings documents. +/// +/// Implemented by the SQLite and PostgreSQL backends. The trait is intentionally +/// minimal — get the whole document, replace it, or merge-patch it — because the +/// document body is opaque to the server. +#[async_trait] +pub trait SettingsStore: Send + Sync { + /// Returns the user's settings document, or `None` if the user has never + /// stored any settings. + async fn get_settings(&self, user_key: &str) -> StorageResult>; + + /// Replaces the user's entire settings document. + /// + /// `document` must be a JSON object; callers are expected to validate this + /// before invoking. When `if_match_version` is `Some`, the write only + /// succeeds if it matches the currently stored version, otherwise a + /// [`ConcurrencyError::OptimisticLockFailure`](crate::error::ConcurrencyError::OptimisticLockFailure) + /// is returned. A `Some(0)` precondition asserts the document does not yet + /// exist. + async fn put_settings( + &self, + user_key: &str, + document: Value, + if_match_version: Option, + ) -> StorageResult; + + /// Applies an [RFC 7386](https://www.rfc-editor.org/rfc/rfc7386) JSON Merge + /// Patch to the user's settings document, creating an empty document first + /// if none exists. + /// + /// The read-modify-write is performed atomically by the backend (under a row + /// lock / transaction). `if_match_version` enforces optimistic locking as in + /// [`put_settings`](Self::put_settings). + async fn patch_settings( + &self, + user_key: &str, + merge_patch: Value, + if_match_version: Option, + ) -> StorageResult; +} + +/// Applies an [RFC 7386](https://www.rfc-editor.org/rfc/rfc7386) JSON Merge +/// Patch to `target`, returning the merged result. +/// +/// Per the specification: a non-object patch replaces the target outright; an +/// object patch is applied member-wise, where a `null` member deletes the +/// corresponding key and any other value is merged recursively. +pub fn apply_merge_patch(target: Value, patch: &Value) -> Value { + match patch { + Value::Object(patch_members) => { + // A non-object target is discarded in favor of an empty object, per + // the spec ("if Target is not an Object, set it to an empty Object"). + let mut merged = match target { + Value::Object(map) => map, + _ => serde_json::Map::new(), + }; + for (key, patch_value) in patch_members { + if patch_value.is_null() { + merged.remove(key); + } else { + let existing = merged.remove(key).unwrap_or(Value::Null); + merged.insert(key.clone(), apply_merge_patch(existing, patch_value)); + } + } + Value::Object(merged) + } + _ => patch.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn merge_patch_sets_and_overwrites_keys() { + let target = json!({"theme": "dark", "defaultTenant": "acme"}); + let patch = json!({"theme": "light"}); + assert_eq!( + apply_merge_patch(target, &patch), + json!({"theme": "light", "defaultTenant": "acme"}) + ); + } + + #[test] + fn merge_patch_null_deletes_key() { + let target = json!({"theme": "dark", "defaultTenant": "acme"}); + let patch = json!({"defaultTenant": null}); + assert_eq!(apply_merge_patch(target, &patch), json!({"theme": "dark"})); + } + + #[test] + fn merge_patch_merges_nested_objects() { + let target = json!({"recentQueries": {"Patient": ["name=smith"]}}); + let patch = json!({"recentQueries": {"Observation": ["code=1234"]}}); + assert_eq!( + apply_merge_patch(target, &patch), + json!({"recentQueries": {"Patient": ["name=smith"], "Observation": ["code=1234"]}}) + ); + } + + #[test] + fn merge_patch_non_object_replaces_target() { + let target = json!({"theme": "dark"}); + let patch = json!("scalar"); + assert_eq!(apply_merge_patch(target, &patch), json!("scalar")); + } + + #[test] + fn merge_patch_replaces_non_object_target_with_object() { + let target = json!("not-an-object"); + let patch = json!({"theme": "dark"}); + assert_eq!(apply_merge_patch(target, &patch), json!({"theme": "dark"})); + } +} diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 92f3f5ae1..436931eae 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -17,7 +17,7 @@ use helios_persistence::core::{ Backend, BackendCapability, BackendKind, BundleEntry, BundleMethod, BundleProvider, BundleResult, ConditionalCreateResult, ConditionalDeleteResult, ConditionalStorage, ConditionalUpdateResult, HistoryParams, IncludeProvider, InstanceHistoryProvider, PatchFormat, - ResourceStorage, RevincludeProvider, SearchProvider, SystemHistoryProvider, + ResourceStorage, RevincludeProvider, SearchProvider, SettingsStore, SystemHistoryProvider, TypeHistoryProvider, VersionedStorage, }; use helios_persistence::error::{ @@ -2554,3 +2554,290 @@ async fn mongodb_integration_resolve_include_and_revinclude() { "search() should populate `included` from revinclude resolution" ); } + +// ============================================================================ +// Per-user settings store +// ============================================================================ + +/// A user key unique to each test, so tests sharing a database don't collide on +/// the single-document-per-user `user_settings` collection. +fn unique_user_key(prefix: &str) -> String { + format!("{}|{}", prefix, uuid::Uuid::new_v4().simple()) +} + +/// Backend provisioning for the settings-store tests. +/// +/// Unlike the rest of this suite — which only runs against an externally +/// supplied mongo (`HFS_TEST_MONGODB_URL`) — these tests also start an ephemeral +/// standalone Mongo testcontainer when no URL is set, so they run (and cover the +/// [`SettingsStore`] impl) in CI where Docker is available, mirroring how the +/// PostgreSQL suite is provisioned. The settings store uses version-conditioned +/// writes rather than multi-document transactions, so a standalone (non +/// replica-set) Mongo is sufficient. When neither a URL nor Docker is available +/// the tests skip. +mod settings_mongo { + use super::{MongoBackend, MongoBackendConfig, build_test_database_name, test_mongo_url}; + use helios_persistence::core::Backend; // brings `initialize` into scope + use testcontainers::ImageExt; + use testcontainers::runners::AsyncRunner; + use testcontainers_modules::mongo::Mongo; + use tokio::sync::OnceCell; + + /// Mongo endpoint shared across the settings tests. Holds the container so it + /// stays alive for the test binary; reaped by the testcontainers watchdog and + /// the CI cleanup step (by `github.run_id` label), matching the PG suite. + struct SharedMongo { + connection_string: String, + _container: Option>, + } + + static SHARED: OnceCell> = OnceCell::const_new(); + + async fn shared() -> Option<&'static SharedMongo> { + SHARED + .get_or_init(|| async { + // Prefer an explicitly provided mongo (fast local runs). + if let Some(url) = test_mongo_url() { + return Some(SharedMongo { + connection_string: url, + _container: None, + }); + } + // Otherwise start an ephemeral standalone Mongo container; if + // Docker is unavailable, `start()` errors and the tests skip. + let run_id = std::env::var("GITHUB_RUN_ID").unwrap_or_default(); + let container = Mongo::default() + .with_label("github.run_id", &run_id) + .start() + .await + .ok()?; + let host = container.get_host().await.ok()?; + let port = container.get_host_port_ipv4(27017).await.ok()?; + Some(SharedMongo { + connection_string: format!("mongodb://{host}:{port}/"), + _container: Some(container), + }) + }) + .await + .as_ref() + } + + /// Returns a schema-initialised backend against the shared mongo, or `None` + /// when no mongo is available (skip). + pub(super) async fn backend(test_name: &str) -> Option { + let shared = shared().await?; + let config = MongoBackendConfig { + connection_string: shared.connection_string.clone(), + database_name: build_test_database_name(test_name), + ..Default::default() + }; + let backend = MongoBackend::new(config).expect("failed to create MongoBackend"); + backend + .initialize() + .await + .expect("failed to initialize MongoDB schema"); + Some(backend) + } +} + +#[tokio::test] +async fn mongodb_integration_settings_get_missing_is_none() { + let Some(backend) = settings_mongo::backend("settings_missing").await else { + eprintln!( + "Skipping mongodb_integration_settings_get_missing_is_none (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let user = unique_user_key("missing"); + assert!(backend.get_settings(&user).await.unwrap().is_none()); +} + +#[tokio::test] +async fn mongodb_integration_settings_put_get_and_version() { + let Some(backend) = settings_mongo::backend("settings_round_trip").await else { + eprintln!( + "Skipping mongodb_integration_settings_put_get_and_version (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let user = unique_user_key("round-trip"); + let doc = json!({"theme": "dark", "recentQueries": {"Patient": ["name=smith"]}}); + + let stored = backend + .put_settings(&user, doc.clone(), None) + .await + .unwrap(); + assert_eq!(stored.version, 1); + + let fetched = backend.get_settings(&user).await.unwrap().unwrap(); + assert_eq!(fetched.document, doc); + assert_eq!(fetched.version, 1); + + // A second unconditional write replaces the document and bumps the version. + let second = backend + .put_settings(&user, json!({"theme": "light"}), None) + .await + .unwrap(); + assert_eq!(second.version, 2); + assert_eq!(second.document, json!({"theme": "light"})); +} + +#[tokio::test] +async fn mongodb_integration_settings_patch_merges_and_deletes() { + let Some(backend) = settings_mongo::backend("settings_patch").await else { + eprintln!( + "Skipping mongodb_integration_settings_patch_merges_and_deletes (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let user = unique_user_key("patch"); + backend + .put_settings( + &user, + json!({"theme": "dark", "defaultTenant": "acme"}), + None, + ) + .await + .unwrap(); + + let patched = backend + .patch_settings( + &user, + json!({"theme": "light", "defaultTenant": null}), + None, + ) + .await + .unwrap(); + assert_eq!(patched.document, json!({"theme": "light"})); + assert_eq!(patched.version, 2); +} + +#[tokio::test] +async fn mongodb_integration_settings_patch_on_missing_creates_document() { + let Some(backend) = settings_mongo::backend("settings_patch_create").await else { + eprintln!( + "Skipping mongodb_integration_settings_patch_on_missing_creates_document (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let user = unique_user_key("patch-create"); + let patched = backend + .patch_settings(&user, json!({"theme": "dark"}), None) + .await + .unwrap(); + assert_eq!(patched.document, json!({"theme": "dark"})); + assert_eq!(patched.version, 1); +} + +#[tokio::test] +async fn mongodb_integration_settings_optimistic_lock() { + let Some(backend) = settings_mongo::backend("settings_lock").await else { + eprintln!( + "Skipping mongodb_integration_settings_optimistic_lock (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let user = unique_user_key("lock"); + + // `Some(0)` asserts "does not exist yet" — succeeds for the first write. + backend + .put_settings(&user, json!({"a": 1}), Some(0)) + .await + .unwrap(); // version 1 + + // A stale precondition against an existing document is rejected. + let err = backend + .put_settings(&user, json!({"a": 2}), Some(0)) + .await + .unwrap_err(); + assert!(matches!( + err, + StorageError::Concurrency(ConcurrencyError::OptimisticLockFailure { .. }) + )); + + // A matching precondition succeeds and bumps the version. + let ok = backend + .put_settings(&user, json!({"a": 2}), Some(1)) + .await + .unwrap(); + assert_eq!(ok.version, 2); +} + +#[tokio::test] +async fn mongodb_integration_settings_concurrent_patches_serialize() { + let Some(backend) = settings_mongo::backend("settings_concurrent").await else { + eprintln!( + "Skipping mongodb_integration_settings_concurrent_patches_serialize (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let backend = Arc::new(backend); + let user = unique_user_key("concurrent"); + + // Deliberately do NOT seed a document: the racers start from version 0, so + // exactly one wins the initial insert and the rest hit the unique-index + // duplicate-key path and retry into the version-conditioned update. This + // exercises both the insert race and the update race in one test. + // + // Fire many unconditional single-key merge-patches concurrently. The + // version-conditioned write + retry loop must serialize them so every key + // survives (no lost updates) despite the read-modify-write race. + let mut handles = Vec::new(); + for i in 0..12 { + let backend = backend.clone(); + let user = user.clone(); + handles.push(tokio::spawn(async move { + backend + .patch_settings(&user, json!({ format!("k{i}"): i }), None) + .await + .unwrap(); + })); + } + for h in handles { + h.await.unwrap(); + } + + let final_doc = backend.get_settings(&user).await.unwrap().unwrap(); + let obj = final_doc.document.as_object().unwrap(); + for i in 0..12 { + assert_eq!( + obj.get(&format!("k{i}")), + Some(&json!(i)), + "key k{i} was lost to a read-modify-write race" + ); + } + // 12 patches, each a distinct successful write from version 0 upward. + assert_eq!(final_doc.version, 12); +} + +/// A row whose `data` is not valid JSON must surface a backend error on read, +/// rather than panicking or silently returning an empty document. Mirrors the +/// SQLite/PostgreSQL `user_settings` decode-error coverage. +#[tokio::test] +async fn mongodb_integration_settings_get_surfaces_decode_error() { + let Some(backend) = settings_mongo::backend("settings_decode_err").await else { + eprintln!( + "Skipping mongodb_integration_settings_get_surfaces_decode_error (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let user = unique_user_key("corrupt"); + + // Write a row whose `data` blob is not valid JSON, bypassing the store. + let client = Client::with_uri_str(&backend.config().connection_string) + .await + .expect("connect raw mongo client"); + let db = client.database(&backend.config().database_name); + db.collection::("user_settings") + .insert_one(doc! { + "user_key": &user, + "data": "not json", + "version": 1_i64, + "updated_at": mongodb::bson::DateTime::from_millis(0), + }) + .await + .expect("insert corrupt user_settings row"); + + let err = backend.get_settings(&user).await.unwrap_err(); + assert!(matches!(err, StorageError::Backend(_))); +} diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index 87b1b6655..516bd92a7 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -689,9 +689,10 @@ mod postgres_integration { use serde_json::json; use helios_persistence::backends::postgres::{PostgresBackend, PostgresConfig}; + use helios_persistence::core::SettingsStore; use helios_persistence::core::history::{HistoryParams, InstanceHistoryProvider}; use helios_persistence::core::{Backend, BackendCapability, BackendKind, ResourceStorage}; - use helios_persistence::error::{ResourceError, StorageError}; + use helios_persistence::error::{ConcurrencyError, ResourceError, StorageError}; use helios_persistence::tenant::{TenantContext, TenantId, TenantPermissions}; use testcontainers::ImageExt; @@ -3846,6 +3847,98 @@ mod postgres_integration { assert!(expired_now.is_empty()); } + // ======================================================================== + // Per-user settings store + // ======================================================================== + + /// A user key unique to each test, so tests sharing the database don't + /// collide on the single-row-per-user `user_settings` table. + fn unique_user_key(prefix: &str) -> String { + format!("{}|{}", prefix, uuid::Uuid::new_v4().simple()) + } + + #[tokio::test] + async fn postgres_integration_settings_get_missing_is_none() { + let backend = create_backend().await; + let user = unique_user_key("missing"); + assert!(backend.get_settings(&user).await.unwrap().is_none()); + } + + #[tokio::test] + async fn postgres_integration_settings_put_get_and_version() { + let backend = create_backend().await; + let user = unique_user_key("round-trip"); + let doc = json!({"theme": "dark", "recentQueries": {"Patient": ["name=smith"]}}); + + let stored = backend + .put_settings(&user, doc.clone(), None) + .await + .unwrap(); + assert_eq!(stored.version, 1); + + let fetched = backend.get_settings(&user).await.unwrap().unwrap(); + assert_eq!(fetched.document, doc); + assert_eq!(fetched.version, 1); + + let second = backend + .put_settings(&user, json!({"theme": "light"}), None) + .await + .unwrap(); + assert_eq!(second.version, 2); + } + + #[tokio::test] + async fn postgres_integration_settings_patch_merges_and_deletes() { + let backend = create_backend().await; + let user = unique_user_key("patch"); + backend + .put_settings( + &user, + json!({"theme": "dark", "defaultTenant": "acme"}), + None, + ) + .await + .unwrap(); + + let patched = backend + .patch_settings( + &user, + json!({"theme": "light", "defaultTenant": null}), + None, + ) + .await + .unwrap(); + assert_eq!(patched.document, json!({"theme": "light"})); + assert_eq!(patched.version, 2); + } + + #[tokio::test] + async fn postgres_integration_settings_optimistic_lock() { + let backend = create_backend().await; + let user = unique_user_key("lock"); + backend + .put_settings(&user, json!({"a": 1}), None) + .await + .unwrap(); // version 1 + + // Stale precondition is rejected. + let err = backend + .put_settings(&user, json!({"a": 2}), Some(0)) + .await + .unwrap_err(); + assert!(matches!( + err, + StorageError::Concurrency(ConcurrencyError::OptimisticLockFailure { .. }) + )); + + // Matching precondition succeeds. + let ok = backend + .put_settings(&user, json!({"a": 2}), Some(1)) + .await + .unwrap(); + assert_eq!(ok.version, 2); + } + // ======================================================================== // _contained / _containedType search // ======================================================================== diff --git a/crates/rest/src/extractors/mod.rs b/crates/rest/src/extractors/mod.rs index 68eb2747a..ffdc55185 100644 --- a/crates/rest/src/extractors/mod.rs +++ b/crates/rest/src/extractors/mod.rs @@ -16,6 +16,7 @@ pub(crate) mod query_pairs; mod search_params; pub mod search_query_builder; mod tenant; +mod user; pub use fhir_resource::FhirResource; pub use fhir_version::FhirVersionExtractor; @@ -26,3 +27,4 @@ pub use search_query_builder::{ unknown_search_params, }; pub use tenant::TenantExtractor; +pub use user::UserKey; diff --git a/crates/rest/src/extractors/user.rs b/crates/rest/src/extractors/user.rs new file mode 100644 index 000000000..2d201ad89 --- /dev/null +++ b/crates/rest/src/extractors/user.rs @@ -0,0 +1,82 @@ +//! User-identity extractor for per-user endpoints. +//! +//! [`UserKey`] derives a stable, opaque key identifying the caller, used to +//! scope the per-user settings store. It reads the authenticated +//! [`Principal`](helios_auth::Principal) injected into request extensions by the +//! auth middleware; when authentication is disabled (no principal present), it +//! falls back to a fixed local key so single-user / development deployments +//! still get a stable place to persist settings. + +use axum::{extract::FromRequestParts, http::request::Parts}; + +/// The key used when no authenticated principal is present (auth disabled). +const LOCAL_USER_KEY: &str = "local|default"; + +/// A stable, opaque identifier for the calling user. +/// +/// Derived as `"{issuer}|{subject}"` from the authenticated principal — the +/// issuer qualifies the subject so identifiers from different identity providers +/// cannot collide — or [`LOCAL_USER_KEY`] when auth is disabled. +#[derive(Debug, Clone)] +pub struct UserKey(pub String); + +impl UserKey { + /// Returns the user key as a string slice. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl FromRequestParts for UserKey +where + S: Send + Sync, +{ + type Rejection = std::convert::Infallible; + + async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { + let key = match parts.extensions.get::() { + Some(principal) => format!("{}|{}", principal.issuer(), principal.subject()), + None => LOCAL_USER_KEY.to_string(), + }; + Ok(UserKey(key)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::http::Request; + use chrono::Utc; + use helios_auth::Principal; + use helios_auth::scope::ScopeSet; + + async fn extract(req: Request<()>) -> UserKey { + let (mut parts, _) = req.into_parts(); + UserKey::from_request_parts(&mut parts, &()) + .await + .expect("infallible") + } + + #[tokio::test] + async fn falls_back_to_local_key_without_principal() { + let key = extract(Request::builder().body(()).unwrap()).await; + assert_eq!(key.as_str(), "local|default"); + } + + #[tokio::test] + async fn derives_issuer_qualified_key_from_principal() { + let principal = Principal { + subject: "user-123".to_string(), + issuer: "https://idp.example.com".to_string(), + tenant_id: None, + scopes: ScopeSet::default(), + jti: None, + expires_at: Utc::now(), + custom_claims: serde_json::Map::new(), + }; + let mut req = Request::builder().body(()).unwrap(); + req.extensions_mut().insert(principal); + let key = extract(req).await; + assert_eq!(key.as_str(), "https://idp.example.com|user-123"); + } +} diff --git a/crates/rest/src/handlers/mod.rs b/crates/rest/src/handlers/mod.rs index 7c52f8a60..d638a9f8f 100644 --- a/crates/rest/src/handlers/mod.rs +++ b/crates/rest/src/handlers/mod.rs @@ -36,6 +36,7 @@ pub mod subscription_event; #[cfg(feature = "subscriptions")] pub mod subscriptions; pub mod update; +pub mod user_settings; pub mod versions; pub mod vread; #[cfg(feature = "subscriptions")] @@ -78,5 +79,6 @@ pub use patch::patch_handler; pub use read::{head_read_handler, read_handler}; pub use search::{search_get_handler, search_post_handler}; pub use update::{conditional_update_handler, update_handler}; +pub use user_settings::{get_user_settings, patch_user_settings, put_user_settings}; pub use versions::versions_handler; pub use vread::vread_handler; diff --git a/crates/rest/src/handlers/user_settings.rs b/crates/rest/src/handlers/user_settings.rs new file mode 100644 index 000000000..2f0de3f32 --- /dev/null +++ b/crates/rest/src/handlers/user_settings.rs @@ -0,0 +1,249 @@ +//! Per-user UI settings handlers. +//! +//! Implements a small `application/json` API for an opaque, per-user settings +//! document (theme, default tenant, active FHIR version, recent queries, …): +//! +//! - `GET /_user/settings` — fetch the document (defaults to `{}`) +//! - `PUT /_user/settings` — replace the whole document +//! - `PATCH /_user/settings` — [RFC 7386] JSON merge-patch a subset of keys +//! +//! The endpoints live under a leading-underscore path so they are authenticated +//! (a [`Principal`](helios_auth::Principal) is injected when auth is enabled) but +//! exempt from FHIR scope checks, and invisible to FHIR machinery +//! (`CapabilityStatement`, search, history, export). +//! +//! Each response carries a weak `ETag` (`W/"{version}"`). Clients may send +//! `If-Match` on `PUT`/`PATCH` for optimistic concurrency, or `If-None-Match` +//! on `GET` for conditional fetches. +//! +//! [RFC 7386]: https://www.rfc-editor.org/rfc/rfc7386 + +use std::sync::Arc; + +use axum::{ + Json, + body::Bytes, + extract::State, + http::{StatusCode, header}, + response::{IntoResponse, Response}, +}; +use helios_persistence::core::{ResourceStorage, SettingsStore, StoredUserSettings}; +use serde_json::Value; + +use crate::error::{RestError, RestResult}; +use crate::extractors::UserKey; +use crate::middleware::conditional::ConditionalHeaders; +use crate::state::AppState; + +/// Handler for `GET /_user/settings`. +/// +/// Returns the caller's settings document, or an empty object (`{}`, version 0) +/// when none has been stored yet, so the UI always receives a usable document. +pub async fn get_user_settings( + State(state): State>, + user: UserKey, + conditional: ConditionalHeaders, +) -> RestResult +where + S: ResourceStorage + Send + Sync, +{ + let store = settings_store(&state)?; + let (document, version) = match store.get_settings(user.as_str()).await? { + Some(stored) => (stored.document, stored.version), + None => (Value::Object(Default::default()), 0), + }; + let etag = weak_etag(version); + + // Honor If-None-Match only when a document actually exists; an empty default + // document (version 0) must never be reported as "not modified". + if version > 0 + && let Some(inm) = conditional.if_none_match() + && (inm == etag || inm == "*") + { + return Ok((StatusCode::NOT_MODIFIED, [(header::ETAG, etag)]).into_response()); + } + + Ok(([(header::ETAG, etag)], Json(document)).into_response()) +} + +/// Handler for `PUT /_user/settings`. +/// +/// Replaces the caller's entire settings document with the request body, which +/// must be a JSON object. An optional `If-Match` header makes the write +/// conditional on the current version. +pub async fn put_user_settings( + State(state): State>, + user: UserKey, + conditional: ConditionalHeaders, + body: Bytes, +) -> RestResult +where + S: ResourceStorage + Send + Sync, +{ + let store = settings_store(&state)?; + let document = parse_object_body(&body)?; + let if_match = parse_if_match_version(&conditional); + let stored = store + .put_settings(user.as_str(), document, if_match) + .await?; + Ok(settings_response(stored)) +} + +/// Handler for `PATCH /_user/settings`. +/// +/// Applies an [RFC 7386] JSON merge-patch (request body, a JSON object) to the +/// caller's settings document — ideal for toggling a single key such as the +/// theme. An optional `If-Match` header makes the write conditional. +/// +/// [RFC 7386]: https://www.rfc-editor.org/rfc/rfc7386 +pub async fn patch_user_settings( + State(state): State>, + user: UserKey, + conditional: ConditionalHeaders, + body: Bytes, +) -> RestResult +where + S: ResourceStorage + Send + Sync, +{ + let store = settings_store(&state)?; + let merge_patch = parse_object_body(&body)?; + let if_match = parse_if_match_version(&conditional); + let stored = store + .patch_settings(user.as_str(), merge_patch, if_match) + .await?; + Ok(settings_response(stored)) +} + +/// Returns the configured settings store, or a `501 Not Implemented` error when +/// the active backend does not provide one. +/// +/// The per-user settings store is implemented by every standalone *primary* +/// backend that offers the required read-modify-write + monotonic-version +/// primitives: SQLite, PostgreSQL, and MongoDB. It is intentionally unavailable +/// on backends that are not a transactional primary FHIR store — notably the +/// S3 object store (whose recommended role is archival and whose concurrency / +/// version story differs; tracked in issue #199) and Elasticsearch (search-only, +/// never a standalone primary). The message names the supported backends so an +/// operator on an unsupported one gets an explained `501` rather than a bare one. +fn settings_store(state: &AppState) -> RestResult<&Arc> +where + S: ResourceStorage + Send + Sync, +{ + state + .settings_store() + .ok_or_else(|| RestError::NotImplemented { + feature: "per-user settings (supported on the SQLite, PostgreSQL, and MongoDB \ + backends; not available on the S3 or Elasticsearch backends)" + .to_string(), + }) +} + +/// Parses and validates a request body as a JSON object. +fn parse_object_body(body: &Bytes) -> RestResult { + if body.is_empty() { + return Err(RestError::BadRequest { + message: "Request body must be a JSON object".to_string(), + }); + } + let value: Value = serde_json::from_slice(body).map_err(|e| RestError::BadRequest { + message: format!("Invalid JSON: {e}"), + })?; + if !value.is_object() { + return Err(RestError::BadRequest { + message: "Settings document must be a JSON object".to_string(), + }); + } + Ok(value) +} + +/// Extracts the version number from an `If-Match` weak ETag (`W/"{n}"`, `"{n}"`, +/// or bare `{n}`). A wildcard (`*`) or absent/unparseable header yields `None`, +/// meaning "no version precondition". +fn parse_if_match_version(conditional: &ConditionalHeaders) -> Option { + let raw = conditional.if_match()?.trim(); + if raw == "*" { + return None; + } + raw.trim_start_matches("W/").trim_matches('"').parse().ok() +} + +/// Builds the success response for a write: the stored document plus its ETag. +fn settings_response(stored: StoredUserSettings) -> Response { + ( + [(header::ETAG, weak_etag(stored.version))], + Json(stored.document), + ) + .into_response() +} + +/// Formats a version number as a weak ETag. +fn weak_etag(version: i64) -> String { + format!("W/\"{version}\"") +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::http::HeaderMap; + + #[test] + fn parse_object_body_rejects_empty() { + let err = parse_object_body(&Bytes::new()).unwrap_err(); + assert!(matches!(err, RestError::BadRequest { .. })); + } + + #[test] + fn parse_object_body_rejects_invalid_json() { + let err = parse_object_body(&Bytes::from_static(b"{ not json")).unwrap_err(); + assert!(matches!(err, RestError::BadRequest { .. })); + } + + #[test] + fn parse_object_body_rejects_non_object() { + let err = parse_object_body(&Bytes::from_static(b"[1, 2, 3]")).unwrap_err(); + assert!(matches!(err, RestError::BadRequest { .. })); + } + + #[test] + fn parse_object_body_accepts_object() { + let value = parse_object_body(&Bytes::from_static(b"{\"theme\":\"dark\"}")).unwrap(); + assert!(value.is_object()); + } + + #[test] + fn parse_if_match_version_variants() { + let with_if_match = |raw: &str| { + let mut headers = HeaderMap::new(); + headers.insert(header::IF_MATCH, raw.parse().unwrap()); + ConditionalHeaders::from_headers(&headers) + }; + assert_eq!(parse_if_match_version(&with_if_match("W/\"5\"")), Some(5)); + assert_eq!(parse_if_match_version(&with_if_match("\"7\"")), Some(7)); + assert_eq!(parse_if_match_version(&with_if_match("9")), Some(9)); + // A wildcard means "no version precondition". + assert_eq!(parse_if_match_version(&with_if_match("*")), None); + // An unparseable value is ignored rather than rejected. + assert_eq!(parse_if_match_version(&with_if_match("garbage")), None); + // An absent header yields no precondition. + assert_eq!( + parse_if_match_version(&ConditionalHeaders::from_headers(&HeaderMap::new())), + None + ); + } + + /// Backends without a settings store surface `501 Not Implemented`. + #[cfg(feature = "sqlite")] + #[test] + fn settings_store_absent_returns_not_implemented() { + use crate::config::ServerConfig; + use helios_persistence::backends::sqlite::SqliteBackend; + + let backend = SqliteBackend::in_memory().expect("in-memory sqlite"); + backend.init_schema().expect("init schema"); + // `AppState::new` leaves the settings store unset. The `Ok` variant + // (`&Arc`) is not `Debug`, so match instead of unwrap. + let state = AppState::new(Arc::new(backend), ServerConfig::default()); + let result = settings_store(&state); + assert!(matches!(result, Err(RestError::NotImplemented { .. }))); + } +} diff --git a/crates/rest/src/lib.rs b/crates/rest/src/lib.rs index 9890aca9a..3a8141e3f 100644 --- a/crates/rest/src/lib.rs +++ b/crates/rest/src/lib.rs @@ -340,6 +340,7 @@ where audit_state, None, None, + None, ) } @@ -378,6 +379,7 @@ where audit_state, Some(bulk_export), None, + None, ) } @@ -419,11 +421,58 @@ where audit_state, bulk_export, bulk_submit, + None, + ) +} + +/// Like [`create_app_with_auth_and_bulk`], but also wires the per-user settings +/// store (used by the `/_user/settings` endpoints). `bulk_export` and +/// `bulk_submit` are each optional, so this single entry point covers every +/// combination for a settings-capable backend (SQLite, PostgreSQL). +#[allow(clippy::too_many_arguments)] +pub fn create_app_with_auth_bulk_and_settings( + storage: Arc, + config: ServerConfig, + auth_config: helios_auth::AuthConfig, + auth_state: Option>, + audit_state: Option>, + bulk_export: Option, + bulk_submit: Option, + settings_store: Option>, +) -> Router +where + S: ResourceStorage + + ConditionalStorage + + SearchProvider + + IncludeProvider + + RevincludeProvider + + InstanceHistoryProvider + + TypeHistoryProvider + + SystemHistoryProvider + + BundleProvider + + helios_persistence::core::ExportDataProvider + + helios_persistence::core::PatientExportProvider + + helios_persistence::core::GroupExportProvider + + Send + + Sync + + 'static, +{ + build_app( + storage, + config, + auth_config, + auth_state, + audit_state, + bulk_export, + bulk_submit, + settings_store, ) } -/// Internal app builder shared by [`create_app_with_auth`] and -/// [`create_app_with_auth_and_bulk_export`]. +/// Internal app builder shared by [`create_app_with_auth`], +/// [`create_app_with_auth_and_bulk_export`], [`create_app_with_auth_and_bulk`], +/// and [`create_app_with_auth_bulk_and_settings`]. +#[allow(clippy::too_many_arguments)] fn build_app( storage: Arc, config: ServerConfig, @@ -432,6 +481,7 @@ fn build_app( audit_state: Option>, bulk_export: Option, bulk_submit: Option, + settings_store: Option>, ) -> Router where S: ResourceStorage @@ -595,6 +645,12 @@ where None => state, }; + // Wire the per-user settings store if provided. + let state = match settings_store { + Some(store) => state.with_settings_store(store), + None => state, + }; + // Inject subscription engine if enabled #[cfg(feature = "subscriptions")] let state = { @@ -963,3 +1019,55 @@ pub fn init_logging(level: &str) { .with(filter) .init(); } + +#[cfg(all(test, feature = "sqlite"))] +mod builder_tests { + use super::*; + use helios_persistence::backends::sqlite::SqliteBackend; + + fn backend() -> Arc { + let backend = SqliteBackend::in_memory().expect("in-memory sqlite"); + backend.init_schema().expect("init schema"); + Arc::new(backend) + } + + /// SOF is disabled so `build_app` skips the in-DB runner / export-controller + /// path and stays a pure, side-effect-free router build. + fn config() -> ServerConfig { + let mut config = ServerConfig::default(); + config.sof_enabled = false; + config + } + + /// The generic bulk builder wires a router with no bundles (bulk export and + /// bulk submit both disabled). + #[tokio::test] + async fn builds_app_with_bulk_builder_and_no_bundles() { + let _app: Router = create_app_with_auth_and_bulk( + backend(), + config(), + helios_auth::AuthConfig::default(), + None, + None, + None, + None, + ); + } + + /// The settings-capable builder wires the settings store into the router. + #[tokio::test] + async fn builds_app_with_settings_store() { + let backend = backend(); + let settings: Arc = backend.clone(); + let _app: Router = create_app_with_auth_bulk_and_settings( + backend, + config(), + helios_auth::AuthConfig::default(), + None, + None, + None, + None, + Some(settings), + ); + } +} diff --git a/crates/rest/src/routing/fhir_routes.rs b/crates/rest/src/routing/fhir_routes.rs index 5e7295262..c92257bd4 100644 --- a/crates/rest/src/routing/fhir_routes.rs +++ b/crates/rest/src/routing/fhir_routes.rs @@ -233,6 +233,14 @@ where get(handlers::smart_discovery::smart_configuration_handler::), ) .route("/_history", get(handlers::history_system_handler::)) + // Per-user UI settings. The leading `_` keeps these authenticated yet + // exempt from FHIR scope checks, and out of the FHIR resource namespace. + .route( + "/_user/settings", + get(handlers::get_user_settings::) + .put(handlers::put_user_settings::) + .patch(handlers::patch_user_settings::), + ) .route("/", post(handlers::batch_handler::)) // Bulk Data Export ($export) — operation routes precede the catch-all. .route( diff --git a/crates/rest/src/state.rs b/crates/rest/src/state.rs index f4fee7956..b65852bdf 100644 --- a/crates/rest/src/state.rs +++ b/crates/rest/src/state.rs @@ -10,7 +10,8 @@ use helios_audit::AuditSink; use helios_auth::AuthConfig; use helios_persistence::core::sof_runner::SofRunner; use helios_persistence::core::{ - BulkExportJobStore, BulkSubmitJobStore, ExportOutputStore, ResourceStorage, SubmitInputFetcher, + BulkExportJobStore, BulkSubmitJobStore, ExportOutputStore, ResourceStorage, SettingsStore, + SubmitInputFetcher, }; use crate::bulk_export_auth::ExportFileAuth; @@ -79,6 +80,12 @@ pub struct AppState { /// Bulk export configuration. bulk_export_config: Arc, + /// Optional per-user UI settings store (theme, default tenant, recent + /// queries, …). Present only for backends that provide one (SQLite, + /// PostgreSQL); `None` otherwise, in which case the settings endpoints + /// report the feature as unavailable. + user_settings: Option>, + /// Bulk submit job-state store (claim + worker storage + lifecycle). bulk_submit_jobs: Option>, @@ -113,6 +120,7 @@ impl Clone for AppState { bulk_export_output: self.bulk_export_output.clone(), bulk_export_file_auth: self.bulk_export_file_auth.clone(), bulk_export_config: Arc::clone(&self.bulk_export_config), + user_settings: self.user_settings.clone(), bulk_submit_jobs: self.bulk_submit_jobs.clone(), bulk_submit_fetcher: self.bulk_submit_fetcher.clone(), bulk_submit_output: self.bulk_submit_output.clone(), @@ -147,6 +155,7 @@ impl AppState { bulk_export_output: None, bulk_export_file_auth: None, bulk_export_config, + user_settings: None, bulk_submit_jobs: None, bulk_submit_fetcher: None, bulk_submit_output: None, @@ -191,6 +200,7 @@ impl AppState { bulk_export_output: None, bulk_export_file_auth: None, bulk_export_config, + user_settings: None, bulk_submit_jobs: None, bulk_submit_fetcher: None, bulk_submit_output: None, @@ -259,6 +269,17 @@ impl AppState { &self.bulk_export_config } + /// Wires the per-user UI settings store. + pub fn with_settings_store(mut self, store: Arc) -> Self { + self.user_settings = Some(store); + self + } + + /// Returns the per-user settings store, if configured. + pub fn settings_store(&self) -> Option<&Arc> { + self.user_settings.as_ref() + } + /// Wires the bulk-submit job store, input fetcher, output store, and file authorizer. pub fn with_bulk_submit( mut self, diff --git a/crates/rest/tests/user_settings.rs b/crates/rest/tests/user_settings.rs new file mode 100644 index 000000000..2d2ea8cca --- /dev/null +++ b/crates/rest/tests/user_settings.rs @@ -0,0 +1,183 @@ +//! End-to-end tests for the per-user UI settings endpoints +//! (`GET`/`PUT`/`PATCH /_user/settings`). +//! +//! These exercise the full REST stack — routing, the `UserKey` extractor, the +//! handlers, and the SQLite-backed [`SettingsStore`] — against an in-memory +//! database, with no authentication configured (so the caller resolves to the +//! `local|default` fallback key). + +use std::sync::Arc; + +use axum::body::Bytes; +use axum::http::{HeaderName, HeaderValue, StatusCode}; +use axum_test::TestServer; +use helios_persistence::backends::sqlite::SqliteBackend; +use helios_persistence::core::SettingsStore; +use helios_rest::ServerConfig; +use serde_json::{Value, json}; + +const IF_MATCH: HeaderName = HeaderName::from_static("if-match"); +const IF_NONE_MATCH: HeaderName = HeaderName::from_static("if-none-match"); +const CONTENT_TYPE: HeaderName = HeaderName::from_static("content-type"); + +/// Builds a test server whose SQLite backend also hosts the settings store. +fn create_test_server() -> TestServer { + let backend = SqliteBackend::in_memory().expect("create in-memory SQLite backend"); + backend.init_schema().expect("init schema"); + let backend = Arc::new(backend); + + let config = ServerConfig { + base_url: "http://localhost:8080".to_string(), + ..ServerConfig::for_testing() + }; + + let settings_store: Arc = backend.clone(); + let state = helios_rest::AppState::new(backend, config).with_settings_store(settings_store); + let app = helios_rest::routing::fhir_routes::create_routes(state); + TestServer::new(app).expect("create test server") +} + +/// Reads the `ETag` response header as an owned string. +fn etag(response: &axum_test::TestResponse) -> String { + response + .headers() + .get("etag") + .expect("ETag header present") + .to_str() + .expect("ETag is valid UTF-8") + .to_string() +} + +#[tokio::test] +async fn get_returns_empty_document_by_default() { + let server = create_test_server(); + + let response = server.get("/_user/settings").await; + + assert_eq!(response.status_code(), StatusCode::OK); + assert_eq!(response.json::(), json!({})); + assert_eq!(etag(&response), "W/\"0\""); +} + +#[tokio::test] +async fn put_then_get_round_trips_document() { + let server = create_test_server(); + let doc = json!({ + "theme": "dark", + "defaultTenant": "acme", + "activeFhirVersion": "R4", + "recentQueries": {"Patient": ["name=smith"]} + }); + + let put = server.put("/_user/settings").json(&doc).await; + assert_eq!(put.status_code(), StatusCode::OK); + assert_eq!(etag(&put), "W/\"1\""); + + let get = server.get("/_user/settings").await; + assert_eq!(get.status_code(), StatusCode::OK); + assert_eq!(get.json::(), doc); + assert_eq!(etag(&get), "W/\"1\""); +} + +#[tokio::test] +async fn patch_merges_a_single_key_and_preserves_others() { + let server = create_test_server(); + server + .put("/_user/settings") + .json(&json!({"theme": "dark", "defaultTenant": "acme"})) + .await; + + // Toggle just the theme via JSON merge-patch. + let patch = server + .patch("/_user/settings") + .json(&json!({"theme": "light"})) + .await; + assert_eq!(patch.status_code(), StatusCode::OK); + assert_eq!(etag(&patch), "W/\"2\""); + + let get = server.get("/_user/settings").await; + assert_eq!( + get.json::(), + json!({"theme": "light", "defaultTenant": "acme"}) + ); +} + +#[tokio::test] +async fn patch_null_deletes_a_key() { + let server = create_test_server(); + server + .put("/_user/settings") + .json(&json!({"theme": "dark", "defaultTenant": "acme"})) + .await; + + server + .patch("/_user/settings") + .json(&json!({"defaultTenant": null})) + .await; + + let get = server.get("/_user/settings").await; + assert_eq!(get.json::(), json!({"theme": "dark"})); +} + +#[tokio::test] +async fn stale_if_match_is_rejected_with_412() { + let server = create_test_server(); + server.put("/_user/settings").json(&json!({"a": 1})).await; // -> version 1 + + // Precondition asserts "does not exist yet", but a document now exists. + let conflict = server + .put("/_user/settings") + .add_header(IF_MATCH, HeaderValue::from_static("W/\"0\"")) + .json(&json!({"a": 2})) + .await; + + assert_eq!(conflict.status_code(), StatusCode::PRECONDITION_FAILED); +} + +#[tokio::test] +async fn matching_if_match_succeeds() { + let server = create_test_server(); + server.put("/_user/settings").json(&json!({"a": 1})).await; // -> version 1 + + let ok = server + .put("/_user/settings") + .add_header(IF_MATCH, HeaderValue::from_static("W/\"1\"")) + .json(&json!({"a": 2})) + .await; + + assert_eq!(ok.status_code(), StatusCode::OK); + assert_eq!(etag(&ok), "W/\"2\""); +} + +#[tokio::test] +async fn non_object_body_is_rejected_with_400() { + let server = create_test_server(); + + let response = server + .put("/_user/settings") + .add_header(CONTENT_TYPE, HeaderValue::from_static("application/json")) + .bytes(Bytes::from_static(b"[1, 2, 3]")) + .await; + + assert_eq!(response.status_code(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn if_none_match_returns_304_for_unchanged_document() { + let server = create_test_server(); + let put = server + .put("/_user/settings") + .json(&json!({"theme": "dark"})) + .await; + let current = etag(&put); + + let not_modified = server + .get("/_user/settings") + .add_header( + IF_NONE_MATCH, + HeaderValue::from_str(¤t).expect("valid header"), + ) + .await; + + assert_eq!(not_modified.status_code(), StatusCode::NOT_MODIFIED); +}