Simplification round: collapse five repeated shapes, and fix the two cache leaks one of them bred

crates/navigator-app/src/appview.rs

@@ -0,0 +1,107 @@
+//! The one way Navigator talks to the AppView's `/api/v1/*` Edge API.
+//!
+//! Three clients grew here independently — IBD exchange, social, recruitment — and each arrived at
+//! the same two shapes: an unauthenticated-looking POST whose body carries the device-key
+//! signature, and a replay-guarded signed GET whose `did`/`ts`/`sig` ride on the query string. The
+//! IBD and social versions were byte-for-byte identical, and the remaining one-off calls in
+//! `sync.rs` / `matching.rs` open-coded the same thing a fourth and fifth time. They are all this
+//! module now, so the error mapping, the signing-query layout, and the non-2xx classification are
+//! decided once.
+//!
+//! What travels: a DID, a timestamp, a signature, and whatever the caller chose to send. Never
+//! genotypes, never coordinates.
+
+use super::*;
+
+/// A transport failure (connection refused, timeout, TLS) on an AppView call.
+///
+/// The AppView is reached with a bare `reqwest` client rather than through the sync engine, but a
+/// network failure means the same thing either way, so it lands in the same error variant the PDS
+/// paths use and the offline indicator already understands.
+pub(crate) fn transport(e: reqwest::Error) -> AppError {
+    AppError::Sync(navigator_sync::SyncError::from(e))
+}
+
+/// Classify a non-2xx AppView response into a user-facing [`AppError::AppView`]. Consumes `resp` to
+/// read the body (so capture the status first at the call site if it is also needed).
+pub(crate) async fn status_error(api: &str, resp: reqwest::Response) -> AppError {
+    let status = resp.status();
+    let body = resp.text().await.unwrap_or_default();
+    match status.as_u16() {
+        403 => AppError::AppView(format!(
+            "{api}: device key not yet registered or verified by the AppView (403)"
+        )),
+        422 => AppError::AppView(format!(
+            "{api}: request rejected, likely clock skew (422) — check the system clock"
+        )),
+        _ => AppError::AppView(format!("{api}: {status}: {body}")),
+    }
+}
+
+impl App {
+    /// The absolute URL of an `/api/v1/<path>` endpoint on the configured AppView.
+    pub(crate) fn appview_url(&self, path: &str) -> String {
+        format!("{}/api/v1/{path}", decodingus_appview_url())
+    }
+
+    /// POST a JSON body to an `/api/v1/<path>` endpoint and return the decoded response.
+    ///
+    /// The signature (and the DID it is over) belongs in `body` — these endpoints authenticate the
+    /// device key per call, not the HTTP request — so this deliberately takes an already-signed
+    /// body rather than signing on the caller's behalf: the canonical string differs per endpoint
+    /// and only the caller knows it.
+    pub(crate) async fn appview_post(
+        &self,
+        path: &str,
+        body: serde_json::Value,
+    ) -> Result<serde_json::Value, AppError> {
+        let resp = self
+            .auth
+            .http
+            .post(self.appview_url(path))
+            .json(&body)
+            .send()
+            .await
+            .map_err(transport)?;
+        if !resp.status().is_success() {
+            return Err(status_error(path, resp).await);
+        }
+        resp.json().await.map_err(transport)
+    }
+
+    /// Device-key-signed GET to an `/api/v1/<path>` endpoint, decoded into `T`.
+    ///
+    /// `build_msg(did, ts)` produces the canonical string to sign — the one thing that varies
+    /// between a poll, a thread read, and an exchange pull. `did`/`ts`/`sig` plus `extra` go on the
+    /// query; the timestamp is what makes the signature replay-guarded.
+    pub(crate) async fn appview_get_signed<T, F>(
+        &self,
+        path: &str,
+        build_msg: F,
+        extra: &[(&str, &str)],
+    ) -> Result<T, AppError>
+    where
+        T: serde::de::DeserializeOwned,
+        F: Fn(&str, i64) -> String,
+    {
+        let did = self.current_account().ok_or(AppError::NotAuthenticated)?;
+        let dev = self.ensure_device_key().await?;
+        let ts = Utc::now().timestamp();
+        let sig = dev.sign(&build_msg(&did, ts));
+        let ts_s = ts.to_string();
+        let mut query: Vec<(&str, &str)> = vec![("did", did.as_str()), ("ts", ts_s.as_str()), ("sig", sig.as_str())];
+        query.extend_from_slice(extra);
+        let resp = self
+            .auth
+            .http
+            .get(self.appview_url(path))
+            .query(&query)
+            .send()
+            .await
+            .map_err(transport)?;
+        if !resp.status().is_success() {
+            return Err(status_error(path, resp).await);
+        }
+        resp.json().await.map_err(transport)
+    }
+}

