diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index 84498d512..fe3ba691d 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -807,7 +807,7 @@ impl MongoBackend { Ok(Some(ids)) } - fn build_search_index_filter( + pub(super) fn build_search_index_filter( &self, tenant_id: &str, resource_type: &str, @@ -1171,7 +1171,7 @@ impl MongoBackend { } } - fn build_resource_filter( + pub(super) fn build_resource_filter( &self, tenant_id: &str, resource_type: &str, @@ -1423,7 +1423,7 @@ impl MongoBackend { Ok(result.resources.items) } - fn build_search_parameters( + pub(super) fn build_search_parameters( &self, tenant: &TenantContext, resource_type: &str, diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index c56d1f7e5..38ee37a39 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -1,6 +1,6 @@ //! ResourceStorage implementation for MongoDB. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use async_trait::async_trait; use chrono::{DateTime, Utc}; @@ -28,7 +28,7 @@ use crate::search::converters::IndexValue; use crate::search::extractor::ExtractedValue; use crate::search::reindex::{ReindexSource, ReindexTarget, ResourcePage}; use crate::tenant::{Operation, TenantContext}; -use crate::types::{CursorValue, Page, PageCursor, PageInfo, StoredResource}; +use crate::types::{CursorValue, Page, PageCursor, PageInfo, SearchQuery, StoredResource}; use super::MongoBackend; @@ -3328,47 +3328,173 @@ impl MongoBackend { return Ok(Vec::new()); } - let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); + if self.is_search_offloaded() { + return self + .if_none_exist_offloaded_scan(db, session, tenant, resource_type, &parsed_params) + .await; + } + + let typed_params = self.build_search_parameters(tenant, resource_type, &parsed_params); + let query = SearchQuery { + resource_type: resource_type.to_string(), + parameters: typed_params, + count: Some(2), + ..Default::default() + }; + let tenant_id = tenant.tenant_id().as_str(); + let search_index = db.collection::(MongoBackend::SEARCH_INDEX_COLLECTION); + let mut matched_ids: Option> = None; + + for param in &query.parameters { + if matches!(param.name.as_str(), "_id" | "_lastUpdated") { + continue; + } + + let filter = self.build_search_index_filter(tenant_id, resource_type, param)?; + let pipeline = vec![ + doc! { "$match": filter }, + doc! { "$group": { "_id": "$resource_id" } }, + doc! { "$limit": 1001_i32 }, + ]; + let cursor = search_index + .aggregate(pipeline) + .session(&mut *session) + .await + .map_err(|e| { + internal_error(format!( + "Failed to query search_index in transaction: {}", + e + )) + })?; + let docs = collect_session_documents(cursor, session).await?; + + if docs.len() > 1000 { + tracing::warn!( + resource_type, + param_name = %param.name, + "ifNoneExist criteria match more than 1000 search_index entries; \ + intersection may be incomplete for this parameter" + ); + } + + let ids: HashSet = docs + .into_iter() + .filter_map(|d| d.get_str("_id").ok().map(str::to_string)) + .collect(); + + if ids.is_empty() { + return Ok(Vec::new()); + } + + matched_ids = Some(match matched_ids { + Some(current) => current.intersection(&ids).cloned().collect(), + None => ids, + }); + + if matched_ids.as_ref().is_some_and(|s| s.is_empty()) { + return Ok(Vec::new()); + } + } + let candidate_ids: Option> = + matched_ids.map(|ids| ids.into_iter().take(2).collect()); + + let filter = self.build_resource_filter( + tenant_id, + resource_type, + &query, + candidate_ids.as_ref(), + None, + )?; + + let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); let cursor = resources - .find(doc! { - "tenant_id": tenant_id, - "resource_type": resource_type, - "is_deleted": false, - }) + .find(filter) + .limit(2) .session(&mut *session) .await .map_err(|e| { - internal_error(format!( - "Failed to query conditional matches in transaction: {}", - e - )) + internal_error(format!("Failed to query resources in transaction: {}", e)) })?; - let docs = collect_session_documents(cursor, session).await?; - let mut matches = Vec::new(); - for doc in docs { - let payload = doc.get_document("data").map_err(|e| { + docs.into_iter() + .map(|doc| document_to_stored_resource(&doc, tenant, resource_type)) + .collect() + } + + async fn if_none_exist_offloaded_scan( + &self, + db: &mongodb::Database, + session: &mut ClientSession, + tenant: &TenantContext, + resource_type: &str, + parsed_params: &[(String, String)], + ) -> StorageResult> { + let tenant_id = tenant.tenant_id().as_str(); + let mut conditions = vec![doc! { + "tenant_id": tenant_id, + "resource_type": resource_type, + "is_deleted": false, + }]; + + for (name, value) in parsed_params { + match name.as_str() { + "_id" => { + conditions.push(doc! { "id": value.as_str() }); + } + "identifier" => { + let mut elem_match = Document::new(); + if let Some((system, val)) = value.split_once('|') { + if !system.is_empty() { + elem_match.insert("system", system); + } + if !val.is_empty() { + elem_match.insert("value", val); + } + } else if !value.is_empty() { + elem_match.insert("value", value.as_str()); + } + if !elem_match.is_empty() { + conditions.push(doc! { "data.identifier": { "$elemMatch": elem_match } }); + } + } + _ => { + tracing::warn!( + resource_type, + param_name = %name, + "ifNoneExist with offloaded search: param cannot be evaluated \ + against the resource collection directly; matches may include \ + false positives" + ); + } + } + } + + let filter = if conditions.len() == 1 { + conditions.remove(0) + } else { + doc! { "$and": Bson::Array(conditions.into_iter().map(Bson::Document).collect()) } + }; + + let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); + let cursor = resources + .find(filter) + .limit(2) + .session(&mut *session) + .await + .map_err(|e| { internal_error(format!( - "Missing payload while matching conditionals: {}", + "Failed to scan resources for offloaded ifNoneExist: {}", e )) })?; - let resource = document_to_value(payload)?; - - if resource_matches_bundle_search_params(&resource, &parsed_params) - && doc - .get_str("resource_type") - .map(|rt| rt == resource_type) - .unwrap_or(true) - { - matches.push(document_to_stored_resource(&doc, tenant, resource_type)?); - } - } + let docs = collect_session_documents(cursor, session).await?; - Ok(matches) + docs.into_iter() + .map(|doc| document_to_stored_resource(&doc, tenant, resource_type)) + .collect() } async fn index_resource_in_bundle_transaction( @@ -3494,93 +3620,6 @@ impl MongoBackend { } } -fn resource_matches_bundle_search_params(resource: &Value, params: &[(String, String)]) -> bool { - params.iter().all(|(name, expected)| match name.as_str() { - "_id" => resource - .get("id") - .and_then(Value::as_str) - .is_some_and(|id| id == expected), - "identifier" => resource_identifier_matches(resource, expected), - _ => resource_field_matches(resource.get(name), expected), - }) -} - -fn resource_identifier_matches(resource: &Value, expected: &str) -> bool { - let Some(identifier_value) = resource.get("identifier") else { - return false; - }; - - let (system, value, has_separator) = if let Some((system, value)) = expected.split_once('|') { - (system, value, true) - } else { - ("", expected, false) - }; - - match identifier_value { - Value::Array(items) => items - .iter() - .any(|item| match_identifier_item(item, system, value, has_separator)), - Value::Object(_) => match_identifier_item(identifier_value, system, value, has_separator), - _ => false, - } -} - -fn match_identifier_item(item: &Value, system: &str, value: &str, has_separator: bool) -> bool { - let item_system = item.get("system").and_then(Value::as_str); - let item_value = item.get("value").and_then(Value::as_str); - - if has_separator { - let system_matches = if system.is_empty() { - true - } else { - item_system == Some(system) - }; - let value_matches = if value.is_empty() { - true - } else { - item_value == Some(value) - }; - - system_matches && value_matches - } else { - item_value == Some(value) - } -} - -fn resource_field_matches(value: Option<&Value>, expected: &str) -> bool { - let Some(value) = value else { - return false; - }; - - match value { - Value::String(s) => s == expected, - Value::Array(items) => items - .iter() - .any(|item| resource_field_matches(Some(item), expected)), - Value::Object(map) => { - if map - .get("reference") - .and_then(Value::as_str) - .is_some_and(|reference| reference == expected) - { - return true; - } - - if map - .get("value") - .and_then(Value::as_str) - .is_some_and(|value| value == expected) - { - return true; - } - - map.values() - .any(|nested| resource_field_matches(Some(nested), expected)) - } - _ => false, - } -} - // ============================================================================ // PurgableStorage // diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 572bcaf1f..882dc4eae 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -4515,3 +4515,269 @@ async fn mongodb_integration_export_until_is_inclusive() { "a resource exactly on the bound is included" ); } + +#[tokio::test] +async fn mongodb_integration_if_none_exist_multi_param_and_semantics() { + let Some(backend) = create_backend_with_full_registry("if_none_exist_multi_param").await else { + eprintln!( + "Skipping mongodb_integration_if_none_exist_multi_param_and_semantics (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + + let tenant = create_tenant("tenant-if-none-exist-multi"); + + backend + .create( + &tenant, + "Patient", + json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-MULTI-1"}], + "active": true + }), + FhirVersion::default(), + ) + .await + .unwrap(); + + backend + .create( + &tenant, + "Patient", + json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-MULTI-1"}], + "active": false + }), + FhirVersion::default(), + ) + .await + .unwrap(); + + let active_entry = BundleEntry { + method: BundleMethod::Post, + url: "Patient".to_string(), + resource: Some(json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-MULTI-1"}], + "active": true + })), + if_match: None, + if_none_match: None, + if_none_exist: Some( + "identifier=http://example.org/mrn|MRN-MULTI-1&active=true".to_string(), + ), + full_url: Some("urn:uuid:active-patient".to_string()), + }; + + let Some(result) = process_transaction_or_skip( + &backend, + &tenant, + vec![active_entry], + "mongodb_integration_if_none_exist_multi_param_and_semantics", + ) + .await + else { + return; + }; + + assert_eq!( + result.entries[0].status, 200, + "both params match the active patient — should not create a duplicate" + ); + + let count = backend.count(&tenant, Some("Patient")).await.unwrap(); + assert_eq!(count, 2, "no third patient should have been created"); +} + +#[tokio::test] +async fn mongodb_integration_if_none_exist_same_transaction_read_your_writes() { + let Some(backend) = create_backend_with_full_registry("if_none_exist_ryw").await else { + eprintln!( + "Skipping mongodb_integration_if_none_exist_same_transaction_read_your_writes (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + + let tenant = create_tenant("tenant-if-none-exist-ryw"); + + let entry = |full_url: &str| BundleEntry { + method: BundleMethod::Post, + url: "Patient".to_string(), + resource: Some(json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-RYW-1"}] + })), + if_match: None, + if_none_match: None, + if_none_exist: Some("identifier=http://example.org/mrn|MRN-RYW-1".to_string()), + full_url: Some(full_url.to_string()), + }; + + let Some(result) = process_transaction_or_skip( + &backend, + &tenant, + vec![entry("urn:uuid:first"), entry("urn:uuid:second")], + "mongodb_integration_if_none_exist_same_transaction_read_your_writes", + ) + .await + else { + return; + }; + + assert_eq!( + result.entries[0].status, 201, + "first entry creates the patient" + ); + assert_eq!( + result.entries[1].status, 200, + "second entry must see the first entry's write via session" + ); + assert_eq!( + result.entries[1].location, result.entries[0].location, + "second entry must resolve to the same resource as the first" + ); + assert_eq!(backend.count(&tenant, Some("Patient")).await.unwrap(), 1); +} + +#[tokio::test] +async fn mongodb_integration_if_none_exist_multiple_matches_rolls_back() { + let Some(backend) = create_backend_with_full_registry("if_none_exist_multi_match").await else { + eprintln!( + "Skipping mongodb_integration_if_none_exist_multiple_matches_rolls_back (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + + let tenant = create_tenant("tenant-if-none-exist-ambiguous"); + + for family in ["One", "Two"] { + backend + .create( + &tenant, + "Patient", + json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-AMB-1"}], + "name": [{"family": family}] + }), + FhirVersion::default(), + ) + .await + .unwrap(); + } + + let entries = vec![ + BundleEntry { + method: BundleMethod::Post, + url: "Patient".to_string(), + resource: Some( + json!({ "resourceType": "Patient", "name": [{"family": "ShouldRollBack"}] }), + ), + if_match: None, + if_none_match: None, + if_none_exist: None, + full_url: None, + }, + BundleEntry { + method: BundleMethod::Post, + url: "Patient".to_string(), + resource: Some(json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-AMB-1"}] + })), + if_match: None, + if_none_match: None, + if_none_exist: Some("identifier=http://example.org/mrn|MRN-AMB-1".to_string()), + full_url: Some("urn:uuid:ambiguous".to_string()), + }, + ]; + + match backend + .process_transaction(&tenant, entries, FhirVersion::default()) + .await + { + Err(helios_persistence::error::TransactionError::UnsupportedIsolationLevel { .. }) => { + eprintln!("Skipping multiple_matches_rolls_back (replica-set required)"); + return; + } + Err(helios_persistence::error::TransactionError::BundleError { index, message }) => { + assert_eq!(index, 1); + assert!( + message.contains("412"), + "expected 412 in message, got: {message}" + ); + } + Err(other) => panic!("unexpected error: {other:?}"), + Ok(_) => panic!("ambiguous ifNoneExist must fail the bundle"), + } + + assert_eq!( + backend.count(&tenant, Some("Patient")).await.unwrap(), + 2, + "the plain create in entry 0 must have been rolled back" + ); +} + +#[tokio::test] +async fn mongodb_integration_if_none_exist_offloaded_search_uses_resource_scan() { + let Some(mut backend) = create_backend_with_full_registry("if_none_exist_offloaded").await + else { + eprintln!( + "Skipping mongodb_integration_if_none_exist_offloaded_search_uses_resource_scan (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + + let tenant = create_tenant("tenant-if-none-exist-offloaded"); + + backend + .create( + &tenant, + "Patient", + json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-OFFL-1"}] + }), + FhirVersion::default(), + ) + .await + .unwrap(); + + backend.set_search_offloaded(true); + + let entry = BundleEntry { + method: BundleMethod::Post, + url: "Patient".to_string(), + resource: Some(json!({ + "resourceType": "Patient", + "identifier": [{"system": "http://example.org/mrn", "value": "MRN-OFFL-1"}] + })), + if_match: None, + if_none_match: None, + if_none_exist: Some("identifier=http://example.org/mrn|MRN-OFFL-1".to_string()), + full_url: Some("urn:uuid:offloaded".to_string()), + }; + + let Some(result) = process_transaction_or_skip( + &backend, + &tenant, + vec![entry], + "mongodb_integration_if_none_exist_offloaded_search_uses_resource_scan", + ) + .await + else { + return; + }; + + assert_eq!( + result.entries[0].status, 200, + "offloaded-search fallback must find the existing resource and suppress the create" + ); + assert_eq!( + backend.count(&tenant, Some("Patient")).await.unwrap(), + 1, + "no duplicate should have been created" + ); +}