From b609b82052fff8b123dd7bfbd6c315d2f2bfc629 Mon Sep 17 00:00:00 2001 From: mauripunzueta Date: Mon, 8 Jun 2026 11:23:37 -0400 Subject: [PATCH 1/9] feat(rest): per-user JSON UI settings store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a simple, extensible per-user settings store for the upcoming web UI (theme/dark mode, default tenant, active FHIR version, recent queries, …). The value is an opaque JSON object, so new settings keys need no schema or code changes — the frontend owns the document shape. Design: - Dedicated `user_settings` table (one opaque JSON document per user), kept separate from the FHIR `resources` table so UI preferences never leak into CapabilityStatement, _history, _search, or $export. Schema migration v10→v11 for both SQLite and PostgreSQL backends. - New `SettingsStore` trait (get/put/patch) in helios-persistence core, plus an RFC 7386 JSON merge-patch helper. Implemented for SQLite and PostgreSQL with transactional read-modify-write and a monotonic `version` for optimistic locking (surfaced as a weak ETag). - Keyed by user only (issuer|subject), tenant-independent; falls back to a fixed `local|default` key when auth is disabled. - REST endpoints GET/PUT/PATCH /_user/settings. The leading-underscore path keeps them authenticated yet exempt from FHIR scope checks and invisible to FHIR machinery. PUT replaces the document; PATCH merge-patches; both honor If-Match, GET honors If-None-Match (304). - Wired into the SQLite and PostgreSQL standalone backends in the hfs binary; other backends report the feature as unavailable (501). Tests: persistence unit tests (merge-patch + SQLite store), PostgreSQL integration tests, REST extractor tests, and an end-to-end HTTP suite covering defaults, round-trip, merge/delete, optimistic locking, validation, and 304. Also refresh the HTS README to reflect completed PostgreSQL backend parity. --- crates/hfs/src/main.rs | 59 ++-- crates/hts/README.md | 13 +- .../persistence/src/backends/postgres/mod.rs | 1 + .../src/backends/postgres/schema.rs | 25 +- .../src/backends/postgres/user_settings.rs | 156 +++++++++++ crates/persistence/src/backends/sqlite/mod.rs | 1 + .../persistence/src/backends/sqlite/schema.rs | 24 +- .../src/backends/sqlite/user_settings.rs | 265 ++++++++++++++++++ crates/persistence/src/core/mod.rs | 2 + crates/persistence/src/core/user_settings.rs | 163 +++++++++++ crates/persistence/tests/postgres_tests.rs | 95 ++++++- crates/rest/src/extractors/mod.rs | 2 + crates/rest/src/extractors/user.rs | 82 ++++++ crates/rest/src/handlers/mod.rs | 2 + crates/rest/src/handlers/user_settings.rs | 171 +++++++++++ crates/rest/src/lib.rs | 57 +++- crates/rest/src/routing/fhir_routes.rs | 8 + crates/rest/src/state.rs | 24 +- crates/rest/tests/user_settings.rs | 183 ++++++++++++ 19 files changed, 1289 insertions(+), 44 deletions(-) create mode 100644 crates/persistence/src/backends/postgres/user_settings.rs create mode 100644 crates/persistence/src/backends/sqlite/user_settings.rs create mode 100644 crates/persistence/src/core/user_settings.rs create mode 100644 crates/rest/src/extractors/user.rs create mode 100644 crates/rest/src/handlers/user_settings.rs create mode 100644 crates/rest/tests/user_settings.rs diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index 9d7b3b403..2e91dbc2e 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -25,18 +25,25 @@ 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, init_logging, -}; +use helios_rest::{AuthMiddlewareState, ServerConfig, StorageBackendMode, init_logging}; 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, DefaultExportWorker, ExportOutputStore, WorkerId, }; #[cfg(any(feature = "sqlite", feature = "postgres"))] use helios_rest::bulk_export_auth::BearerScopeAuth; +// Settings-capable standalone backends (SQLite, PostgreSQL). #[cfg(any(feature = "sqlite", feature = "postgres"))] +use helios_rest::create_app_with_auth_bulk_export_and_settings; +// Composite/secondary backends (MongoDB, Elasticsearch, S3) that do not host a +// settings store and so use the plain app builders. +#[cfg(any(feature = "mongodb", feature = "elasticsearch", feature = "s3"))] +use helios_rest::create_app_with_auth; +#[cfg(any(feature = "mongodb", feature = "elasticsearch"))] use helios_rest::create_app_with_auth_and_bulk_export; #[cfg(feature = "sqlite")] @@ -841,26 +848,17 @@ async fn start_sqlite( let serve_audit_state = audit_state.clone(); let backend = Arc::new(create_sqlite_backend(&config)?); - if let Some(bundle) = build_bulk_export(&config, backend.clone(), backend.clone()).await? { - let app = create_app_with_auth_and_bulk_export( - backend, - config.clone(), - auth_config, - auth_state, - audit_state, - 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") - }), + // The SQLite backend also hosts the per-user settings store. + let settings_store: Option> = Some(backend.clone()); + let bundle = build_bulk_export(&config, backend.clone(), backend.clone()).await?; + let app = create_app_with_auth_bulk_export_and_settings( + backend, config.clone(), auth_config, auth_state, audit_state, + bundle, + settings_store, ); serve(app, &config, serve_audit_state).await } @@ -1272,26 +1270,17 @@ async fn start_postgres( let backend = Arc::new(backend); let serve_audit_state = audit_state.clone(); - if let Some(bundle) = build_bulk_export(&config, backend.clone(), backend.clone()).await? { - let app = create_app_with_auth_and_bulk_export( - backend, - config.clone(), - auth_config, - auth_state, - audit_state, - 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") - }), + // The PostgreSQL backend also hosts the per-user settings store. + let settings_store: Option> = Some(backend.clone()); + let bundle = build_bulk_export(&config, backend.clone(), backend.clone()).await?; + let app = create_app_with_auth_bulk_export_and_settings( + backend, config.clone(), auth_config, auth_state, audit_state, + bundle, + settings_store, ); serve(app, &config, serve_audit_state).await } diff --git a/crates/hts/README.md b/crates/hts/README.md index 4f1f6430e..4ab7aa1ca 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 @@ -402,10 +402,17 @@ The `value_set_expansions` table acts as a write-through cache: the first `$expa ### 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/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 4aa3fca8d..b068449b3 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 = 10; +pub const SCHEMA_VERSION: i32 = 11; /// Initialize the database schema. pub async fn initialize_schema(client: &deadpool_postgres::Client) -> StorageResult<()> { @@ -273,6 +273,7 @@ async fn migrate_schema( 7 => migrate_v7_to_v8(client).await?, 8 => migrate_v8_to_v9(client).await?, 9 => migrate_v9_to_v10(client).await?, + 10 => migrate_v10_to_v11(client).await?, _ => { return Err(pg_error(format!("Unknown schema version: {}", version))); } @@ -676,6 +677,28 @@ async fn migrate_v9_to_v10(client: &deadpool_postgres::Client) -> StorageResult< Ok(()) } +/// v10 -> v11: 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_v10_to_v11(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 v10->v11 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 b1bc4e410..02681782c 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 = 10; +pub const SCHEMA_VERSION: i32 = 11; /// Initialize the database schema. pub fn initialize_schema(conn: &Connection) -> StorageResult<()> { @@ -267,6 +267,7 @@ fn migrate_schema(conn: &Connection, from_version: i32) -> StorageResult<()> { 7 => migrate_v7_to_v8(conn)?, 8 => migrate_v8_to_v9(conn)?, 9 => migrate_v9_to_v10(conn)?, + 10 => migrate_v10_to_v11(conn)?, _ => { return Err(crate::error::StorageError::Backend( crate::error::BackendError::Internal { @@ -1005,6 +1006,27 @@ fn migrate_v9_to_v10(conn: &Connection) -> StorageResult<()> { Ok(()) } +/// Migrate from schema version 10 to version 11. +/// +/// 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_v10_to_v11(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..af20a6cc1 --- /dev/null +++ b/crates/persistence/src/backends/sqlite/user_settings.rs @@ -0,0 +1,265 @@ +//! 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); + } +} diff --git a/crates/persistence/src/core/mod.rs b/crates/persistence/src/core/mod.rs index 5c45d5e84..f187902a6 100644 --- a/crates/persistence/src/core/mod.rs +++ b/crates/persistence/src/core/mod.rs @@ -99,6 +99,7 @@ pub mod history; pub mod search; pub mod storage; pub mod transaction; +pub mod user_settings; pub mod versioned; // Re-export main types @@ -144,4 +145,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/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index f487929f6..e6406cbb5 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -686,9 +686,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; @@ -3671,4 +3672,96 @@ mod postgres_integration { // Only completed/error/cancelled jobs can expire — these are accepted. 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); + } } 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 8549f79fe..bd099ad23 100644 --- a/crates/rest/src/handlers/mod.rs +++ b/crates/rest/src/handlers/mod.rs @@ -32,6 +32,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")] @@ -70,5 +71,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..6cc46bc54 --- /dev/null +++ b/crates/rest/src/handlers/user_settings.rs @@ -0,0 +1,171 @@ +//! 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 (e.g. MongoDB, S3, Elasticsearch). +fn settings_store(state: &AppState) -> RestResult<&Arc> +where + S: ResourceStorage + Send + Sync, +{ + state + .settings_store() + .ok_or_else(|| RestError::NotImplemented { + feature: "per-user settings (requires the SQLite or PostgreSQL backend)".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}\"") +} diff --git a/crates/rest/src/lib.rs b/crates/rest/src/lib.rs index 3e721233d..73115f8df 100644 --- a/crates/rest/src/lib.rs +++ b/crates/rest/src/lib.rs @@ -312,6 +312,7 @@ where auth_state, audit_state, None, + None, ) } @@ -349,11 +350,56 @@ where auth_state, audit_state, Some(bulk_export), + None, + ) +} + +/// Like [`create_app_with_auth_and_bulk_export`], but also wires the per-user +/// settings store (used by the `/_user/settings` endpoints). `bulk_export` is +/// optional so this single entry point covers both bulk-enabled and bulk-less +/// deployments of a settings-capable backend (SQLite, PostgreSQL). +#[allow(clippy::too_many_arguments)] +pub fn create_app_with_auth_bulk_export_and_settings( + storage: Arc, + config: ServerConfig, + auth_config: helios_auth::AuthConfig, + auth_state: Option>, + audit_state: Option>, + bulk_export: 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, + 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`], and +/// [`create_app_with_auth_bulk_export_and_settings`]. +#[allow(clippy::too_many_arguments)] fn build_app( storage: Arc, config: ServerConfig, @@ -361,6 +407,7 @@ fn build_app( auth_state: Option>, audit_state: Option>, bulk_export: Option, + settings_store: Option>, ) -> Router where S: ResourceStorage @@ -418,6 +465,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 = { diff --git a/crates/rest/src/routing/fhir_routes.rs b/crates/rest/src/routing/fhir_routes.rs index f91502ff2..318ec60d2 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 323d1034c..d0fd68630 100644 --- a/crates/rest/src/state.rs +++ b/crates/rest/src/state.rs @@ -8,7 +8,9 @@ use std::sync::Arc; use helios_audit::AuditSink; use helios_auth::AuthConfig; -use helios_persistence::core::{BulkExportJobStore, ExportOutputStore, ResourceStorage}; +use helios_persistence::core::{ + BulkExportJobStore, ExportOutputStore, ResourceStorage, SettingsStore, +}; use crate::bulk_export_auth::ExportFileAuth; use crate::config::{BulkExportConfig, ServerConfig}; @@ -68,6 +70,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>, } // Manually implement Clone since S is wrapped in Arc and doesn't need to be Clone @@ -86,6 +94,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(), } } } @@ -112,6 +121,7 @@ impl AppState { bulk_export_output: None, bulk_export_file_auth: None, bulk_export_config, + user_settings: None, } } @@ -148,6 +158,7 @@ impl AppState { bulk_export_output: None, bulk_export_file_auth: None, bulk_export_config, + user_settings: None, } } @@ -184,6 +195,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() + } + /// Sets the subscription engine on this AppState. #[cfg(feature = "subscriptions")] pub fn with_subscription_engine( 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); +} From f9de007debf6931262dfbf32d6ffcbc104d4fb0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Fri, 3 Jul 2026 11:37:48 -0400 Subject: [PATCH 2/9] style: rustfmt import ordering in hfs main after merge --- crates/hfs/src/main.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index c20599d0f..f92f493cd 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -40,9 +40,9 @@ use helios_rest::bulk_export_auth::BearerScopeAuth; // Settings-capable standalone backends (SQLite, PostgreSQL) also 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; -#[cfg(any(feature = "sqlite", feature = "postgres"))] use helios_rest::create_app_with_auth_and_bulk; +#[cfg(any(feature = "sqlite", feature = "postgres"))] +use helios_rest::create_app_with_auth_bulk_and_settings; // Composite/secondary backends (MongoDB, Elasticsearch, S3) that do not host a // settings store and so use the plain app builder. #[cfg(any(feature = "mongodb", feature = "elasticsearch", feature = "s3"))] From 3273daa249c28805e294ed48df016567dd2646f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Fri, 3 Jul 2026 12:10:12 -0400 Subject: [PATCH 3/9] test(user-settings): cover handler helpers, sqlite error paths, app builders Address Codecov patch-coverage gaps on PR #151: - handlers/user_settings.rs: unit-test parse_object_body (empty/invalid/ non-object), parse_if_match_version (wildcard/unparseable/absent), and the 501-when-no-store branch - sqlite/user_settings.rs: error paths for a corrupt JSON row (decode -> backend_err) and an unparseable stored timestamp (falls back to now) - rest/lib.rs: builder_tests exercising create_app_with_auth_and_bulk and create_app_with_auth_bulk_and_settings (SOF disabled for a side-effect-free router build) --- .../src/backends/sqlite/user_settings.rs | 40 +++++++++++ crates/rest/src/handlers/user_settings.rs | 66 +++++++++++++++++++ crates/rest/src/lib.rs | 51 ++++++++++++++ 3 files changed, 157 insertions(+) diff --git a/crates/persistence/src/backends/sqlite/user_settings.rs b/crates/persistence/src/backends/sqlite/user_settings.rs index af20a6cc1..f88bc0f79 100644 --- a/crates/persistence/src/backends/sqlite/user_settings.rs +++ b/crates/persistence/src/backends/sqlite/user_settings.rs @@ -262,4 +262,44 @@ mod tests { .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/rest/src/handlers/user_settings.rs b/crates/rest/src/handlers/user_settings.rs index 6cc46bc54..aa0d5ed9d 100644 --- a/crates/rest/src/handlers/user_settings.rs +++ b/crates/rest/src/handlers/user_settings.rs @@ -169,3 +169,69 @@ fn settings_response(stored: StoredUserSettings) -> Response { 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. + let state = AppState::new(Arc::new(backend), ServerConfig::default()); + let err = settings_store(&state).unwrap_err(); + assert!(matches!(err, RestError::NotImplemented { .. })); + } +} diff --git a/crates/rest/src/lib.rs b/crates/rest/src/lib.rs index 4c1d0b86a..2d74c0654 100644 --- a/crates/rest/src/lib.rs +++ b/crates/rest/src/lib.rs @@ -958,3 +958,54 @@ 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, + Some(settings), + ); + } +} From d502f7b0f557478ad3d5727052fe25823df9b9e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Fri, 3 Jul 2026 12:13:56 -0400 Subject: [PATCH 4/9] test(user-settings): fix builder arg count and non-Debug unwrap_err --- crates/rest/src/handlers/user_settings.rs | 7 ++++--- crates/rest/src/lib.rs | 1 + 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/rest/src/handlers/user_settings.rs b/crates/rest/src/handlers/user_settings.rs index aa0d5ed9d..f55c64bde 100644 --- a/crates/rest/src/handlers/user_settings.rs +++ b/crates/rest/src/handlers/user_settings.rs @@ -229,9 +229,10 @@ mod tests { let backend = SqliteBackend::in_memory().expect("in-memory sqlite"); backend.init_schema().expect("init schema"); - // `AppState::new` leaves the settings store unset. + // `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 err = settings_store(&state).unwrap_err(); - assert!(matches!(err, RestError::NotImplemented { .. })); + 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 2d74c0654..a8e8e8a0d 100644 --- a/crates/rest/src/lib.rs +++ b/crates/rest/src/lib.rs @@ -1005,6 +1005,7 @@ mod builder_tests { None, None, None, + None, Some(settings), ); } From fe36ad641244f7606b2a25861f2fc9074763c906 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 6 Jul 2026 09:59:35 -0400 Subject: [PATCH 5/9] feat(user-settings): add MongoDB SettingsStore; explain 501 on S3/ES MongoDB is a full standalone primary FHIR backend, so a mongodb-only deployment must not have its /_user/settings endpoints silently 501. Implement SettingsStore for MongoBackend using a version-conditioned update: read the current document, compute the new one, then persist with a filter pinned to the read version. A concurrent writer that bumped the version makes the update match nothing, which is retried for unconditional writes and surfaced as an OptimisticLockFailure for If-Match writes. A Some(0) precondition maps to an insert whose duplicate-key error (from the new unique index on user_key) becomes the same lock failure. This yields the same read-modify-write + monotonic-version semantics as the SQLite/Postgres stores without requiring multi-document transactions. - schema: unique idx_user_settings_key index; bump SCHEMA_VERSION 4 -> 5, wired into both initialize and migrate paths. - hfs: wire the settings store into start_mongodb and start_mongodb_elasticsearch (the latter sources settings from the underlying Mongo primary behind the composite storage). - tests: MongoDB integration suite mirroring the SQLite/Postgres coverage plus a concurrent-patch test asserting no lost updates. S3 remains unimplemented (its recommended role is Archive and object-store optimistic locking needs conditional PutObject); tracked in #199. The /_user/settings 501 message now names the supported backends and explicitly excludes S3/Elasticsearch so operators get an explained response rather than a bare 501. --- crates/hfs/src/main.rs | 82 +++--- .../persistence/src/backends/mongodb/mod.rs | 1 + .../src/backends/mongodb/schema.rs | 25 +- .../src/backends/mongodb/user_settings.rs | 266 ++++++++++++++++++ crates/persistence/tests/mongodb_tests.rs | 180 +++++++++++- crates/rest/src/handlers/user_settings.rs | 15 +- 6 files changed, 525 insertions(+), 44 deletions(-) create mode 100644 crates/persistence/src/backends/mongodb/user_settings.rs diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index f92f493cd..3cced1237 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -512,32 +512,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 } @@ -1778,32 +1779,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/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/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 31ba821f2..de044dbc7 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::{ @@ -2372,3 +2372,181 @@ 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()) +} + +#[tokio::test] +async fn mongodb_integration_settings_get_missing_is_none() { + let Some(backend) = create_backend("settings_missing").await else { + eprintln!( + "Skipping mongodb_integration_settings_get_missing_is_none (set 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) = create_backend("settings_round_trip").await else { + eprintln!( + "Skipping mongodb_integration_settings_put_get_and_version (set 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) = create_backend("settings_patch").await else { + eprintln!( + "Skipping mongodb_integration_settings_patch_merges_and_deletes (set 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) = create_backend("settings_patch_create").await else { + eprintln!( + "Skipping mongodb_integration_settings_patch_on_missing_creates_document (set 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) = create_backend("settings_lock").await else { + eprintln!( + "Skipping mongodb_integration_settings_optimistic_lock (set 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) = create_backend("settings_concurrent").await else { + eprintln!( + "Skipping mongodb_integration_settings_concurrent_patches_serialize (set HFS_TEST_MONGODB_URL)" + ); + return; + }; + let backend = Arc::new(backend); + let user = unique_user_key("concurrent"); + + // Seed an empty document so all racers start from version 1. + backend.put_settings(&user, json!({}), None).await.unwrap(); + + // 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" + ); + } + // Initial put (v1) + 12 successful patches. + assert_eq!(final_doc.version, 13); +} diff --git a/crates/rest/src/handlers/user_settings.rs b/crates/rest/src/handlers/user_settings.rs index f55c64bde..2f0de3f32 100644 --- a/crates/rest/src/handlers/user_settings.rs +++ b/crates/rest/src/handlers/user_settings.rs @@ -115,7 +115,16 @@ where } /// Returns the configured settings store, or a `501 Not Implemented` error when -/// the active backend does not provide one (e.g. MongoDB, S3, Elasticsearch). +/// 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, @@ -123,7 +132,9 @@ where state .settings_store() .ok_or_else(|| RestError::NotImplemented { - feature: "per-user settings (requires the SQLite or PostgreSQL backend)".to_string(), + feature: "per-user settings (supported on the SQLite, PostgreSQL, and MongoDB \ + backends; not available on the S3 or Elasticsearch backends)" + .to_string(), }) } From e26dd429a57c8a1214722feb65d8e9cf72eea5fe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 6 Jul 2026 10:22:14 -0400 Subject: [PATCH 6/9] test(user-settings): run MongoDB SettingsStore tests via testcontainer The MongoDB integration suite is gated on HFS_TEST_MONGODB_URL and skips in CI, so the new mongodb/user_settings.rs landed at 0% patch coverage and dragged the PR's Codecov patch report down to ~81%. Provision the settings-store tests from an ephemeral standalone Mongo testcontainer when no external URL is set (preferring the URL when it is), mirroring how the PostgreSQL suite is provisioned. Docker is available in the coverage/test-rust jobs, so these tests now actually run and exercise the SettingsStore impl. The store uses version-conditioned writes rather than multi-document transactions, so a standalone (non replica-set) Mongo is sufficient. If neither a URL nor Docker is available the tests skip, so no environment is hard-broken. - Cargo: enable the testcontainers-modules `mongo` feature (test-only). --- crates/persistence/Cargo.toml | 2 +- crates/persistence/tests/mongodb_tests.rs | 98 ++++++++++++++++++++--- 2 files changed, 87 insertions(+), 13 deletions(-) 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/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index de044dbc7..019996e64 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -2383,11 +2383,85 @@ 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 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) = create_backend("settings_missing").await else { + let Some(backend) = settings_mongo::backend("settings_missing").await else { eprintln!( - "Skipping mongodb_integration_settings_get_missing_is_none (set HFS_TEST_MONGODB_URL)" + "Skipping mongodb_integration_settings_get_missing_is_none (requires Docker or HFS_TEST_MONGODB_URL)" ); return; }; @@ -2397,9 +2471,9 @@ async fn mongodb_integration_settings_get_missing_is_none() { #[tokio::test] async fn mongodb_integration_settings_put_get_and_version() { - let Some(backend) = create_backend("settings_round_trip").await else { + let Some(backend) = settings_mongo::backend("settings_round_trip").await else { eprintln!( - "Skipping mongodb_integration_settings_put_get_and_version (set HFS_TEST_MONGODB_URL)" + "Skipping mongodb_integration_settings_put_get_and_version (requires Docker or HFS_TEST_MONGODB_URL)" ); return; }; @@ -2427,9 +2501,9 @@ async fn mongodb_integration_settings_put_get_and_version() { #[tokio::test] async fn mongodb_integration_settings_patch_merges_and_deletes() { - let Some(backend) = create_backend("settings_patch").await else { + let Some(backend) = settings_mongo::backend("settings_patch").await else { eprintln!( - "Skipping mongodb_integration_settings_patch_merges_and_deletes (set HFS_TEST_MONGODB_URL)" + "Skipping mongodb_integration_settings_patch_merges_and_deletes (requires Docker or HFS_TEST_MONGODB_URL)" ); return; }; @@ -2457,9 +2531,9 @@ async fn mongodb_integration_settings_patch_merges_and_deletes() { #[tokio::test] async fn mongodb_integration_settings_patch_on_missing_creates_document() { - let Some(backend) = create_backend("settings_patch_create").await else { + let Some(backend) = settings_mongo::backend("settings_patch_create").await else { eprintln!( - "Skipping mongodb_integration_settings_patch_on_missing_creates_document (set HFS_TEST_MONGODB_URL)" + "Skipping mongodb_integration_settings_patch_on_missing_creates_document (requires Docker or HFS_TEST_MONGODB_URL)" ); return; }; @@ -2474,9 +2548,9 @@ async fn mongodb_integration_settings_patch_on_missing_creates_document() { #[tokio::test] async fn mongodb_integration_settings_optimistic_lock() { - let Some(backend) = create_backend("settings_lock").await else { + let Some(backend) = settings_mongo::backend("settings_lock").await else { eprintln!( - "Skipping mongodb_integration_settings_optimistic_lock (set HFS_TEST_MONGODB_URL)" + "Skipping mongodb_integration_settings_optimistic_lock (requires Docker or HFS_TEST_MONGODB_URL)" ); return; }; @@ -2508,9 +2582,9 @@ async fn mongodb_integration_settings_optimistic_lock() { #[tokio::test] async fn mongodb_integration_settings_concurrent_patches_serialize() { - let Some(backend) = create_backend("settings_concurrent").await else { + let Some(backend) = settings_mongo::backend("settings_concurrent").await else { eprintln!( - "Skipping mongodb_integration_settings_concurrent_patches_serialize (set HFS_TEST_MONGODB_URL)" + "Skipping mongodb_integration_settings_concurrent_patches_serialize (requires Docker or HFS_TEST_MONGODB_URL)" ); return; }; From 6c549dc06714ffa28bb38151690cf768656007a9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 6 Jul 2026 10:34:49 -0400 Subject: [PATCH 7/9] fix(test): import Backend trait in settings_mongo test module The new settings_mongo helper module called MongoBackend::initialize(), which is provided by the core `Backend` trait. That trait is in scope at the top of mongodb_tests.rs but not inside the nested module, so clippy and Test Rust failed to compile the test binary (E0599). Import the trait in the module. --- crates/persistence/tests/mongodb_tests.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 019996e64..8bb731908 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -2395,6 +2395,7 @@ fn unique_user_key(prefix: &str) -> String { /// 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; From 13147a8db46df30f042f4ca31dab8d171dfa7ca2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 6 Jul 2026 12:04:09 -0400 Subject: [PATCH 8/9] test(user-settings): cover Mongo insert race and decode-error paths Lift mongodb/user_settings.rs coverage toward its SQLite/Postgres siblings by exercising two more real branches: - Drop the seed in the concurrent-patch test so the racers start from version 0: exactly one wins the initial insert and the rest hit the unique-index duplicate-key path and retry into the version-conditioned update, covering both races in one test. - Add a decode-error test that inserts a row whose `data` is not valid JSON and asserts get_settings surfaces a backend error, mirroring the existing SQLite/PostgreSQL corrupt-row coverage. --- crates/persistence/tests/mongodb_tests.rs | 44 ++++++++++++++++++++--- 1 file changed, 39 insertions(+), 5 deletions(-) diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 8bb731908..495583a90 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -2592,9 +2592,11 @@ async fn mongodb_integration_settings_concurrent_patches_serialize() { let backend = Arc::new(backend); let user = unique_user_key("concurrent"); - // Seed an empty document so all racers start from version 1. - backend.put_settings(&user, json!({}), None).await.unwrap(); - + // 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. @@ -2622,6 +2624,38 @@ async fn mongodb_integration_settings_concurrent_patches_serialize() { "key k{i} was lost to a read-modify-write race" ); } - // Initial put (v1) + 12 successful patches. - assert_eq!(final_doc.version, 13); + // 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(_))); } From 57e1b09b43c46a4936bb9d78c87cbd8aaf2dde56 Mon Sep 17 00:00:00 2001 From: smunini Date: Wed, 8 Jul 2026 07:55:34 -0400 Subject: [PATCH 9/9] fix(hfs): wire per-user settings store into SQLite+ES and Postgres+ES composites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SQLite+Elasticsearch and PostgreSQL+Elasticsearch composite startup paths served their apps through the plain `create_app_with_auth` builder and never passed a settings store, so `/_user/settings` returned 501 even though the SQLite/Postgres primary fully implements `SettingsStore` — the same backend that is settings-capable in the standalone `start_sqlite`/ `start_postgres` paths. Wire the settings store from the underlying primary (mirroring `start_mongodb_elasticsearch`, which already did this from its `mongo` primary) and route both composites through `create_app_with_auth_bulk_and_settings`. Narrow the now-unused import gates: `create_app_with_auth` is only used by the S3 paths (`#[cfg(feature = "s3")]`), and `create_app_with_auth_and_bulk` only by the S3+ES sidecar path (`#[cfg(all(s3, elasticsearch, sqlite))]`). Verified compiling: sqlite; sqlite+es; postgres+es; s3+es+sqlite; s3. --- crates/hfs/src/main.rs | 70 ++++++++++++++++++------------------------ 1 file changed, 30 insertions(+), 40 deletions(-) diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index 5f3528739..019f26ce1 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -37,15 +37,19 @@ use helios_persistence::core::{ }; #[cfg(any(feature = "sqlite", feature = "postgres"))] use helios_rest::bulk_export_auth::BearerScopeAuth; -// Settings-capable standalone backends (SQLite, PostgreSQL) also host the -// per-user settings store, wired alongside bulk export/submit. -#[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; -// Composite/secondary backends (MongoDB, Elasticsearch, S3) that do not host a -// settings store and so use the plain app builder. -#[cfg(any(feature = "mongodb", feature = "elasticsearch", feature = "s3"))] +// 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")] @@ -1418,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 } @@ -1643,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 }