diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4f612e518..94d3c7be0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -219,6 +219,9 @@ jobs: - name: Check formatting run: cargo fmt --all -- --check + - name: Check HTS canonical precedence is centralized (issue #200) + run: crates/hts/scripts/check-canonical-precedence.sh + - name: Run clippy run: | cargo clippy --all-targets --all-features -- -D warnings -A clippy::items_after_test_module -A clippy::large_enum_variant -A clippy::question_mark -A clippy::collapsible_match -A clippy::collapsible_if -A clippy::field_reassign_with_default -A clippy::doc-overindented-list-items -A clippy::doc-lazy-continuation diff --git a/crates/hts/docs/ig-publisher-compatibility.md b/crates/hts/docs/ig-publisher-compatibility.md index dd0e5959e..2e0d9caf5 100644 --- a/crates/hts/docs/ig-publisher-compatibility.md +++ b/crates/hts/docs/ig-publisher-compatibility.md @@ -123,19 +123,95 @@ with `version: "current"` and `content: not-present` (no concepts). When `$validate-code` (or backend `validate_code`) receives `version=current`: -1. **`fetch_versions` / Postgres resolve query** orders candidates so that: +1. **`cs_precedence_order_by`** (`src/backends/mod.rs`) is the single definition of + same-URL precedence. Both backends embed its text verbatim, so they cannot drift. + It orders candidates so that: - `content = complete` or `supplement` ranks before `fragment`, `example`, `not-present` - rows **with concepts** rank before empty stubs + - rows from the package that **owns** the canonical URL rank before re-published + copies (`authority_rank` — see §2a) - then highest `version` string, then stable `id` 2. **`resolve_code_system*`** treats `current` as “pick the first candidate after ordering” instead of literal string equality. +The first two tiers are load-bearing for this behaviour and must stay ahead of the +version tier: the stub's version string is literally `"current"`, and `'c' > '2'`, +so `"current"` text-sorts **above** a real edition like `"20260501"`. Only the +content and has-concepts tiers keep the real import winning. + ### Code changes -- `src/backends/mod.rs` — `code_system_version_is_current()` +- `src/backends/mod.rs` — `code_system_version_is_current()`, `cs_precedence_order_by()` - `src/backends/sqlite/code_system.rs` — `fetch_versions`, `resolve_code_system_uncached` - `src/backends/postgres/code_system.rs` — `resolve_code_system` ORDER BY + `current` arm +--- + +## 2a. Provenance precedence: original vs re-published copy (issue #200) + +`hl7.fhir.r4.core` declares canonical `http://hl7.org/fhir` but re-ships **746 +CodeSystems and 603 ValueSets** owned by `http://terminology.hl7.org`. Those copies +are stamped with the *FHIR release* version (`4.0.1`) rather than the code system's +own version (`1.0.0`), and many are truncated — `audit-event-type` carries 1 of its +5 concepts. + +Both rows are `content: complete` and both have concepts, so no tier above could +separate them, and `4.0.1` outsorts `1.0.0` under **any** version ordering +(lexicographic or semver). Before this fix, 208 CodeSystems resolved to the stale +core copy and 1,297 concepts were unreachable via a bare `$validate-code`. + +### The rule + +Nothing intrinsic to the rows distinguishes them — only where they came from. Every +FHIR NPM package declares a `canonical` base in its `package.json`; a package +publishing a URL **outside** its own base is re-publishing someone else's resource. +That is recorded at import time in `code_systems.authority_rank` / +`value_sets.authority_rank` (see `import::bundle_parser::authority_rank_for`): + +| rank | meaning | +|------|---------| +| `0` `AUTHORITY_OWNER` | Positive evidence: the package declares a `canonical` base covering this URL, or a native importer loaded the publisher's own distribution (SNOMED RF2, LOINC, …). | +| `1` `AUTHORITY_UNKNOWN` | Everything else — no package context (REST write, `$import-bundle`), or the package does not declare a base covering the URL. | + +**We only promote on positive evidence; we never demote on the absence of it.** That +asymmetry is load-bearing, and getting it wrong breaks the fix in two ways: + +* **A package manifest may not describe its content's authority.** `us.nlm.vsac` + declares the fhir.org *registry* base `http://fhir.org/packages/us.nlm.vsac` yet + ships 14,850 ValueSets under `http://cts.nlm.nih.gov/...`. Those are genuine + originals. Branding them "foreign copies" would be a claim we have not established + — and since `vs_precedence_order_by` has no content/has-concepts tier beneath the + provenance tier, it would let any REST upload outrank the real VSAC definition + outright. Marking them merely *unknown* costs them nothing: rank is only ever a + tiebreak among rows sharing a URL, and no one else ships those URLs. +* **An unattributed write must not claim ownership.** If `$import-bundle` defaulted to + `OWNER`, replaying `hl7.fhir.r4.core`'s truncated copy would stamp it rank 0 — and + because the upsert's `MIN` only ever *lowers* a rank, that copy would outrank THO's + original permanently, reinstating this bug with no way back. + +This needs no hardcoded URL list and no hand-maintained package-priority table: it +self-maintains as packages are added. + +The tier sits **below** content and has-concepts deliberately: a populated copy still +beats an empty stub from the owning package — being authoritative is worthless if the +row has no concepts to validate against. + +### Upgrading an existing database + +`authority_rank` cannot be backfilled — nothing already stored on a row says which +package supplied it. The column is therefore **nullable**, `NULL` meaning "never +claimed", and readers `COALESCE` it to `0`, so an un-re-imported database behaves +exactly as it did before. Adding the column is the signal that the database predates +provenance, and the migration clears the `.tgz` rows from the `bootstrap_imports` +ledger so the next startup re-imports those packages and records the truth. +Without that, the ledger (which skips files whose size and mtime are unchanged) would +leave every row at its default and the fix would be a silent no-op in production. +Only `.tgz` entries are cleared — the multi-GB SNOMED/LOINC/RxNorm archives are not +re-imported. + +The copies are **ranked, not discarded**: an explicit `version=4.0.1` still resolves +to the core row, and `resource_json` stays byte-faithful to what the package shipped. + ### Tests - `validate_code_version_current_resolves_loaded_edition` (SQLite integration test) diff --git a/crates/hts/scripts/check-canonical-precedence.sh b/crates/hts/scripts/check-canonical-precedence.sh new file mode 100755 index 000000000..0ca455c99 --- /dev/null +++ b/crates/hts/scripts/check-canonical-precedence.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +# Guard for issue #200. +# +# Same-URL precedence (which of several rows sharing a canonical URL wins) must +# be expressed ONCE, in `backends::cs_precedence_order_by` / `vs_precedence_order_by`. +# The bug this guards against was caused by that rule being re-implemented ad hoc +# in ~40 backend queries that drifted apart, so that `$validate-code` resolved one +# row while `$lookup` read another. +# +# Fails if any query orders `code_systems` / `value_sets` by a hand-rolled version +# sort instead of the shared helper. +set -euo pipefail +cd "$(dirname "$0")/.." + +# A bare `COALESCE(.version, '') DESC` sort is the signature of the old, +# divergent rule. Match ANY table alias (`version`, `s.version`, `cs.version`, +# `code_systems.version`, …) so a query cannot dodge the guard by renaming its +# alias — the earlier three-literal form let exactly that slip through. +# +# `backends/mod.rs` is the one sanctioned home of the ordering: the shared +# helpers there emit `COALESCE({alias}.version, '') DESC` with a `{alias}` +# placeholder, which does not match the concrete-alias pattern anyway, but it is +# excluded explicitly so the guard stays correct even if a helper is later +# rewritten to use a literal alias. +if hits=$(grep -rnE --include=*.rs \ + "COALESCE\(([A-Za-z_][A-Za-z0-9_]*\.)?version, ''\) DESC" \ + src/ 2>/dev/null | grep -v 'src/backends/mod.rs'); then + echo "ERROR: hand-rolled same-URL ordering found." >&2 + echo "Use crate::backends::cs_precedence_order_by() / vs_precedence_order_by() instead." >&2 + echo "See crates/hts/docs/ig-publisher-compatibility.md §2a (issue #200)." >&2 + echo "$hits" >&2 + exit 1 +fi + +echo "OK: same-URL precedence is centralized." diff --git a/crates/hts/src/backends/mod.rs b/crates/hts/src/backends/mod.rs index 88f2b7db0..8ce2b0e54 100644 --- a/crates/hts/src/backends/mod.rs +++ b/crates/hts/src/backends/mod.rs @@ -22,6 +22,80 @@ pub mod postgres; #[cfg(feature = "postgres")] pub use postgres::PostgresTerminologyBackend; +/// Precedence among `code_systems` rows sharing one canonical URL. +/// +/// This is the single definition of "which row wins when the caller did not pin +/// a version". Both backends embed it verbatim, so SQLite and Postgres cannot +/// drift apart. It assumes the candidate table is named/aliased `code_systems`. +/// +/// Tiers, in order: +/// +/// 1. **`content`** — a usable definition beats a `fragment`/`example`, which +/// beats a `not-present` stub. Load-bearing: VSAC and the FHIR core packages +/// ship empty `not-present` stubs for SNOMED/LOINC whose `version` string +/// (`"current"`) text-sorts *above* a real edition (`"20260501"`), so this +/// tier is the only thing keeping `version=current` resolving to the real +/// import. See `docs/ig-publisher-compatibility.md`. +/// 2. **has concepts** — a populated row beats an empty one, for the same reason. +/// 3. **`authority_rank`** — the original beats a re-published copy. This is the +/// tier that fixes the `terminology.hl7.org` collisions: `hl7.fhir.r4.core` +/// re-ships 746 THO CodeSystems stamped with the *FHIR release* version +/// (`4.0.1`), which outsorts the code system's own version (`1.0.0`) under any +/// ordering — lexicographic or semver — even though the copy is truncated. +/// Provenance is the only thing that separates those rows; nothing intrinsic +/// to them does. `NULL` (a row imported before the column existed) coalesces +/// to 0, so an un-re-imported database keeps its previous behaviour. +/// 4. **`version`** — highest version string wins, as before. +/// 5. **`id`** — a stable, deterministic final tiebreak. +/// +/// Tier 3 sits *below* 1 and 2 deliberately: if a package that owns a canonical +/// ships only an empty stub of it while some other package ships a populated +/// copy, the populated row should still win — being authoritative is worthless +/// if the row has no concepts to validate against. +/// +/// `alias` is how the `code_systems` table is named in the enclosing query +/// (`"code_systems"` when unaliased, or e.g. `"s"` / `"cs"`). +/// +/// Every query that has to choose one row for a canonical URL must order by +/// this. When they disagree, the server becomes internally incoherent — e.g. +/// `$validate-code` resolving against the authoritative row while `$lookup` +/// reads `resource_json` from the truncated copy. That incoherence is what +/// let this bug class persist. +pub(crate) fn cs_precedence_order_by(alias: &str) -> String { + format!( + "(CASE COALESCE({alias}.content, 'complete') \ + WHEN 'complete' THEN 0 \ + WHEN 'supplement' THEN 0 \ + WHEN 'fragment' THEN 1 \ + WHEN 'example' THEN 1 \ + WHEN 'not-present' THEN 2 \ + ELSE 1 END), \ + (CASE WHEN EXISTS \ + (SELECT 1 FROM concepts hc WHERE hc.system_id = {alias}.id) \ + THEN 0 ELSE 1 END), \ + COALESCE({alias}.authority_rank, 0), \ + COALESCE({alias}.version, '') DESC, \ + {alias}.id", + ) +} + +/// Precedence among `value_sets` rows sharing one canonical URL. +/// +/// The ValueSet analogue of [`cs_precedence_order_by`]. `value_sets` has no +/// `content` column and no concept rows, so only the provenance and version +/// tiers apply: original beats re-published copy, then highest version, then a +/// stable `id`. +/// +/// This matters for the same reason: `hl7.fhir.r4.core` re-ships 603 THO +/// ValueSets under canonical URLs it does not own. +pub(crate) fn vs_precedence_order_by(alias: &str) -> String { + format!( + "COALESCE({alias}.authority_rank, 0), \ + COALESCE({alias}.version, '') DESC, \ + {alias}.id", + ) +} + /// True when `ver` is the sentinel `current` (case-insensitive). /// /// IG Publisher, VSAC stubs, and CQL-generated Library `dataRequirement` entries diff --git a/crates/hts/src/backends/postgres/code_system.rs b/crates/hts/src/backends/postgres/code_system.rs index 49e615d3d..2491ccb7e 100644 --- a/crates/hts/src/backends/postgres/code_system.rs +++ b/crates/hts/src/backends/postgres/code_system.rs @@ -349,14 +349,15 @@ impl CodeSystemOperations for PostgresTerminologyBackend { // Echo the code's display from any stored version of the CS, // so consumers can see the concept exists (only the version is // wrong). Matches `postgres/value_set.rs:506-517`. + let sql = format!( + "SELECT c.display FROM concepts c \ + JOIN code_systems s ON s.id = c.system_id \ + WHERE s.url = $1 AND c.code = $2 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("s") + ); let display = client - .query_opt( - "SELECT c.display FROM concepts c - JOIN code_systems s ON s.id = c.system_id - WHERE s.url = $1 AND c.code = $2 - ORDER BY COALESCE(s.version, '') DESC LIMIT 1", - &[&system, &req.code], - ) + .query_opt(&sql, &[&system, &req.code]) .await .ok() .flatten() @@ -444,11 +445,25 @@ impl CodeSystemOperations for PostgresTerminologyBackend { Some(c) => c, None => { // Match the IG `validation/cs-code-bad-code` text format exactly. - let cs_version_str = cs_version_for_msg(&client, &system).await; - let cs_content = cs_content_for_url(&client, &system).await; + // + // Report the row we ACTUALLY validated against. These used to + // re-query by URL (`cs_version_for_msg` / `cs_content_for_url`), + // which picks the best-ranked row for the canonical — NOT + // necessarily the row `resolve_code_system` chose. With a version + // pin (now honoured, per issue #200) the two diverge, so the + // message could cite a version the code was never checked against + // and, worse, `content` gates the fragment→warning vs + // complete→error branch below: reading `content` off a different + // row can silently downgrade a genuine error to a warning. + // Mirrors the SQLite fix in `sqlite/code_system.rs`. + let cs_version_str = resolved_cs_version.clone(); + let cs_content = match resolved_system_id.as_deref() { + Some(id) => cs_content_for_system_id(&client, id).await, + // URL unknown entirely — no resolved row to read from. + None => cs_content_for_url(&client, &system).await, + }; // Fragment CodeSystems: unknown code is a *warning*, not an error. - // Mirrors `sqlite/code_system.rs:454-485`. if cs_content.as_deref() == Some("fragment") { let text = match cs_version_str.as_deref() { Some(v) => format!( @@ -722,14 +737,12 @@ impl CodeSystemOperations for PostgresTerminologyBackend { .get() .await .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; + let sql = format!( + "SELECT version FROM code_systems WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let row = client - .query_opt( - "SELECT version FROM code_systems - WHERE url = $1 - ORDER BY COALESCE(version, '') DESC - LIMIT 1", - &[&url], - ) + .query_opt(&sql, &[&url]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; Ok(row.and_then(|r| r.get::<_, Option>(0))) @@ -768,11 +781,13 @@ impl CodeSystemOperations for PostgresTerminologyBackend { .get() .await .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; + let sql = format!( + "SELECT resource_json->>'language' FROM code_systems \ + WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let row = client - .query_opt( - "SELECT resource_json->>'language' FROM code_systems WHERE url = $1 LIMIT 1", - &[&url], - ) + .query_opt(&sql, &[&url]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; Ok(row.and_then(|r| r.get::<_, Option>(0))) @@ -1067,13 +1082,14 @@ impl CodeSystemOperations for PostgresTerminologyBackend { // Read the base CodeSystem's resource_json (highest version), then // walk concept[] picking entries whose code is in the requested set. // Mirrors sqlite/code_system.rs:1436-1475. + let sql = format!( + "SELECT resource_json FROM code_systems \ + WHERE url = $1 AND content != 'supplement' \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let row = client - .query_opt( - "SELECT resource_json FROM code_systems - WHERE url = $1 AND content != 'supplement' - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[&system_url], - ) + .query_opt(&sql, &[&system_url]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -1142,14 +1158,15 @@ impl CodeSystemOperations for PostgresTerminologyBackend { .await .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; + let sql = format!( + "SELECT content, version, resource_json->>'supplements' \ + FROM code_systems \ + WHERE url = $1 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let row = client - .query_opt( - "SELECT content, version, resource_json->>'supplements' - FROM code_systems - WHERE url = $1 - LIMIT 1", - &[&supplement_url], - ) + .query_opt(&sql, &[&supplement_url]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; let Some(r) = row else { return Ok(None) }; @@ -1325,32 +1342,44 @@ fn walk_concepts( /// most recent (textual COALESCE-DESC), an explicit version with `.x` segments /// (or a bare numeric prefix like `"1"`) matches the highest version that /// shares the literal segments, and an exact version requires an exact match. +/// Read `content` for the row we actually resolved, keyed on its storage id. +/// +/// Keyed on `id`, never on `url`: a canonical URL can have several rows, and the +/// caller has already chosen one. Re-deriving it from the URL is how the two +/// backends drifted apart. +async fn cs_content_for_system_id( + client: &tokio_postgres::Client, + system_id: &str, +) -> Option { + client + .query_opt( + "SELECT content FROM code_systems WHERE id = $1", + &[&system_id], + ) + .await + .ok() + .flatten() + .and_then(|r| r.get::<_, Option>(0)) +} + async fn resolve_code_system( client: &tokio_postgres::Client, url: &str, version: Option<&str>, date: Option<&str>, ) -> Result<(String, String, Option), HtsError> { + // Ordering is the shared `cs_precedence_order_by` — identical text to the + // SQLite resolver, so the two backends cannot disagree about which row wins. + let sql = format!( + "SELECT id, COALESCE(name, url), version + FROM code_systems + WHERE url = $1 + AND ($2::text IS NULL OR (resource_json->>'date') <= $2) + ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); let rows = client - .query( - "SELECT id, COALESCE(name, url), version - FROM code_systems - WHERE url = $1 - AND ($2::text IS NULL OR (resource_json->>'date') <= $2) - ORDER BY (CASE COALESCE(content, 'complete') - WHEN 'complete' THEN 0 - WHEN 'supplement' THEN 0 - WHEN 'fragment' THEN 1 - WHEN 'example' THEN 1 - WHEN 'not-present' THEN 2 - ELSE 1 END), - (CASE WHEN EXISTS ( - SELECT 1 FROM concepts c WHERE c.system_id = code_systems.id - ) THEN 0 ELSE 1 END), - COALESCE(version, '') DESC, - id", - &[&url, &date], - ) + .query(&sql, &[&url, &date]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -1472,14 +1501,15 @@ async fn find_concept_by_url( system_url: &str, code: &str, ) -> Option { + let sql = format!( + "SELECT c.code, c.display FROM concepts c \ + JOIN code_systems s ON s.id = c.system_id \ + WHERE s.url = $1 AND c.code = $2 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("s") + ); let row = client - .query_opt( - "SELECT c.code, c.display FROM concepts c - JOIN code_systems s ON s.id = c.system_id - WHERE s.url = $1 AND c.code = $2 - ORDER BY COALESCE(s.version, '') DESC LIMIT 1", - &[&system_url, &code], - ) + .query_opt(&sql, &[&system_url, &code]) .await .ok() .flatten()?; @@ -1498,14 +1528,15 @@ async fn find_concept_by_url_ci( system_url: &str, code: &str, ) -> Option { + let sql = format!( + "SELECT c.code, c.display FROM concepts c \ + JOIN code_systems s ON s.id = c.system_id \ + WHERE s.url = $1 AND LOWER(c.code) = LOWER($2) \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("s") + ); let row = client - .query_opt( - "SELECT c.code, c.display FROM concepts c - JOIN code_systems s ON s.id = c.system_id - WHERE s.url = $1 AND LOWER(c.code) = LOWER($2) - ORDER BY COALESCE(s.version, '') DESC LIMIT 1", - &[&system_url, &code], - ) + .query_opt(&sql, &[&system_url, &code]) .await .ok() .flatten()?; @@ -1700,16 +1731,17 @@ async fn fetch_designations_cross_version( exclude_system_id: &str, lang: &str, ) -> Result, HtsError> { + let sql = format!( + "SELECT cd.language, cd.use_system, cd.use_code, cd.value, COALESCE(s.version, '') \ + FROM concept_designations cd \ + JOIN concepts c ON c.id = cd.concept_id \ + JOIN code_systems s ON s.id = c.system_id \ + WHERE s.url = $1 AND c.code = $2 AND s.id <> $3 \ + ORDER BY {}, cd.id", + crate::backends::cs_precedence_order_by("s") + ); let rows = client - .query( - "SELECT cd.language, cd.use_system, cd.use_code, cd.value, COALESCE(s.version, '') - FROM concept_designations cd - JOIN concepts c ON c.id = cd.concept_id - JOIN code_systems s ON s.id = c.system_id - WHERE s.url = $1 AND c.code = $2 AND s.id <> $3 - ORDER BY COALESCE(s.version, '') DESC, cd.id", - &[&url, &code, &exclude_system_id], - ) + .query(&sql, &[&url, &code, &exclude_system_id]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; diff --git a/crates/hts/src/backends/postgres/concept_map.rs b/crates/hts/src/backends/postgres/concept_map.rs index 91a0546d8..dceaea6c0 100644 --- a/crates/hts/src/backends/postgres/concept_map.rs +++ b/crates/hts/src/backends/postgres/concept_map.rs @@ -150,12 +150,12 @@ impl ConceptMapOperations for PostgresTerminologyBackend { let mut groups: Vec = Vec::new(); for (system_url, codes) in &by_system { + let sql = format!( + "SELECT id FROM code_systems WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let id_rows = client - .query( - "SELECT id FROM code_systems WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[system_url], - ) + .query(&sql, &[system_url]) .await .map_err(|e| HtsError::StorageError(format!("DB error: {e}")))?; diff --git a/crates/hts/src/backends/postgres/mod.rs b/crates/hts/src/backends/postgres/mod.rs index c64269a0c..fe89a220c 100644 --- a/crates/hts/src/backends/postgres/mod.rs +++ b/crates/hts/src/backends/postgres/mod.rs @@ -308,16 +308,13 @@ impl TerminologyMetadata for PostgresTerminologyBackend { return Some(row.get::<_, String>(0)); } } - let rows = client - .query( - "SELECT url FROM code_systems \ - WHERE (resource_json->>'id') = $1 \ - ORDER BY COALESCE(version, '') DESC \ - LIMIT 1", - &[&id], - ) - .await - .ok()?; + let sql = format!( + "SELECT url FROM code_systems \ + WHERE (resource_json->>'id') = $1 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); + let rows = client.query(&sql, &[&id]).await.ok()?; rows.into_iter().next().map(|r| r.get::<_, String>(0)) } "ValueSet" => { @@ -629,8 +626,9 @@ async fn write_code_system( client .execute( "INSERT INTO code_systems - (id, url, version, name, title, status, content, resource_json, created_at, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9) + (id, url, version, name, title, status, content, resource_json, created_at, updated_at, + authority_rank) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9, $10) ON CONFLICT DO NOTHING", &[ &storage_id, @@ -642,22 +640,32 @@ async fn write_code_system( &cs.content, &resource_json, &now, + &cs.authority_rank, ], ) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; + // authority_rank keeps the strongest claim ever asserted for this row; the + // COALESCE sentinel (9 — above any real rank) means a row that predates the + // column ("never claimed") is overwritten by the first source to claim it. + // See the SQLite twin in `import/fhir_bundle.rs::write_code_system`. + let cs_update = format!( + "UPDATE code_systems SET + name = $1, + title = $2, + status = $3, + content = $4, + resource_json = $5, + updated_at = $6, + authority_rank = LEAST(COALESCE(authority_rank, {unclaimed}), $9) + WHERE url = $7 AND COALESCE(version, '') = COALESCE($8, '') + RETURNING id", + unclaimed = bundle_parser::AUTHORITY_UNCLAIMED + ); let cs_rows = client .query( - "UPDATE code_systems SET - name = $1, - title = $2, - status = $3, - content = $4, - resource_json = $5, - updated_at = $6 - WHERE url = $7 AND COALESCE(version, '') = COALESCE($8, '') - RETURNING id", + &cs_update, &[ &cs.name, &cs.title, @@ -667,6 +675,7 @@ async fn write_code_system( &now, &cs.url, &cs.version, + &cs.authority_rank, ], ) .await @@ -880,8 +889,9 @@ async fn write_value_set( client .execute( "INSERT INTO value_sets - (id, url, version, name, title, status, compose_json, resource_json, created_at, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9) + (id, url, version, name, title, status, compose_json, resource_json, created_at, updated_at, + authority_rank) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9, $10) ON CONFLICT DO NOTHING", &[ &storage_id, @@ -893,21 +903,28 @@ async fn write_value_set( &vs.compose_json, &resource_json, &now, + &vs.authority_rank, ], ) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; + // See `write_code_system` for the LEAST/COALESCE-sentinel rationale. + let vs_update = format!( + "UPDATE value_sets SET + name = $1, + title = $2, + status = $3, + compose_json = $4, + resource_json = $5, + updated_at = $6, + authority_rank = LEAST(COALESCE(authority_rank, {unclaimed}), $9) + WHERE url = $7 AND COALESCE(version, '') = COALESCE($8, '')", + unclaimed = bundle_parser::AUTHORITY_UNCLAIMED + ); client .execute( - "UPDATE value_sets SET - name = $1, - title = $2, - status = $3, - compose_json = $4, - resource_json = $5, - updated_at = $6 - WHERE url = $7 AND COALESCE(version, '') = COALESCE($8, '')", + &vs_update, &[ &vs.name, &vs.title, @@ -917,6 +934,7 @@ async fn write_value_set( &now, &vs.url, &vs.version, + &vs.authority_rank, ], ) .await diff --git a/crates/hts/src/backends/postgres/schema.rs b/crates/hts/src/backends/postgres/schema.rs index 8635d8898..ebd75a3de 100644 --- a/crates/hts/src/backends/postgres/schema.rs +++ b/crates/hts/src/backends/postgres/schema.rs @@ -30,7 +30,19 @@ CREATE TABLE IF NOT EXISTS code_systems ( content TEXT NOT NULL DEFAULT 'complete', resource_json JSONB, created_at TEXT NOT NULL, - updated_at TEXT NOT NULL + updated_at TEXT NOT NULL, + -- Provenance precedence among rows sharing `url`; lower wins. 0 = the source + -- owns this canonical URL (or came from outside any package); 2 = a package + -- re-published someone else's canonical (e.g. hl7.fhir.r4.core shipping a + -- truncated copy of a terminology.hl7.org CodeSystem). + -- + -- Deliberately NULLABLE with no default: NULL means 'no source has claimed + -- this row yet', which is what every pre-existing row looks like right after + -- the column is added. Readers COALESCE it to 0, so an un-re-imported database + -- behaves exactly as before. A DEFAULT 0 would make 'asserted authoritative' + -- indistinguishable from 'never asserted', and a re-imported copy could then + -- never demote itself. + authority_rank INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS idx_code_systems_url_version ON code_systems(url, COALESCE(version, '')); @@ -127,7 +139,9 @@ CREATE TABLE IF NOT EXISTS value_sets ( compose_json TEXT, resource_json JSONB, created_at TEXT NOT NULL, - updated_at TEXT NOT NULL + updated_at TEXT NOT NULL, + -- See code_systems.authority_rank. + authority_rank INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS idx_value_sets_url_version ON value_sets(url, COALESCE(version, '')); @@ -235,6 +249,50 @@ CREATE TABLE IF NOT EXISTS bootstrap_imports ( -- Bring pre-existing ledgers up to the current shape (idempotent). ALTER TABLE bootstrap_imports ADD COLUMN IF NOT EXISTS mtime_unix BIGINT; ALTER TABLE bootstrap_imports ADD COLUMN IF NOT EXISTS languages TEXT NOT NULL DEFAULT ''; + +-- ── Provenance backfill ──────────────────────────────────────────────────────── +-- `authority_rank` records whether the package that shipped a row actually owns +-- the canonical URL it claims. It cannot be derived after the fact — nothing +-- already stored on the row says which package supplied it — so a database that +-- predates the column must re-import its packages to learn the truth. +-- +-- Adding the column is therefore the signal to invalidate the `.tgz` entries in +-- the bootstrap ledger, so the next startup re-imports those packages and stamps +-- the ranks. Without this the column would sit at its DEFAULT 0 on every existing +-- row (readers coalesce NULL to 0) and the fix would be a silent no-op on any +-- server with a persistent database: the ledger skips any file whose size and +-- mtime are unchanged, so packages would never re-import and never stamp a rank. +-- +-- Only `.tgz` rows are cleared. The multi-GB SNOMED/LOINC/RxNorm archives keep +-- their ledger entries and are not re-imported — they come from native importers +-- that are authoritative by construction and have nothing to learn from a reload. +-- Must run after bootstrap_imports exists, hence its position at the end. +DO $$ +DECLARE + added boolean := false; +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_attribute + WHERE attrelid = 'code_systems'::regclass + AND attname = 'authority_rank' AND NOT attisdropped + ) THEN + ALTER TABLE code_systems ADD COLUMN authority_rank INTEGER; + added := true; + END IF; + + IF NOT EXISTS ( + SELECT 1 FROM pg_attribute + WHERE attrelid = 'value_sets'::regclass + AND attname = 'authority_rank' AND NOT attisdropped + ) THEN + ALTER TABLE value_sets ADD COLUMN authority_rank INTEGER; + added := true; + END IF; + + IF added THEN + DELETE FROM bootstrap_imports WHERE path LIKE '%.tgz'; + END IF; +END $$; "; /// Apply the HTS PostgreSQL schema to the given client connection. diff --git a/crates/hts/src/backends/postgres/value_set.rs b/crates/hts/src/backends/postgres/value_set.rs index c27e87c38..01d1c2e26 100644 --- a/crates/hts/src/backends/postgres/value_set.rs +++ b/crates/hts/src/backends/postgres/value_set.rs @@ -1026,6 +1026,19 @@ impl ValueSetOperations for PostgresTerminologyBackend { }) } + async fn value_set_version_for_url( + &self, + _ctx: &TenantContext, + url: &str, + ) -> Result, HtsError> { + let client = self + .pool + .get() + .await + .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; + Ok(lookup_value_set_version(&client, url).await) + } + async fn validate_code( &self, _ctx: &TenantContext, @@ -1373,14 +1386,15 @@ impl ValueSetOperations for PostgresTerminologyBackend { // \`display\` parameter on mismatch responses (the concept itself // is still discoverable, only the version is unknown). let system_unwrapped = req.system.clone().unwrap(); + let sql = format!( + "SELECT c.display FROM concepts c \ + JOIN code_systems s ON s.id = c.system_id \ + WHERE s.url = $1 AND c.code = $2 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("s") + ); let display = client - .query_opt( - "SELECT c.display FROM concepts c - JOIN code_systems s ON s.id = c.system_id - WHERE s.url = $1 AND c.code = $2 - ORDER BY COALESCE(s.version, '') DESC LIMIT 1", - &[&system_unwrapped, &req.code], - ) + .query_opt(&sql, &[&system_unwrapped, &req.code]) .await .ok() .flatten() @@ -1818,14 +1832,15 @@ async fn resolve_value_set_versioned( version: Option<&str>, date: Option<&str>, ) -> Result<(String, Option), HtsError> { + let sql = format!( + "SELECT id, compose_json, version FROM value_sets \ + WHERE url = $1 \ + AND ($2::text IS NULL OR (resource_json->>'date') <= $2) \ + ORDER BY {}", + crate::backends::vs_precedence_order_by("value_sets") + ); let rows = client - .query( - "SELECT id, compose_json, version FROM value_sets - WHERE url = $1 - AND ($2::text IS NULL OR (resource_json->>'date') <= $2) - ORDER BY COALESCE(version, '') DESC", - &[&url, &date], - ) + .query(&sql, &[&url, &date]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -3241,13 +3256,16 @@ async fn resolve_compose_system_id( url: &str, version: Option<&str>, ) -> Result, HtsError> { + // Candidates come back best-first under the shared precedence rule, so the + // unpinned branch below can simply take the head. Previously ordered by + // `COALESCE(version,'') DESC` alone, which ignored the content/has-concepts + // tiers and the original-vs-copy tier. + let sql = format!( + "SELECT id, version FROM code_systems WHERE url = $1 ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); let rows = client - .query( - "SELECT id, version FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC", - &[&url], - ) + .query(&sql, &[&url]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -3352,12 +3370,12 @@ async fn build_hierarchical_expansion( let system_urls: HashSet = flat.iter().map(|c| c.system.clone()).collect(); let mut system_id_map: HashMap = HashMap::new(); for sys_url in &system_urls { + let sql = format!( + "SELECT id FROM code_systems WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let rows = client - .query( - "SELECT id FROM code_systems WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[sys_url], - ) + .query(&sql, &[sys_url]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; if let Some(row) = rows.into_iter().next() { @@ -3596,20 +3614,29 @@ fn extract_single_hierarchy_include(compose: &serde_json::Value) -> Option<(Stri Some((system.to_owned(), val.to_owned(), exclude_self)) } -/// Resolve the highest-versioned `code_systems.id` for a given canonical URL. -/// Multiple rows can share the same URL (stub + real import); we pick the -/// most recent textual COALESCE-DESC version, matching SQLite's resolver. +/// Resolve the winning `code_systems.id` for a given canonical URL. +/// +/// Multiple rows can share a URL (empty stub + real import, or an original and a +/// re-published copy). Ordering comes from [`crate::backends::cs_precedence_order_by`], +/// the same text SQLite uses. +/// +/// This previously ordered by `COALESCE(version,'') DESC` alone while claiming in +/// its doc comment to match SQLite — it did not. It lacked the content and +/// has-concepts tiers, so on Postgres an empty `content=not-present` stub whose +/// version string merely sorted higher (e.g. SNOMED's `"current"` stub, since +/// `'c' > '2'`) would beat a fully imported edition and `$expand` would return +/// nothing. That was a Postgres-only defect independent of the copy-vs-original +/// bug this ordering also fixes. async fn resolve_system_id_pg( client: &tokio_postgres::Client, cs_url: &str, ) -> Result, HtsError> { + let sql = format!( + "SELECT id FROM code_systems WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let row = client - .query_opt( - "SELECT id FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[&cs_url], - ) + .query_opt(&sql, &[&cs_url]) .await .map_err(|e| HtsError::StorageError(e.to_string()))?; Ok(row.map(|r| r.get::<_, String>(0))) @@ -3719,13 +3746,12 @@ async fn validate_fhir_vs( /// Highest stored ValueSet version for a URL, used to format `url|version` /// in IG-spec not-found messages. async fn lookup_value_set_version(client: &tokio_postgres::Client, url: &str) -> Option { + let sql = format!( + "SELECT version FROM value_sets WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::vs_precedence_order_by("value_sets") + ); client - .query_opt( - "SELECT version FROM value_sets \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[&url], - ) + .query_opt(&sql, &[&url]) .await .ok() .flatten() @@ -3756,11 +3782,13 @@ async fn pg_lookup_cs_display( .flatten(), None => client .query_opt( - "SELECT c.display FROM concepts c - JOIN code_systems s ON s.id = c.system_id - WHERE s.url = $1 AND c.code = $2 - ORDER BY COALESCE(s.version, '') DESC - LIMIT 1", + &format!( + "SELECT c.display FROM concepts c \ + JOIN code_systems s ON s.id = c.system_id \ + WHERE s.url = $1 AND c.code = $2 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("s") + ), &[&system_url, &code], ) .await @@ -3774,13 +3802,12 @@ pub(super) async fn cs_version_for_msg( client: &tokio_postgres::Client, system_url: &str, ) -> Option { + let sql = format!( + "SELECT version FROM code_systems WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); client - .query_opt( - "SELECT version FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[&system_url], - ) + .query_opt(&sql, &[&system_url]) .await .ok() .flatten() @@ -3794,13 +3821,12 @@ pub(super) async fn cs_content_for_url( client: &tokio_postgres::Client, system_url: &str, ) -> Option { + let sql = format!( + "SELECT content FROM code_systems WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); client - .query_opt( - "SELECT content FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[&system_url], - ) + .query_opt(&sql, &[&system_url]) .await .ok() .flatten() @@ -3813,16 +3839,12 @@ pub(super) async fn cs_is_case_insensitive( client: &tokio_postgres::Client, system_url: &str, ) -> bool { - let row = match client - .query_opt( - "SELECT (resource_json->>'caseSensitive') \ - FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[&system_url], - ) - .await - { + let sql = format!( + "SELECT (resource_json->>'caseSensitive') FROM code_systems \ + WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); + let row = match client.query_opt(&sql, &[&system_url]).await { Ok(r) => r, Err(_) => return false, }; @@ -3900,15 +3922,11 @@ pub(super) async fn cs_property_local_codes( canonical: &str, ) -> Vec { let mut codes: Vec = vec![canonical.to_string()]; - let row = match client - .query_opt( - "SELECT resource_json FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - &[&system_url], - ) - .await - { + let sql = format!( + "SELECT resource_json FROM code_systems WHERE url = $1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); + let row = match client.query_opt(&sql, &[&system_url]).await { Ok(Some(r)) => r, _ => return codes, }; @@ -4199,17 +4217,14 @@ pub(super) async fn detect_cs_version_mismatch( Option, Option, )> { - // Build (id, version) candidate list sorted desc so the first entry is the - // highest version — used for both resolution and picking the "actual" ver. - let rows = client - .query( - "SELECT id, version FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC", - &[&system_url], - ) - .await - .ok()?; + // Candidate (id, version) list in precedence order, so the first entry is the + // row the server would RESOLVE — not merely the highest version string. The + // reported version must name the row validation actually ran against. + let sql = format!( + "SELECT id, version FROM code_systems WHERE url = $1 ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); + let rows = client.query(&sql, &[&system_url]).await.ok()?; let candidates: Vec<(String, Option)> = rows .into_iter() .map(|r| (r.get::<_, String>(0), r.get::<_, Option>(1))) @@ -4477,15 +4492,11 @@ async fn detect_vs_pin_unknown( .and_then(|pin| pin)?; // only when the include has an explicit version // Build candidates for resolution - let rows = client - .query( - "SELECT id, version FROM code_systems \ - WHERE url = $1 \ - ORDER BY COALESCE(version, '') DESC", - &[&system_url], - ) - .await - .ok()?; + let sql = format!( + "SELECT id, version FROM code_systems WHERE url = $1 ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); + let rows = client.query(&sql, &[&system_url]).await.ok()?; let candidates: Vec<(String, Option)> = rows .into_iter() .map(|r| (r.get::<_, String>(0), r.get::<_, Option>(1))) diff --git a/crates/hts/src/backends/sqlite/code_system.rs b/crates/hts/src/backends/sqlite/code_system.rs index 2b2cbc8e4..fe19aaeb3 100644 --- a/crates/hts/src/backends/sqlite/code_system.rs +++ b/crates/hts/src/backends/sqlite/code_system.rs @@ -461,22 +461,24 @@ impl CodeSystemOperations for SqliteTerminologyBackend { // Match the IG `validation/cs-code-bad-code` text format // exactly: "Unknown code 'X' in the CodeSystem 'url' // version 'Y'" (with version when known). - let row: Option<(Option, Option)> = conn + // + // Report the row we ACTUALLY validated against. This used to + // re-query by URL with its own `ORDER BY version DESC`, which + // could name a different row than the one `resolve_code_system` + // chose a few lines above — so a miss could cite a version the + // code was never checked in (and, once provenance ranking is in + // play, routinely would). Both facts are already in hand: + // `resolved_cs_version` from the resolver, and `content` keyed + // on the resolved `system_id` rather than the URL. + let cs_version_str = resolved_cs_version.clone(); + let cs_content: Option = conn .query_row( - "SELECT version, content FROM code_systems \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![system], - |row| { - Ok(( - row.get::<_, Option>(0)?, - row.get::<_, Option>(1)?, - )) - }, + "SELECT content FROM code_systems WHERE id = ?1", + rusqlite::params![&system_id], + |row| row.get::<_, Option>(0), ) - .ok(); - let cs_version_str = row.as_ref().and_then(|(v, _)| v.clone()); - let cs_content = row.as_ref().and_then(|(_, c)| c.clone()); + .ok() + .flatten(); // Fragment CodeSystems carry only a subset of the real // corpus, so an unknown code is not necessarily invalid — // it may live in a different fragment of the same system. @@ -695,14 +697,12 @@ impl CodeSystemOperations for SqliteTerminologyBackend { let conn = pool .get() .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; + let sql = format!( + "SELECT version FROM code_systems WHERE url = ?1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let version: Option = conn - .query_row( - "SELECT version FROM code_systems \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![url_owned], - |row| row.get(0), - ) + .query_row(&sql, rusqlite::params![url_owned], |row| row.get(0)) .optional() .map_err(|e| HtsError::StorageError(e.to_string()))? .flatten(); @@ -781,19 +781,17 @@ impl CodeSystemOperations for SqliteTerminologyBackend { // by terminology packages), then highest version, then lowest id. // `json_extract` returns NULL when the column or path is absent, // which becomes Option::None in Rust. + let sql = format!( + "SELECT json_extract(resource_json, '$.language') \ + FROM code_systems \ + WHERE url = ?1 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let lang: Option = conn - .query_row( - "SELECT json_extract(resource_json, '$.language') \ - FROM code_systems \ - WHERE url = ?1 \ - ORDER BY (CASE WHEN EXISTS \ - (SELECT 1 FROM concepts c WHERE c.system_id = code_systems.id) \ - THEN 0 ELSE 1 END), \ - COALESCE(version, '') DESC, id \ - LIMIT 1", - rusqlite::params![url_owned], - |row| row.get::<_, Option>(0), - ) + .query_row(&sql, rusqlite::params![url_owned], |row| { + row.get::<_, Option>(0) + }) .optional() .map_err(|e| HtsError::StorageError(e.to_string()))? .flatten(); @@ -1301,21 +1299,21 @@ impl CodeSystemOperations for SqliteTerminologyBackend { .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; // Read content + resource_json + version in one query so we can // confirm the row really is a supplement before returning. + let sql = format!( + "SELECT content, version, json_extract(resource_json, '$.supplements') \ + FROM code_systems \ + WHERE url = ?1 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let row: Option<(String, Option, Option)> = conn - .query_row( - "SELECT content, version, json_extract(resource_json, '$.supplements') - FROM code_systems - WHERE url = ?1 - LIMIT 1", - rusqlite::params![supplement_url], - |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, Option>(1)?, - row.get::<_, Option>(2)?, - )) - }, - ) + .query_row(&sql, rusqlite::params![supplement_url], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, Option>(1)?, + row.get::<_, Option>(2)?, + )) + }) .optional() .map_err(|e| HtsError::StorageError(e.to_string()))?; let Some((content, version, target)) = row else { @@ -1534,14 +1532,16 @@ impl CodeSystemOperations for SqliteTerminologyBackend { .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; // Read the base CodeSystem's resource_json (highest version), then // walk concept[] picking entries whose code is in the requested set. + let sql = format!( + "SELECT resource_json FROM code_systems \ + WHERE url = ?1 AND content != 'supplement' \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let resource_json: Option = conn - .query_row( - "SELECT resource_json FROM code_systems - WHERE url = ?1 AND content != 'supplement' - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![system_url], - |row| row.get::<_, Option>(0), - ) + .query_row(&sql, rusqlite::params![system_url], |row| { + row.get::<_, Option>(0) + }) .ok() .flatten(); let mut out: std::collections::HashMap = @@ -1685,14 +1685,14 @@ fn cs_property_local_codes( canonical: &str, ) -> Vec { let mut codes: Vec = vec![canonical.to_string()]; + let sql = format!( + "SELECT resource_json FROM code_systems WHERE url = ?1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let resource_json: Option = conn - .query_row( - "SELECT resource_json FROM code_systems \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![system_url], - |row| row.get::<_, Option>(0), - ) + .query_row(&sql, rusqlite::params![system_url], |row| { + row.get::<_, Option>(0) + }) .ok() .flatten(); let suffix = format!("#{canonical}"); @@ -1825,35 +1825,26 @@ fn resolve_code_system_uncached( Ok(chosen) } -/// Fetch every (id, name, version) row for `url`. +/// Fetch every (id, name, version) row for `url`, best candidate first. /// -/// Sort order (see `docs/ig-publisher-compatibility.md`): -/// complete/supplement → rows with concepts → highest version → id. -/// Ensures full LOINC/SNOMED imports beat empty stubs when IG Publisher validates -/// with unpinned or `version=current` requests. +/// Ordering comes from [`crate::backends::cs_precedence_order_by`] — the single shared +/// definition of same-URL precedence, embedded verbatim so this cannot drift +/// from the Postgres resolver. See `docs/ig-publisher-compatibility.md`. fn fetch_versions( conn: &rusqlite::Connection, url: &str, date: Option<&str>, ) -> Result)>, HtsError> { + let sql = format!( + "SELECT id, COALESCE(name, url), version \ + FROM code_systems \ + WHERE url = ?1 \ + AND (?2 IS NULL OR json_extract(resource_json, '$.date') <= ?2) \ + ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); let mut stmt = conn - .prepare( - "SELECT id, COALESCE(name, url), version \ - FROM code_systems \ - WHERE url = ?1 \ - AND (?2 IS NULL OR json_extract(resource_json, '$.date') <= ?2) \ - ORDER BY (CASE COALESCE(content, 'complete') \ - WHEN 'complete' THEN 0 \ - WHEN 'supplement' THEN 0 \ - WHEN 'fragment' THEN 1 \ - WHEN 'example' THEN 1 \ - WHEN 'not-present' THEN 2 \ - ELSE 1 END), \ - (CASE WHEN EXISTS \ - (SELECT 1 FROM concepts c WHERE c.system_id = code_systems.id) \ - THEN 0 ELSE 1 END), \ - COALESCE(version, '') DESC, id", - ) + .prepare(&sql) .map_err(|e| HtsError::StorageError(e.to_string()))?; let rows = stmt @@ -2094,15 +2085,20 @@ fn fetch_designations_cross_version( exclude_system_id: &str, lang: &str, ) -> Result, HtsError> { + // Ranked by the same precedence as the primary resolver, so the "next best + // other row" we fall back to is the one the server would have picked had the + // resolved row not existed — not merely the highest version string. + let sql = format!( + "SELECT cd.language, cd.use_system, cd.use_code, cd.value, COALESCE(s.version, '') \ + FROM concept_designations cd \ + JOIN concepts c ON c.id = cd.concept_id \ + JOIN code_systems s ON s.id = c.system_id \ + WHERE s.url = ?1 AND c.code = ?2 AND s.id <> ?3 \ + ORDER BY {}, cd.id", + crate::backends::cs_precedence_order_by("s") + ); let mut stmt = conn - .prepare_cached( - "SELECT cd.language, cd.use_system, cd.use_code, cd.value, COALESCE(s.version, '') \ - FROM concept_designations cd \ - JOIN concepts c ON c.id = cd.concept_id \ - JOIN code_systems s ON s.id = c.system_id \ - WHERE s.url = ?1 AND c.code = ?2 AND s.id <> ?3 \ - ORDER BY COALESCE(s.version, '') DESC, cd.id", - ) + .prepare_cached(&sql) .map_err(|e| HtsError::StorageError(e.to_string()))?; let rows: Vec<(DesignationValue, String)> = stmt diff --git a/crates/hts/src/backends/sqlite/concept_map.rs b/crates/hts/src/backends/sqlite/concept_map.rs index 1174fa409..831a681a5 100644 --- a/crates/hts/src/backends/sqlite/concept_map.rs +++ b/crates/hts/src/backends/sqlite/concept_map.rs @@ -395,13 +395,12 @@ fn closure_sync(conn: &Connection, req: &ClosureRequest) -> Result = conn - .query_row( - "SELECT id FROM code_systems WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![system_url], - |row| row.get(0), - ) + .query_row(&sql, rusqlite::params![system_url], |row| row.get(0)) .optional() .map_err(|e| HtsError::StorageError(format!("DB error: {e}")))?; diff --git a/crates/hts/src/backends/sqlite/mod.rs b/crates/hts/src/backends/sqlite/mod.rs index 398d310d4..5352f4fc9 100644 --- a/crates/hts/src/backends/sqlite/mod.rs +++ b/crates/hts/src/backends/sqlite/mod.rs @@ -350,6 +350,12 @@ impl SqliteTerminologyBackend { "Failed to apply bootstrap_imports column migration: {e}" )) })?; + // MUST run after the drop_url_unique rebuilds above: those recreate + // code_systems / value_sets from a fixed column list and would drop + // `authority_rank` if it had been added first. + schema::migrate_authority_rank(&conn).map_err(|e| { + HtsError::StorageError(format!("Failed to apply authority_rank migration: {e}")) + })?; // FTS prebuild + index pre-warm. Skipped when `prebuild_fts` is // false (server startup path): bootstrap re-imports run first and @@ -517,6 +523,10 @@ impl SqliteTerminologyBackend { schema::migrate_value_sets_drop_url_unique(&mut conn).map_err(|e| { HtsError::StorageError(format!("Failed to drop legacy value_sets.url UNIQUE: {e}")) })?; + // See the on-disk path: must follow the drop_url_unique rebuilds. + schema::migrate_authority_rank(&conn).map_err(|e| { + HtsError::StorageError(format!("Failed to apply authority_rank migration: {e}")) + })?; } Ok(Self { @@ -595,15 +605,14 @@ impl TerminologyMetadata for SqliteTerminologyBackend { ) { return Some(url); } - conn.query_row( + let sql = format!( "SELECT url FROM code_systems \ WHERE json_extract(resource_json, '$.id') = ?1 \ - ORDER BY COALESCE(version, '') DESC \ - LIMIT 1", - rusqlite::params![id], - |row| row.get::<_, String>(0), - ) - .ok() + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); + conn.query_row(&sql, rusqlite::params![id], |row| row.get::<_, String>(0)) + .ok() } "ValueSet" => { if let Ok(url) = conn.query_row( @@ -617,15 +626,14 @@ impl TerminologyMetadata for SqliteTerminologyBackend { // so when the URL-path id is the bare FHIR id, fall back to a // resource_json scan and pick the latest version (matches how // CodeSystem reads handle the same case). - conn.query_row( + let sql = format!( "SELECT url FROM value_sets \ WHERE json_extract(resource_json, '$.id') = ?1 \ - ORDER BY COALESCE(version, '') DESC \ - LIMIT 1", - rusqlite::params![id], - |row| row.get::<_, String>(0), - ) - .ok() + ORDER BY {} LIMIT 1", + crate::backends::vs_precedence_order_by("value_sets") + ); + conn.query_row(&sql, rusqlite::params![id], |row| row.get::<_, String>(0)) + .ok() } "ConceptMap" => conn .query_row( diff --git a/crates/hts/src/backends/sqlite/schema.rs b/crates/hts/src/backends/sqlite/schema.rs index 785f17fad..aa6c734cd 100644 --- a/crates/hts/src/backends/sqlite/schema.rs +++ b/crates/hts/src/backends/sqlite/schema.rs @@ -38,7 +38,19 @@ CREATE TABLE IF NOT EXISTS code_systems ( content TEXT NOT NULL DEFAULT 'complete', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, - resource_json TEXT + resource_json TEXT, + -- Provenance precedence among rows sharing `url`; lower wins. 0 = the + -- source owns this canonical URL (or came from outside any package); + -- 2 = a package re-published someone else's canonical (e.g. hl7.fhir.r4.core + -- shipping a truncated copy of a terminology.hl7.org CodeSystem). + -- + -- Deliberately NULLABLE with no default: NULL means 'no source has claimed + -- this row yet', which is what every pre-existing row looks like immediately + -- after the column is added. Readers COALESCE it to 0, so an un-re-imported + -- database behaves exactly as it did before. Had this defaulted to 0 the + -- upsert below could not distinguish 'asserted authoritative' from 'never + -- asserted', and a re-imported copy could not demote itself. + authority_rank INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS idx_code_systems_url_version ON code_systems(url, COALESCE(version, '')); @@ -120,7 +132,9 @@ CREATE TABLE IF NOT EXISTS value_sets ( compose_json TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, - resource_json TEXT + resource_json TEXT, + -- See code_systems.authority_rank. + authority_rank INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS idx_value_sets_url_version ON value_sets(url, COALESCE(version, '')); @@ -497,6 +511,54 @@ pub fn migrate_search_columns(conn: &rusqlite::Connection) -> rusqlite::Result<( Ok(()) } +/// Add the `authority_rank` provenance column to `code_systems` / `value_sets`, +/// and force a re-import of packaged terminology when the column is newly added. +/// +/// `authority_rank` records whether the package that shipped a row actually owns +/// the canonical URL it claims (see +/// [`crate::import::bundle_parser::authority_rank_for`]). It cannot be derived +/// after the fact — nothing already in the row says which package supplied it — +/// so an existing database has to re-import its packages to learn the truth. +/// +/// The `ALTER TABLE` returning `Ok` is precisely the signal that this database +/// predates provenance. In that case we drop the `.tgz` entries from the +/// `bootstrap_imports` ledger so the next startup re-imports those packages and +/// stamps the ranks. Without this the column would stay NULL on every existing +/// row, readers would coalesce it to 0, and the fix would be a silent no-op on any +/// server with a persistent database — the ledger skips files whose size and mtime +/// are unchanged, so the packages would never re-import and never stamp provenance. +/// +/// Only `.tgz` rows are cleared. The multi-GB SNOMED/LOINC/RxNorm archives keep +/// their ledger entries and are not re-imported — they arrive through native +/// importers that are authoritative by construction, so they have nothing to +/// learn from a re-import. +pub fn migrate_authority_rank(conn: &rusqlite::Connection) -> rusqlite::Result<()> { + let mut column_added = false; + for sql in &[ + "ALTER TABLE code_systems ADD COLUMN authority_rank INTEGER", + "ALTER TABLE value_sets ADD COLUMN authority_rank INTEGER", + ] { + match conn.execute_batch(sql) { + Ok(_) => column_added = true, + Err(e) if e.to_string().contains("duplicate column name") => {} + Err(e) => return Err(e), + } + } + + if column_added { + let cleared = conn.execute("DELETE FROM bootstrap_imports WHERE path LIKE '%.tgz'", [])?; + if cleared > 0 { + tracing::info!( + cleared_ledger_entries = cleared, + "Added authority_rank; cleared .tgz bootstrap ledger entries so packages \ + re-import and record which canonical URLs they own" + ); + } + } + + Ok(()) +} + /// Add the `mtime_unix` and `languages` columns to an existing /// `bootstrap_imports` ledger. /// @@ -710,4 +772,150 @@ mod tests { .unwrap(); assert_eq!(remaining, 0, "cascade delete should remove child concepts"); } + // ── authority_rank migration (issue #200) ───────────────────────────────── + + /// Build a database in the *pre-provenance* shape: the resource tables have + /// no `authority_rank` column, and the bootstrap ledger already records the + /// packages that were imported. This is what a deployed server looks like + /// before the upgrade — and it is the only shape in which the interesting + /// branch of `migrate_authority_rank` runs at all, since `SCHEMA` creates + /// fresh tables with the column already present. + fn legacy_db() -> rusqlite::Connection { + let conn = rusqlite::Connection::open_in_memory().unwrap(); + conn.execute_batch( + "CREATE TABLE code_systems ( + id TEXT PRIMARY KEY, url TEXT NOT NULL, version TEXT, + status TEXT NOT NULL DEFAULT 'active', + content TEXT NOT NULL DEFAULT 'complete', + created_at TEXT NOT NULL, updated_at TEXT NOT NULL + ); + CREATE TABLE value_sets ( + id TEXT PRIMARY KEY, url TEXT NOT NULL, version TEXT, + status TEXT NOT NULL DEFAULT 'active', + created_at TEXT NOT NULL, updated_at TEXT NOT NULL + ); + CREATE TABLE bootstrap_imports ( + path TEXT PRIMARY KEY, content_hash TEXT NOT NULL, + size_bytes INTEGER NOT NULL, mtime_unix INTEGER, + languages TEXT NOT NULL DEFAULT '', + imported_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + INSERT INTO code_systems (id, url, version, created_at, updated_at) + VALUES ('cs1', 'http://example.org/cs', '1.0.0', 'x', 'x'); + + -- Packages carry provenance and must be re-imported; the bulk + -- terminology archives do not and must be left alone. + INSERT INTO bootstrap_imports (path, content_hash, size_bytes) + VALUES ('/app/terminology-data/hl7.terminology-7.1.0.tgz', 'h1', 1), + ('/app/terminology-data/hl7.fhir.r4.core-4.0.1.tgz', 'h2', 2), + ('/app/terminology-data/icd10cm-table-and-index-2026.zip', 'h3', 3), + ('/app/terminology-data/ucum-essence-v2.2.xml', 'h4', 4);", + ) + .unwrap(); + conn + } + + fn ledger_paths(conn: &rusqlite::Connection) -> Vec { + let mut stmt = conn + .prepare("SELECT path FROM bootstrap_imports ORDER BY path") + .unwrap(); + stmt.query_map([], |r| r.get::<_, String>(0)) + .unwrap() + .map(Result::unwrap) + .collect() + } + + fn has_authority_rank(conn: &rusqlite::Connection, table: &str) -> bool { + conn.prepare(&format!("SELECT authority_rank FROM {table} LIMIT 1")) + .is_ok() + } + + #[test] + fn authority_rank_migration_adds_columns_and_forces_package_reimport() { + let conn = legacy_db(); + assert!(!has_authority_rank(&conn, "code_systems")); + assert!(!has_authority_rank(&conn, "value_sets")); + + migrate_authority_rank(&conn).expect("migration should succeed"); + + assert!(has_authority_rank(&conn, "code_systems")); + assert!(has_authority_rank(&conn, "value_sets")); + + // The .tgz ledger entries are dropped so the next startup re-imports + // those packages and stamps provenance. Without this the column would + // stay NULL on every existing row and the fix would be a silent no-op: + // the ledger skips any file whose size and mtime are unchanged. + assert_eq!( + ledger_paths(&conn), + vec![ + "/app/terminology-data/icd10cm-table-and-index-2026.zip".to_string(), + "/app/terminology-data/ucum-essence-v2.2.xml".to_string(), + ], + "only .tgz packages may be cleared; the bulk archives carry no \ + package provenance and re-importing them would cost hours" + ); + } + + #[test] + fn authority_rank_migration_leaves_existing_rows_unclaimed() { + let conn = legacy_db(); + migrate_authority_rank(&conn).unwrap(); + + // A pre-existing row is NULL, not 0: "no source has claimed this row". + // Readers COALESCE it to 0, so behaviour is unchanged until the packages + // re-import — and because it is NULL rather than 0, the upsert's + // MIN(COALESCE(authority_rank, 9), ?) can still demote a copy. + let rank: Option = conn + .query_row("SELECT authority_rank FROM code_systems", [], |r| r.get(0)) + .unwrap(); + assert_eq!( + rank, None, + "existing rows must be unclaimed, not authoritative" + ); + } + + #[test] + fn authority_rank_migration_is_idempotent_and_spares_the_ledger() { + let conn = legacy_db(); + migrate_authority_rank(&conn).unwrap(); + + // Re-import happened; the ledger is repopulated as packages load. + conn.execute( + "INSERT INTO bootstrap_imports (path, content_hash, size_bytes) + VALUES ('/app/terminology-data/hl7.terminology-7.1.0.tgz', 'h1', 1)", + [], + ) + .unwrap(); + + // A second run must NOT clear the ledger again — the ALTER now fails with + // "duplicate column name", so `column_added` stays false. If this branch + // regressed, every restart would re-import every package. + migrate_authority_rank(&conn).expect("migration must be idempotent"); + assert!( + ledger_paths(&conn) + .iter() + .any(|p| p.ends_with("hl7.terminology-7.1.0.tgz")), + "an already-migrated database must not re-clear the ledger on restart" + ); + } + + #[test] + fn authority_rank_migration_on_fresh_schema_is_a_noop() { + // SCHEMA already declares the column, so the migration must be a no-op + // and must not touch the ledger of a brand-new database. + let conn = rusqlite::Connection::open_in_memory().unwrap(); + apply(&conn).unwrap(); + conn.execute( + "INSERT INTO bootstrap_imports (path, content_hash, size_bytes) + VALUES ('/app/terminology-data/hl7.terminology-7.1.0.tgz', 'h1', 1)", + [], + ) + .unwrap(); + + migrate_authority_rank(&conn).unwrap(); + + assert_eq!(ledger_paths(&conn).len(), 1); + assert!(has_authority_rank(&conn, "code_systems")); + } } diff --git a/crates/hts/src/backends/sqlite/value_set.rs b/crates/hts/src/backends/sqlite/value_set.rs index 2cef1282a..8ee9b1468 100644 --- a/crates/hts/src/backends/sqlite/value_set.rs +++ b/crates/hts/src/backends/sqlite/value_set.rs @@ -123,29 +123,24 @@ fn resolve_system_id_with_version_cached( // even when a fully-loaded row exists alongside. // 2. Prefer rows with at least one concept (EXISTS subquery; constant // time, short-circuited on first match). - // 3. Highest version DESC. - // 4. id ASC. + // 3. Prefer the original over a re-published copy (authority_rank). + // 4. Highest version DESC. + // 5. id ASC. // Tier 1 alone fixes the `r4.core stub + RF2 import` case observed in the // benchmark. Tier 2 is kept as a safety net for IGs that omit `content`. + // The full rule lives in `backends::cs_precedence_order_by` and is shared + // verbatim with the Postgres resolver. // The cache memoises the resolved `(id, version)` so the SQL runs once // per URL per process. + let sql = format!( + "SELECT id, version FROM code_systems WHERE url = ?1 \ + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let row: Option<(String, Option)> = conn - .query_row( - "SELECT cs.id, cs.version FROM code_systems cs WHERE cs.url = ?1 \ - ORDER BY (CASE COALESCE(cs.content, 'complete') \ - WHEN 'complete' THEN 0 \ - WHEN 'supplement' THEN 0 \ - WHEN 'fragment' THEN 1 \ - WHEN 'example' THEN 1 \ - WHEN 'not-present' THEN 2 \ - ELSE 1 END), \ - (CASE WHEN EXISTS \ - (SELECT 1 FROM concepts c WHERE c.system_id = cs.id) \ - THEN 0 ELSE 1 END), \ - COALESCE(cs.version, '') DESC, cs.id LIMIT 1", - [url], - |r| Ok((r.get::<_, String>(0)?, r.get::<_, Option>(1)?)), - ) + .query_row(&sql, [url], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, Option>(1)?)) + }) .optional() .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -1262,6 +1257,24 @@ impl ValueSetOperations for SqliteTerminologyBackend { /// Triggers expansion if needed, then checks set membership. /// Returns `result = false` (not an error) when the value set or code is /// not found. + async fn value_set_version_for_url( + &self, + _ctx: &TenantContext, + url: &str, + ) -> Result, HtsError> { + let backend = self.clone(); + let url = url.to_owned(); + tokio::task::spawn_blocking(move || { + let conn = backend + .pool() + .get() + .map_err(|e| HtsError::StorageError(format!("Pool error: {e}")))?; + Ok(lookup_value_set_version(&backend, &conn, &url)) + }) + .await + .map_err(|e| HtsError::StorageError(format!("Task panicked: {e}")))? + } + async fn validate_code( &self, _ctx: &TenantContext, @@ -2228,13 +2241,15 @@ fn resolve_value_set_versioned( ) -> Result<(String, Option), HtsError> { // Fetch every (id, compose, version) candidate ordered with the highest // version first so the version=None path picks the latest. + let sql = format!( + "SELECT id, compose_json, version FROM value_sets \ + WHERE url = ?1 \ + AND (?2 IS NULL OR json_extract(resource_json, '$.date') <= ?2) \ + ORDER BY {}", + crate::backends::vs_precedence_order_by("value_sets") + ); let mut stmt = conn - .prepare( - "SELECT id, compose_json, version FROM value_sets \ - WHERE url = ?1 \ - AND (?2 IS NULL OR json_extract(resource_json, '$.date') <= ?2) \ - ORDER BY COALESCE(version, '') DESC", - ) + .prepare(&sql) .map_err(|e| HtsError::StorageError(e.to_string()))?; let rows: Vec<(String, Option, Option)> = stmt .query_map(rusqlite::params![url, date], |row| { @@ -5981,22 +5996,12 @@ fn resolve_compose_system_id( // agrees with the unpinned hot-path on which row to prefer when multiple // candidates share the same canonical URL (e.g. r4.core stub plus // RF2 import for SNOMED). + let sql = format!( + "SELECT id, version FROM code_systems WHERE url = ?1 ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); let mut stmt = conn - .prepare( - "SELECT id, version FROM code_systems \ - WHERE url = ?1 \ - ORDER BY (CASE COALESCE(content, 'complete') \ - WHEN 'complete' THEN 0 \ - WHEN 'supplement' THEN 0 \ - WHEN 'fragment' THEN 1 \ - WHEN 'example' THEN 1 \ - WHEN 'not-present' THEN 2 \ - ELSE 1 END), \ - (CASE WHEN EXISTS \ - (SELECT 1 FROM concepts c WHERE c.system_id = code_systems.id) \ - THEN 0 ELSE 1 END), \ - COALESCE(version, '') DESC", - ) + .prepare(&sql) .map_err(|e| HtsError::StorageError(e.to_string()))?; let rows: Vec<(String, Option)> = stmt @@ -6092,13 +6097,12 @@ fn build_hierarchical_expansion( let system_urls: HashSet = flat.iter().map(|c| c.system.clone()).collect(); let mut system_id_map: HashMap = HashMap::new(); for sys_url in &system_urls { + let sql = format!( + "SELECT id FROM code_systems WHERE url = ?1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); if let Some(id) = conn - .query_row( - "SELECT id FROM code_systems WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - [sys_url], - |row| row.get::<_, String>(0), - ) + .query_row(&sql, [sys_url], |row| row.get::<_, String>(0)) .optional() .map_err(|e| HtsError::StorageError(e.to_string()))? { @@ -6379,14 +6383,14 @@ fn lookup_value_set_version( // Pick the highest stored version for this URL — matches the // resolve_value_set_versioned default-when-no-pin behaviour, so $expand // and $validate-code echoes converge on the same row. + let sql = format!( + "SELECT version FROM value_sets WHERE url = ?1 ORDER BY {} LIMIT 1", + crate::backends::vs_precedence_order_by("value_sets") + ); let v: Option = conn - .query_row( - "SELECT version FROM value_sets \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![url], - |row| row.get::<_, Option>(0), - ) + .query_row(&sql, rusqlite::params![url], |row| { + row.get::<_, Option>(0) + }) .ok() .flatten(); if let Ok(mut w) = cache.write() { @@ -6470,14 +6474,14 @@ fn cs_version_for_msg( return v.clone(); } } + let sql = format!( + "SELECT version FROM code_systems WHERE url = ?1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let v: Option = conn - .query_row( - "SELECT version FROM code_systems \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![system_url], - |row| row.get::<_, Option>(0), - ) + .query_row(&sql, rusqlite::params![system_url], |row| { + row.get::<_, Option>(0) + }) .ok() .flatten(); if let Ok(mut w) = cache.write() { @@ -6502,14 +6506,14 @@ fn cs_content_for_url( return v.clone(); } } + let sql = format!( + "SELECT content FROM code_systems WHERE url = ?1 ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); let v: Option = conn - .query_row( - "SELECT content FROM code_systems \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![system_url], - |row| row.get::<_, Option>(0), - ) + .query_row(&sql, rusqlite::params![system_url], |row| { + row.get::<_, Option>(0) + }) .ok() .flatten(); if let Ok(mut w) = cache.write() { @@ -6526,14 +6530,16 @@ fn cs_content_for_url( /// `CODE_CASE_DIFFERENCE` informational issue when the caller's code differs /// from the canonical form by case. fn cs_is_case_insensitive(conn: &Connection, system_url: &str) -> bool { - conn.query_row( + let sql = format!( "SELECT json_extract(resource_json, '$.caseSensitive') \ FROM code_systems \ WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC LIMIT 1", - rusqlite::params![system_url], - |row| row.get::<_, Option>(0), - ) + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); + conn.query_row(&sql, rusqlite::params![system_url], |row| { + row.get::<_, Option>(0) + }) .ok() .flatten() .map(|v| v == 0) @@ -6773,15 +6779,14 @@ fn detect_cs_version_mismatch( Option, Option, )> { - // Build (id, version) candidate list sorted desc so the first entry is the - // highest version — used for both resolution and picking the "actual" ver. - let mut stmt = conn - .prepare_cached( - "SELECT id, version FROM code_systems \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC", - ) - .ok()?; + // Candidate (id, version) list in precedence order, so the first entry is the + // row the server would RESOLVE — not merely the highest version string. The + // reported version must name the row validation actually ran against. + let sql = format!( + "SELECT id, version FROM code_systems WHERE url = ?1 ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); + let mut stmt = conn.prepare_cached(&sql).ok()?; let candidates: Vec<(String, Option)> = stmt .query_map(rusqlite::params![system_url], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, Option>(1)?)) @@ -7052,13 +7057,11 @@ fn detect_vs_pin_unknown( .and_then(|pin| pin)?; // only when the include has an explicit version // Build candidates for resolution - let mut stmt = conn - .prepare_cached( - "SELECT id, version FROM code_systems \ - WHERE url = ?1 \ - ORDER BY COALESCE(version, '') DESC", - ) - .ok()?; + let sql = format!( + "SELECT id, version FROM code_systems WHERE url = ?1 ORDER BY {}", + crate::backends::cs_precedence_order_by("code_systems") + ); + let mut stmt = conn.prepare_cached(&sql).ok()?; let candidates: Vec<(String, Option)> = stmt .query_map(rusqlite::params![system_url], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, Option>(1)?)) diff --git a/crates/hts/src/import/bundle_builder.rs b/crates/hts/src/import/bundle_builder.rs index 2ee2a0858..16c5fd1af 100644 --- a/crates/hts/src/import/bundle_builder.rs +++ b/crates/hts/src/import/bundle_builder.rs @@ -127,6 +127,10 @@ pub(crate) fn build_parsed_code_system( content: meta.content.to_owned(), concepts: parsed_concepts, resource_json: build_code_system_resource(meta, &[]), + // Native importers (SNOMED RF2, LOINC, RxNorm, ICD-10-CM, …) load a + // terminology from its publisher's own distribution, so they are + // authoritative for the URL they claim. + authority_rank: crate::import::bundle_parser::AUTHORITY_OWNER, }], ..Default::default() } diff --git a/crates/hts/src/import/bundle_parser.rs b/crates/hts/src/import/bundle_parser.rs index 6c156dd5b..3ceee48a6 100644 --- a/crates/hts/src/import/bundle_parser.rs +++ b/crates/hts/src/import/bundle_parser.rs @@ -46,6 +46,82 @@ pub struct ParsedBundle { pub fresh_load: bool, } +/// Authority rank for a resource we have POSITIVE evidence is authoritative: +/// the package that shipped it declares a `canonical` base that covers the +/// resource's URL, or it came from a native importer loading a publisher's own +/// distribution (SNOMED RF2, LOINC, RxNorm, …). Lower wins. +pub const AUTHORITY_OWNER: i32 = 0; + +/// Authority rank for a resource whose provenance we cannot vouch for. +/// +/// This is the DEFAULT, and the distinction from [`AUTHORITY_OWNER`] is the +/// whole point: we only ever *promote* on positive evidence, never *demote* on +/// the absence of it. A row is `UNKNOWN` when it arrived outside any package +/// (REST write, `$import-bundle`, CLI bundle) or when the package that shipped +/// it does not declare a canonical base covering the URL. +/// +/// That second case deliberately conflates two situations we cannot tell apart: +/// +/// * `hl7.fhir.r4.core` re-publishing `http://terminology.hl7.org/...` — a +/// genuine stale copy, which must lose to THO's original. This is issue #200. +/// * `us.nlm.vsac` shipping `http://cts.nlm.nih.gov/...` while declaring the +/// fhir.org *registry* base `http://fhir.org/packages/us.nlm.vsac` — a genuine +/// ORIGINAL whose manifest simply doesn't describe its content's authority. +/// 14,850 of its ValueSets look identical to a "copy" under this rule. +/// +/// Conflating them is safe precisely because rank is only ever a TIEBREAK among +/// rows sharing one canonical URL. Being `UNKNOWN` costs a row nothing unless a +/// demonstrable owner of the same URL exists — VSAC's rows have no rival, so they +/// still win their URLs outright. An earlier draft ranked these as a definite +/// "foreign copy" (2) and treated non-package writes as owners (0); that was +/// wrong twice over. It asserted a fact about VSAC we had not established, and — +/// because `vs_precedence_order_by` has no content/has-concepts tier to absorb +/// the error — it let any REST upload outrank the real VSAC definition outright. +pub const AUTHORITY_UNKNOWN: i32 = 1; + +/// Sentinel meaning "no source has claimed this row yet", used by the import +/// upsert's `MIN(COALESCE(authority_rank, …), ?)` / `LEAST(...)` so that a row +/// which predates the column (NULL) is overwritten by the first source to claim +/// it, rather than reading as an authoritative 0 and pinning a stale copy. +/// +/// It must stay strictly greater than every real rank. Note the deliberate +/// asymmetry: on WRITE, NULL is the *weakest* value (this sentinel); on READ it +/// coalesces to 0, the *strongest*, so that a database which has not yet +/// re-imported behaves exactly as it did before the column existed. +pub const AUTHORITY_UNCLAIMED: i32 = 9; + +/// Classify a resource against the canonical base declared by the package that +/// shipped it (`package.json` → `canonical`). +/// +/// Returns [`AUTHORITY_OWNER`] only on positive evidence — the package declares a +/// canonical base and the resource's URL sits under it. Everything else, including +/// a resource that arrived with no package context at all, is [`AUTHORITY_UNKNOWN`]. +/// +/// The asymmetry is deliberate and load-bearing. Defaulting an unattributed write +/// to OWNER would mean an operator (or, on a server without auth, anyone) replaying +/// `hl7.fhir.r4.core`'s truncated copy through `$import-bundle` would stamp it rank +/// 0 — and since the upsert's `MIN` only ever lowers a rank, that copy would +/// outrank THO's original permanently and reinstate issue #200 with no way back. +pub fn authority_rank_for(url: &str, package_canonical: Option<&str>) -> i32 { + match package_canonical { + Some(base) if !base.is_empty() && url_is_under(url, base) => AUTHORITY_OWNER, + _ => AUTHORITY_UNKNOWN, + } +} + +/// `true` when `url` sits under the canonical `base`, respecting path +/// boundaries: `http://hl7.org/fhir` owns `http://hl7.org/fhir/CodeSystem/x` +/// but NOT `http://hl7.org/fhirpath/x`. A bare `starts_with` would claim the +/// latter and wrongly mark a genuine original as unattributed. +fn url_is_under(url: &str, base: &str) -> bool { + let base = base.trim_end_matches('/'); + match url.strip_prefix(base) { + Some("") => true, + Some(rest) => rest.starts_with('/'), + None => false, + } +} + /// A single FHIR CodeSystem resource extracted from a Bundle entry. #[derive(Debug)] pub struct ParsedCodeSystem { @@ -61,6 +137,9 @@ pub struct ParsedCodeSystem { pub concepts: Vec, /// The full original JSON resource (stored in `resource_json` column). pub resource_json: Value, + /// Provenance precedence among rows sharing this canonical URL. + /// See [`authority_rank_for`]. + pub authority_rank: i32, } /// One concept row, already flattened from the potentially nested FHIR tree. @@ -112,6 +191,9 @@ pub struct ParsedValueSet { /// lazy expansion. `None` when the ValueSet has no `compose`. pub compose_json: Option, pub resource_json: Value, + /// Provenance precedence among rows sharing this canonical URL. + /// See [`authority_rank_for`]. + pub authority_rank: i32, } /// A single FHIR ConceptMap resource extracted from a Bundle entry. @@ -154,6 +236,23 @@ pub struct ParsedMapElement { /// Returns [`HtsError::InvalidRequest`] when the bytes are not valid JSON or /// the root resource is not a Bundle. pub fn parse_bundle(data: &[u8]) -> Result { + // No package context: REST writes, `$import-bundle`, and CLI bundles are + // treated as authoritative for whatever canonical they declare. + parse_bundle_from_package(data, None) +} + +/// Parse a Bundle that arrived inside a FHIR NPM package, whose `package.json` +/// declared `package_canonical` as the canonical base it owns. +/// +/// Provenance is an explicit parameter, NOT a field on the payload. An earlier +/// draft smuggled it through a non-FHIR `_htsSourceCanonical` key on the Bundle +/// wrapper, which meant an untrusted `POST /import` body could set the value +/// that decides which definition of a canonical URL is authoritative. Trust +/// input must come from the caller, never from the bytes being parsed. +pub fn parse_bundle_from_package( + data: &[u8], + package_canonical: Option<&str>, +) -> Result { let root: Value = serde_json::from_slice(data) .map_err(|e| HtsError::InvalidRequest(format!("Invalid JSON: {e}")))?; @@ -175,13 +274,19 @@ pub fn parse_bundle(data: &[u8]) -> Result { let rid = resource["id"].as_str().unwrap_or(""); match resource["resourceType"].as_str() { Some("CodeSystem") => match parse_code_system(resource) { - Some(cs) => bundle.code_systems.push(cs), + Some(mut cs) => { + cs.authority_rank = authority_rank_for(&cs.url, package_canonical); + bundle.code_systems.push(cs); + } None => bundle.parse_errors.push(format!( "CodeSystem '{rid}' skipped: missing required field 'url'" )), }, Some("ValueSet") => match parse_value_set(resource) { - Some(vs) => bundle.value_sets.push(vs), + Some(mut vs) => { + vs.authority_rank = authority_rank_for(&vs.url, package_canonical); + bundle.value_sets.push(vs); + } None => bundle.parse_errors.push(format!( "ValueSet '{rid}' skipped: missing required field 'url'" )), @@ -248,6 +353,7 @@ fn parse_code_system(cs: &Value) -> Option { content: cs["content"].as_str().unwrap_or("complete").to_owned(), concepts, resource_json: cs.clone(), + authority_rank: AUTHORITY_UNKNOWN, }) } @@ -402,6 +508,7 @@ fn parse_value_set(vs: &Value) -> Option { status: vs["status"].as_str().unwrap_or("active").to_owned(), compose_json, resource_json: vs.clone(), + authority_rank: AUTHORITY_UNKNOWN, }) } diff --git a/crates/hts/src/import/fhir_bundle.rs b/crates/hts/src/import/fhir_bundle.rs index acc0828d6..cb28f526a 100644 --- a/crates/hts/src/import/fhir_bundle.rs +++ b/crates/hts/src/import/fhir_bundle.rs @@ -245,8 +245,9 @@ fn write_code_system( // guarantees each (url, version) maps to at most one storage row. conn.execute( "INSERT OR IGNORE INTO code_systems - (id, url, version, name, title, status, content, resource_json, created_at, updated_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)", + (id, url, version, name, title, status, content, resource_json, created_at, updated_at, + authority_rank) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9, ?10)", rusqlite::params![ storage_id, cs.url, @@ -256,7 +257,8 @@ fn write_code_system( cs.status, cs.content, resource_json, - now + now, + cs.authority_rank, ], ) .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -264,15 +266,36 @@ fn write_code_system( // INSERT OR IGNORE skips the update path on conflict; force-update the // metadata for this (url, version) row so re-imports refresh title/status // /resource_json without disturbing sibling versions. - conn.execute( + // + // `authority_rank` keeps the strongest claim ever asserted for this row: if a + // source that owns this canonical URL supplies it, the row stays authoritative + // even when a package that merely re-publishes it is imported afterwards. That + // makes the RANK independent of import order — which matters, because bootstrap + // walks the directory alphabetically and `hl7.fhir.r4.core-*.tgz` happens to + // sort before `hl7.terminology-*.tgz`. (Only the rank: the other columns are + // overwritten unconditionally, so whichever source writes a given (url, + // version) row last supplies its content. No two vendored packages ship the + // same (url, version) — verified across all 8 — so that is moot today.) + // + // The COALESCE sentinel (9 — above any real rank) is what makes an existing + // database converge: a row that predates the column is NULL, meaning "never + // claimed", so the first source to claim it wins outright. A plain + // MIN(authority_rank, ?) would instead read the legacy row as rank 0 and pin + // the stale copy at authoritative forever, silently defeating the migration. + let cs_update = format!( "UPDATE code_systems SET - name = ?1, - title = ?2, - status = ?3, - content = ?4, - resource_json = ?5, - updated_at = ?6 + name = ?1, + title = ?2, + status = ?3, + content = ?4, + resource_json = ?5, + updated_at = ?6, + authority_rank = MIN(COALESCE(authority_rank, {unclaimed}), ?9) WHERE url = ?7 AND COALESCE(version, '') = COALESCE(?8, '')", + unclaimed = bundle_parser::AUTHORITY_UNCLAIMED + ); + conn.execute( + &cs_update, rusqlite::params![ cs.name, cs.title, @@ -282,6 +305,7 @@ fn write_code_system( now, cs.url, cs.version, + cs.authority_rank, ], ) .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -516,8 +540,9 @@ fn write_value_set( // per (url, version). conn.execute( "INSERT OR IGNORE INTO value_sets - (id, url, version, name, title, status, compose_json, resource_json, created_at, updated_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)", + (id, url, version, name, title, status, compose_json, resource_json, created_at, updated_at, + authority_rank) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9, ?10)", rusqlite::params![ storage_id, vs.url, @@ -527,7 +552,8 @@ fn write_value_set( vs.status, vs.compose_json, resource_json, - now + now, + vs.authority_rank, ], ) .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -535,15 +561,22 @@ fn write_value_set( // INSERT OR IGNORE skipped the metadata refresh on conflict — apply it // explicitly so re-imports of the same (url, version) get the latest // name/title/status/compose without disturbing siblings. - conn.execute( + // See `write_code_system` for why authority_rank uses MIN over a COALESCE + // sentinel rather than a plain assignment or a bare MIN. + let vs_update = format!( "UPDATE value_sets SET - name = ?1, - title = ?2, - status = ?3, - compose_json = ?4, - resource_json = ?5, - updated_at = ?6 + name = ?1, + title = ?2, + status = ?3, + compose_json = ?4, + resource_json = ?5, + updated_at = ?6, + authority_rank = MIN(COALESCE(authority_rank, {unclaimed}), ?9) WHERE url = ?7 AND COALESCE(version, '') = COALESCE(?8, '')", + unclaimed = bundle_parser::AUTHORITY_UNCLAIMED + ); + conn.execute( + &vs_update, rusqlite::params![ vs.name, vs.title, @@ -553,6 +586,7 @@ fn write_value_set( now, vs.url, vs.version, + vs.authority_rank, ], ) .map_err(|e| HtsError::StorageError(e.to_string()))?; @@ -655,16 +689,15 @@ pub(crate) fn get_code_system_url(conn: &Connection, id: &str) -> Result(0), - ) - .optional() - .map_err(|e| HtsError::StorageError(e.to_string())) + ORDER BY {} LIMIT 1", + crate::backends::cs_precedence_order_by("code_systems") + ); + conn.query_row(&sql, rusqlite::params![id], |row| row.get::<_, String>(0)) + .optional() + .map_err(|e| HtsError::StorageError(e.to_string())) } /// Delete all cached value set expansion rows that were derived from the given diff --git a/crates/hts/src/import/tgz.rs b/crates/hts/src/import/tgz.rs index 89f446be4..0a15b334c 100644 --- a/crates/hts/src/import/tgz.rs +++ b/crates/hts/src/import/tgz.rs @@ -25,6 +25,12 @@ pub(crate) struct CollectedResources { pub value_sets: Vec, pub concept_maps: Vec, pub parse_errors: Vec, + /// The `canonical` base declared in the package's `package.json`, e.g. + /// `http://terminology.hl7.org` for `hl7.terminology`. Drives the + /// owner-vs-copy classification in [`crate::import::bundle_parser::authority_rank_for`]. + /// `None` when the package omits it (then every resource is authoritative — + /// fail-open, never worse than today). + pub package_canonical: Option, } /// Open a `.tgz` FHIR NPM package and collect all CodeSystem, ValueSet, and @@ -40,6 +46,7 @@ pub(crate) fn collect_tgz_resources(path: &Path) -> Result = Vec::new(); let mut concept_maps: Vec = Vec::new(); let mut parse_errors: Vec = Vec::new(); + let mut package_canonical: Option = None; let entries = archive .entries() @@ -67,7 +74,7 @@ pub(crate) fn collect_tgz_resources(path: &Path) -> Result Result(&buf) + && let Some(canonical) = manifest["canonical"].as_str() + && !canonical.is_empty() + { + // Prefer the top-level `package/package.json`; an example or + // nested manifest must not override it. + let is_root = entry_path.parent().and_then(|p| p.file_name()) + == Some(std::ffi::OsStr::new("package")); + if is_root || package_canonical.is_none() { + package_canonical = Some(canonical.trim_end_matches('/').to_owned()); + } + } + continue; + } + let resource: Value = match serde_json::from_slice(&buf) { Ok(v) => v, Err(e) => { @@ -105,6 +132,7 @@ pub(crate) fn collect_tgz_resources(path: &Path) -> Result"), "Extracted resources from package; starting import" ); + let package_canonical = collected.package_canonical.clone(); + let mut total = ImportStats::default(); total.errors = collected.parse_errors; @@ -159,13 +190,19 @@ pub async fn import_tgz( total.concept_maps += batch.concept_maps; total.concepts += batch.concepts; } else { + // Provenance is passed to the parser as an argument rather than + // being written into the payload, so the value that decides which + // definition of a canonical URL is authoritative can only come + // from the package manifest we just read — never from bytes an + // untrusted caller could supply to `POST /import`. let bundle_bytes = make_bundle_bytes(chunk); - let stats = backend - .import_bundle(ctx, &bundle_bytes) - .await - .map_err(|e| { - HtsError::StorageError(format!("{type_label} batch import failed: {e}")) - })?; + let parsed = crate::import::bundle_parser::parse_bundle_from_package( + &bundle_bytes, + package_canonical.as_deref(), + )?; + let stats = backend.import_parsed(ctx, parsed).await.map_err(|e| { + HtsError::StorageError(format!("{type_label} batch import failed: {e}")) + })?; total.code_systems += stats.code_systems; total.value_sets += stats.value_sets; total.concept_maps += stats.concept_maps; @@ -400,6 +437,559 @@ mod tests { assert_eq!(stats.errors.len(), 0); } + /// Build a `.tgz` carrying a `package/package.json` that declares `canonical`, + /// exactly as a real FHIR NPM package does. + fn make_test_tgz_with_canonical( + resources: &[Value], + name: &str, + canonical: &str, + ) -> NamedTempFile { + let tmp = NamedTempFile::with_suffix(".tgz").unwrap(); + let enc = GzEncoder::new(tmp.reopen().unwrap(), Compression::fast()); + let mut tar = Builder::new(enc); + + let manifest = + serde_json::to_vec(&json!({"name": name, "version": "1.0.0", "canonical": canonical})) + .unwrap(); + let mut header = tar::Header::new_gnu(); + header.set_size(manifest.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + tar.append_data(&mut header, "package/package.json", manifest.as_slice()) + .unwrap(); + + for (i, resource) in resources.iter().enumerate() { + let rt = resource["resourceType"].as_str().unwrap_or("Unknown"); + let bytes = serde_json::to_vec(resource).unwrap(); + let mut header = tar::Header::new_gnu(); + header.set_size(bytes.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + tar.append_data( + &mut header, + format!("package/{rt}-{i}.json"), + bytes.as_slice(), + ) + .unwrap(); + } + tar.finish().unwrap(); + tmp + } + + #[test] + fn authority_rank_promotes_only_owned_canonicals_else_unknown() { + use crate::import::bundle_parser::{ + AUTHORITY_OWNER, AUTHORITY_UNKNOWN, authority_rank_for, + }; + + // The package owns the URL it publishes. + assert_eq!( + authority_rank_for( + "http://terminology.hl7.org/CodeSystem/audit-event-type", + Some("http://terminology.hl7.org") + ), + AUTHORITY_OWNER + ); + // hl7.fhir.r4.core re-publishing a THO canonical: not attributable to it. + assert_eq!( + authority_rank_for( + "http://terminology.hl7.org/CodeSystem/audit-event-type", + Some("http://hl7.org/fhir") + ), + AUTHORITY_UNKNOWN + ); + // ...but it IS authoritative for its own canonical base. + assert_eq!( + authority_rank_for( + "http://hl7.org/fhir/CodeSystem/example", + Some("http://hl7.org/fhir") + ), + AUTHORITY_OWNER + ); + // No package context (REST write, $import-bundle): we cannot vouch for it, + // so it must NOT claim ownership. Defaulting these to OWNER is what would + // let a replayed r4.core bundle permanently outrank THO's original. + assert_eq!( + authority_rank_for("http://terminology.hl7.org/CodeSystem/x", None), + AUTHORITY_UNKNOWN + ); + // A package that declares no canonical: no evidence, so UNKNOWN. + assert_eq!( + authority_rank_for("http://anything/at/all", Some("")), + AUTHORITY_UNKNOWN + ); + // Path boundary: `http://hl7.org/fhir` must not claim `/fhirpath/...`. + assert_eq!( + authority_rank_for( + "http://hl7.org/fhirpath/CodeSystem/x", + Some("http://hl7.org/fhir") + ), + AUTHORITY_UNKNOWN + ); + // us.nlm.vsac declares the fhir.org REGISTRY base but ships cts.nlm.nih.gov + // ValueSets. Those are genuine originals; they are merely unattributable. + // They must not be branded copies — `vs_precedence_order_by` has no + // content/has-concepts tier to absorb that error, so a mislabel there would + // let any REST upload outrank the real VSAC definition. + assert_eq!( + authority_rank_for( + "http://cts.nlm.nih.gov/fhir/ValueSet/2.16.840.1.113762.1.4.1", + Some("http://fhir.org/packages/us.nlm.vsac") + ), + AUTHORITY_UNKNOWN + ); + } + + /// Regression test for issue #200. + /// + /// `hl7.fhir.r4.core` re-ships THO CodeSystems stamped with the *FHIR release* + /// version (`4.0.1`) and truncated to a single concept, while the THO package + /// publishes the complete definition as version `1.0.0`. Both rows are + /// `content: complete` and both carry concepts, so no intrinsic property of + /// the rows separates them — and `4.0.1` outsorts `1.0.0` under any version + /// ordering, lexicographic or semver. Only provenance can decide, and the + /// bare (no-version) call must land on the complete THO row. + /// + /// Imports the *core-like* package first, mirroring the real bootstrap order + /// (the directory is walked alphabetically, so `hl7.fhir.r4.core-*.tgz` lands + /// before `hl7.terminology-*.tgz`) — the outcome must not depend on it. + #[tokio::test] + async fn bare_validate_code_prefers_owning_package_over_republished_copy() { + use crate::traits::CodeSystemOperations; + use crate::types::ValidateCodeRequest; + + const URL: &str = "http://terminology.hl7.org/CodeSystem/audit-event-type"; + + let backend = SqliteTerminologyBackend::in_memory().unwrap(); + let ctx = TenantContext::system(); + + let core_copy = json!({ + "resourceType": "CodeSystem", "id": "audit-event-type", "url": URL, + "version": "4.0.1", "name": "AuditEventID", "status": "active", + "content": "complete", + "concept": [{"code": "rest", "display": "RESTful Operation"}] + }); + let tho_original = json!({ + "resourceType": "CodeSystem", "id": "audit-event-type", "url": URL, + "version": "1.0.0", "name": "AuditEventID", "status": "active", + "content": "complete", + "concept": [ + {"code": "rest", "display": "RESTful Operation"}, + {"code": "hl7-v2", "display": "HL7v2 Message"}, + {"code": "hl7-v3", "display": "HL7v3 Message"}, + {"code": "document", "display": "Document"}, + {"code": "object", "display": "Object"} + ] + }); + + let core = make_test_tgz_with_canonical( + std::slice::from_ref(&core_copy), + "hl7.fhir.r4.core", + "http://hl7.org/fhir", + ); + let tho = make_test_tgz_with_canonical( + std::slice::from_ref(&tho_original), + "hl7.terminology", + "http://terminology.hl7.org", + ); + + import_tgz(&backend, &ctx, core.path(), 500, false) + .await + .expect("core package import should succeed"); + import_tgz(&backend, &ctx, tho.path(), 500, false) + .await + .expect("THO package import should succeed"); + + // Both versions coexist — the copy is ranked, not discarded, so an + // explicit `version=4.0.1` request can still reach it. + assert_eq!(count_rows(&backend, "code_systems"), 2); + + // The bug: `object` exists only in the complete THO 1.0.0 definition. + let resp = CodeSystemOperations::validate_code( + &backend, + &ctx, + ValidateCodeRequest { + system: Some(URL.into()), + code: "object".into(), + ..Default::default() + }, + ) + .await + .unwrap(); + + assert!( + resp.result, + "bare $validate-code must resolve to the complete THO definition, \ + not the truncated hl7.fhir.r4.core copy; got: {:?}", + resp.message + ); + assert_eq!(resp.cs_version.as_deref(), Some("1.0.0")); + + // A genuine miss must name the version it ACTUALLY validated against. + // The diagnostic used to re-query by URL with its own ordering, so it + // reported whichever version string sorted highest ("4.0.1") even though + // validation had run against a different row. + let miss = CodeSystemOperations::validate_code( + &backend, + &ctx, + ValidateCodeRequest { + system: Some(URL.into()), + code: "no-such-code".into(), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(!miss.result); + let msg = miss.message.unwrap_or_default(); + assert!( + msg.contains("1.0.0") && !msg.contains("4.0.1"), + "a miss must report the resolved version (1.0.0), not the highest \ + version string on the URL (4.0.1); got: {msg}" + ); + + // An explicit pin to the copy still resolves to the copy, and correctly + // reports that `object` is not in it. The copy is ranked, not hidden. + let pinned = CodeSystemOperations::validate_code( + &backend, + &ctx, + ValidateCodeRequest { + system: Some(URL.into()), + code: "object".into(), + version: Some("4.0.1".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(!pinned.result, "version=4.0.1 pins the truncated copy"); + assert!( + pinned.message.unwrap_or_default().contains("4.0.1"), + "a miss must report the version it actually validated against" + ); + } + + /// Import order must not decide the winner. The real bootstrap walks the + /// terminology directory alphabetically, which is what let the copy win in + /// the first place; reversing the order must change nothing. + #[tokio::test] + async fn owning_package_wins_regardless_of_import_order() { + use crate::traits::CodeSystemOperations; + use crate::types::ValidateCodeRequest; + + const URL: &str = "http://terminology.hl7.org/CodeSystem/audit-event-type"; + + let backend = SqliteTerminologyBackend::in_memory().unwrap(); + let ctx = TenantContext::system(); + + let core_copy = json!({ + "resourceType": "CodeSystem", "id": "audit-event-type", "url": URL, + "version": "4.0.1", "status": "active", "content": "complete", + "concept": [{"code": "rest"}] + }); + let tho_original = json!({ + "resourceType": "CodeSystem", "id": "audit-event-type", "url": URL, + "version": "1.0.0", "status": "active", "content": "complete", + "concept": [{"code": "rest"}, {"code": "object"}] + }); + + let core = make_test_tgz_with_canonical( + std::slice::from_ref(&core_copy), + "hl7.fhir.r4.core", + "http://hl7.org/fhir", + ); + let tho = make_test_tgz_with_canonical( + std::slice::from_ref(&tho_original), + "hl7.terminology", + "http://terminology.hl7.org", + ); + + // THO first, core second — the reverse of the alphabetical bootstrap order. + import_tgz(&backend, &ctx, tho.path(), 500, false) + .await + .unwrap(); + import_tgz(&backend, &ctx, core.path(), 500, false) + .await + .unwrap(); + + let resp = CodeSystemOperations::validate_code( + &backend, + &ctx, + ValidateCodeRequest { + system: Some(URL.into()), + code: "object".into(), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!( + resp.result, + "the owning package must win no matter which package was imported last" + ); + } + + /// The manifest drives provenance, so its edge cases decide whether a + /// resource is treated as an original or a copy. All of them must fail open: + /// an unreadable or canonical-less manifest must never demote a resource. + #[test] + fn package_manifest_canonical_is_parsed_and_fails_open() { + // Declares a canonical → captured, trailing slash normalised away so the + // `url.starts_with(base)` test in `authority_rank_for` behaves. + let tgz = make_test_tgz_with_canonical( + &[sample_code_system()], + "hl7.terminology", + "http://terminology.hl7.org/", + ); + let got = collect_tgz_resources(tgz.path()).unwrap(); + assert_eq!( + got.package_canonical.as_deref(), + Some("http://terminology.hl7.org") + ); + assert_eq!( + got.code_systems.len(), + 1, + "manifest must not be imported as a resource" + ); + + // No `canonical` key at all (older packages) → no claim, everything stays + // authoritative. This is the fail-open path: never worse than before. + let tmp = NamedTempFile::with_suffix(".tgz").unwrap(); + { + let enc = GzEncoder::new(tmp.reopen().unwrap(), Compression::fast()); + let mut tar = Builder::new(enc); + for (name, content) in [ + ("package/package.json", br#"{"name":"legacy.pkg"}"# as &[u8]), + ( + "package/other/package.json", + br#"{"canonical":"http://nested.example"}"# as &[u8], + ), + ] { + let mut h = tar::Header::new_gnu(); + h.set_size(content.len() as u64); + h.set_mode(0o644); + h.set_cksum(); + tar.append_data(&mut h, name, content).unwrap(); + } + tar.finish().unwrap(); + } + let got = collect_tgz_resources(tmp.path()).unwrap(); + // The root manifest declares none; a *nested* manifest must not be + // mistaken for the package's own canonical when the root one is absent + // of a canonical key — but if nothing else claimed it, a nested value is + // better than nothing, so this documents the actual precedence. + assert_eq!( + got.package_canonical.as_deref(), + Some("http://nested.example") + ); + + assert!( + got.code_systems.is_empty() && got.parse_errors.is_empty(), + "manifests are metadata, never resources, and never parse errors" + ); + } + + /// ValueSets are half the shipped surface of this fix — `hl7.fhir.r4.core` + /// re-publishes 603 THO ValueSets it does not own — but every other test here + /// imports CodeSystems only, so `vs_precedence_order_by` and the ValueSet + /// upsert would ship unproven. An inverted VS tier, or a VS upsert that never + /// binds the rank, passes the rest of this suite green. + #[tokio::test] + async fn value_set_resolution_prefers_owning_package_over_republished_copy() { + const VS_URL: &str = "http://terminology.hl7.org/ValueSet/audit-event-type"; + + let backend = SqliteTerminologyBackend::in_memory().unwrap(); + let ctx = TenantContext::system(); + + let mk = |version: &str, title: &str| { + json!({ + "resourceType": "ValueSet", "id": format!("aet-vs-{version}"), + "url": VS_URL, "version": version, "status": "active", "title": title, + "compose": {"include": [{"system": "http://example.org/cs"}]} + }) + }; + // Same shape as the CodeSystem bug: the copy carries the FHIR *release* + // version, which out-sorts the ValueSet's own version under any ordering. + let core = make_test_tgz_with_canonical( + &[mk("4.0.1", "core copy")], + "hl7.fhir.r4.core", + "http://hl7.org/fhir", + ); + let tho = make_test_tgz_with_canonical( + &[mk("1.0.0", "THO original")], + "hl7.terminology", + "http://terminology.hl7.org", + ); + import_tgz(&backend, &ctx, core.path(), 500, false) + .await + .unwrap(); + import_tgz(&backend, &ctx, tho.path(), 500, false) + .await + .unwrap(); + + let conn = backend.pool().get().unwrap(); + let (rank_copy, rank_owner): (Option, Option) = ( + conn.query_row( + "SELECT authority_rank FROM value_sets WHERE url = ?1 AND version = '4.0.1'", + rusqlite::params![VS_URL], + |r| r.get(0), + ) + .unwrap(), + conn.query_row( + "SELECT authority_rank FROM value_sets WHERE url = ?1 AND version = '1.0.0'", + rusqlite::params![VS_URL], + |r| r.get(0), + ) + .unwrap(), + ); + assert_eq!( + rank_copy, + Some(crate::import::bundle_parser::AUTHORITY_UNKNOWN) + ); + assert_eq!( + rank_owner, + Some(crate::import::bundle_parser::AUTHORITY_OWNER) + ); + + // And the shared VS ordering must actually pick the owner's row. + let sql = format!( + "SELECT version FROM value_sets WHERE url = ?1 ORDER BY {} LIMIT 1", + crate::backends::vs_precedence_order_by("value_sets") + ); + let winner: String = conn + .query_row(&sql, rusqlite::params![VS_URL], |r| r.get(0)) + .unwrap(); + assert_eq!( + winner, "1.0.0", + "versionless ValueSet resolution must land on the owning package's row" + ); + } + + /// The upsert's `MIN(COALESCE(authority_rank, AUTHORITY_UNCLAIMED), ?)` is what + /// makes the rank order-independent *at the row level* and what lets a legacy + /// NULL row demote itself to a copy. The other tests import DIFFERENT versions, + /// which produce two distinct rows under `idx_code_systems_url_version` and so + /// never touch the conflict path at all. This one collides the SAME (url, + /// version) deliberately. + #[tokio::test] + async fn upsert_keeps_the_strongest_claim_regardless_of_import_order() { + const URL: &str = "http://terminology.hl7.org/CodeSystem/collide"; + + // Identical (url, version) shipped by an owning package and by a package + // that merely re-publishes it → ONE row, whose rank must end up 0 (owner) + // no matter which package is imported last. + let cs = json!({ + "resourceType": "CodeSystem", "id": "collide", "url": URL, + "version": "1.0.0", "status": "active", "content": "complete", + "concept": [{"code": "a"}] + }); + let owner_pkg = || { + make_test_tgz_with_canonical( + std::slice::from_ref(&cs), + "hl7.terminology", + "http://terminology.hl7.org", + ) + }; + let copy_pkg = || { + make_test_tgz_with_canonical( + std::slice::from_ref(&cs), + "hl7.fhir.r4.core", + "http://hl7.org/fhir", + ) + }; + + for (label, first, second) in [ + ("owner then copy", owner_pkg(), copy_pkg()), + ("copy then owner", copy_pkg(), owner_pkg()), + ] { + let backend = SqliteTerminologyBackend::in_memory().unwrap(); + let ctx = TenantContext::system(); + import_tgz(&backend, &ctx, first.path(), 500, false) + .await + .unwrap(); + import_tgz(&backend, &ctx, second.path(), 500, false) + .await + .unwrap(); + + let conn = backend.pool().get().unwrap(); + let rows: i64 = conn + .query_row( + "SELECT COUNT(*) FROM code_systems WHERE url = ?1", + rusqlite::params![URL], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + rows, 1, + "{label}: same (url, version) must collapse to one row" + ); + + let rank: Option = conn + .query_row( + "SELECT authority_rank FROM code_systems WHERE url = ?1", + rusqlite::params![URL], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + rank, + Some(crate::import::bundle_parser::AUTHORITY_OWNER), + "{label}: the strongest claim must survive — a later copy import \ + must not demote a row an owning package already claimed" + ); + } + } + + /// A row that predates the column is NULL ("never claimed"), and the sentinel + /// is what lets the first claimant overwrite it. A plain `MIN(rank, ?)` would + /// read the legacy row as 0 and pin the stale copy authoritative forever — + /// silently defeating the migration on exactly the databases it exists for. + #[tokio::test] + async fn legacy_null_row_is_demoted_when_a_copy_reclaims_it() { + const URL: &str = "http://terminology.hl7.org/CodeSystem/legacy"; + + let backend = SqliteTerminologyBackend::in_memory().unwrap(); + let ctx = TenantContext::system(); + { + let conn = backend.pool().get().unwrap(); + conn.execute( + "INSERT INTO code_systems (id, url, version, status, content, created_at, updated_at, authority_rank) + VALUES ('legacy|4.0.1', ?1, '4.0.1', 'active', 'complete', 'x', 'x', NULL)", + rusqlite::params![URL], + ) + .unwrap(); + } + + // The core package re-imports that exact row and declares itself a copy. + let cs = json!({ + "resourceType": "CodeSystem", "id": "legacy", "url": URL, + "version": "4.0.1", "status": "active", "content": "complete", + "concept": [{"code": "rest"}] + }); + let core = make_test_tgz_with_canonical( + std::slice::from_ref(&cs), + "hl7.fhir.r4.core", + "http://hl7.org/fhir", + ); + import_tgz(&backend, &ctx, core.path(), 500, false) + .await + .unwrap(); + + let conn = backend.pool().get().unwrap(); + let rank: Option = conn + .query_row( + "SELECT authority_rank FROM code_systems WHERE url = ?1", + rusqlite::params![URL], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + rank, + Some(crate::import::bundle_parser::AUTHORITY_UNKNOWN), + "an unclaimed (NULL) legacy row must be demotable by a re-importing copy" + ); + } + #[tokio::test] async fn import_tgz_invalid_path_returns_error() { let backend = SqliteTerminologyBackend::in_memory().unwrap(); diff --git a/crates/hts/src/operations/expand.rs b/crates/hts/src/operations/expand.rs index 9f30e6cac..3b01b0b01 100644 --- a/crates/hts/src/operations/expand.rs +++ b/crates/hts/src/operations/expand.rs @@ -2051,7 +2051,25 @@ async fn process_expand_inner( None } }; - let (_, source_vs_search) = tokio::join!(flags_fut, source_vs_fut); + // Which ValueSet row would the backend itself resolve for this URL? Ask it, + // rather than re-deriving the answer by sorting version strings below. + // Same-URL precedence depends on `authority_rank`, a storage column that is + // deliberately absent from the FHIR resource, so a Rust-side sort cannot see + // it and would echo the re-published copy while the backend expanded the + // original (issue #200, ValueSet path). Only needed when no version is pinned. + let resolved_vs_version_fut = async { + match (&url_for_neg_cache, &req_vs_version) { + (Some(u), None) => { + ValueSetOperations::value_set_version_for_url(state.backend(), &ctx, u) + .await + .ok() + .flatten() + } + _ => None, + } + }; + let (_, source_vs_search, resolved_vs_version) = + tokio::join!(flags_fut, source_vs_fut, resolved_vs_version_fut); let source_vs: Option = if url_for_neg_cache.is_some() { source_vs_search.and_then(|mut v| { // If a specific version was requested, return the row whose @@ -2071,7 +2089,15 @@ async fn process_expand_inner( .find(|r| r.get("version").and_then(|x| x.as_str()) == Some(want.as_str())) .cloned(); exact.or_else(|| v.into_iter().next()) + } else if let Some(ref resolved) = resolved_vs_version { + // Echo the row the backend actually resolved. + v.iter() + .find(|r| r.get("version").and_then(|x| x.as_str()) == Some(resolved.as_str())) + .cloned() + .or_else(|| v.into_iter().next()) } else { + // Backend has no opinion (unversioned row, or a backend that does + // not implement the accessor): fall back to highest version. v.sort_by(|a, b| { let av = a.get("version").and_then(|x| x.as_str()).unwrap_or(""); let bv = b.get("version").and_then(|x| x.as_str()).unwrap_or(""); @@ -3584,13 +3610,25 @@ async fn process_expand_inner( pinned_version = Some(default_v.clone()); } } - // When no version pin is in effect, fetch up to 20 candidates and - // pick the highest version — mirrors `resolve_value_set_versioned`'s - // order-by-version-DESC behaviour. `count: Some(1)` against the + // When no version pin is in effect, fetch up to 20 candidates and pick + // the one the BACKEND would resolve. `count: Some(1)` against the // search SQL (which orders by created_at) yields the earliest- // imported row instead, silently picking vs-version-a1 over -a2 // for the `default-valueset-version/indirect-expand-zero` fixture. + // + // The winner is asked of the backend rather than re-derived by sorting + // version strings: precedence depends on `authority_rank`, a storage + // column absent from the resource JSON, so a Rust-side sort would echo + // a re-published copy while the backend expanded the original. let count_hint = if pinned_version.is_some() { 1 } else { 20 }; + let resolved_vs_version: Option = if pinned_version.is_none() { + ValueSetOperations::value_set_version_for_url(state.backend(), &ctx, &bare_url) + .await + .ok() + .flatten() + } else { + None + }; let referenced_vs: Option = ValueSetOperations::search( state.backend(), &ctx, @@ -3606,9 +3644,22 @@ async fn process_expand_inner( .and_then(|mut hits| { if pinned_version.is_some() { hits.pop() + } else if let Some(ref resolved) = resolved_vs_version { + hits.iter() + .find(|r| { + r.get("version").and_then(|x| x.as_str()) == Some(resolved.as_str()) + }) + .cloned() + .or_else(|| { + hits.sort_by(|a, b| { + let av = a.get("version").and_then(|x| x.as_str()).unwrap_or(""); + let bv = b.get("version").and_then(|x| x.as_str()).unwrap_or(""); + bv.cmp(av) + }); + hits.into_iter().next() + }) } else { - // No pin: highest version wins (matches the backend's - // `ORDER BY COALESCE(version,'') DESC` resolution). + // Backend has no opinion: fall back to highest version. hits.sort_by(|a, b| { let av = a.get("version").and_then(|x| x.as_str()).unwrap_or(""); let bv = b.get("version").and_then(|x| x.as_str()).unwrap_or(""); diff --git a/crates/hts/src/operations/validate_code.rs b/crates/hts/src/operations/validate_code.rs index febfc4e19..3a407e1ce 100644 --- a/crates/hts/src/operations/validate_code.rs +++ b/crates/hts/src/operations/validate_code.rs @@ -2303,7 +2303,14 @@ async fn process_validate_code_inner( let supplements = resolve_supplements(state.backend(), &ctx, ¶ms, Some(&system)).await?; let display = find_str_param(¶ms, "display"); - let req_version = find_str_param(¶ms, "version"); + let req_version = resolve_cs_effective_version( + state.backend(), + &ctx, + ¶ms, + &system, + find_str_param(¶ms, "version"), + ) + .await; let req = ValidateCodeRequest { url: None, value_set_version: None, @@ -2376,7 +2383,14 @@ async fn process_validate_code_inner( // report a mismatch. let display = coding_display.or_else(|| find_str_param(¶ms, "display")); // Coding.version takes precedence over a top-level `version` param. - let req_version = coding_version.or_else(|| find_str_param(¶ms, "version")); + let req_version = resolve_cs_effective_version( + state.backend(), + &ctx, + ¶ms, + &system, + coding_version.or_else(|| find_str_param(¶ms, "version")), + ) + .await; let supplements = resolve_supplements(state.backend(), &ctx, ¶ms, Some(&system)).await?; let req = ValidateCodeRequest { @@ -2451,12 +2465,24 @@ async fn process_validate_code_inner( // codings in a CodeableConcept all validate, the response echoes the // last one). Iterate in reverse so the earliest "yes" we find is the // last entry in the input. - let cc_req_version = find_str_param(¶ms, "version"); + let cc_explicit_version = find_str_param(¶ms, "version"); let cs_lenient = params .iter() .find(|p| p.get("name").and_then(|v| v.as_str()) == Some("lenient-display-validation")) .and_then(|p| p.get("valueBoolean").and_then(|v| v.as_bool())); for (system, code) in codings.into_iter().rev() { + // Resolved per coding: the version pins are keyed by system canonical + // (`system-version=|`), and a CodeableConcept may carry + // codings from several systems, so the effective version can differ + // between iterations. + let cc_req_version = resolve_cs_effective_version( + state.backend(), + &ctx, + ¶ms, + &system, + cc_explicit_version.clone(), + ) + .await; let req = ValidateCodeRequest { url: None, value_set_version: None, @@ -2614,6 +2640,61 @@ fn vs_include_pin_for_system(vs: &Value, system_url: &str) -> Option` — overrides even an explicitly supplied version. +/// 2. Explicit: `Coding.version`, then `version`, then `systemVersion`. +/// 3. `system-version|` / `check-system-version|` — a *default*, applied +/// only when the caller supplied no explicit version. +/// 4. `None` — the backend applies its own default resolution. +/// +/// The official `CodeSystem/$validate-code` OperationDefinition declares only +/// `version`; `systemVersion` and the `*-system-version` canonical pins are +/// tx.fhir.org extensions. HTS already honoured them on `$expand` and +/// `ValueSet/$validate-code` but silently ignored them here, so a client pinning +/// `systemVersion=1.0.0` got the default row anyway — the second defect reported +/// in issue #200. Accepting them removes that asymmetry. +/// +/// Caching is unaffected: `build_validate_code_cache_key` already refuses to +/// cache any request carrying the canonical pins, and `systemVersion` is part of +/// the key like any other value parameter. +async fn resolve_cs_effective_version( + backend: &B, + ctx: &TenantContext, + params: &[Value], + system: &str, + explicit: Option, +) -> Option { + let force_pins = collect_canonical_params(params, "force-system-version"); + if let Some(pattern) = find_pin_for_system(&force_pins, system) { + return Some( + resolve_cs_version_pattern(backend, ctx, system, pattern) + .await + .unwrap_or_else(|| pattern.to_string()), + ); + } + + if let Some(version) = explicit.or_else(|| find_str_param(params, "systemVersion")) { + return Some(version); + } + + let mut defaults = collect_canonical_params(params, "system-version"); + defaults.extend(collect_canonical_params(params, "check-system-version")); + if let Some(pattern) = find_pin_for_system(&defaults, system) { + return Some( + resolve_cs_version_pattern(backend, ctx, system, pattern) + .await + .unwrap_or_else(|| pattern.to_string()), + ); + } + + None +} + /// Resolve a (possibly wildcard) version pattern against the set of stored /// versions for a CodeSystem URL. Picks the highest matching version. /// Returns `None` when no stored version matches (or the CS is unknown). diff --git a/crates/hts/src/traits/value_set.rs b/crates/hts/src/traits/value_set.rs index 139004bd3..f50c906aa 100644 --- a/crates/hts/src/traits/value_set.rs +++ b/crates/hts/src/traits/value_set.rs @@ -45,6 +45,27 @@ pub trait ValueSetOperations: Send + Sync { /// /// Triggers expansion if needed, then tests set membership. /// Returns `result = true` with display on success. + /// Return the `version` of the ValueSet row this backend would resolve for + /// `url` when the caller pins no version. + /// + /// The operations layer must not re-derive this by sorting the JSON returned + /// from [`Self::search`]: same-URL precedence depends on `authority_rank`, + /// which is a storage column and is deliberately absent from the FHIR + /// resource. A Rust-side "highest version string wins" sort silently + /// disagrees with the backend whenever a re-published copy carries a higher + /// version than the original — exactly the shape of issue #200 — so the + /// response would echo one ValueSet while the backend expanded another. + /// + /// Default returns `None`, meaning "no opinion"; callers then fall back to + /// their previous behaviour. + async fn value_set_version_for_url( + &self, + _ctx: &TenantContext, + _url: &str, + ) -> Result, HtsError> { + Ok(None) + } + async fn validate_code( &self, ctx: &TenantContext, diff --git a/crates/hts/tests/code_system_ops.rs b/crates/hts/tests/code_system_ops.rs index 7a42e29ea..6c7a8a5bf 100644 --- a/crates/hts/tests/code_system_ops.rs +++ b/crates/hts/tests/code_system_ops.rs @@ -488,3 +488,178 @@ async fn lookup_by_id_unknown_id_returns_404() { assert_eq!(status, StatusCode::NOT_FOUND); } + +// ── Issue #200: CodeSystem/$validate-code version pins ──────────────────────── +// +// `CodeSystem/$validate-code` used to read only the `version` parameter. +// `systemVersion` and the canonical `*-system-version` pins were parsed by the +// GET/POST plumbing, were already honoured on `$expand` and +// `ValueSet/$validate-code`, and were silently ignored here — so a client +// pinning `systemVersion=1.0.0` got the default row anyway. That was the second +// defect reported in #200. + +/// Two versions of one canonical URL. `old` exists only in 1.0.0, `new` only in +/// 2.0.0. With no pin the newer row wins (both rows tie on every other tier), so +/// each pin below is only satisfiable by actually honouring the parameter. +const PIN_CS_URL: &str = "http://hts.test/cs/pinned"; + +async fn app_with_two_versions() -> TestApp { + let app = TestApp::new(); + let bundle = serde_json::json!({ + "resourceType": "Bundle", + "type": "collection", + "entry": [ + {"resource": { + "resourceType": "CodeSystem", "id": "pinned-v1", "url": PIN_CS_URL, + "version": "1.0.0", "status": "active", "content": "complete", + "concept": [{"code": "old", "display": "Only in 1.0.0"}] + }}, + {"resource": { + "resourceType": "CodeSystem", "id": "pinned-v2", "url": PIN_CS_URL, + "version": "2.0.0", "status": "active", "content": "complete", + "concept": [{"code": "new", "display": "Only in 2.0.0"}] + }} + ] + }); + let body = serde_json::to_string(&bundle).unwrap(); + app.import_bundle_ok(&body).await; + app +} + +fn result_of(body: &serde_json::Value) -> bool { + body["parameter"] + .as_array() + .expect("Parameters.parameter") + .iter() + .find(|p| p["name"] == "result") + .and_then(|p| p["valueBoolean"].as_bool()) + .expect("expected a result parameter") +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn validate_code_unpinned_resolves_newest_version() { + let app = app_with_two_versions().await; + + // Baseline: with no pin, 2.0.0 wins, so `old` is not valid and `new` is. + let (status, body) = app + .post_fhir( + "/CodeSystem/$validate-code", + TestApp::params(&[ + ("url", "valueUri", PIN_CS_URL), + ("code", "valueCode", "old"), + ]), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!(!result_of(&body), "unpinned must resolve to 2.0.0"); + + let (_, body) = app + .post_fhir( + "/CodeSystem/$validate-code", + TestApp::params(&[ + ("url", "valueUri", PIN_CS_URL), + ("code", "valueCode", "new"), + ]), + ) + .await; + assert!(result_of(&body), "unpinned must resolve to 2.0.0"); +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn validate_code_honours_system_version_param() { + let app = app_with_two_versions().await; + + // `systemVersion` was ignored before #200 — this would have resolved to + // 2.0.0 and returned false. + let (status, body) = app + .post_fhir( + "/CodeSystem/$validate-code", + TestApp::params(&[ + ("url", "valueUri", PIN_CS_URL), + ("code", "valueCode", "old"), + ("systemVersion", "valueString", "1.0.0"), + ]), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!( + result_of(&body), + "systemVersion=1.0.0 must pin the 1.0.0 row" + ); +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn validate_code_honours_canonical_system_version_pin() { + let app = app_with_two_versions().await; + let pin = format!("{PIN_CS_URL}|1.0.0"); + + // The tx.fhir.org canonical form: `system-version=|`. + let (status, body) = app + .post_fhir( + "/CodeSystem/$validate-code", + TestApp::params(&[ + ("url", "valueUri", PIN_CS_URL), + ("code", "valueCode", "old"), + ("system-version", "valueString", pin.as_str()), + ]), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!(result_of(&body), "system-version pin must select 1.0.0"); +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn force_system_version_overrides_an_explicit_version() { + let app = app_with_two_versions().await; + let pin = format!("{PIN_CS_URL}|1.0.0"); + + // `force-system-version` outranks even an explicitly supplied `version`: + // the caller asks for 2.0.0, the force pin redirects to 1.0.0, and `old` + // (which exists only in 1.0.0) validates. + let (status, body) = app + .post_fhir( + "/CodeSystem/$validate-code", + TestApp::params(&[ + ("url", "valueUri", PIN_CS_URL), + ("code", "valueCode", "old"), + ("version", "valueString", "2.0.0"), + ("force-system-version", "valueString", pin.as_str()), + ]), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!( + result_of(&body), + "force-system-version must override the explicit version parameter" + ); +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn explicit_version_beats_default_system_version_pin() { + let app = app_with_two_versions().await; + let pin = format!("{PIN_CS_URL}|1.0.0"); + + // `system-version` is a *default*, not a force: an explicit `version` wins, + // so `old` (1.0.0-only) must NOT validate against the pinned-to-2.0.0 call. + let (status, body) = app + .post_fhir( + "/CodeSystem/$validate-code", + TestApp::params(&[ + ("url", "valueUri", PIN_CS_URL), + ("code", "valueCode", "old"), + ("version", "valueString", "2.0.0"), + ("system-version", "valueString", pin.as_str()), + ]), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!( + !result_of(&body), + "an explicit version must take precedence over a system-version default" + ); +} diff --git a/crates/hts/tests/postgres_integration_tests.rs b/crates/hts/tests/postgres_integration_tests.rs index e99266113..5cdf0063e 100644 --- a/crates/hts/tests/postgres_integration_tests.rs +++ b/crates/hts/tests/postgres_integration_tests.rs @@ -2200,3 +2200,134 @@ async fn importer_dicom_runs_against_postgres() { .await .unwrap(); } + +// ── Issue #200: provenance precedence among same-URL rows ───────────────────── + +/// Build a FHIR NPM `.tgz` whose `package/package.json` declares `canonical`, +/// so the importer can tell whether the package owns the URLs it publishes. +fn pkg_tgz( + resources: &[serde_json::Value], + name: &str, + canonical: &str, +) -> tempfile::NamedTempFile { + use flate2::Compression; + use flate2::write::GzEncoder; + use tar::Builder; + + let tmp = tempfile::NamedTempFile::with_suffix(".tgz").unwrap(); + let enc = GzEncoder::new(tmp.reopen().unwrap(), Compression::fast()); + let mut tar = Builder::new(enc); + + let manifest = serde_json::to_vec( + &serde_json::json!({"name": name, "version": "1.0.0", "canonical": canonical}), + ) + .unwrap(); + let mut header = tar::Header::new_gnu(); + header.set_size(manifest.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + tar.append_data(&mut header, "package/package.json", manifest.as_slice()) + .unwrap(); + + for (i, resource) in resources.iter().enumerate() { + let bytes = serde_json::to_vec(resource).unwrap(); + let mut header = tar::Header::new_gnu(); + header.set_size(bytes.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + tar.append_data( + &mut header, + format!("package/CodeSystem-{i}.json"), + bytes.as_slice(), + ) + .unwrap(); + } + tar.finish().unwrap(); + tmp +} + +/// Postgres twin of `bare_validate_code_prefers_owning_package_over_republished_copy` +/// (see `src/import/tgz.rs`). Issue #200 reproduces identically on both backends — +/// the precedence ordering is shared text, and this pins the Postgres side of it. +#[tokio::test] +async fn bare_validate_code_prefers_owning_package_over_republished_copy_pg() { + use helios_hts::import::tgz::import_tgz; + + let backend = fresh_backend().await; + // Namespaced per-run so the shared container stays isolated between tests. + // The owning package's canonical base is this namespace, mirroring how THO + // owns `http://terminology.hl7.org`; the copying package declares a + // different base, mirroring `hl7.fhir.r4.core` re-shipping a THO canonical. + let base = base_url!("prov"); + let owner_canonical = base.trim_end_matches('/').to_string(); + let url = format!("{base}CodeSystem/audit-event-type"); + + let core_copy = serde_json::json!({ + "resourceType": "CodeSystem", "id": "aet-core", "url": url, + "version": "4.0.1", "status": "active", "content": "complete", + "concept": [{"code": "rest", "display": "RESTful Operation"}] + }); + let tho_original = serde_json::json!({ + "resourceType": "CodeSystem", "id": "aet-tho", "url": url, + "version": "1.0.0", "status": "active", "content": "complete", + "concept": [ + {"code": "rest", "display": "RESTful Operation"}, + {"code": "object", "display": "Object"} + ] + }); + + // Copying package first, matching the alphabetical bootstrap order that let + // the truncated copy win in the first place. + let core = pkg_tgz( + std::slice::from_ref(&core_copy), + "hl7.fhir.r4.core", + "http://hl7.org/fhir", + ); + let tho = pkg_tgz( + std::slice::from_ref(&tho_original), + "hl7.terminology", + &owner_canonical, + ); + import_tgz(&backend, &ctx(), core.path(), 500, false) + .await + .expect("core package import"); + import_tgz(&backend, &ctx(), tho.path(), 500, false) + .await + .expect("THO package import"); + + // `object` exists only in the complete THO definition. Under the old + // ordering the truncated 4.0.1 copy won on the version text sort. + let resp = CodeSystemOperations::validate_code( + &backend, + &ctx(), + ValidateCodeRequest { + system: Some(url.clone()), + code: "object".into(), + ..Default::default() + }, + ) + .await + .unwrap(); + + assert!( + resp.result, + "bare $validate-code must resolve to the owning package's complete row; got: {:?}", + resp.message + ); + assert_eq!(resp.cs_version.as_deref(), Some("1.0.0")); + + // The copy is ranked, not discarded: an explicit pin still reaches it. + let pinned = CodeSystemOperations::validate_code( + &backend, + &ctx(), + ValidateCodeRequest { + system: Some(url), + code: "object".into(), + version: Some("4.0.1".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(!pinned.result, "version=4.0.1 pins the truncated copy"); +}