crates/navigator-app/src/commands.rs

@@ -67,13 +67,10 @@ impl App {
             }
         }
         let b = Biosample {
-            guid: SampleGuid(Uuid::new_v4()),
             sample_accession,
-            donor_identifier: donor_identifier.into(),
-            description: None,
-            center_name: None,
             sex,
             project_id,
+            ..Biosample::new(SampleGuid(Uuid::new_v4()), donor_identifier)
         };
         biosample::create(self.store.pool(), &b).await?;
         Ok(b)
@@ -378,9 +375,11 @@ impl App {
         for dna in ["Y", "Mt", "Auto"] {
             consensus_profile::delete(pool, biosample, dna).await?;
         }
-        consensus_painting::delete(pool, biosample).await?;
-        consensus_roh::delete(pool, biosample).await?;
-        consensus_archaic::delete(pool, biosample).await?;
+        // Every signature-keyed cache, from the one list — this used to name three of the four by
+        // hand and leave the Tier-B archaic segments behind, still keyed to a deleted alignment.
+        for cache in sig_cache::ALL {
+            cache.delete(pool, biosample).await?;
+        }
         // The audit log describes the consensus we just wiped; clear it so deleting the last run
         // can't leave a stale RUN_RECORDED history pointing at gone alignments. It is re-appended
         // when the consensus is next rebuilt from any remaining calls.

crates/navigator-app/src/haplogroup.rs

@@ -2473,7 +2473,7 @@ impl App {
     /// DecodingUs Y-DNA tree-with-variants JSON from our AppView (`/api/v1/y-tree/full`),
     /// host from [`decodingus_appview_url`]. On-disk cached like the FTDNA tree.
     pub(crate) async fn fetch_decodingus_y_tree(&self) -> Result<String, AppError> {
-        let url = format!("{}/api/v1/y-tree/full", decodingus_appview_url());
+        let url = self.appview_url("y-tree/full");
         self.fetch_tree(&url, "decodingus-ytree.json").await
     }
 
@@ -2483,7 +2483,7 @@ impl App {
     /// (~577, plus local indels), so callers must remap onto rCRS via [`mt_tree_rcrs`]. On-disk
     /// cached like the other trees.
     pub(crate) async fn fetch_decodingus_mt_tree(&self) -> Result<String, AppError> {
-        let url = format!("{}/api/v1/mt-tree/full", decodingus_appview_url());
+        let url = self.appview_url("mt-tree/full");
         self.fetch_tree(&url, "decodingus-mttree.json").await
     }
 
@@ -2564,7 +2564,7 @@ impl App {
     /// cached like the trees (7-day TTL + offline fallback). Looked up locally so a batch import
     /// makes one network call, not one per sample.
     async fn fetch_lab_instruments(&self) -> Result<Vec<SequencerLabInfo>, AppError> {
-        let url = format!("{}/api/v1/sequencer/lab-instruments", decodingus_appview_url());
+        let url = self.appview_url("sequencer/lab-instruments");
         let json = self.fetch_tree(&url, "sequencer-lab-instruments.json").await?;
         serde_json::from_str(&json).map_err(|e| AppError::Import(format!("parsing lab-instruments: {e}")))
     }
@@ -3324,13 +3324,13 @@ impl App {
         let Some(row) = consensus_profile::get(self.store.pool(), biosample_guid, "Auto").await? else {
             return Ok(None);
         };
-        let Some(p) = consensus_painting::get(self.store.pool(), biosample_guid).await? else {
+        let Some(p) = sig_cache::PAINTING.get(self.store.pool(), biosample_guid).await? else {
             return Ok(None);
         };
-        if p.consensus_sig != row.last_reconciled_at {
+        if p.sig != row.last_reconciled_at {
             return Ok(None); // painted from an older consensus — stale
         }
-        Ok(Some(parse_painting_json(&p.segments)?))
+        Ok(Some(parse_painting_json(&p.payload)?))
     }
 
     /// Paint each chromosome with local ancestry from the subject's **consensus** — no BAM walk. The
@@ -3448,14 +3448,15 @@ impl App {
         };
 
         // Cache keyed to the consensus signature so it's reused until the consensus is rebuilt.
-        consensus_painting::upsert(
-            self.store.pool(),
-            biosample_guid,
-            &sig,
-            &serde_json::to_string(&result)?,
-            &Utc::now().to_rfc3339(),
-        )
-        .await?;
+        sig_cache::PAINTING
+            .upsert(
+                self.store.pool(),
+                biosample_guid,
+                &sig,
+                &serde_json::to_string(&result)?,
+                &Utc::now().to_rfc3339(),
+            )
+            .await?;
         Ok(result)
     }
 
@@ -3538,11 +3539,11 @@ impl App {
         let Some(row) = consensus_profile::get(self.store.pool(), biosample_guid, "Auto").await? else {
             return Ok(None);
         };
-        let Some(r) = consensus_roh::get(self.store.pool(), biosample_guid).await? else {
+        let Some(r) = sig_cache::ROH.get(self.store.pool(), biosample_guid).await? else {
             return Ok(None);
         };
-        if r.consensus_sig == row.last_reconciled_at {
-            Ok(Some(serde_json::from_str(&r.roh)?))
+        if r.sig == row.last_reconciled_at {
+            Ok(Some(serde_json::from_str(&r.payload)?))
         } else {
             Ok(None) // computed from an older consensus — stale
         }
@@ -3561,9 +3562,9 @@ impl App {
         let sig = row.last_reconciled_at.clone();
 
         // Cache hit (same consensus signature) → return without recomputing.
-        if let Some(r) = consensus_roh::get(self.store.pool(), biosample_guid).await? {
-            if r.consensus_sig == sig {
-                return Ok(serde_json::from_str(&r.roh)?);
+        if let Some(r) = sig_cache::ROH.get(self.store.pool(), biosample_guid).await? {
+            if r.sig == sig {
+                return Ok(serde_json::from_str(&r.payload)?);
             }
         }
 
@@ -3583,14 +3584,15 @@ impl App {
         .await?;
 
         // Cache keyed to the consensus signature so it's reused until the consensus is rebuilt.
-        consensus_roh::upsert(
-            self.store.pool(),
-            biosample_guid,
-            &sig,
-            &serde_json::to_string(&result)?,
-            &Utc::now().to_rfc3339(),
-        )
-        .await?;
+        sig_cache::ROH
+            .upsert(
+                self.store.pool(),
+                biosample_guid,
+                &sig,
+                &serde_json::to_string(&result)?,
+                &Utc::now().to_rfc3339(),
+            )
+            .await?;
         Ok(result)
     }
 
@@ -3674,15 +3676,18 @@ impl App {
         if !crate::ARCHAIC_SEGMENTS_ENABLED {
             return Ok(None);
         }
-        let Some(row) = consensus_archaic_segments::get(self.store.pool(), biosample_guid).await? else {
+        let Some(row) = sig_cache::ARCHAIC_SEGMENTS
+            .get(self.store.pool(), biosample_guid)
+            .await?
+        else {
             return Ok(None);
         };
         let Some(aln) = self.alignment_with_diploid_calls(biosample_guid).await? else {
             return Ok(None);
         };
         let contigs = crate::called_diploid_contigs(&self.store, aln).await?;
-        if row.source_sig == archaic_segment_sig(aln, &contigs) {
-            Ok(Some(serde_json::from_str(&row.segments)?))
+        if row.sig == archaic_segment_sig(aln, &contigs) {
+            Ok(Some(serde_json::from_str(&row.payload)?))
         } else {
             Ok(None)
         }
@@ -3735,9 +3740,12 @@ impl App {
         // Computed from the contigs actually cached, so a later genome-wide pass invalidates a
         // partial result instead of inheriting it.
         let sig = archaic_segment_sig(aln, &crate::called_diploid_contigs(&self.store, aln).await?);
-        if let Some(row) = consensus_archaic_segments::get(self.store.pool(), biosample_guid).await? {
-            if row.source_sig == sig {
-                return Ok(serde_json::from_str(&row.segments)?);
+        if let Some(row) = sig_cache::ARCHAIC_SEGMENTS
+            .get(self.store.pool(), biosample_guid)
+            .await?
+        {
+            if row.sig == sig {
+                return Ok(serde_json::from_str(&row.payload)?);
             }
         }
 
@@ -3833,14 +3841,15 @@ impl App {
         })
         .await??;
 
-        consensus_archaic_segments::upsert(
-            self.store.pool(),
-            biosample_guid,
-            &sig,
-            &serde_json::to_string(&result)?,
-            &Utc::now().to_rfc3339(),
-        )
-        .await?;
+        sig_cache::ARCHAIC_SEGMENTS
+            .upsert(
+                self.store.pool(),
+                biosample_guid,
+                &sig,
+                &serde_json::to_string(&result)?,
+                &Utc::now().to_rfc3339(),
+            )
+            .await?;
         Ok(result)
     }
 
@@ -4013,13 +4022,13 @@ impl App {
         let Some(row) = consensus_profile::get(self.store.pool(), biosample_guid, "Auto").await? else {
             return Ok(None);
         };
-        let Some(r) = consensus_archaic::get(self.store.pool(), biosample_guid).await? else {
+        let Some(r) = sig_cache::ARCHAIC.get(self.store.pool(), biosample_guid).await? else {
             return Ok(None);
         };
         // Prefix match: the stored sig is "<consensus>:<panel16>", so a consensus change or a panel
         // rebuild both read as stale.
-        if r.consensus_sig.starts_with(&row.last_reconciled_at) {
-            Ok(Some(serde_json::from_str(&r.archaic)?))
+        if r.sig.starts_with(&row.last_reconciled_at) {
+            Ok(Some(serde_json::from_str(&r.payload)?))
         } else {
             Ok(None) // computed from an older consensus — stale
         }
@@ -4058,9 +4067,9 @@ impl App {
         let panel_fingerprint = navigator_analysis::manifest::sha256_hex(&bytes);
         let sig = format!("{}:{}", row.last_reconciled_at, &panel_fingerprint[..16]);
 
-        if let Some(r) = consensus_archaic::get(self.store.pool(), biosample_guid).await? {
-            if r.consensus_sig == sig {
-                return Ok(serde_json::from_str(&r.archaic)?);
+        if let Some(r) = sig_cache::ARCHAIC.get(self.store.pool(), biosample_guid).await? {
+            if r.sig == sig {
+                return Ok(serde_json::from_str(&r.payload)?);
             }
         }
         let panel = ArchaicMarkerPanel::from_bytes(&bytes)?;
@@ -4106,14 +4115,15 @@ impl App {
             result.cohort = Some(cohort);
         }
 
-        consensus_archaic::upsert(
-            self.store.pool(),
-            biosample_guid,
-            &sig,
-            &serde_json::to_string(&result)?,
-            &Utc::now().to_rfc3339(),
-        )
-        .await?;
+        sig_cache::ARCHAIC
+            .upsert(
+                self.store.pool(),
+                biosample_guid,
+                &sig,
+                &serde_json::to_string(&result)?,
+                &Utc::now().to_rfc3339(),
+            )
+            .await?;
         Ok(result)
     }