diff --git a/crates/buzz-core/src/desktop_stop.rs b/crates/buzz-core/src/desktop_stop.rs new file mode 100644 index 00000000000..4f431655e99 --- /dev/null +++ b/crates/buzz-core/src/desktop_stop.rs @@ -0,0 +1,220 @@ +//! Immutable, owner-to-self Desktop Stop messages. Profiles are not authority. +use nostr::{nips::nip44, Event, EventBuilder, Keys, Kind, PublicKey, Tag}; +use serde::{Deserialize, Serialize}; + +use crate::kind::{KIND_DESKTOP_STOP, KIND_DESKTOP_STOP_RESULT}; + +/// One agent on one Desktop in one community; never a caller-selected process. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct StopTarget { + /// Schema version. Old exact-run commands are not accepted here. + pub v: u8, + /// Canonical community WebSocket URL. + pub community: String, + /// Installation coordinate from the private Desktop inventory. + pub desktop: String, + /// Agent public key. The receiver independently verifies local ownership. + pub agent: String, +} + +/// Ordinary Desktop outcome, not a stronger process-termination certificate. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum StopOutcome { + /// Ordinary Stop returned success. + Stopped, + /// Ordinary Stop returned an error. No automatic retry of the effect. + Failed, + /// Interrupted, stale or evicted request; never inferred success. + Unknown, +} + +/// Correlates exactly one immutable request with its Desktop's result. +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct StopResult { + /// Original target, not mutable current routing. + pub target: StopTarget, + /// Signed request event ID. + pub request: String, + /// No diagnostic paths, credentials or process details on the wire. + pub outcome: StopOutcome, +} + +fn hex(value: &str, len: usize) -> bool { + value.len() == len + && value + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +/// Check public shape before storage, without decrypting content. +pub fn validate_envelope(event: &Event) -> Result<(), &'static str> { + let kind = event.kind.as_u16() as u32; + let tags: Vec<_> = event.tags.iter().map(|t| t.as_slice()).collect(); + let result = kind == KIND_DESKTOP_STOP_RESULT; + if !matches!(kind, KIND_DESKTOP_STOP | KIND_DESKTOP_STOP_RESULT) + || !(132..=4096).contains(&event.content.len()) + || tags.len() != if result { 2 } else { 1 } + || tags[0].len() != 2 + || tags[0][0] != "d" + || !hex(&tags[0][1], 32) + || (result && (tags[1].len() != 2 || tags[1][0] != "e" || !hex(&tags[1][1], 64))) + { + return Err("invalid Desktop Stop envelope"); + } + Ok(()) +} + +fn sign(value: &T, keys: &Keys, kind: u32, tags: Vec) -> Result { + let ciphertext = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + serde_json::to_string(value).map_err(|e| e.to_string())?, + nip44::Version::V2, + ) + .map_err(|e| e.to_string())?; + EventBuilder::new(Kind::Custom(kind as u16), ciphertext) + .tags(tags) + .sign_with_keys(keys) + .map_err(|e| e.to_string()) +} + +fn read( + event: &Event, + keys: &Keys, + kind: u32, +) -> Result { + validate_envelope(event)?; + event + .verify() + .map_err(|_| "invalid Desktop Stop signature")?; + if event.pubkey != keys.public_key() || event.kind.as_u16() as u32 != kind { + return Err("foreign Desktop Stop message".into()); + } + let plaintext = nip44::decrypt(keys.secret_key(), &keys.public_key(), &event.content) + .map_err(|_| "Desktop Stop decryption failed")?; + serde_json::from_str(&plaintext).map_err(|_| "invalid Desktop Stop payload".into()) +} + +impl StopTarget { + /// Validate the decrypted target against the captured community. + pub fn validate(&self, community: &str) -> Result<(), &'static str> { + if self.v != 1 + || self.community != community + || community.is_empty() + || community.len() > 512 + || !hex(&self.desktop, 32) + || !hex(&self.agent, 64) + || PublicKey::from_hex(&self.agent).is_err() + { + return Err("invalid Desktop Stop target"); + } + Ok(()) + } + + /// Produce a new immutable Stop. Transport retries must reuse this event. + pub fn sign(&self, keys: &Keys) -> Result { + self.validate(&self.community)?; + sign( + self, + keys, + KIND_DESKTOP_STOP, + vec![Tag::identifier(&self.desktop)], + ) + } + + /// Authenticate, decrypt and bind a Stop to its signed host coordinate. + pub fn read(event: &Event, keys: &Keys, community: &str) -> Result { + let target: Self = read(event, keys, KIND_DESKTOP_STOP)?; + target.validate(community)?; + if event.tags.identifier() != Some(target.desktop.as_str()) { + return Err("Desktop Stop routing mismatch".into()); + } + Ok(target) + } +} + +impl StopResult { + /// Sign the saved ordinary Stop result without exposing local diagnostics. + pub fn sign(&self, keys: &Keys) -> Result { + self.target.validate(&self.target.community)?; + if !hex(&self.request, 64) { + return Err("invalid Stop request ID".into()); + } + sign( + self, + keys, + KIND_DESKTOP_STOP_RESULT, + vec![ + Tag::identifier(&self.target.desktop), + Tag::parse(["e", &self.request]).map_err(|e| e.to_string())?, + ], + ) + } + + /// Check all correlation fields against the original authenticated request. + pub fn read( + event: &Event, + keys: &Keys, + request: &Event, + community: &str, + ) -> Result { + let target = StopTarget::read(request, keys, community)?; + let result: Self = read(event, keys, KIND_DESKTOP_STOP_RESULT)?; + if result.target != target + || result.request != request.id.to_hex() + || event.tags.identifier() != Some(target.desktop.as_str()) + || event.tags.iter().nth(1).and_then(|t| t.content()) != Some(result.request.as_str()) + { + return Err("Desktop Stop result correlation mismatch".into()); + } + Ok(result) + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn private_immutable_stop_and_exact_result_correlation() { + let keys = Keys::generate(); + let target = StopTarget { + v: 1, + community: "wss://one.example".into(), + desktop: "a".repeat(32), + agent: Keys::generate().public_key().to_hex(), + }; + let request = target.sign(&keys).unwrap(); + assert_eq!( + StopTarget::read(&request, &keys, &target.community).unwrap(), + target + ); + assert!(!request.content.contains(&target.agent)); + assert!(StopTarget::read(&request, &Keys::generate(), &target.community).is_err()); + assert!(StopTarget::read(&request, &keys, "wss://other.example").is_err()); + let result = StopResult { + target: target.clone(), + request: request.id.to_hex(), + outcome: StopOutcome::Stopped, + } + .sign(&keys) + .unwrap(); + assert_eq!( + StopResult::read(&result, &keys, &request, &target.community) + .unwrap() + .outcome, + StopOutcome::Stopped + ); + let another = target.sign(&keys).unwrap(); + assert!(StopResult::read(&result, &keys, &another, &target.community).is_err()); + let mut tampered = request.clone(); + tampered.content.push('x'); + assert!(StopTarget::read(&tampered, &keys, &target.community).is_err()); + for kind in [KIND_DESKTOP_STOP, KIND_DESKTOP_STOP_RESULT] { + assert!(crate::kind::AUTHOR_ONLY_KINDS.contains(&kind)); + assert!(!crate::kind::is_parameterized_replaceable(kind)); + } + } +} diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 4d309c3b609..5df974c36cb 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -125,6 +125,11 @@ pub const KIND_DESKTOP_OBSERVATION: u32 = 30181; /// Owner-private built-in runtime facts per Desktop, not agent readiness. pub const KIND_DESKTOP_CAPABILITIES: u32 = 30182; +/// Immutable owner-private Desktop Stop request (not a replaceable profile). +pub const KIND_DESKTOP_STOP: u32 = 50180; +/// Owner-private ordinary Desktop Stop outcome, correlated by request event ID. +pub const KIND_DESKTOP_STOP_RESULT: u32 = 50181; + /// Kinds whose stored events are readable only by their author. /// /// The relay must never reveal the existence, count, tags, content, schedule, @@ -141,6 +146,8 @@ pub const AUTHOR_ONLY_KINDS: &[u32] = &[ KIND_DESKTOP_PROFILE, KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_CAPABILITIES, + KIND_DESKTOP_STOP, + KIND_DESKTOP_STOP_RESULT, ]; /// Kinds that require a result-level read gate beyond the filter-layer @@ -673,6 +680,8 @@ pub const ALL_KINDS: &[u32] = &[ KIND_DESKTOP_PROFILE, KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_CAPABILITIES, + KIND_DESKTOP_STOP, + KIND_DESKTOP_STOP_RESULT, KIND_REPORT, KIND_PRODUCT_FEEDBACK, KIND_NIP29_PUT_USER, diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index ee4bb5df789..4b3874be0f0 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -13,6 +13,7 @@ pub mod desktop_capabilities; pub mod desktop_observation; /// Owner-private Desktop display profiles. pub mod desktop_profile; +pub mod desktop_stop; /// NIP-AE Agent Engrams — slug grammar, conversation key, d-tag derivation, /// body parse/serialize, envelope build/validate, head selection. pub mod engram; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 7da5d0d0c68..35261123b61 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -702,7 +702,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 47); + assert_eq!(migrations.len(), 48); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -911,7 +911,7 @@ mod postgres_tests { assert!(migrations[32].sql.as_str().contains("search_tsv")); assert!(!migrations[0].sql.as_str().contains("30179")); assert!(include_str!("../../../../schema/schema.sql").contains( - "kind IN (1059, 30179, 30180, 30181, 30182, 30300, 30350, 30622, 44100, 44101, 44200)" + "kind IN (1059, 30179, 30180, 30181, 30182, 30300, 30350, 30622, 44100, 44101, 44200, 50180, 50181)" )); // Public push-gateway authority is intentionally deployment-global and @@ -2394,6 +2394,8 @@ mod postgres_tests { (4_u8, 30_180_i32), (5_u8, 30_181_i32), (6_u8, 30_182_i32), + (7_u8, 50_180_i32), + (8_u8, 50_181_i32), ] { sqlx::query( "INSERT INTO events \ @@ -2429,7 +2431,9 @@ mod postgres_tests { (30_180, true), (30_181, true), (30_182, true), - (30_350, true) + (30_350, true), + (50_180, true), + (50_181, true) ] ); @@ -2454,7 +2458,9 @@ mod postgres_tests { (30_180, Some(true)), (30_181, Some(true)), (30_182, Some(true)), - (30_350, None) + (30_350, None), + (50_180, Some(true)), + (50_181, Some(true)) ] ); @@ -2493,6 +2499,15 @@ mod postgres_tests { "0047 must change brownfield capability FTS" ); + run_migrations_through(&pool, 47).await.unwrap(); + let stop_indexed: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE kind IN (50180, 50181) AND search_tsv IS NOT NULL", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(stop_indexed, 2, "0048 must change brownfield Stop FTS"); + run_migrations(&pool) .await .expect("apply remaining migrations to populated database"); @@ -2511,7 +2526,9 @@ mod postgres_tests { (30_180, None), (30_181, None), (30_182, None), - (30_350, None) + (30_350, None), + (50_180, None), + (50_181, None) ] ); let gin_exists: bool = sqlx::query_scalar( diff --git a/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs b/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs index 80f4619c583..ac35a0dd12e 100644 --- a/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs +++ b/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs @@ -2,7 +2,10 @@ use super::postgres_tests::bridge_handler_test_state; use super::*; use axum::{body::Body, http::Request}; -use buzz_core::kind::{KIND_DESKTOP_CAPABILITIES, KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_PROFILE}; +use buzz_core::kind::{ + KIND_DESKTOP_CAPABILITIES, KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_PROFILE, KIND_DESKTOP_STOP, + KIND_DESKTOP_STOP_RESULT, +}; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; use tower::ServiceExt; @@ -82,6 +85,13 @@ async fn desktop_capabilities_authenticated_owner_query_and_private_storage() { assert_private_desktop(KIND_DESKTOP_CAPABILITIES).await; } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desktop_stop_authenticated_owner_query_and_private_storage() { + assert_private_desktop(KIND_DESKTOP_STOP).await; + assert_private_desktop(KIND_DESKTOP_STOP_RESULT).await; +} + async fn assert_private_desktop(kind: u32) { let mut state = bridge_handler_test_state() .await @@ -103,7 +113,26 @@ async fn assert_private_desktop(kind: u32) { ) .unwrap(); let id = profile.id.clone(); - let event = if kind == KIND_DESKTOP_PROFILE { + let event = if matches!(kind, KIND_DESKTOP_STOP | KIND_DESKTOP_STOP_RESULT) { + let target = buzz_core::desktop_stop::StopTarget { + v: 1, + community: format!("wss://{host}"), + desktop: id.clone(), + agent: Keys::generate().public_key().to_hex(), + }; + let request = target.sign(&owner).unwrap(); + if kind == KIND_DESKTOP_STOP { + request + } else { + buzz_core::desktop_stop::StopResult { + target, + request: request.id.to_hex(), + outcome: buzz_core::desktop_stop::StopOutcome::Stopped, + } + .sign(&owner) + .unwrap() + } + } else if kind == KIND_DESKTOP_PROFILE { profile.sign(&owner).unwrap() } else if kind == KIND_DESKTOP_CAPABILITIES { buzz_core::desktop_capabilities::DesktopCapabilities::new(profile, vec![]) @@ -284,3 +313,100 @@ async fn aged_desktop_profile_retries_through_production_ingest_without_resignin assert_eq!(status, StatusCode::BAD_REQUEST, "{result}"); } } + +/// Same-ID transport retry must redeliver to a currently connected Desktop, +/// even when the first attempt arrived while it was absent. It cannot re-date, +/// re-store or expose the request to another owner/community. +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn desktop_stop_retry_redelivers_exact_event_only_to_owner() { + use nostr::Filter; + use std::sync::atomic::AtomicU8; + use tokio::sync::{mpsc, Mutex}; + use tokio_util::sync::CancellationToken; + let mut state = bridge_handler_test_state() + .await + .expect("test infrastructure"); + Arc::make_mut(&mut Arc::get_mut(&mut state).unwrap().config).require_auth_token = true; + let host = format!("stop-retry-{}.example", uuid::Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .unwrap() + .id; + let owner = Keys::generate(); + let outsider = Keys::generate(); + let target = buzz_core::desktop_stop::StopTarget { + v: 1, + community: format!("wss://{host}"), + desktop: uuid::Uuid::new_v4().simple().to_string(), + agent: Keys::generate().public_key().to_hex(), + }; + let prepared = target.sign(&owner).unwrap(); + let event = EventBuilder::new(prepared.kind, &prepared.content) + .tags(prepared.tags.iter().cloned()) + .custom_created_at(Timestamp::from(Timestamp::now().as_secs() - 86_400)) + .sign_with_keys(&owner) + .unwrap(); + let raw = json!(event); + let (status, result) = post(&state, &host, "/events", &owner, raw.clone(), true).await; + assert_eq!(status, StatusCode::OK, "{result}"); + assert_eq!(result["accepted"], true); + let mut receivers = vec![]; + for (who, tenant) in [ + (&owner, community), + (&outsider, community), + ( + &owner, + buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::new_v4()), + ), + ] { + let conn = uuid::Uuid::new_v4(); + let (tx, rx) = mpsc::channel(16); + let (ctrl, _) = mpsc::channel(16); + state.conn_manager.register( + conn, + tx, + ctrl, + None, + CancellationToken::new(), + tenant, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(std::collections::HashMap::new())), + 3, + ); + state + .conn_manager + .set_authenticated_pubkey(conn, who.public_key().to_bytes().to_vec()); + state.sub_registry.register_scoped( + tenant, + conn, + "stop".into(), + vec![Filter::new().kind(Kind::Custom(KIND_DESKTOP_STOP as u16))], + None, + ); + receivers.push(rx); + } + // Quiesce the asynchronous first dispatch before measuring the retry. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + for rx in &mut receivers { + drain(rx); + } + for _ in 0..2 { + let (status, result) = post(&state, &host, "/events", &owner, raw.clone(), true).await; + assert_eq!(status, StatusCode::OK, "{result}"); + assert_eq!(result["message"], "duplicate:"); + let frames = drain(&mut receivers[0]); + assert_eq!(frames.len(), 1, "{frames:?}"); + assert_eq!(frames[0][2], raw); + assert!(drain(&mut receivers[1]).is_empty()); + assert!(drain(&mut receivers[2]).is_empty()); + } + let (status, result) = post(&state, &host, "/events", &outsider, raw, true).await; + assert_eq!(status, StatusCode::FORBIDDEN, "{result}"); + assert!(drain(&mut receivers[0]).is_empty()); + let (_, rows) = post(&state, &host, "/query", &owner, + json!([{"kinds":[KIND_DESKTOP_STOP],"authors":[owner.public_key().to_hex()], "ids":[event.id.to_hex()]}]), true).await; + assert_eq!(rows.as_array().unwrap().len(), 1); +} diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 832b5678c9d..c4e4d1f86c9 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -337,6 +337,32 @@ pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pub } } +/// Retry delivery of an already stored Stop, without audit/workflow effects. +/// Admission and durable effect deduplication remain the Desktop's authority. +pub(crate) async fn redeliver_desktop_stop( + tenant: &TenantContext, + state: &Arc, + event: &nostr::Event, +) { + state.mark_local_event(tenant.community(), &event.id); + if let Err(error) = state + .pubsub + .publish_event(tenant, EventTopic::Global, event) + .await + { + state + .local_event_ids + .invalidate(&(tenant.community(), event.id.to_bytes())); + warn!(event_id = %event.id, %error, "Desktop Stop redelivery to peers failed"); + } + fan_out_event_to_local_subscribers( + state, + tenant.community(), + &StoredEvent::new(event.clone(), None), + ) + .await; +} + /// Schedule post-commit delivery/side effects for a stored event. /// /// This intentionally returns after only the bounded audit enqueue has completed: @@ -2220,6 +2246,12 @@ mod tests { assert_author_only_fanout(buzz_core::kind::KIND_DESKTOP_OBSERVATION).await; } + #[tokio::test] + async fn desktop_stop_delivers_to_author_only() { + assert_author_only_fanout(buzz_core::kind::KIND_DESKTOP_STOP).await; + assert_author_only_fanout(buzz_core::kind::KIND_DESKTOP_STOP_RESULT).await; + } + #[tokio::test] async fn desktop_capabilities_delivers_to_author_only() { assert_author_only_fanout(buzz_core::kind::KIND_DESKTOP_CAPABILITIES).await; diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 9d4e9ad5dc4..112bea02085 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -36,7 +36,10 @@ use buzz_core::kind::{ RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; -use buzz_core::kind::{KIND_DESKTOP_CAPABILITIES, KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_PROFILE}; +use buzz_core::kind::{ + KIND_DESKTOP_CAPABILITIES, KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_PROFILE, KIND_DESKTOP_STOP, + KIND_DESKTOP_STOP_RESULT, +}; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; use buzz_core::CommunityId; @@ -437,7 +440,7 @@ fn map_push_accept_error(error: super::push_lease::AcceptError) -> IngestError { /// Returns `Err` for unknown kinds — the relay rejects them. fn required_scope_for_kind(kind: u32, event: &Event) -> Result { match kind { - KIND_PROFILE | KIND_DESKTOP_PROFILE | KIND_DESKTOP_OBSERVATION | KIND_DESKTOP_CAPABILITIES => Ok(Scope::UsersWrite), + KIND_PROFILE | KIND_DESKTOP_PROFILE | KIND_DESKTOP_OBSERVATION | KIND_DESKTOP_CAPABILITIES | KIND_DESKTOP_STOP | KIND_DESKTOP_STOP_RESULT => Ok(Scope::UsersWrite), KIND_TEXT_NOTE | KIND_LONG_FORM => Ok(Scope::MessagesWrite), KIND_CONTACT_LIST | KIND_READ_STATE | KIND_USER_STATUS | KIND_AGENT_ENGRAM | KIND_EVENT_REMINDER | KIND_PERSONA | KIND_TEAM | KIND_MANAGED_AGENT @@ -661,6 +664,8 @@ pub(crate) fn is_global_only_kind(kind: u32) -> bool { | KIND_DESKTOP_PROFILE | KIND_DESKTOP_OBSERVATION | KIND_DESKTOP_CAPABILITIES + | KIND_DESKTOP_STOP + | KIND_DESKTOP_STOP_RESULT | KIND_TEAM_CATALOG // NIP-34: git events use `a` tags (repo reference), not `h` tags (channel scope). // Parameterized replaceable kinds are keyed by (pubkey, kind, d_tag). @@ -2171,15 +2176,20 @@ pub async fn ingest_event( result } -// Profiles and capability facts are durable records, not freshness signals. A Desktop may +// Profiles, capabilities and immutable Stop messages are not freshness signals. A Desktop may // first publish its immutable signed record long after an offline startup. // Only their past-age bound is waived; future drift and all other admission // checks still apply. Observation/presence kinds must retain their own window. fn timestamp_within_ingest_window(kind: u32, event_ts: u64, now: u64) -> bool { const MAX_TIMESTAMP_DRIFT_SECS: u64 = 900; event_ts <= now.saturating_add(MAX_TIMESTAMP_DRIFT_SECS) - && (matches!(kind, KIND_DESKTOP_PROFILE | KIND_DESKTOP_CAPABILITIES) - || now.saturating_sub(event_ts) <= MAX_TIMESTAMP_DRIFT_SECS) + && (matches!( + kind, + KIND_DESKTOP_PROFILE + | KIND_DESKTOP_CAPABILITIES + | KIND_DESKTOP_STOP + | KIND_DESKTOP_STOP_RESULT + ) || now.saturating_sub(event_ts) <= MAX_TIMESTAMP_DRIFT_SECS) } async fn ingest_event_inner( @@ -2794,6 +2804,11 @@ async fn ingest_event_inner( } } + if matches!(kind_u32, KIND_DESKTOP_STOP | KIND_DESKTOP_STOP_RESULT) { + buzz_core::desktop_stop::validate_envelope(&event) + .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; + } + if kind_u32 == KIND_DESKTOP_CAPABILITIES { buzz_core::desktop_capabilities::validate_envelope(&event) .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; @@ -3232,6 +3247,12 @@ async fn ingest_event_inner( }; if !was_inserted { + // Stop is a one-shot owned by Desktop, not a replaceable projection. + // Explicit transport retry must reach a live receiver even after an ACK + // or its result was lost. Never replay history or repeat relay effects. + if kind_u32 == KIND_DESKTOP_STOP { + super::event::redeliver_desktop_stop(tenant, state, &stored_event.event).await; + } return Ok(IngestResult { event_id: event_id_hex, accepted: true, @@ -3345,6 +3366,8 @@ mod postgres_tests { for kind in [ KIND_DESKTOP_PROFILE, KIND_DESKTOP_CAPABILITIES, + KIND_DESKTOP_STOP, + KIND_DESKTOP_STOP_RESULT, 30181, KIND_PROFILE, KIND_EVENT_REMINDER, @@ -3362,7 +3385,13 @@ mod postgres_tests { ] { assert_eq!( timestamp_within_ingest_window(kind, timestamp, now), - if matches!(kind, KIND_DESKTOP_PROFILE | KIND_DESKTOP_CAPABILITIES) { + if matches!( + kind, + KIND_DESKTOP_PROFILE + | KIND_DESKTOP_CAPABILITIES + | KIND_DESKTOP_STOP + | KIND_DESKTOP_STOP_RESULT + ) { profile } else { ordinary diff --git a/migrations/0048_desktop_stop_fts.sql b/migrations/0048_desktop_stop_fts.sql new file mode 100644 index 00000000000..98116efbda9 --- /dev/null +++ b/migrations/0048_desktop_stop_fts.sql @@ -0,0 +1,26 @@ +-- Owner-private Desktop Stop requests and results must not enter legacy ciphertext search indexes. +-- Like 0033, this rewrites events under ACCESS EXCLUSIVE; schedule accordingly. +DO $$ +DECLARE + existing_expression TEXT; +BEGIN + SELECT pg_get_expr(d.adbin, d.adrelid) + INTO existing_expression + FROM pg_attrdef d + JOIN pg_attribute a + ON a.attrelid = d.adrelid + AND a.attnum = d.adnum + WHERE d.adrelid = 'events'::regclass + AND a.attname = 'search_tsv'; + + IF existing_expression IS NULL THEN + RAISE EXCEPTION 'events.search_tsv generated expression not found'; + END IF; + + ALTER TABLE events DROP COLUMN search_tsv; + EXECUTE format( + 'ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (CASE WHEN kind IN (50180, 50181) THEN NULL::tsvector ELSE (%s) END) STORED', + existing_expression + ); + CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv); +END $$; diff --git a/schema/schema.sql b/schema/schema.sql index d78b3036d97..0a20e940285 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -221,7 +221,7 @@ CREATE TABLE events ( -- never matches `@@`. -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033). search_tsv TSVECTOR GENERATED ALWAYS AS ( - CASE WHEN kind IN (1059, 30179, 30180, 30181, 30182, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector + CASE WHEN kind IN (1059, 30179, 30180, 30181, 30182, 30300, 30350, 30622, 44100, 44101, 44200, 50180, 50181) THEN NULL::tsvector ELSE to_tsvector('simple', content) END ) STORED,