diff --git a/crates/persistence/src/backends/postgres/search/query_builder.rs b/crates/persistence/src/backends/postgres/search/query_builder.rs index 3fdbeaffc..f949e0935 100644 --- a/crates/persistence/src/backends/postgres/search/query_builder.rs +++ b/crates/persistence/src/backends/postgres/search/query_builder.rs @@ -758,15 +758,76 @@ impl PostgresQueryBuilder { Some(combined) } + /// Builds a `last_updated` comparison against the `resources` column. + /// + /// Binds real timestamps — `last_updated` is `TIMESTAMPTZ`, and a text + /// parameter fails serialization outright (#871) — and applies the same + /// precision-range semantics as [`Self::build_date_condition`]: `eq` at + /// day precision means `[day, day+1)`, not a scalar equality that nothing + /// can ever hit. fn build_last_updated_condition(values: &[SearchValue], offset: usize) -> Option { let mut conditions = Vec::new(); - for (i, value) in values.iter().enumerate() { - let param_num = offset + i + 1; - let op = Self::prefix_to_operator(&value.prefix); - conditions.push(SqlFragment::with_params( - format!("last_updated {} ${}", op, param_num), - vec![SqlParam::text(&value.value)], - )); + let mut next = offset; + for value in values { + let (start, end) = date_precision_range(&value.value); + let degenerate = start == end; + let (sql, params): (String, Vec) = match value.prefix { + SearchPrefix::Eq if !degenerate => { + next += 1; + let a = next; + next += 1; + ( + format!("last_updated >= ${a} AND last_updated < ${next}"), + vec![SqlParam::Timestamp(start), SqlParam::Timestamp(end)], + ) + } + SearchPrefix::Ne if !degenerate => { + next += 1; + let a = next; + next += 1; + ( + format!("(last_updated < ${a} OR last_updated >= ${next})"), + vec![SqlParam::Timestamp(start), SqlParam::Timestamp(end)], + ) + } + SearchPrefix::Gt | SearchPrefix::Sa if !degenerate => { + next += 1; + ( + format!("last_updated >= ${next}"), + vec![SqlParam::Timestamp(end)], + ) + } + SearchPrefix::Lt | SearchPrefix::Eb if !degenerate => { + next += 1; + ( + format!("last_updated < ${next}"), + vec![SqlParam::Timestamp(start)], + ) + } + SearchPrefix::Ge if !degenerate => { + next += 1; + ( + format!("last_updated >= ${next}"), + vec![SqlParam::Timestamp(start)], + ) + } + SearchPrefix::Le if !degenerate => { + next += 1; + ( + format!("last_updated < ${next}"), + vec![SqlParam::Timestamp(end)], + ) + } + other => { + let op = Self::prefix_to_operator(&other); + next += 1; + ( + format!("last_updated {op} ${next}"), + vec![SqlParam::Timestamp(start)], + ) + } + }; + conditions.push(SqlFragment::with_params(sql, params)); } if conditions.is_empty() { return None; @@ -2555,6 +2616,55 @@ mod tests { assert_eq!(frag.params.len(), 1); } + #[test] + fn last_updated_binds_timestamps_with_precision_ranges() { + // #871: a text param bound against the TIMESTAMPTZ column fails + // serialization, and day-precision eq must mean [day, day+1). + let query = SearchQuery::new("Patient").with_parameter(special_param( + "_lastUpdated", + vec![SearchValue::eq("2026-09-01")], + )); + let frag = PostgresQueryBuilder::build_search_query(&query, 2) + .expect("_lastUpdated must produce a condition"); + + assert!( + frag.sql + .contains("last_updated >= $3 AND last_updated < $4"), + "{}", + frag.sql + ); + assert_eq!(frag.params.len(), 2); + assert!( + frag.params + .iter() + .all(|p| matches!(p, SqlParam::Timestamp(_))), + "must bind timestamps, not text: {:?}", + frag.params + ); + } + + #[test] + fn last_updated_gt_excludes_the_named_day() { + let query = SearchQuery::new("Patient").with_parameter(special_param( + "_lastUpdated", + vec![SearchValue { + prefix: SearchPrefix::Gt, + value: "2026-09-01".to_string(), + }], + )); + let frag = PostgresQueryBuilder::build_search_query(&query, 2) + .expect("_lastUpdated gt must produce a condition"); + + assert!(frag.sql.contains("last_updated >= $3"), "{}", frag.sql); + assert_eq!(frag.params.len(), 1); + match &frag.params[0] { + SqlParam::Timestamp(ts) => { + assert_eq!(ts.to_rfc3339(), "2026-09-02T00:00:00+00:00"); + } + other => panic!("must bind the period end as a timestamp: {other:?}"), + } + } + #[test] fn text_or_list_placeholders_are_gap_free() { // Two terms OR together, and a blank one must not consume a placeholder diff --git a/crates/persistence/src/core/search.rs b/crates/persistence/src/core/search.rs index 6fc339893..ab8675d7f 100644 --- a/crates/persistence/src/core/search.rs +++ b/crates/persistence/src/core/search.rs @@ -118,7 +118,7 @@ impl SearchResult { bundle = bundle.with_entry( BundleEntry::match_entry( format!("{}/{}", base_url, url), - resource.content().clone(), + resource.content_with_meta(), ) .with_score(score), ); @@ -127,7 +127,7 @@ impl SearchResult { for resource in &self.included { bundle = bundle.with_entry(BundleEntry::include_entry( format!("{}/{}", base_url, resource.url()), - resource.content().clone(), + resource.content_with_meta(), )); } @@ -158,8 +158,11 @@ impl SearchResult { let url = resource.url(); let score = scores.get(&url).copied(); bundle = bundle.with_entry( - BundleEntry::match_entry(format!("{}/{}", base_url, url), resource.into_content()) - .with_score(score), + BundleEntry::match_entry( + format!("{}/{}", base_url, url), + resource.into_content_with_meta(), + ) + .with_score(score), ); } @@ -167,7 +170,7 @@ impl SearchResult { let url = resource.url(); bundle = bundle.with_entry(BundleEntry::include_entry( format!("{}/{}", base_url, url), - resource.into_content(), + resource.into_content_with_meta(), )); } diff --git a/crates/persistence/src/core/transaction.rs b/crates/persistence/src/core/transaction.rs index 957bc236e..806e8d696 100644 --- a/crates/persistence/src/core/transaction.rs +++ b/crates/persistence/src/core/transaction.rs @@ -329,7 +329,7 @@ impl BundleEntryResult { location: Some(resource.versioned_url()), etag: Some(resource.etag().to_string()), last_modified: Some(resource.last_modified().to_rfc3339()), - resource: Some(resource.into_content()), + resource: Some(resource.content_with_meta()), outcome: None, } } @@ -341,7 +341,7 @@ impl BundleEntryResult { location: None, etag: Some(resource.etag().to_string()), last_modified: Some(resource.last_modified().to_rfc3339()), - resource: Some(resource.into_content()), + resource: Some(resource.content_with_meta()), outcome: None, } } diff --git a/crates/persistence/src/types/stored_resource.rs b/crates/persistence/src/types/stored_resource.rs index 885d61e5a..0de193194 100644 --- a/crates/persistence/src/types/stored_resource.rs +++ b/crates/persistence/src/types/stored_resource.rs @@ -190,6 +190,47 @@ impl StoredResource { &self.content } + /// Returns the content with server-populated `meta.versionId` and + /// `meta.lastUpdated` merged in from this row's version and timestamp. + /// + /// Stored resources persist the content as submitted — the version and + /// timestamp live in their own columns — so bodies read back without this + /// carry no server metadata even though the ETag/Last-Modified headers do + /// (#873). Client-supplied `meta` members (`profile`, `tag`, `security`) + /// are preserved; `versionId`/`lastUpdated` are overwritten because the + /// server's row is authoritative for both. + pub fn content_with_meta(&self) -> Value { + Self::merge_meta(self.content.clone(), &self.version_id, self.last_modified) + } + + /// Consumes self and returns the content with server-populated + /// `meta.versionId` / `meta.lastUpdated` merged in. + /// + /// This is [`Self::content_with_meta`] without the clone — for callers + /// that are about to drop the `StoredResource` anyway (see + /// `SearchResult::into_bundle`). + pub fn into_content_with_meta(self) -> Value { + Self::merge_meta(self.content, &self.version_id, self.last_modified) + } + + fn merge_meta(mut content: Value, version_id: &str, last_modified: DateTime) -> Value { + if let Value::Object(obj) = &mut content { + let meta = obj + .entry("meta") + .or_insert_with(|| Value::Object(serde_json::Map::new())); + if let Value::Object(meta) = meta { + meta.insert("versionId".into(), Value::String(version_id.to_string())); + meta.insert( + "lastUpdated".into(), + Value::String( + last_modified.to_rfc3339_opts(chrono::SecondsFormat::Millis, true), + ), + ); + } + } + content + } + /// Returns a mutable reference to the resource content. pub fn content_mut(&mut self) -> &mut Value { &mut self.content diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index 40e92be1c..99eb060ed 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -715,17 +715,27 @@ mod query_builder_tests { #[test] fn test_prefix_operators() { - // Test all prefix-to-operator mappings by using date search - let prefixes_and_ops = vec![ - (SearchPrefix::Eq, "="), - (SearchPrefix::Ne, "!="), - (SearchPrefix::Gt, ">"), - (SearchPrefix::Lt, "<"), - (SearchPrefix::Ge, ">="), - (SearchPrefix::Le, "<="), + // Day-precision prefixes compare against the value's [day, day+1) + // range (#871), matching the date-parameter semantics from #463: eq + // must mean "inside the named day", gt must exclude it entirely. + let cases = vec![ + ( + SearchPrefix::Eq, + "last_updated >= $1 AND last_updated < $2", + 2, + ), + ( + SearchPrefix::Ne, + "(last_updated < $1 OR last_updated >= $2)", + 2, + ), + (SearchPrefix::Gt, "last_updated >= $1", 1), + (SearchPrefix::Lt, "last_updated < $1", 1), + (SearchPrefix::Ge, "last_updated >= $1", 1), + (SearchPrefix::Le, "last_updated < $1", 1), ]; - for (prefix, expected_op) in prefixes_and_ops { + for (prefix, expected_sql, expected_params) in cases { let query = SearchQuery::new("Patient").with_parameter(SearchParameter { name: "_lastUpdated".to_string(), param_type: SearchParamType::Date, @@ -739,15 +749,37 @@ mod query_builder_tests { assert!(result.is_some(), "Failed for prefix {:?}", prefix); let fragment = result.unwrap(); assert!( - fragment - .sql - .contains(&format!("last_updated {} $", expected_op)), - "Expected operator '{}' for prefix {:?}, got SQL: {}", - expected_op, + fragment.sql.contains(expected_sql), + "Expected '{}' for prefix {:?}, got SQL: {}", + expected_sql, prefix, fragment.sql ); + assert_eq!( + fragment.params.len(), + expected_params, + "param count for prefix {:?}", + prefix + ); } + + // A full-precision instant is a degenerate range and falls back to + // scalar comparison. + let query = SearchQuery::new("Patient").with_parameter(SearchParameter { + name: "_lastUpdated".to_string(), + param_type: SearchParamType::Date, + modifier: None, + values: vec![SearchValue::new(SearchPrefix::Gt, "2024-01-01T10:00:00Z")], + chain: vec![], + components: vec![], + }); + let fragment = PostgresQueryBuilder::build_search_query(&query, 0) + .expect("full-precision instant must build"); + assert!( + fragment.sql.contains("last_updated > $1"), + "full-precision gt must stay scalar, got SQL: {}", + fragment.sql + ); } } @@ -5927,6 +5959,46 @@ mod postgres_integration { "expected a backend error from an unmigrated database, got {create:?}" ); } + /// Claims manifests in a loop until the lease for `target` comes back, + /// holding any other lease it picks up along the way (so the loop cannot + /// re-claim the same foreign manifest) and returning those to the queue + /// once the target is held. Robust to concurrent tests sharing the + /// testcontainers PostgreSQL instance — the submit-side twin of + /// `claim_specific` for exports. + async fn claim_specific_manifest( + backend: &PostgresBackend, + worker_id: &helios_persistence::core::WorkerId, + submission_id: &helios_persistence::core::SubmissionId, + target_manifest_id: &str, + lease_duration: std::time::Duration, + ) -> helios_persistence::core::ManifestLease { + use helios_persistence::core::SubmitClaimStrategy; + + let mut held = Vec::new(); + let mut found = None; + for _ in 0..100 { + match backend + .claim_next_manifest(worker_id, lease_duration) + .await + .unwrap() + { + Some(lease) + if lease.submission_id == *submission_id + && lease.manifest_id == target_manifest_id => + { + found = Some(lease); + break; + } + Some(other) => held.push(other), + None => tokio::time::sleep(std::time::Duration::from_millis(20)).await, + } + } + for other in held { + let _ = SubmitClaimStrategy::release(backend, other).await; + } + found.expect("never claimed the expected manifest") + } + /// The `import` / `metadata` kickoff directives must survive the PostgreSQL /// round-trip and reach the worker, and `merge` must actually merge. /// @@ -5938,8 +6010,7 @@ mod postgres_integration { async fn postgres_bulk_submit_import_directives_round_trip() { use helios_persistence::core::{ BulkProcessingOptions, BulkSubmitProvider, IMPORT_MODE_PARAMETER_URL, ImportMode, - ManifestFetchParams, NdjsonEntry, SubmissionId, SubmitClaimStrategy, - SubmitWorkerStorage, + ManifestFetchParams, NdjsonEntry, SubmissionId, SubmitWorkerStorage, }; let backend = create_backend().await; @@ -5971,15 +6042,25 @@ mod postgres_integration { .await .unwrap(); - // Claim the manifest the way the worker does, then read its view back. - let lease = backend - .claim_next_manifest( - &helios_persistence::core::WorkerId::new("pg-import-worker"), - std::time::Duration::from_secs(60), - ) - .await - .unwrap() - .expect("claimable manifest"); + // Claim *this* manifest the way the worker does, then read its view + // back. The claim queue is cross-tenant and ordered by `added_at`, and + // the test binary shares one container database, so a plain + // `claim_next_manifest` can hand back another test's manifest — and + // an unleased `processing` manifest (the synchronous `process_entries` + // path leaves one behind) is reclaimable, so "move it out of pending" + // elsewhere is no defence. Loop until ours comes back, returning + // anything else to the queue. + let lease = claim_specific_manifest( + &backend, + &helios_persistence::core::WorkerId::new(format!( + "pg-import-worker-{}", + uuid::Uuid::new_v4() + )), + &sub_id, + &manifest.manifest_id, + std::time::Duration::from_secs(60), + ) + .await; let view = backend.get_manifest_for_worker(&lease).await.unwrap(); assert_eq!(view.import_directives, directives); assert_eq!(view.metadata, metadata); @@ -6055,10 +6136,11 @@ mod postgres_integration { .add_manifest(&tenant, &sub_id, Some("https://provider/b.json"), None) .await .unwrap(); - // Move the manifest out of `pending` right away: the test binary - // shares one container database, and a concurrently running test that - // calls claim_next_manifest would otherwise claim this one (the claim - // queue is cross-tenant by design). + // Mark the manifest `processing` right away. Note this is not a + // defence against other tests' `claim_next_manifest` calls — an + // unleased `processing` manifest is reclaimable, so a claimant may + // still pick this one up transiently — which is why claiming tests use + // `claim_specific_manifest` and hand foreign manifests back. backend .process_entries( &tenant, diff --git a/crates/rest/src/handlers/create.rs b/crates/rest/src/handlers/create.rs index 0be6d78ee..518138f22 100644 --- a/crates/rest/src/handlers/create.rs +++ b/crates/rest/src/handlers/create.rs @@ -294,11 +294,10 @@ fn build_create_response( } _ => { // Default: return=representation - format_resource_response(status, header_map, stored.content(), format).map_err(|_| { - RestError::InternalError { + format_resource_response(status, header_map, &stored.content_with_meta(), format) + .map_err(|_| RestError::InternalError { message: "Failed to serialize response".to_string(), - } - }) + }) } } } @@ -314,9 +313,14 @@ fn build_existing_response( match prefer.return_preference() { Some("minimal") => Ok((StatusCode::OK, header_map).into_response()), - _ => format_resource_response(StatusCode::OK, header_map, stored.content(), format) - .map_err(|_| RestError::InternalError { - message: "Failed to serialize response".to_string(), - }), + _ => format_resource_response( + StatusCode::OK, + header_map, + &stored.content_with_meta(), + format, + ) + .map_err(|_| RestError::InternalError { + message: "Failed to serialize response".to_string(), + }), } } diff --git a/crates/rest/src/handlers/history.rs b/crates/rest/src/handlers/history.rs index 26bdd4f12..ff5e82d21 100644 --- a/crates/rest/src/handlers/history.rs +++ b/crates/rest/src/handlers/history.rs @@ -117,7 +117,7 @@ where // Deleted versions carry no resource body. if entry.method != HistoryMethod::Delete { - bundle_entry["resource"] = resource.content().clone(); + bundle_entry["resource"] = resource.content_with_meta(); } bundle_entry diff --git a/crates/rest/src/handlers/patch.rs b/crates/rest/src/handlers/patch.rs index 164813334..7e12f1933 100644 --- a/crates/rest/src/handlers/patch.rs +++ b/crates/rest/src/handlers/patch.rs @@ -324,6 +324,6 @@ fn build_patch_response( }); Ok((StatusCode::OK, header_map, Json(outcome)).into_response()) } - _ => Ok((StatusCode::OK, header_map, Json(stored.content().clone())).into_response()), + _ => Ok((StatusCode::OK, header_map, Json(stored.content_with_meta())).into_response()), } } diff --git a/crates/rest/src/handlers/read.rs b/crates/rest/src/handlers/read.rs index 21c507f71..171e0a2d7 100644 --- a/crates/rest/src/handlers/read.rs +++ b/crates/rest/src/handlers/read.rs @@ -142,7 +142,7 @@ where .get("_elements") .map(|v| v.split(',').map(|s| s.trim()).collect()); - let mut content = stored.content().clone(); + let mut content = stored.content_with_meta(); let mut subsetted = false; if let Some(mode) = summary_mode { diff --git a/crates/rest/src/handlers/subscription_event.rs b/crates/rest/src/handlers/subscription_event.rs index 8df93fa63..404fbb2f9 100644 --- a/crates/rest/src/handlers/subscription_event.rs +++ b/crates/rest/src/handlers/subscription_event.rs @@ -30,7 +30,7 @@ pub fn emit_subscription_event( resource_id: stored.id().to_string(), version_id: stored.version_id().to_string(), event_type, - resource: Some(stored.content().clone()), + resource: Some(stored.content_with_meta()), previous_resource: None, timestamp: chrono::Utc::now(), }; diff --git a/crates/rest/src/handlers/update.rs b/crates/rest/src/handlers/update.rs index ba1c2ad8e..fd5095f7b 100644 --- a/crates/rest/src/handlers/update.rs +++ b/crates/rest/src/handlers/update.rs @@ -499,10 +499,9 @@ fn build_update_response( } }) } - _ => format_resource_response(status, header_map, stored.content(), format).map_err(|_| { - RestError::InternalError { + _ => format_resource_response(status, header_map, &stored.content_with_meta(), format) + .map_err(|_| RestError::InternalError { message: "Failed to serialize response".to_string(), - } - }), + }), } } diff --git a/crates/rest/src/handlers/vread.rs b/crates/rest/src/handlers/vread.rs index c8b33b93d..808b30328 100644 --- a/crates/rest/src/handlers/vread.rs +++ b/crates/rest/src/handlers/vread.rs @@ -133,10 +133,15 @@ where "Returning resource version" ); - format_resource_response(StatusCode::OK, headers, stored.content(), negotiated.format) - .map_err(|_| RestError::InternalError { - message: "Failed to serialize response".to_string(), - }) + format_resource_response( + StatusCode::OK, + headers, + &stored.content_with_meta(), + negotiated.format, + ) + .map_err(|_| RestError::InternalError { + message: "Failed to serialize response".to_string(), + }) } None => Err(RestError::VersionNotFound { resource_type, diff --git a/crates/rest/tests/search_integration.rs b/crates/rest/tests/search_integration.rs index 2e1ba9a9d..76fb81040 100644 --- a/crates/rest/tests/search_integration.rs +++ b/crates/rest/tests/search_integration.rs @@ -3016,3 +3016,93 @@ mod date_precision { ); } } + +/// #873: response bodies must carry server-populated `meta.versionId` and +/// `meta.lastUpdated` on every read path. The row's version and timestamp were +/// header-only (ETag / Last-Modified), so search entries, reads, vreads, and +/// history bundles all returned resources with no server metadata — visibly, +/// the Resources workspace's UPDATED column rendered blank on every backend. +mod server_meta { + use super::*; + + #[tokio::test] + async fn test_bodies_carry_version_and_last_updated_on_every_read_path() { + let (server, _backend) = create_test_server().await; + let tenant_header = HeaderValue::from_static("test-tenant"); + + // Create: the response body already carries v1 meta, with + // client-supplied meta members preserved. + let created = server + .post("/Patient") + .add_header(X_TENANT_ID, tenant_header.clone()) + .json(&json!({ + "resourceType": "Patient", + "meta": {"profile": ["http://example.org/StructureDefinition/p"]}, + "name": [{"family": "Metadata"}] + })) + .await; + created.assert_status(StatusCode::CREATED); + let body: Value = created.json(); + assert_eq!(body["meta"]["versionId"], "1"); + assert!(body["meta"]["lastUpdated"].is_string()); + assert_eq!( + body["meta"]["profile"][0], "http://example.org/StructureDefinition/p", + "client-supplied meta members must survive the injection" + ); + let id = body["id"].as_str().expect("created id").to_string(); + + // Update to v2. + let mut updated_body = body.clone(); + updated_body["name"][0]["family"] = json!("Metadata2"); + server + .put(&format!("/Patient/{id}")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .json(&updated_body) + .await + .assert_status_ok(); + + // Read: current version's meta. + let read: Value = server + .get(&format!("/Patient/{id}")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .await + .json(); + assert_eq!(read["meta"]["versionId"], "2"); + assert!(read["meta"]["lastUpdated"].is_string()); + + // Vread: each version reports its own meta. + let v1: Value = server + .get(&format!("/Patient/{id}/_history/1")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .await + .json(); + assert_eq!(v1["meta"]["versionId"], "1"); + + // History bundle entries carry their version's meta. + let history: Value = server + .get(&format!("/Patient/{id}/_history")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .await + .json(); + let versions: Vec<&str> = history["entry"] + .as_array() + .expect("history entries") + .iter() + .map(|e| e["resource"]["meta"]["versionId"].as_str().unwrap_or("")) + .collect(); + assert!( + versions.contains(&"1") && versions.contains(&"2"), + "history entries must carry per-version meta, got {versions:?}" + ); + + // Search: match entries carry meta. + let search: Value = server + .get("/Patient?family=Metadata2") + .add_header(X_TENANT_ID, tenant_header) + .await + .json(); + let entry = &get_bundle_entries(&search)[0]["resource"]; + assert_eq!(entry["meta"]["versionId"], "2"); + assert!(entry["meta"]["lastUpdated"].is_string()); + } +}