From f3cfadc9821f9bb20470a64ece8ccb3d29909f9d Mon Sep 17 00:00:00 2001 From: tornquist Date: Fri, 21 Aug 2026 01:21:01 +0000 Subject: [PATCH] fix(workflow): keep one state head across relay keys Signed-off-by: tornquist --- crates/buzz-cli/src/commands/workflows.rs | 68 +++++++++++++++++-- crates/buzz-db/src/migration.rs | 2 + .../buzz-relay/src/handlers/workflow_state.rs | 3 +- desktop/src-tauri/src/commands/workflows.rs | 33 +++++++-- .../src-tauri/src/commands/workflows_tests.rs | 51 ++++++++++++++ docs/nips/NIP-WF.md | 2 +- migrations/0032_workflow_revision_event.sql | 7 ++ schema/schema.sql | 3 + 8 files changed, 154 insertions(+), 15 deletions(-) diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index 61033680f4..48a3c74274 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -1,4 +1,4 @@ -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use sha2::{Digest, Sha256}; @@ -21,16 +21,32 @@ fn json_tag(event: &serde_json::Value, name: &str) -> Option { } fn current_workflow_events(events: &[serde_json::Value]) -> Vec { - let projected: HashSet = events + let mut state_heads: HashMap = HashMap::new(); + for event in events .iter() .filter(|event| event.get("kind").and_then(|value| value.as_u64()) == Some(30623)) - .filter_map(|event| json_tag(event, "d")) - .collect(); + { + let Some(workflow_id) = json_tag(event, "d") else { + continue; + }; + match state_heads.get(&workflow_id) { + Some(current) if !json_workflow_state_is_newer(event, current) => {} + _ => { + state_heads.insert(workflow_id, event); + } + } + } + let projected: HashSet = state_heads.keys().cloned().collect(); events .iter() .filter( |event| match event.get("kind").and_then(|value| value.as_u64()) { - Some(30623) => json_tag(event, "status").as_deref() != Some("deleted"), + Some(30623) => json_tag(event, "d").is_some_and(|workflow_id| { + state_heads + .get(&workflow_id) + .is_some_and(|head| head.get("id") == event.get("id")) + && json_tag(event, "status").as_deref() != Some("deleted") + }), Some(30620) => json_tag(event, "d").is_some_and(|id| !projected.contains(&id)), _ => false, }, @@ -39,6 +55,30 @@ fn current_workflow_events(events: &[serde_json::Value]) -> Vec bool { + let candidate_created = candidate + .get("created_at") + .and_then(serde_json::Value::as_u64) + .unwrap_or_default(); + let current_created = current + .get("created_at") + .and_then(serde_json::Value::as_u64) + .unwrap_or_default(); + let candidate_id = candidate + .get("id") + .and_then(serde_json::Value::as_str) + .unwrap_or_default(); + let current_id = current + .get("id") + .and_then(serde_json::Value::as_str) + .unwrap_or_default(); + candidate_created > current_created + || (candidate_created == current_created && candidate_id < current_id) +} + fn workflow_owner(event: &serde_json::Value) -> Option { if event.get("kind").and_then(|value| value.as_u64()) == Some(30623) { json_tag(event, "owner") @@ -347,4 +387,22 @@ mod tests { }); assert!(current_workflow_events(&[legacy, deleted]).is_empty()); } + + #[test] + fn relay_state_fold_selects_one_head_after_key_rotation() { + let id = uuid::Uuid::new_v4().to_string(); + let old = serde_json::json!({ + "id": "bb", "kind": 30623, "pubkey": "old-relay", "created_at": 100, "content": "old", + "tags": [["d", id.clone()], ["owner", "11"], ["e", "aa"], ["status", "active"]] + }); + let current = serde_json::json!({ + "id": "dd", "kind": 30623, "pubkey": "new-relay", "created_at": 200, "content": "new", + "tags": [["d", id], ["owner", "11"], ["e", "cc"], ["status", "active"]] + }); + + assert_eq!( + current_workflow_events(&[old, current.clone()]), + vec![current] + ); + } } diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 0ec091e96c..515929c672 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -1071,9 +1071,11 @@ mod tests { assert!(sql.contains("WHERE kind = 30620")); assert!(sql.contains("definition.id")); assert!(sql.contains("octet_length(revision_event_id) = 32")); + assert!(sql.contains("events_live_workflow_state_coordinate_idx")); let desired_schema = include_str!("../../../schema/schema.sql"); assert!(desired_schema.contains("revision_event_id BYTEA CHECK")); assert!(desired_schema.contains("octet_length(revision_event_id) = 32")); + assert!(desired_schema.contains("events_live_workflow_state_coordinate_idx")); } #[test] diff --git a/crates/buzz-relay/src/handlers/workflow_state.rs b/crates/buzz-relay/src/handlers/workflow_state.rs index 68ef0dbe86..63df7ef0d3 100644 --- a/crates/buzz-relay/src/handlers/workflow_state.rs +++ b/crates/buzz-relay/src/handlers/workflow_state.rs @@ -93,12 +93,11 @@ pub(crate) async fn store_workflow_state( sqlx::query( "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \ + WHERE community_id = $1 AND kind = $2 AND d_tag = $3 \ AND deleted_at IS NULL", ) .bind(community_id.as_uuid()) .bind(KIND_WORKFLOW_STATE as i32) - .bind(pubkey.as_slice()) .bind(&workflow_id) .execute(&mut **tx) .await?; diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index 2361f4bf18..86c6268501 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -1,4 +1,4 @@ -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use serde::Serialize; use serde_json::Value; @@ -460,18 +460,31 @@ fn webhook_secret_from_message(message: &str) -> Option { /// Fold relay-authored current state over legacy owner-authored definitions. /// A deleted state is authoritative and suppresses the legacy fallback. fn workflows_from_events(events: &[nostr::Event]) -> Vec { - let projected_ids: HashSet = events - .iter() - .filter(|event| event.kind.as_u16() == 30623) - .filter_map(|event| tag_value(event, "d")) - .collect(); + let mut state_heads: HashMap = HashMap::new(); + for event in events.iter().filter(|event| event.kind.as_u16() == 30623) { + let Some(workflow_id) = tag_value(event, "d") else { + continue; + }; + match state_heads.get(&workflow_id) { + Some(current) if !workflow_state_is_newer(event, current) => {} + _ => { + state_heads.insert(workflow_id, event); + } + } + } + let projected_ids: HashSet = state_heads.keys().cloned().collect(); events .iter() .filter(|event| { let is_state = event.kind.as_u16() == 30623; if is_state { - return tag_value(event, "status").as_deref() != Some("deleted"); + return tag_value(event, "d").is_some_and(|workflow_id| { + state_heads + .get(&workflow_id) + .is_some_and(|head| head.id == event.id) + && tag_value(event, "status").as_deref() != Some("deleted") + }); } event.kind.as_u16() == 30620 && tag_value(event, "d").is_some_and(|id| !projected_ids.contains(&id)) @@ -480,6 +493,12 @@ fn workflows_from_events(events: &[nostr::Event]) -> Vec { .collect() } +fn workflow_state_is_newer(candidate: &nostr::Event, current: &nostr::Event) -> bool { + candidate.created_at > current.created_at + || (candidate.created_at == current.created_at + && candidate.id.to_hex() < current.id.to_hex()) +} + /// Convert a workflow definition or relay state event into a [`WorkflowWire`]. fn workflow_from_event(ev: &nostr::Event) -> WorkflowWire { let id = tag_value(ev, "d").unwrap_or_default(); diff --git a/desktop/src-tauri/src/commands/workflows_tests.rs b/desktop/src-tauri/src/commands/workflows_tests.rs index f2c5934bd3..ad1b9ffa9e 100644 --- a/desktop/src-tauri/src/commands/workflows_tests.rs +++ b/desktop/src-tauri/src/commands/workflows_tests.rs @@ -27,6 +27,19 @@ fn wf_state( revision: &nostr::EventId, status: &str, yaml: &str, +) -> nostr::Event { + wf_state_at(d, h, owner, revision, status, yaml, 100) +} + +#[allow(clippy::too_many_arguments)] +fn wf_state_at( + d: &str, + h: &str, + owner: &nostr::PublicKey, + revision: &nostr::EventId, + status: &str, + yaml: &str, + created_at: u64, ) -> nostr::Event { let relay = Keys::generate(); let owner = owner.to_hex(); @@ -42,6 +55,7 @@ fn wf_state( .map(|tag| Tag::parse(tag).expect("parse tag")) .collect(); EventBuilder::new(Kind::Custom(30623), yaml) + .custom_created_at(nostr::Timestamp::from(created_at)) .tags(tags) .sign_with_keys(&relay) .expect("sign") @@ -100,6 +114,43 @@ fn deleted_relay_state_suppresses_legacy_definition() { assert!(workflows_from_events(&[legacy, deleted]).is_empty()); } +#[test] +fn relay_state_fold_selects_one_head_after_key_rotation() { + let owner = Keys::generate(); + let first_request = EventBuilder::new(Kind::Custom(46021), "first") + .custom_created_at(nostr::Timestamp::from(100)) + .tags([]) + .sign_with_keys(&Keys::generate()) + .expect("sign first request"); + let second_request = EventBuilder::new(Kind::Custom(46021), "second") + .custom_created_at(nostr::Timestamp::from(200)) + .tags([]) + .sign_with_keys(&Keys::generate()) + .expect("sign second request"); + let first = wf_state_at( + WF, + CHAN, + &owner.public_key(), + &first_request.id, + "active", + "name: old\nsteps: []\n", + 100, + ); + let second = wf_state_at( + WF, + CHAN, + &owner.public_key(), + &second_request.id, + "active", + "name: new\nsteps: []\n", + 200, + ); + + let workflows = workflows_from_events(&[first, second]); + assert_eq!(workflows.len(), 1); + assert_eq!(workflows[0].revision, second_request.id.to_hex()); +} + #[test] fn definition_is_parsed_into_object_with_nested_fields() { let ev = wf_event(WF, CHAN, YAML); diff --git a/docs/nips/NIP-WF.md b/docs/nips/NIP-WF.md index d1fb217781..7f1a2c7dc2 100644 --- a/docs/nips/NIP-WF.md +++ b/docs/nips/NIP-WF.md @@ -55,7 +55,7 @@ After accepting a create, update, or deletion request, the relay publishes kind } ``` -Only the configured relay identity is authoritative for kind `30623`. Clients must not infer workflow ownership from the state event's relay pubkey; they use the `owner` tag. +Kind `30623` is a relay-owned projection. Relay signing-key rotation replaces the existing workflow coordinate instead of creating a second head; clients select one deterministic current head per `d` tag. Clients must not infer workflow ownership from the state event's relay pubkey; they use the `owner` tag. A deleted state has empty content and `status=deleted`. Clients prefer valid relay-signed `30623` state over legacy `30620` definitions and hide deleted state. diff --git a/migrations/0032_workflow_revision_event.sql b/migrations/0032_workflow_revision_event.sql index 077708235f..72da2e6751 100644 --- a/migrations/0032_workflow_revision_event.sql +++ b/migrations/0032_workflow_revision_event.sql @@ -26,3 +26,10 @@ WHERE definition.community_id = workflow.community_id ALTER TABLE workflows ADD CONSTRAINT workflows_revision_event_id_length CHECK (revision_event_id IS NULL OR octet_length(revision_event_id) = 32); + +-- Workflow state is a relay-owned projection whose coordinate survives relay +-- signing-key rotation. There may be only one live head per workflow, +-- regardless of which relay key authored it. +CREATE UNIQUE INDEX events_live_workflow_state_coordinate_idx + ON events (community_id, kind, d_tag) + WHERE kind = 30623 AND d_tag IS NOT NULL AND deleted_at IS NULL; diff --git a/schema/schema.sql b/schema/schema.sql index ab866d9863..70c4dd6a2e 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -270,6 +270,9 @@ CREATE INDEX idx_events_addressable CREATE INDEX idx_events_parameterized ON events (community_id, kind, pubkey, d_tag, created_at DESC, id) WHERE d_tag IS NOT NULL AND deleted_at IS NULL; +CREATE UNIQUE INDEX events_live_workflow_state_coordinate_idx + ON events (community_id, kind, d_tag) + WHERE kind = 30623 AND d_tag IS NOT NULL AND deleted_at IS NULL; CREATE INDEX idx_events_not_before ON events (community_id, not_before) WHERE not_before IS NOT NULL AND deleted_at IS NULL AND delivered_at IS NULL; -- Full-text search. Minimal GIN over the generated tsvector; community scoping