diff --git a/crates/buzz-cli/Cargo.toml b/crates/buzz-cli/Cargo.toml index 1476e60bfd4..59d1bb2cee6 100644 --- a/crates/buzz-cli/Cargo.toml +++ b/crates/buzz-cli/Cargo.toml @@ -91,3 +91,5 @@ rand = { workspace = true } tempfile = "3" # Minimal HTTP test server for retry/policy integration tests axum = { workspace = true } +# `test-util` enables paused-time control for deterministic timeout tests +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] } diff --git a/crates/buzz-cli/src/commands/channels.rs b/crates/buzz-cli/src/commands/channels.rs index 7ad051ef9fc..72168793588 100644 --- a/crates/buzz-cli/src/commands/channels.rs +++ b/crates/buzz-cli/src/commands/channels.rs @@ -1,6 +1,9 @@ -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; -use buzz_core::kind::{KIND_MANAGED_AGENT, KIND_TEAM}; +use buzz_core::kind::{ + KIND_MANAGED_AGENT, KIND_PRESENCE_SNAPSHOT, KIND_PRESENCE_UPDATE, KIND_TEAM, +}; +use chrono::DateTime; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -10,6 +13,7 @@ use crate::client::{ }; use crate::commands::agents::fetch_archived_snapshot; use crate::commands::channel_templates::{self, ChannelTemplateRecord, TemplateAgentRoster}; +use crate::commands::users::presence_subject; use crate::error::CliError; use crate::validate::{parse_uuid, read_or_stdin, validate_hex64, validate_uuid}; @@ -474,14 +478,318 @@ async fn scan_managed_agents_by_owner( Ok(found) } +/// Best-effort hints for a candidate agent pubkey, used to annotate the +/// duplicate-instance error. Gathered from relay presence and kind:0 lookups +/// before cardinality runs — both are optional so a lookup failure never +/// becomes a new failure mode. +#[derive(Debug, Clone, PartialEq, Eq)] +struct CandidateHint { + /// Latest presence status from kind:40902 (`"online"`, `"offline"`, or + /// whatever string the relay holds). `None` if the lookup failed or + /// returned no event. + presence: Option, + /// `created_at` timestamp from the agent's kind:0 profile event — the + /// time of the last profile update (kind:0 is replaceable; desktop + /// republishes it on rename and profile reconciliation). `None` if the + /// lookup failed or returned nothing. + profile_updated_at: Option, +} + +/// Fetch best-effort presence (kind:40902) and kind:0 metadata for each +/// pubkey in `pubkeys`. Each query is bounded *independently* by `timeout` and +/// the two outcomes are joined, so a lookup that completes survives a sibling +/// that hangs (see [`join_bounded_queries`]). Returns a map from pubkey to +/// hints; pubkeys with failed or absent lookups are absent from the map rather +/// than causing an error — callers must handle the missing-hint case. On +/// timeout or relay error, returns whatever partial hints were collected +/// (possibly an empty map) so the caller can still print bare pubkeys promptly. +/// +/// Only called when duplicate candidates have been detected: happy-path +/// resolutions perform zero hint queries. +async fn fetch_candidate_hints( + client: &BuzzClient, + pubkeys: &[String], + timeout: std::time::Duration, +) -> HashMap { + if pubkeys.is_empty() { + return HashMap::new(); + } + + // Presence: kind:40902, relay-synthesized on demand. + let presence_filter = serde_json::json!({ + "kinds": [KIND_PRESENCE_SNAPSHOT], + "authors": pubkeys, + "limit": pubkeys.len(), + }); + // Profile: kind:0 replaceable head per author. + let profile_filter = serde_json::json!({ + "kinds": [0], + "authors": pubkeys, + "limit": pubkeys.len(), + }); + + let (presence_result, profile_result) = join_bounded_queries( + timeout, + client.query(&presence_filter), + client.query(&profile_filter), + ) + .await; + + hints_from_results(pubkeys, presence_result, profile_result) +} + +/// Run two relay queries concurrently, bounding *each* independently by +/// `timeout` and joining the outcomes. A per-query timeout maps to `Err`, so a +/// completed lookup is never discarded because its sibling hung — the fail-soft +/// contract requires partial enrichment to survive. The whole call still +/// returns within `timeout` because neither branch can outlast it. +async fn join_bounded_queries( + timeout: std::time::Duration, + presence: P, + profile: Q, +) -> (Result, Result) +where + P: std::future::Future>, + Q: std::future::Future>, +{ + tokio::join!( + async { + tokio::time::timeout(timeout, presence) + .await + .unwrap_or_else(|_| Err(CliError::Other("presence hint timeout".to_string()))) + }, + async { + tokio::time::timeout(timeout, profile) + .await + .unwrap_or_else(|_| Err(CliError::Other("profile hint timeout".to_string()))) + }, + ) +} + +/// Convert the raw presence and profile query outcomes into a hint map. +/// +/// A presence response is trusted as a **complete snapshot** for the requested +/// `pubkeys` only when it parses as a JSON array in which *every* element is a +/// relay-synthesized presence event — a complete signed [`nostr::Event`] of +/// kind [`KIND_PRESENCE_UPDATE`] carrying exactly one `p` tag whose subject is +/// one of the requested `pubkeys` (see [`trusted_presence_snapshot`]). The +/// relay drops the Redis +/// presence key when an identity goes offline, so a trusted snapshot that omits +/// a requested pubkey means that pubkey is offline — exactly the stale +/// duplicate an operator needs flagged. Omitted pubkeys are therefore seeded as +/// `offline`, then returned statuses overlay the seed. +/// +/// Anything less than a fully trusted array — a failed/timed-out query, invalid +/// top-level JSON, or an array containing any element that is not such an event +/// (a vacuous object, `[{}]`, `[null]`, an event of the wrong kind, or one for +/// an unrequested subject) — makes presence enrichment untrusted: no offline +/// seeding and no presence labels at all. A completed profile sibling still +/// contributes its hints in that case. This refuses to invent an `offline` +/// label from a response we cannot trust (a relay-side fake-empty success or a +/// partially malformed body). Kept separate from IO so the trust boundary is +/// directly unit-testable without a relay. +fn hints_from_results( + pubkeys: &[String], + presence_result: Result, + profile_result: Result, +) -> HashMap { + let (offline_seed, presence_events): (&[String], Vec) = + match trusted_presence_snapshot(pubkeys, presence_result) { + Some(events) => (pubkeys, events), + None => (&[], Vec::new()), + }; + let profile_events: Vec = profile_result + .ok() + .and_then(|r| serde_json::from_str(&r).ok()) + .unwrap_or_default(); + + build_hint_map(offline_seed, &presence_events, &profile_events) +} + +/// Validate a presence query outcome as a trustworthy complete snapshot. +/// +/// Returns the parsed events only when the body parses as a JSON array and +/// *every* element is a relay-synthesized presence snapshot for the requested +/// set: a complete, well-formed [`nostr::Event`] of kind +/// [`KIND_PRESENCE_UPDATE`] carrying exactly one `p` tag whose subject is one of +/// `pubkeys`. A failed query, non-array JSON, or any element that is not such +/// an event yields `None` — the caller must then treat presence as untrusted +/// and never infer `offline`. +/// +/// Parsing each element as a full event (not just checking two fields) is what +/// stops a vacuous object like `{"pubkey":"…","content":"online"}` — which +/// lacks `id`/`sig`/`kind`/`created_at` — from masquerading as a snapshot; the +/// kind check rejects a fully-shaped event of the wrong kind; and validating the +/// *sole* `p`-tag subject (the exact value the consumer reads) rejects an event +/// for an unrequested subject as well as a mixed-tag event that would pass a +/// weaker "any `p` tag is requested" check yet overlay a different subject +/// downstream. Any of these would otherwise re-enable false `offline` seeding +/// from an untrustworthy body. +fn trusted_presence_snapshot( + pubkeys: &[String], + presence_result: Result, +) -> Option> { + let events: Vec = presence_result + .ok() + .and_then(|r| serde_json::from_str(&r).ok())?; + let requested: HashSet<&str> = pubkeys.iter().map(String::as_str).collect(); + let all_trusted = events.iter().all(|value| { + // Must parse as a complete signed event of the presence-update kind. + let Ok(event) = serde_json::from_value::(value.clone()) else { + return false; + }; + event.kind == nostr::Kind::Custom(KIND_PRESENCE_UPDATE as u16) + // Require the *sole* `p`-tag subject — the one `build_hint_map` + // consumes via `presence_subject` — to be requested. Reading the + // same single subject the consumer reads is what prevents a + // mixed-tag event (`[["p",""],["p",""]]`) + // from passing here yet overlaying a different subject downstream. + && sole_p_tag_subject(value).is_some_and(|s| requested.contains(s)) + }); + all_trusted.then_some(events) +} + +/// The subject of the event's single `p` tag, or `None` unless there is exactly +/// one `p` tag carrying a string subject. The relay synthesizes presence +/// snapshots with exactly one `p` tag (the subject); requiring exactly one keeps +/// this validator reading the same subject that `presence_subject` (which takes +/// the first `p` tag) consumes in `build_hint_map`, so a mixed- or +/// malformed-tag event cannot pass validation and then overlay a different +/// subject. +fn sole_p_tag_subject(event: &serde_json::Value) -> Option<&str> { + let tags = event.get("tags")?.as_array()?; + let mut p_subjects = tags + .iter() + .filter_map(|tag| match tag.as_array()?.as_slice() { + [name, subject, ..] if name == "p" => Some(subject.as_str()), + _ => None, + }); + let first = p_subjects.next()?; + if p_subjects.next().is_some() { + return None; // more than one `p` tag → outside the single-subject contract + } + first // the sole `p` tag's subject, or `None` if it was not a string +} + +/// Pure response-to-map conversion: takes the raw presence (kind:40902) and +/// profile (kind:0) event slices returned by the relay and builds the +/// per-pubkey hint map. Extracted as a sync function so it is directly +/// unit-testable without a relay. +/// +/// `offline_seed` names the pubkeys whose presence was requested via a +/// response the caller trusts as a complete snapshot; each is pre-labeled +/// `offline` before overlaying returned statuses, so a duplicate the relay +/// omitted (its Redis key was dropped on going offline) is still flagged +/// `offline` rather than left blank. Pass an empty slice when the presence +/// response failed, timed out, or was malformed — never infer offline then. +/// +/// Presence subject is the `p`-tag value when present (relay signs the event +/// and embeds the agent pubkey there), otherwise the event author. +fn build_hint_map( + offline_seed: &[String], + presence_events: &[serde_json::Value], + profile_events: &[serde_json::Value], +) -> HashMap { + let mut hints: HashMap = HashMap::new(); + + // Seed requested pubkeys as offline: a trusted snapshot that omits a + // requested pubkey means that identity is offline. + for pubkey in offline_seed { + hints + .entry(pubkey.clone()) + .or_insert(CandidateHint { + presence: None, + profile_updated_at: None, + }) + .presence = Some("offline".to_string()); + } + + for event in presence_events { + let subject = presence_subject(event).to_string(); + if subject.is_empty() { + continue; + } + let status = event + .get("content") + .and_then(|v| v.as_str()) + .map(str::to_string); + // Only overlay a real status string; a returned event with no readable + // content must not erase an offline seed for the same pubkey. + if let Some(status) = status { + hints + .entry(subject) + .or_insert(CandidateHint { + presence: None, + profile_updated_at: None, + }) + .presence = Some(status); + } + } + + for event in profile_events { + let Some(pubkey) = event + .get("pubkey") + .and_then(|v| v.as_str()) + .map(str::to_string) + else { + continue; + }; + let profile_updated_at = event.get("created_at").and_then(|v| v.as_u64()); + hints + .entry(pubkey) + .or_insert(CandidateHint { + presence: None, + profile_updated_at: None, + }) + .profile_updated_at = profile_updated_at; + } + + hints +} + +/// Format a single candidate pubkey for the duplicate-instance error, +/// appending available hint fields in brackets. Pure and testable. +/// +/// Examples: +/// - `"aaa…bbb [online, profile updated 2024-01-15]"` +/// - `"aaa…bbb [offline]"` +/// - `"aaa…bbb [profile updated 2024-01-15]"` +/// - `"aaa…bbb"` (no hint at all) +fn format_candidate(pubkey: &str, hint: Option<&CandidateHint>) -> String { + let Some(h) = hint else { + return pubkey.to_string(); + }; + let mut parts: Vec = Vec::new(); + if let Some(status) = &h.presence { + parts.push(status.clone()); + } + if let Some(ts) = h.profile_updated_at { + // Use chrono for safe conversion; omit the date if the timestamp is + // out of range rather than panicking or printing garbage. + if let Some(dt) = DateTime::from_timestamp(ts as i64, 0) { + parts.push(format!("profile updated {}", dt.format("%Y-%m-%d"))); + } + } + if parts.is_empty() { + pubkey.to_string() + } else { + format!("{pubkey} [{}]", parts.join(", ")) + } +} + /// Apply the F4 cardinality rule per persona slug: zero live instances is a /// known skip (cold-start provisioning is desktop-only, out of scope), one is /// added, more than one is a hard error listing candidate pubkeys — matching /// all instances silently would risk adding a stale or wrong instance. Pure /// and independent of the relay so it's directly unit-testable. +/// +/// `hints` is best-effort decoration gathered by the async caller before this +/// function runs: absent entries are silently omitted from the error, never a +/// new failure mode. fn apply_cardinality_rule( slugs: &[String], found: &[ResolvedAgent], + hints: &HashMap, ) -> Result { let mut agents = Vec::new(); let mut skipped = Vec::new(); @@ -491,7 +799,10 @@ fn apply_cardinality_rule( [] => skipped.push(slug.clone()), [one] => agents.push((*one).clone()), many => { - let candidates: Vec<&str> = many.iter().map(|a| a.pubkey.as_str()).collect(); + let candidates: Vec = many + .iter() + .map(|a| format_candidate(&a.pubkey, hints.get(&a.pubkey))) + .collect(); return Err(CliError::Usage(format!( "persona '{slug}' has {} live instances for this owner ({}); \ pass a template with a single instance per persona, or resolve \ @@ -531,6 +842,7 @@ fn resolve_roster_with_archive_filter( slugs: &[String], found: Vec, archived_result: Result, CliError>, + hints: &HashMap, ) -> Result { let (archived, archive_state_warning) = match archived_result { Ok(pubkeys) => (pubkeys.into_iter().collect::>(), None), @@ -550,7 +862,7 @@ fn resolve_roster_with_archive_filter( } } - let resolved = apply_cardinality_rule(slugs, &live_found).map_err(|e| { + let resolved = apply_cardinality_rule(slugs, &live_found, hints).map_err(|e| { match (e, &archive_state_warning) { (CliError::Usage(msg), Some(warning)) => { CliError::Usage(format!("{msg} (warning: {warning})")) @@ -594,13 +906,74 @@ fn finalize_roster_resolution( slugs: &[String], found: Vec, archived_result: Result, CliError>, + hints: &HashMap, warn_sink: &mut dyn std::io::Write, ) -> Result { if let Err(e) = &archived_result { let warning = archive_snapshot_warning(e); let _ = writeln!(warn_sink, "{}", serde_json::json!({"warning": warning})); } - resolve_roster_with_archive_filter(slugs, found, archived_result) + resolve_roster_with_archive_filter(slugs, found, archived_result, hints) +} + +/// Post-fetch stage of [`build_roster_resolution`]: given the already-fetched +/// `found` and `archived_result`, identifies duplicate live instances, calls +/// `fetch_hints` only for their pubkeys, then delegates to +/// [`finalize_roster_resolution`]. +/// +/// Accepting `fetch_hints` as a generic async closure makes this function +/// directly testable without a relay: tests pass a recording closure that +/// asserts the exact pubkey set and returns a controlled hint map. +/// +/// - **Happy path** (no duplicates): `fetch_hints` is never called. +/// - **Trusted archive archives one of a pair**: only the surviving live pair +/// triggers `fetch_hints`; archived instances are not fetched for. +/// - **Untrusted archive** (`archived_result: Err`): all found instances are +/// conservatively treated as live for duplicate detection. +async fn assemble_roster_resolution( + slugs: &[String], + found: Vec, + archived_result: Result, CliError>, + fetch_hints: F, + warn_sink: &mut dyn std::io::Write, +) -> Result +where + F: FnOnce(Vec) -> Fut, + Fut: std::future::Future>, +{ + // Determine which pubkeys belong to duplicate live instances after archive + // filtering. Untrusted archive (Err) → empty archived set → conservative. + let duplicate_pubkeys: Vec = { + let archived_set: HashSet<&str> = match &archived_result { + Ok(keys) => keys.iter().map(String::as_str).collect(), + Err(_) => HashSet::new(), + }; + let live: Vec<&ResolvedAgent> = found + .iter() + .filter(|a| !archived_set.contains(a.pubkey.as_str())) + .collect(); + let mut slug_count: HashMap<&str, Vec<&str>> = HashMap::new(); + for a in &live { + slug_count + .entry(a.persona_id.as_str()) + .or_default() + .push(a.pubkey.as_str()); + } + slug_count + .into_values() + .filter(|pks| pks.len() > 1) + .flatten() + .map(str::to_string) + .collect() + }; + + let hints = if duplicate_pubkeys.is_empty() { + HashMap::new() + } else { + fetch_hints(duplicate_pubkeys).await + }; + + finalize_roster_resolution(slugs, found, archived_result, &hints, warn_sink) } /// Resolve a template's roster against the relay: expand team entries into @@ -610,6 +983,11 @@ fn finalize_roster_resolution( /// for the pure filter+cardinality core and the fail-open contract). Runs /// entirely before any channel-creation side effect — a cardinality error /// aborts with nothing created. +/// +/// Hint fetching is zero-cost on the happy path: [`assemble_roster_resolution`] +/// only invokes the hint fetcher when duplicate live instances are detected +/// after archive filtering. Queries run concurrently and are bounded by a +/// 3-second timeout; on expiry the error prints with bare pubkeys. async fn build_roster_resolution( client: &BuzzClient, owner: &str, @@ -640,10 +1018,22 @@ async fn build_roster_resolution( } let slug_set: HashSet<&str> = slugs.iter().map(String::as_str).collect(); - let found = scan_managed_agents_by_owner(client, owner, &slug_set).await?; - - let archived_result = fetch_archived_snapshot(client).await; - finalize_roster_resolution(&slugs, found, archived_result, &mut std::io::stderr()) + let (found, archived_result) = tokio::join!( + scan_managed_agents_by_owner(client, owner, &slug_set), + fetch_archived_snapshot(client), + ); + let found = found?; + + assemble_roster_resolution( + &slugs, + found, + archived_result, + |pks| async move { + fetch_candidate_hints(client, &pks, std::time::Duration::from_secs(3)).await + }, + &mut std::io::stderr(), + ) + .await } /// `buzz channels create --template `: load a desktop-local channel @@ -1196,19 +1586,32 @@ pub async fn dispatch_canvas(cmd: crate::CanvasCmd, client: &BuzzClient) -> Resu #[cfg(test)] mod tests { use super::{ - apply_cardinality_rule, build_template_report, cmd_set_add_policy, - finalize_roster_resolution, name_matches, resolve_roster_with_archive_filter, - validate_ttl_seconds, validate_update_channel_fields, ArchivedExclusion, ChannelSummary, - ResolvedAgent, RosterResolution, SkippedSlug, + apply_cardinality_rule, assemble_roster_resolution, build_hint_map, build_template_report, + cmd_set_add_policy, fetch_candidate_hints, finalize_roster_resolution, format_candidate, + hints_from_results, join_bounded_queries, name_matches, resolve_roster_with_archive_filter, + validate_ttl_seconds, validate_update_channel_fields, ArchivedExclusion, CandidateHint, + ChannelSummary, ResolvedAgent, RosterResolution, SkippedSlug, }; use crate::client::BuzzClient; use crate::CliError; use serde_json::json; + use std::collections::HashMap; fn event(tags: serde_json::Value) -> serde_json::Value { json!({ "tags": tags }) } + fn no_hints() -> HashMap { + HashMap::new() + } + + fn hint(presence: Option<&str>, profile_updated_at: Option) -> CandidateHint { + CandidateHint { + presence: presence.map(str::to_string), + profile_updated_at, + } + } + #[test] fn from_event_extracts_known_tags() { let ev = event(json!([ @@ -1429,7 +1832,8 @@ mod tests { #[test] fn cardinality_zero_instances_is_skipped_not_error() { let slugs = vec!["builtin:fizz".to_string()]; - let resolved = apply_cardinality_rule(&slugs, &[]).expect("zero instances is not fatal"); + let resolved = + apply_cardinality_rule(&slugs, &[], &no_hints()).expect("zero instances is not fatal"); assert!(resolved.agents.is_empty()); assert_eq!(resolved.skipped, vec!["builtin:fizz".to_string()]); } @@ -1438,7 +1842,8 @@ mod tests { fn cardinality_one_instance_is_added() { let slugs = vec!["builtin:fizz".to_string()]; let found = vec![agent("builtin:fizz", "a".repeat(64).as_str())]; - let resolved = apply_cardinality_rule(&slugs, &found).expect("single instance resolves"); + let resolved = + apply_cardinality_rule(&slugs, &found, &no_hints()).expect("single instance resolves"); assert_eq!(resolved.agents.len(), 1); assert_eq!(resolved.agents[0].persona_id, "builtin:fizz"); assert!(resolved.skipped.is_empty()); @@ -1451,7 +1856,7 @@ mod tests { agent("builtin:fizz", &"a".repeat(64)), agent("builtin:fizz", &"b".repeat(64)), ]; - let err = apply_cardinality_rule(&slugs, &found).unwrap_err(); + let err = apply_cardinality_rule(&slugs, &found, &no_hints()).unwrap_err(); assert!(matches!(err, CliError::Usage(_))); let msg = err.to_string(); assert!(msg.contains("builtin:fizz")); @@ -1475,13 +1880,14 @@ mod tests { agent("builtin:duplicated", &"b".repeat(64)), agent("builtin:duplicated", &"c".repeat(64)), ]; - let err = apply_cardinality_rule(&slugs, &found).unwrap_err(); + let err = apply_cardinality_rule(&slugs, &found, &no_hints()).unwrap_err(); assert!(err.to_string().contains("builtin:duplicated")); } #[test] fn cardinality_empty_roster_resolves_to_empty_lists() { - let resolved = apply_cardinality_rule(&[], &[]).expect("empty roster is not fatal"); + let resolved = + apply_cardinality_rule(&[], &[], &no_hints()).expect("empty roster is not fatal"); assert!(resolved.agents.is_empty()); assert!(resolved.skipped.is_empty()); } @@ -1495,7 +1901,7 @@ mod tests { agent("builtin:fizz", &"a".repeat(64)), agent("builtin:unrelated", &"z".repeat(64)), ]; - let resolved = apply_cardinality_rule(&slugs, &found).expect("resolves"); + let resolved = apply_cardinality_rule(&slugs, &found, &no_hints()).expect("resolves"); assert_eq!(resolved.agents.len(), 1); assert_eq!(resolved.agents[0].persona_id, "builtin:fizz"); } @@ -1514,9 +1920,13 @@ mod tests { agent("builtin:fizz", &live_pk), agent("builtin:fizz", &archived_pk), ]; - let resolution = - resolve_roster_with_archive_filter(&slugs, found, Ok(vec![archived_pk.clone()])) - .expect("resolves to the single live instance"); + let resolution = resolve_roster_with_archive_filter( + &slugs, + found, + Ok(vec![archived_pk.clone()]), + &no_hints(), + ) + .expect("resolves to the single live instance"); assert_eq!(resolution.agents.len(), 1); assert_eq!(resolution.agents[0].pubkey, live_pk); assert!(resolution.skipped.is_empty()); @@ -1539,9 +1949,13 @@ mod tests { let pk1 = "a".repeat(64); let pk2 = "b".repeat(64); let found = vec![agent("builtin:fizz", &pk1), agent("builtin:fizz", &pk2)]; - let resolution = - resolve_roster_with_archive_filter(&slugs, found, Ok(vec![pk1.clone(), pk2.clone()])) - .expect("all-archived is a skip, not an error"); + let resolution = resolve_roster_with_archive_filter( + &slugs, + found, + Ok(vec![pk1.clone(), pk2.clone()]), + &no_hints(), + ) + .expect("all-archived is a skip, not an error"); assert!(resolution.agents.is_empty()); assert_eq!( resolution.skipped, @@ -1558,8 +1972,9 @@ mod tests { // Zero live instances (nothing to archive) must not be confused // with "all instances archived" — no exclusions were made. let slugs = vec!["builtin:fizz".to_string()]; - let resolution = resolve_roster_with_archive_filter(&slugs, vec![], Ok(vec![])) - .expect("zero instances is not fatal"); + let resolution = + resolve_roster_with_archive_filter(&slugs, vec![], Ok(vec![]), &no_hints()) + .expect("zero instances is not fatal"); assert!(resolution.agents.is_empty()); assert_eq!( resolution.skipped, @@ -1580,8 +1995,9 @@ mod tests { let pk = "a".repeat(64); let found = vec![agent("builtin:fizz", &pk)]; let archived_err = CliError::Other("relay info document missing 'self' field".into()); - let resolution = resolve_roster_with_archive_filter(&slugs, found, Err(archived_err)) - .expect("fails open — resolution still succeeds"); + let resolution = + resolve_roster_with_archive_filter(&slugs, found, Err(archived_err), &no_hints()) + .expect("fails open — resolution still succeeds"); assert_eq!(resolution.agents.len(), 1); assert_eq!(resolution.agents[0].pubkey, pk); assert!(resolution.archived_excluded.is_empty()); @@ -1604,7 +2020,7 @@ mod tests { agent("builtin:fizz", &"b".repeat(64)), ]; let archived_err = CliError::Other("query failure".into()); - let err = resolve_roster_with_archive_filter(&slugs, found, Err(archived_err)) + let err = resolve_roster_with_archive_filter(&slugs, found, Err(archived_err), &no_hints()) .expect_err("ambiguity error must still propagate"); assert!(matches!(err, CliError::Usage(_))); let msg = err.to_string(); @@ -1626,7 +2042,7 @@ mod tests { let slugs = vec!["builtin:fizz".to_string()]; let pk = "a".repeat(64); let found = vec![agent("builtin:fizz", &pk)]; - let resolution = resolve_roster_with_archive_filter(&slugs, found, Ok(vec![])) + let resolution = resolve_roster_with_archive_filter(&slugs, found, Ok(vec![]), &no_hints()) .expect("resolves with nothing archived"); assert!(resolution.archived_excluded.is_empty()); let serialized = serde_json::to_value(&resolution.archived_excluded).unwrap(); @@ -1656,8 +2072,9 @@ mod tests { let found = vec![agent("builtin:fizz", &pk)]; let archived_err = CliError::Other("relay info document missing 'self' field".into()); let mut sink: Vec = Vec::new(); - let resolution = finalize_roster_resolution(&slugs, found, Err(archived_err), &mut sink) - .expect("fails open — resolution still succeeds"); + let resolution = + finalize_roster_resolution(&slugs, found, Err(archived_err), &no_hints(), &mut sink) + .expect("fails open — resolution still succeeds"); let sink_text = String::from_utf8(sink).expect("sink is UTF-8"); let lines: Vec<&str> = sink_text.lines().collect(); @@ -1700,8 +2117,9 @@ mod tests { ]; let archived_err = CliError::Other("query failure".into()); let mut sink: Vec = Vec::new(); - let err = finalize_roster_resolution(&slugs, found, Err(archived_err), &mut sink) - .expect_err("ambiguity error must still propagate"); + let err = + finalize_roster_resolution(&slugs, found, Err(archived_err), &no_hints(), &mut sink) + .expect_err("ambiguity error must still propagate"); let sink_text = String::from_utf8(sink).expect("sink is UTF-8"); assert_eq!( @@ -1747,4 +2165,768 @@ mod tests { "no warning key expected: {report}" ); } + + // --- Candidate hint formatting --- + + #[test] + fn format_candidate_no_hint_returns_bare_pubkey() { + let pk = "a".repeat(64); + assert_eq!(format_candidate(&pk, None), pk); + } + + #[test] + fn format_candidate_presence_only_appends_status() { + let pk = "a".repeat(64); + let h = hint(Some("offline"), None); + let formatted = format_candidate(&pk, Some(&h)); + assert!(formatted.contains(&pk), "pubkey must appear: {formatted}"); + assert!( + formatted.contains("[offline]"), + "presence status must appear: {formatted}" + ); + } + + #[test] + fn format_candidate_provisioned_at_only_appends_date() { + let pk = "b".repeat(64); + // 2024-01-15 = 1705276800 seconds since epoch + let h = hint(None, Some(1_705_276_800)); + let formatted = format_candidate(&pk, Some(&h)); + assert!(formatted.contains(&pk), "pubkey must appear: {formatted}"); + assert!( + formatted.contains("profile updated 2024-01-15"), + "date must appear: {formatted}" + ); + } + + #[test] + fn format_candidate_both_hints_appends_both() { + let pk = "c".repeat(64); + let h = hint(Some("online"), Some(1_705_276_800)); + let formatted = format_candidate(&pk, Some(&h)); + assert!(formatted.contains(&pk), "pubkey must appear: {formatted}"); + assert!( + formatted.contains("online"), + "presence must appear: {formatted}" + ); + assert!( + formatted.contains("profile updated 2024-01-15"), + "date must appear: {formatted}" + ); + } + + #[test] + fn format_candidate_empty_hint_fields_returns_bare_pubkey() { + // Both hint fields None — same output as no hint at all. + let pk = "d".repeat(64); + let h = hint(None, None); + assert_eq!(format_candidate(&pk, Some(&h)), pk); + } + + #[test] + fn cardinality_error_includes_hint_when_provided() { + // When hints are present, the duplicate-instance error must include + // the presence and provisioned-at decoration in its candidate list. + let pk_a = "a".repeat(64); + let pk_b = "b".repeat(64); + let slugs = vec!["builtin:fizz".to_string()]; + let found = vec![agent("builtin:fizz", &pk_a), agent("builtin:fizz", &pk_b)]; + let mut hints = HashMap::new(); + hints.insert(pk_a.clone(), hint(Some("offline"), Some(1_705_276_800))); + hints.insert(pk_b.clone(), hint(Some("online"), None)); + + let err = apply_cardinality_rule(&slugs, &found, &hints).unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains(&pk_a), "pk_a must appear: {msg}"); + assert!(msg.contains(&pk_b), "pk_b must appear: {msg}"); + assert!(msg.contains("offline"), "offline status must appear: {msg}"); + assert!( + msg.contains("profile updated 2024-01-15"), + "provisioned date must appear: {msg}" + ); + assert!(msg.contains("online"), "online status must appear: {msg}"); + } + + #[test] + fn cardinality_error_falls_back_to_bare_pubkey_when_hint_missing() { + // A missing hint entry in the map must not cause a panic or omit + // the pubkey from the error — it must print as a bare pubkey. + let pk_a = "a".repeat(64); + let pk_b = "b".repeat(64); + let slugs = vec!["builtin:fizz".to_string()]; + let found = vec![agent("builtin:fizz", &pk_a), agent("builtin:fizz", &pk_b)]; + // Only pk_a has a hint; pk_b is absent from the map. + let mut hints = HashMap::new(); + hints.insert(pk_a.clone(), hint(Some("offline"), None)); + + let err = apply_cardinality_rule(&slugs, &found, &hints).unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains(&pk_a), "pk_a must appear: {msg}"); + assert!( + msg.contains(&pk_b), + "pk_b must appear as bare pubkey: {msg}" + ); + // pk_b has no hint — it must not appear as "[online]" or "[offline]" + // but must still appear in the candidate list. + assert!( + !msg.contains(&format!("{pk_b} [")), + "pk_b must not have hint brackets: {msg}" + ); + } + + // --- build_hint_map boundary tests --- + + #[test] + fn build_hint_map_uses_p_tag_over_author_for_presence() { + // Relay signs presence events with its own key; the agent pubkey is in + // the `p` tag. The relay author must NOT be used as the map key. + let relay_pk = "r".repeat(64); + let agent_pk = "a".repeat(64); + let presence = vec![json!({ + "pubkey": relay_pk, + "content": "online", + "tags": [["p", agent_pk]], + })]; + let map = build_hint_map(&[], &presence, &[]); + assert!( + !map.contains_key(&relay_pk), + "relay author must not be the key: {map:?}" + ); + assert!( + map.contains_key(&agent_pk), + "agent p-tag must be key: {map:?}" + ); + assert_eq!( + map[&agent_pk].presence.as_deref(), + Some("online"), + "presence status preserved" + ); + } + + #[test] + fn build_hint_map_presence_failure_profile_survives() { + // If presence lookup fails (empty slice), profile hints must still be + // populated from the profile events alone. + let pk = "b".repeat(64); + let profile = vec![json!({ + "pubkey": pk, + "created_at": 1_705_276_800_u64, + })]; + let map = build_hint_map(&[], &[], &profile); + assert!(map.contains_key(&pk), "pubkey must be in map: {map:?}"); + assert_eq!( + map[&pk].profile_updated_at, + Some(1_705_276_800), + "profile timestamp preserved" + ); + assert!( + map[&pk].presence.is_none(), + "presence must be absent when lookup failed" + ); + } + + #[test] + fn build_hint_map_profile_failure_presence_survives() { + // If profile lookup fails (empty slice), presence hints must still be + // populated from the presence events alone. + let pk = "c".repeat(64); + let presence = vec![json!({ + "pubkey": pk, + "content": "offline", + "tags": [], + })]; + let map = build_hint_map(&[], &presence, &[]); + assert!(map.contains_key(&pk), "pubkey must be in map: {map:?}"); + assert_eq!( + map[&pk].presence.as_deref(), + Some("offline"), + "presence status preserved" + ); + assert!( + map[&pk].profile_updated_at.is_none(), + "profile_updated_at must be absent when lookup failed" + ); + } + + #[test] + fn build_hint_map_malformed_entries_are_skipped() { + // Presence events missing both pubkey and p-tag are skipped without + // panicking; profile events missing pubkey are skipped too. + let malformed_presence = vec![ + json!({"content": "online"}), // no pubkey, no p-tag + json!({"pubkey": null, "content": "online", "tags": []}), + ]; + let malformed_profile = vec![ + json!({"created_at": 1_705_276_800_u64}), // no pubkey + json!({"pubkey": null, "created_at": 1_705_276_800_u64}), + ]; + let map = build_hint_map(&[], &malformed_presence, &malformed_profile); + assert!( + map.is_empty(), + "malformed entries must yield empty map: {map:?}" + ); + } + + #[test] + fn build_hint_map_both_failures_yield_empty_map() { + // Both slices empty simulates a total timeout / relay error. + let map = build_hint_map(&[], &[], &[]); + assert!(map.is_empty(), "empty inputs must yield empty map"); + } + + // --- assemble_roster_resolution wiring tests --- + // These tests exercise the conditional-fetch logic directly, proving: + // (a) the fetcher is called only when duplicate live instances exist, and + // (b) the exact pubkey set passed to the fetcher matches the live duplicates. + // Using a recording closure instead of a real relay means these run + // synchronously fast and catch the wiring even without a relay. + + /// Helper: make a `ResolvedAgent` with the given persona and pubkey. + fn owned_agent(persona_id: &str, pubkey: &str) -> ResolvedAgent { + ResolvedAgent { + persona_id: persona_id.to_string(), + pubkey: pubkey.to_string(), + } + } + + #[tokio::test] + async fn assemble_roster_resolution_duplicate_pair_invokes_fetcher_with_their_pubkeys() { + // Two live instances for the same slug — fetcher must be called with + // exactly those two pubkeys. + let pk_a = "a".repeat(64); + let pk_b = "b".repeat(64); + let slugs = vec!["sietch:agent".to_string()]; + let found = vec![ + owned_agent("sietch:agent", &pk_a), + owned_agent("sietch:agent", &pk_b), + ]; + + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + let fetcher_invoked = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&fetcher_invoked); + let result = assemble_roster_resolution( + &slugs, + found, + Ok(vec![]), // trusted empty archive: both are live + |pks| async move { + flag.store(true, Ordering::Relaxed); + // Verify the fetcher receives exactly the duplicate pubkeys. + let mut sorted = pks.clone(); + sorted.sort(); + assert_eq!(sorted.len(), 2, "exactly 2 duplicate pubkeys expected"); + HashMap::new() + }, + &mut std::io::sink(), + ) + .await; + + assert!( + fetcher_invoked.load(Ordering::Relaxed), + "fetcher must be called for a duplicate pair" + ); + // Both pubkeys appear in the cardinality error (bare, since the fetcher returned empty). + let err = result.unwrap_err().to_string(); + assert!(err.contains(&pk_a), "pk_a must appear in error: {err}"); + assert!(err.contains(&pk_b), "pk_b must appear in error: {err}"); + } + + #[tokio::test] + async fn assemble_roster_resolution_single_instance_never_invokes_fetcher() { + // All slugs have exactly one live instance — fetcher must NOT be called. + // If it is called, the `panic!` fires. + let pk = "c".repeat(64); + let slugs = vec!["sietch:agent".to_string()]; + let found = vec![owned_agent("sietch:agent", &pk)]; + + let result = assemble_roster_resolution( + &slugs, + found, + Ok(vec![]), + |_pks| async move { + panic!("fetcher must not be called on a single-instance roster"); + #[allow(unreachable_code)] + HashMap::::new() + }, + &mut std::io::sink(), + ) + .await; + + assert!( + result.is_ok(), + "single instance resolves cleanly: {result:?}" + ); + } + + #[tokio::test] + async fn assemble_roster_resolution_trusted_archive_removes_duplicate_suppresses_fetcher() { + // pk_a is archived. Only pk_b remains live — no duplicate, so the + // fetcher must NOT be called. + let pk_a = "d".repeat(64); + let pk_b = "e".repeat(64); + let slugs = vec!["sietch:agent".to_string()]; + let found = vec![ + owned_agent("sietch:agent", &pk_a), + owned_agent("sietch:agent", &pk_b), + ]; + + let result = assemble_roster_resolution( + &slugs, + found, + Ok(vec![pk_a.clone()]), // pk_a archived + |_pks| async move { + panic!("fetcher must not be called when archive resolves the duplicate"); + #[allow(unreachable_code)] + HashMap::::new() + }, + &mut std::io::sink(), + ) + .await; + + assert!( + result.is_ok(), + "archive resolves duplicate cleanly: {result:?}" + ); + } + + #[tokio::test] + async fn assemble_roster_resolution_untrusted_archive_invokes_fetcher_conservatively() { + // Archive snapshot is Err (untrusted). Both instances are treated as + // live conservatively → fetcher must be called. + let pk_a = "f".repeat(64); + let pk_b = "g".repeat(64); + let slugs = vec!["sietch:agent".to_string()]; + let found = vec![ + owned_agent("sietch:agent", &pk_a), + owned_agent("sietch:agent", &pk_b), + ]; + + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + let fetcher_invoked = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&fetcher_invoked); + let result = assemble_roster_resolution( + &slugs, + found, + Err(CliError::Other("snapshot unavailable".to_string())), + |pks| async move { + flag.store(true, Ordering::Relaxed); + let _ = pks; + HashMap::::new() + }, + &mut std::io::sink(), + ) + .await; + + assert!( + fetcher_invoked.load(Ordering::Relaxed), + "fetcher must be called under untrusted archive" + ); + // Error still surfaces (bare pubkeys, plus the archive warning embedded). + assert!( + result.is_err(), + "untrusted archive + duplicates is still an error" + ); + } + + // --- hints_from_results offline-seeding boundary --- + // A successful presence snapshot is complete: the relay drops the Redis + // presence key on offline, so a requested pubkey the snapshot omits is + // offline. A failed/malformed presence response must NOT infer offline. + + /// Serialize presence/profile events the way the relay returns them. + fn events_json(events: &[serde_json::Value]) -> String { + serde_json::to_string(events).unwrap() + } + + /// Build a relay-shaped presence snapshot event: a real signed + /// `nostr::Event` of kind `KIND_PRESENCE_UPDATE` whose `p` tag names + /// `subject`, matching exactly what `synthesize_presence` produces. Signed + /// by an arbitrary "relay" key so its author differs from the subject. + fn presence_event(subject: &str, status: &str) -> serde_json::Value { + let relay_keys = + nostr::Keys::parse("0000000000000000000000000000000000000000000000000000000000000002") + .expect("valid relay test key"); + let event = nostr::EventBuilder::new( + nostr::Kind::Custom(buzz_core::kind::KIND_PRESENCE_UPDATE as u16), + status, + ) + .tags([nostr::Tag::parse(["p", subject]).expect("valid p tag")]) + .sign_with_keys(&relay_keys) + .expect("signing presence event"); + serde_json::to_value(&event).expect("event to json") + } + + /// Build a presence event carrying the given `p`-tag subjects in order, + /// signed by a relay key. Used to construct off-contract multi-`p`-tag + /// events the relay never emits but a hostile responder could. + fn presence_event_with_p_tags(subjects: &[&str], status: &str) -> serde_json::Value { + let relay_keys = + nostr::Keys::parse("0000000000000000000000000000000000000000000000000000000000000002") + .expect("valid relay test key"); + let tags: Vec = subjects + .iter() + .map(|s| nostr::Tag::parse(["p", s]).expect("valid p tag")) + .collect(); + let event = nostr::EventBuilder::new( + nostr::Kind::Custom(buzz_core::kind::KIND_PRESENCE_UPDATE as u16), + status, + ) + .tags(tags) + .sign_with_keys(&relay_keys) + .expect("signing presence event"); + serde_json::to_value(&event).expect("event to json") + } + + #[test] + fn hints_from_results_successful_partial_snapshot_seeds_absent_as_offline() { + let online_pk = "a".repeat(64); + let absent_pk = "b".repeat(64); + let pubkeys = vec![online_pk.clone(), absent_pk.clone()]; + // Snapshot returns only the online instance; absent_pk is omitted. + let presence = events_json(&[presence_event(&online_pk, "online")]); + + let map = hints_from_results(&pubkeys, Ok(presence), Ok("[]".to_string())); + + assert_eq!( + map[&online_pk].presence.as_deref(), + Some("online"), + "returned status must overlay the seed" + ); + assert_eq!( + map[&absent_pk].presence.as_deref(), + Some("offline"), + "a requested pubkey omitted from a successful snapshot is offline" + ); + } + + #[test] + fn hints_from_results_successful_empty_snapshot_seeds_all_offline() { + let pk_a = "c".repeat(64); + let pk_b = "d".repeat(64); + let pubkeys = vec![pk_a.clone(), pk_b.clone()]; + + // Empty-but-successful snapshot: every requested pubkey is offline. + let map = hints_from_results(&pubkeys, Ok("[]".to_string()), Ok("[]".to_string())); + + assert_eq!(map[&pk_a].presence.as_deref(), Some("offline")); + assert_eq!(map[&pk_b].presence.as_deref(), Some("offline")); + } + + #[test] + fn hints_from_results_failed_presence_yields_no_offline_label() { + let pk = "e".repeat(64); + let pubkeys = vec![pk.clone()]; + let profile = + events_json(&[json!({ "pubkey": pk.clone(), "created_at": 1_705_276_800_u64 })]); + + let map = hints_from_results( + &pubkeys, + Err(CliError::Other("presence hint timeout".to_string())), + Ok(profile), + ); + + assert!( + map[&pk].presence.is_none(), + "a failed presence lookup must never be inferred as offline" + ); + assert_eq!( + map[&pk].profile_updated_at, + Some(1_705_276_800), + "the completed profile lookup must survive the failed presence sibling" + ); + } + + #[test] + fn hints_from_results_malformed_presence_yields_no_offline_label() { + let pk = "f".repeat(64); + let pubkeys = vec![pk.clone()]; + + // Unparseable presence body → not a trusted snapshot → no seeding. + let map = hints_from_results( + &pubkeys, + Ok("not json".to_string()), + Err(CliError::Other("profile hint timeout".to_string())), + ); + + assert!( + map.get(&pk).is_none_or(|h| h.presence.is_none()), + "malformed presence must not infer offline: {map:?}" + ); + } + + #[test] + fn hints_from_results_malformed_element_makes_snapshot_untrusted() { + // A body that parses as an array but contains a malformed element + // (`{}`, `null`, or a contentless event) is NOT an authoritative + // snapshot: it must seed nothing, while the profile sibling survives. + let requested = "a".repeat(64); + let subject = "b".repeat(64); + let pubkeys = vec![requested.clone(), subject.clone()]; + let profile = + events_json(&[json!({ "pubkey": requested.clone(), "created_at": 1_705_276_800_u64 })]); + + for bad_body in [ + "[{}]".to_string(), + "[null]".to_string(), + // A well-formed subject but non-string (unreadable) content. + events_json(&[json!({ + "pubkey": "r".repeat(64), + "content": 42, + "tags": [["p", subject.clone()]], + })]), + ] { + let map = hints_from_results(&pubkeys, Ok(bad_body.clone()), Ok(profile.clone())); + + assert!( + map.values().all(|h| h.presence.is_none()), + "malformed element {bad_body} must yield no presence labels: {map:?}" + ); + assert_eq!( + map[&requested].profile_updated_at, + Some(1_705_276_800), + "the completed profile sibling must still contribute hints: {map:?}" + ); + } + } + + #[test] + fn hints_from_results_relay_error_response_seeds_nothing() { + // The relay surfaces a Redis-outage presence lookup as a non-2xx error, + // which the CLI query returns as `Err`. That must seed nothing — a + // backend failure is not an authoritative all-offline snapshot. + let pk = "c".repeat(64); + let pubkeys = vec![pk.clone()]; + + let map = hints_from_results( + &pubkeys, + Err(CliError::Other("presence lookup: redis down".to_string())), + Ok("[]".to_string()), + ); + + assert!( + map.get(&pk).is_none_or(|h| h.presence.is_none()), + "a relay-side presence failure must never be inferred as offline: {map:?}" + ); + } + + #[test] + fn hints_from_results_vacuous_object_makes_snapshot_untrusted() { + // A syntactically-valid array whose element carries a plausible subject + // and string content but is NOT a complete signed event (no id/sig/kind + // /created_at) must not be trusted as a snapshot — otherwise it would + // re-seed every requested candidate `offline` from an unverifiable body. + let requested = "a".repeat(64); + let pubkeys = vec![requested.clone()]; + let profile = + events_json(&[json!({ "pubkey": requested.clone(), "created_at": 1_705_276_800_u64 })]); + let vacuous = events_json(&[json!({ "pubkey": requested.clone(), "content": "online" })]); + + let map = hints_from_results(&pubkeys, Ok(vacuous), Ok(profile)); + + assert!( + map.values().all(|h| h.presence.is_none()), + "a vacuous non-event object must yield no presence labels: {map:?}" + ); + assert_eq!( + map[&requested].profile_updated_at, + Some(1_705_276_800), + "the completed profile sibling must still contribute hints: {map:?}" + ); + } + + #[test] + fn hints_from_results_unrequested_subject_makes_snapshot_untrusted() { + // A fully-shaped, correctly-signed presence event whose subject is NOT + // one of the requested pubkeys is not a snapshot of the requested set; + // trusting it would seed the requested duplicates `offline` from an + // answer about someone else entirely. + let requested = "a".repeat(64); + let other = "b".repeat(64); + let pubkeys = vec![requested.clone()]; + let profile = + events_json(&[json!({ "pubkey": requested.clone(), "created_at": 1_705_276_800_u64 })]); + let presence = events_json(&[presence_event(&other, "online")]); + + let map = hints_from_results(&pubkeys, Ok(presence), Ok(profile)); + + assert!( + map.values().all(|h| h.presence.is_none()), + "an event for an unrequested subject must yield no presence labels: {map:?}" + ); + assert_eq!( + map[&requested].profile_updated_at, + Some(1_705_276_800), + "the completed profile sibling must still contribute hints: {map:?}" + ); + } + + #[test] + fn hints_from_results_mixed_p_tags_makes_snapshot_untrusted() { + // The relay emits exactly one `p` tag per presence event. A hostile + // responder could return `[["p",""],["p",""]]`: + // a weaker "any requested `p` tag" gate would accept it, but the + // consumer reads the FIRST `p` tag (the unrequested subject) — so it + // would overlay the wrong subject and leave the requested candidate + // falsely seeded `offline`. Requiring exactly one `p`-tag subject that + // is requested rejects both an unrequested-first ordering and any event + // carrying more than one `p` tag. + let requested = "a".repeat(64); + let unrequested = "b".repeat(64); + let pubkeys = vec![requested.clone()]; + let profile = + events_json(&[json!({ "pubkey": requested.clone(), "created_at": 1_705_276_800_u64 })]); + + // Case 1: an unrequested `p` tag before a requested one. + let mixed = events_json(&[presence_event_with_p_tags( + &[&unrequested, &requested], + "online", + )]); + // Case 2: a valid requested `p` tag plus a second (also requested) — + // still off-contract: more than one `p` tag. + let two_requested = events_json(&[presence_event_with_p_tags( + &[&requested, &requested], + "online", + )]); + + for body in [mixed, two_requested] { + let map = hints_from_results(&pubkeys, Ok(body.clone()), Ok(profile.clone())); + + assert!( + map.values().all(|h| h.presence.is_none()), + "a multi-`p`-tag event must yield no presence labels: {map:?}" + ); + assert_eq!( + map[&requested].profile_updated_at, + Some(1_705_276_800), + "the completed profile sibling must still contribute hints: {map:?}" + ); + } + } + + #[tokio::test(start_paused = true)] + async fn join_bounded_queries_completed_presence_survives_hung_profile() { + let timeout = std::time::Duration::from_secs(3); + let (presence, profile) = join_bounded_queries( + timeout, + // Presence completes immediately. + async { Ok::("[]".to_string()) }, + // Profile hangs past the timeout. + async { + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + Ok::("[]".to_string()) + }, + ) + .await; + + assert!( + presence.is_ok(), + "the completed presence lookup must be retained, not discarded by the hung sibling" + ); + assert!(profile.is_err(), "the hung profile lookup must time out"); + } + + #[tokio::test(start_paused = true)] + async fn join_bounded_queries_completed_profile_survives_hung_presence() { + let timeout = std::time::Duration::from_secs(3); + let (presence, profile) = join_bounded_queries( + timeout, + async { + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + Ok::("[]".to_string()) + }, + async { Ok::("[]".to_string()) }, + ) + .await; + + assert!(presence.is_err(), "the hung presence lookup must time out"); + assert!( + profile.is_ok(), + "the completed profile lookup must be retained despite the hung presence sibling" + ); + } + + /// Production-wiring seam: drive `fetch_candidate_hints` itself against a + /// controlled `/query` server where the presence query completes and the + /// profile query hangs past the timeout. The completed presence hint (an + /// `online` overlay plus `offline` seeds for the requested pubkeys) must + /// survive. This is what protects the `fetch_candidate_hints` call site: if + /// the old shared `timeout(join!(...))` is restored, the hung profile query + /// discards the completed presence result and the map comes back empty. + #[tokio::test] + async fn fetch_candidate_hints_completed_presence_survives_hung_profile_query() { + use axum::{extract::State, routing::post, Router}; + use serde_json::Value; + use std::net::SocketAddr; + use tokio::net::TcpListener; + + let online_pk = "a".repeat(64); + let offline_pk = "b".repeat(64); + + // Server dispatches on filter kind: presence (40902) returns one online + // event immediately; profile (kind 0) hangs well past the timeout. + let online_for_server = online_pk.clone(); + let app = Router::new() + .route( + "/query", + post(move |State(()): State<()>, body: axum::body::Bytes| { + let online_pk = online_for_server.clone(); + async move { + let filters: Vec = serde_json::from_slice(&body).unwrap_or_default(); + let kind = filters + .first() + .and_then(|f| f.get("kinds")) + .and_then(|k| k.as_array()) + .and_then(|k| k.first()) + .and_then(Value::as_u64); + if kind == Some(0) { + // Profile query hangs past the 100ms test timeout. + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + } + let body = + serde_json::to_string(&vec![presence_event(&online_pk, "online")]) + .unwrap(); + axum::response::Response::builder() + .header("content-type", "application/json") + .body(axum::body::Body::from(body)) + .unwrap() + } + }), + ) + .with_state(()); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr: SocketAddr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let keys = + nostr::Keys::parse("0000000000000000000000000000000000000000000000000000000000000001") + .expect("valid test key"); + let client = BuzzClient::new(format!("http://{addr}"), keys, None, None) + .expect("client construction should not fail"); + + let map = fetch_candidate_hints( + &client, + &[online_pk.clone(), offline_pk.clone()], + std::time::Duration::from_millis(100), + ) + .await; + + // Presence completed: online overlay present, absent pubkey seeded offline. + assert_eq!( + map[&online_pk].presence.as_deref(), + Some("online"), + "the completed presence result must survive the hung profile query: {map:?}" + ); + assert_eq!( + map[&offline_pk].presence.as_deref(), + Some("offline"), + "the trusted snapshot must seed the absent candidate offline: {map:?}" + ); + // Profile hung → no profile timestamps. + assert!( + map.values().all(|h| h.profile_updated_at.is_none()), + "the hung profile query must contribute nothing: {map:?}" + ); + } } diff --git a/crates/buzz-cli/src/commands/users.rs b/crates/buzz-cli/src/commands/users.rs index 7c15d285a0d..bb2d45dbf1b 100644 --- a/crates/buzz-cli/src/commands/users.rs +++ b/crates/buzz-cli/src/commands/users.rs @@ -485,7 +485,7 @@ pub async fn cmd_get_presence(client: &BuzzClient, pubkeys_csv: &str) -> Result< Ok(()) } -fn presence_subject(event: &serde_json::Value) -> &str { +pub(crate) fn presence_subject(event: &serde_json::Value) -> &str { event .get("tags") .and_then(|tags| tags.as_array()) diff --git a/crates/buzz-pubsub/src/presence.rs b/crates/buzz-pubsub/src/presence.rs index e0c9dfd6c9b..a3490222715 100644 --- a/crates/buzz-pubsub/src/presence.rs +++ b/crates/buzz-pubsub/src/presence.rs @@ -109,6 +109,26 @@ mod tests { TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(id)), host) } + #[tokio::test] + async fn get_presence_bulk_surfaces_connection_failure_as_error() { + // A backend outage must surface as `Err`, not a silently-empty `Ok`. + // `synthesize_presence` relies on this to return an error response + // rather than a fake-empty "all offline" snapshot on a Redis failure. + // Pool points at a closed port so the connection attempt fails. + let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("pool builds lazily"); + let ctx = ctx(0xaaaa, "a.example"); + let pubkey = make_pubkey(); + + let result = get_presence_bulk(&pool, &ctx, &[pubkey]).await; + + assert!( + result.is_err(), + "a connection failure must surface as Err, got {result:?}" + ); + } + #[test] fn presence_ttl_is_three_one_minute_heartbeat_windows() { assert_eq!(PRESENCE_TTL_SECS, 180); diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 2545cd819dd..5dbb2aaf50c 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1125,8 +1125,10 @@ async fn query_events_authed( .await; } - if let Some(presence_events) = synthesize_presence(state, tenant, &filters).await { - return Ok(Json(Value::Array(presence_events))); + if let Some(presence_result) = + synthesize_presence(&state.pubsub, &state.relay_keypair, tenant, &filters).await + { + return presence_result.map(|events| Json(Value::Array(events))); } let mut events: Vec = Vec::new(); @@ -2176,12 +2178,19 @@ pub async fn workflow_webhook( /// presence from Redis instead of querying the DB (ephemeral events are never /// stored, and kind:40902 snapshots are relay-generated on demand). /// -/// Returns `Some(events)` if handled, `None` to fall through to normal query. +/// Returns `None` when the filters are not a presence query (fall through to +/// the normal query path). Returns `Some(Ok(events))` when a presence snapshot +/// was produced — an empty vec is an authoritative "all offline" answer. +/// Returns `Some(Err(_))` when the backing Redis lookup failed: callers must +/// propagate that as an error response rather than a fake-empty success, so a +/// consumer cannot mistake a backend outage for an authoritative snapshot. +#[allow(clippy::type_complexity)] async fn synthesize_presence( - state: &AppState, + pubsub: &buzz_pubsub::PubSubManager, + relay_keypair: &nostr::Keys, tenant: &buzz_core::tenant::TenantContext, filters: &[nostr::Filter], -) -> Option> { +) -> Option, (StatusCode, Json)>> { use buzz_core::kind::{KIND_PRESENCE_SNAPSHOT, KIND_PRESENCE_UPDATE}; // Only intercept if every filter targets kind:20001 or 40902 with authors. @@ -2201,22 +2210,23 @@ async fn synthesize_presence( } if all_pubkeys.is_empty() { - return Some(Vec::new()); + return Some(Ok(Vec::new())); } // Dedup pubkeys. all_pubkeys.sort_by_key(|pk| pk.to_hex()); all_pubkeys.dedup(); - // Look up Redis. - let presence_map = state - .pubsub - .get_presence_bulk(tenant, &all_pubkeys) - .await - .unwrap_or_default(); + // Look up Redis. A lookup failure must surface as an error, not a + // fake-empty success — otherwise a Redis outage is indistinguishable from + // an authoritative all-offline snapshot to the consumer. + let presence_map = match pubsub.get_presence_bulk(tenant, &all_pubkeys).await { + Ok(map) => map, + Err(e) => return Some(Err(internal_error(&format!("presence lookup: {e}")))), + }; if presence_map.is_empty() { - return Some(Vec::new()); + return Some(Ok(Vec::new())); } // Synthesize kind:20001 events signed by the relay. @@ -2228,20 +2238,30 @@ async fn synthesize_presence( let mut events = Vec::with_capacity(presence_map.len()); for (pubkey_hex, status) in &presence_map { // Build a synthetic event: relay-signed, content = status, p-tag = subject. - let tags = vec![nostr::Tag::parse(["p", pubkey_hex]).ok()?]; - let event = - nostr::EventBuilder::new(nostr::Kind::Custom(KIND_PRESENCE_UPDATE as u16), status) - .tags(tags) - .custom_created_at(nostr::Timestamp::from(now)) - .sign_with_keys(&state.relay_keypair) - .ok()?; + // A build/sign failure here is an internal fault, not a "not a presence + // query" signal, so surface it as an error rather than falling through. + let tags = match nostr::Tag::parse(["p", pubkey_hex]) { + Ok(tag) => vec![tag], + Err(e) => return Some(Err(internal_error(&format!("presence tag: {e}")))), + }; + let event = match nostr::EventBuilder::new( + nostr::Kind::Custom(KIND_PRESENCE_UPDATE as u16), + status, + ) + .tags(tags) + .custom_created_at(nostr::Timestamp::from(now)) + .sign_with_keys(relay_keypair) + { + Ok(event) => event, + Err(e) => return Some(Err(internal_error(&format!("presence sign: {e}")))), + }; if let Ok(v) = serde_json::to_value(&event) { events.push(v); } } - Some(events) + Some(Ok(events)) } // ── Moderation queue reads (L6 — Quinn) ─────────────────────────────────────── @@ -2499,6 +2519,48 @@ mod tests { assert!(!has_mixed_search_filters(&filters)); } + /// Production-wiring seam for the Redis-outage boundary. Drives the real + /// `synthesize_presence` with a `PubSubManager` whose pool points at a + /// closed port, so the `get_presence_bulk` lookup fails. A presence-snapshot + /// filter must yield `Some(Err(500))` — never `Some(Ok([]))`, which would + /// let a consumer mistake a backend outage for an authoritative all-offline + /// snapshot. Restoring `unwrap_or_default()` inside `synthesize_presence` + /// turns this red (it would return `Some(Ok([]))`), which is what protects + /// the error-mapping seam Thufir found otherwise mutation-unprotected. + #[tokio::test] + async fn synthesize_presence_surfaces_redis_failure_as_error_response() { + use buzz_core::kind::KIND_PRESENCE_SNAPSHOT; + + // Pool at a closed port: get_presence_bulk's connection attempt fails. + let dead_pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("pool builds lazily"); + let pubsub = buzz_pubsub::PubSubManager::new("redis://127.0.0.1:1", dead_pool) + .await + .expect("PubSubManager::new performs no IO"); + let relay_keypair = Keys::generate(); + let tenant = fresh_tenant("relay.example"); + + // A presence-snapshot query for a concrete author reaches the Redis + // lookup (an empty author set would short-circuit to an empty snapshot). + let filters = vec![nostr::Filter::new() + .kind(Kind::Custom(KIND_PRESENCE_SNAPSHOT as u16)) + .author(Keys::generate().public_key())]; + + let result = synthesize_presence(&pubsub, &relay_keypair, &tenant, &filters).await; + + match result { + Some(Err((status, _))) => assert_eq!( + status, + StatusCode::INTERNAL_SERVER_ERROR, + "a Redis lookup failure must surface as HTTP 500" + ), + other => panic!( + "a Redis outage must yield Some(Err(500)), not a fake-empty success: {other:?}" + ), + } + } + #[test] fn thread_aux_query_targets_root_and_replies() { let tenant = fresh_tenant("relay.example");