jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-04 16:31:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add owner-authorized manual workflow triggers

Browse Source 
Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-08-27 17:35:42 -04:00
parent d2a80d8199
commit 46b2f7cb5d
12 changed files with 844 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
crates/buzz-cli/src/commands/workflows.rs
View File

@@ -162,6 +162,22 @@ pub async fn cmd_delete_workflow(client: &BuzzClient, workflow_id: &str) -> Resu
Ok(())
}
async fn current_workflow_revision(
client: &BuzzClient,
workflow_id: &str,
) -> Result<String, CliError> {
let response = client
.get_authed(&format!("/workflows/{workflow_id}/revision"))
.await?;
let event: serde_json::Value = serde_json::from_str(&response)
.map_err(|e| CliError::Other(format!("invalid workflow revision response: {e}")))?;
event
.get("id")
.and_then(|id| id.as_str())
.map(str::to_owned)
.ok_or_else(|| CliError::NotFound(format!("workflow {workflow_id} not found")))
}
/// Trigger a workflow — sign and submit a kind:46020 event.
///
/// When `inputs` is provided, it is parsed as a JSON object and used as the
@@ -172,6 +188,7 @@ pub async fn cmd_trigger_workflow(
inputs: Option<&str>,
) -> Result<(), CliError> {
let wf_uuid = parse_uuid(workflow_id)?;
let revision = current_workflow_revision(client, workflow_id).await?;
if let Some(raw) = inputs {
// Parse and validate it is a JSON object, then build the event manually
@@ -183,8 +200,12 @@ pub async fn cmd_trigger_workflow(
}
let content = serde_json::to_string(&parsed).unwrap_or_default();
use nostr::{EventBuilder, Kind, Tag};
let tags = vec![Tag::parse(["d", &wf_uuid.to_string()])
.map_err(|e| CliError::Other(format!("tag error: {e}")))?];
let tags = vec![
Tag::parse(["d", &wf_uuid.to_string()])
.map_err(|e| CliError::Other(format!("tag error: {e}")))?,
Tag::parse(["e", revision.as_str()])
.map_err(|e| CliError::Other(format!("tag error: {e}")))?,
];
let builder = EventBuilder::new(
Kind::Custom(buzz_sdk::kind::KIND_WORKFLOW_TRIGGER as u16),
&content,
@@ -194,7 +215,7 @@ pub async fn cmd_trigger_workflow(
let resp = client.submit_event(event).await?;
println!("{}", normalize_write_response(&resp));
} else {
let builder = buzz_sdk::build_workflow_trigger(wf_uuid).map_err(sdk_err)?;
let builder = buzz_sdk::build_workflow_trigger(wf_uuid, &revision).map_err(sdk_err)?;
let event = client.sign_event(builder)?;
let resp = client.submit_event(event).await?;
println!("{}", normalize_write_response(&resp));

4
crates/buzz-db/src/store/channel.rs
View File

@@ -21,8 +21,8 @@ pub use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole};
// Keep the established channel module paths compatible while membership SQL
// and invariants live in their dedicated store module.
pub use crate::channel_members::{
add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members,
get_member_count, get_member_counts_bulk, get_member_role, get_members, get_members_bulk,
acquire_channel_membership_lock, add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members,
get_member_count, get_member_counts_bulk, get_member_role, get_member_role_in_transaction, get_members, get_members_bulk,
get_users_bulk, is_member, list_large_channel_rosters_needing_reconciliation,
lock_member_snapshot, membership_pairs, remove_member, verify_channel_roster_fence_behavior,
verify_channel_roster_fence_catalog, AccessibleChannel, BotChannelEntry, BotMemberRecord,

26
crates/buzz-db/src/store/channel_members.rs
View File

@@ -178,7 +178,7 @@ pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result
/// Take the per-channel membership lock. MUST be the first statement in the
/// transaction that then reads roles/owner counts and writes membership, so the
/// whole check-then-write sequence is atomic against a concurrent one.
async fn acquire_channel_membership_lock(
pub async fn acquire_channel_membership_lock(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
channel_id: Uuid,
@@ -3210,3 +3210,27 @@ mod postgres_tests {
drop_scratch_db(&admin, pool, &scratch_name).await;
}
}
/// Get an active member role using the caller's transaction.
///
/// Authorization callers must acquire [`acquire_channel_membership_lock`] first
/// and hold the transaction through the dependent mutation.
pub async fn get_member_role_in_transaction(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<Option<String>> {
let row = sqlx::query(
"SELECT cm.role::text AS role FROM channel_members cm \
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.fetch_optional(&mut **tx)
.await?;
Ok(row.map(|r| r.try_get("role")).transpose()?)
}

32
crates/buzz-db/src/store/event.rs
View File

@@ -1164,6 +1164,26 @@ pub(crate) async fn get_event_by_id_with_operation(
}
}
/// Fetch a single live event by ID on the caller's transaction.
pub async fn get_event_by_id_in_transaction(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
id_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
let row = sqlx::query(
"SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(id_bytes)
.fetch_optional(&mut **tx)
.await?;
match row {
Some(r) => row_to_stored_event(r),
None => Ok(None),
}
}
/// Fetches the latest global (non-channel, `channel_id IS NULL`) replaceable event
/// for a (kind, pubkey) pair.
///
@@ -1528,6 +1548,18 @@ pub async fn insert_event_with_thread_metadata(
}
impl Db {
/// Fetch a live event by ID using the caller's transaction.
#[datastore_span(name = "get_event_by_id_in_transaction", system = "postgresql")]
pub async fn get_event_by_id_in_transaction(
&self,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
community_id: CommunityId,
id_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
crate::event::get_event_by_id_in_transaction(tx, community_id, id_bytes).await
}
/// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate.
#[datastore_span(name = "insert_event", system = "postgresql")]
pub async fn insert_event(

89
crates/buzz-db/src/store/workflow.rs
View File

@@ -409,6 +409,33 @@ pub async fn get_workflow(
row_to_workflow_record(row)
}
/// Fetch and share-lock one workflow on an existing transaction.
///
/// Definition replacement updates this row, so holding this lock through run
/// creation keeps revision validation and the dependent run atomic.
pub async fn get_workflow_for_share_in_transaction(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
community_id: CommunityId,
id: Uuid,
) -> Result<WorkflowRecord> {
let row = sqlx::query(
r#"
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE community_id = $1 AND id = $2
FOR SHARE
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.fetch_optional(&mut **tx)
.await?
.ok_or_else(|| DbError::NotFound(format!("workflow {id}")))?;
row_to_workflow_record(row)
}
/// List workflows for a channel, ordered newest first.
///
/// `limit` is capped at [`LIST_MAX_LIMIT`]. Pass `None` to use [`LIST_DEFAULT_LIMIT`].
@@ -817,6 +844,34 @@ pub async fn delete_workflow_for_owner(
// -- Workflow Run CRUD --------------------------------------------------------
/// Insert a new exact-revision workflow run on the caller's transaction.
pub async fn create_workflow_run_in_transaction(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
community_id: CommunityId,
workflow_id: Uuid,
definition_event_id: &[u8],
trigger_event_id: Option<&[u8]>,
trigger_context: Option<&serde_json::Value>,
) -> Result<Uuid> {
let id = Uuid::new_v4();
sqlx::query(
r#"
INSERT INTO workflow_runs
(community_id, id, workflow_id, definition_event_id, status, trigger_event_id, current_step, execution_trace, trigger_context)
VALUES ($1, $2, $3, $4, 'pending', $5, 0, '[]', $6)
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.bind(workflow_id)
.bind(definition_event_id)
.bind(trigger_event_id)
.bind(trigger_context)
.execute(&mut **tx)
.await?;
Ok(id)
}
/// Insert a new workflow run. Returns the new run's UUID.
///
/// `trigger_context` is the serialized `TriggerContext` for this run. It is stored
@@ -1297,6 +1352,40 @@ pub async fn find_by_owner_and_name(
// -- Run and approval Db API --------------------------------------------------
impl Db {
/// Fetch and share-lock one workflow on an existing transaction.
#[datastore_span(name = "get_workflow_for_share_in_transaction", system = "postgresql")]
pub async fn get_workflow_for_share_in_transaction(
&self,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
community_id: CommunityId,
id: Uuid,
) -> Result<crate::workflow::WorkflowRecord> {
crate::workflow::get_workflow_for_share_in_transaction(tx, community_id, id).await
}
/// Create an exact-revision workflow run on an existing transaction.
#[datastore_span(name = "create_workflow_run_in_transaction", system = "postgresql")]
pub async fn create_workflow_run_in_transaction(
&self,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
community_id: CommunityId,
workflow_id: Uuid,
definition_event_id: &[u8],
trigger_event_id: Option<&[u8]>,
trigger_context: Option<&serde_json::Value>,
) -> Result<Uuid> {
crate::workflow::create_workflow_run_in_transaction(
tx,
community_id,
workflow_id,
definition_event_id,
trigger_event_id,
trigger_context,
)
.await
}
/// Create a new workflow run.
#[datastore_span(name = "create_workflow_run", system = "postgresql")]
pub async fn create_workflow_run(

66
crates/buzz-relay/src/api/workflows.rs
View File

@@ -46,6 +46,7 @@ async fn authorize_workflow_read(
path: &str,
raw_query: Option<&str>,
workflow_id: Uuid,
allow_immutable_owner: bool,
) -> Result<TenantContext, (StatusCode, Json<Value>)> {
let raw_host = headers
.get(axum::http::header::HOST)
@@ -99,15 +100,59 @@ async fn authorize_workflow_read(
.await
.map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?;
if !accessible.contains(&channel_id) {
return Err(api_error(
StatusCode::FORBIDDEN,
"workflow is not accessible",
));
let controls = allow_immutable_owner
&& (workflow.owner_pubkey == pubkey_bytes
|| state
.db
.is_agent_owner(tenant.community(), &workflow.owner_pubkey, &pubkey_bytes)
.await
.map_err(|error| internal_error(&format!("workflow owner lookup: {error}")))?);
if !controls {
return Err(api_error(
StatusCode::FORBIDDEN,
"workflow is not accessible",
));
}
}
Ok(tenant)
}
/// `GET /workflows/{workflow_id}/revision` — current signed revision for an
/// authorized channel reader or the managed agent's immutable human owner.
///
/// This narrow endpoint does not grant channel visibility; it returns only the
/// exact owner-signed definition event needed to construct a revision-bound
/// manual trigger.
pub async fn workflow_revision(
State(state): State<Arc<AppState>>,
Path(workflow_id): Path<Uuid>,
headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let path = format!("/workflows/{workflow_id}/revision");
let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, true).await?;
let workflow = state
.db
.get_workflow(tenant.community(), workflow_id)
.await
.map_err(|_| api_error(StatusCode::NOT_FOUND, "workflow not found"))?;
let revision = workflow.definition_event_id.as_deref().ok_or_else(|| {
api_error(
StatusCode::CONFLICT,
"owner-signed workflow revision is unavailable",
)
})?;
let event = state
.db
.get_event_by_id(tenant.community(), revision)
.await
.map_err(|error| internal_error(&format!("get workflow revision: {error}")))?
.ok_or_else(|| api_error(StatusCode::CONFLICT, "workflow revision is unavailable"))?;
Ok(Json(serde_json::to_value(event.event).map_err(
|error| internal_error(&format!("serialize workflow revision: {error}")),
)?))
}
/// `GET /workflows/{workflow_id}/runs` — one authorized, keyset-paginated page.
pub async fn workflow_runs(
State(state): State<Arc<AppState>>,
@@ -131,8 +176,15 @@ pub async fn workflow_runs(
}
let path = format!("/workflows/{workflow_id}/runs");
let tenant =
authorize_workflow_read(&state, &headers, &path, raw_query.as_deref(), workflow_id).await?;
let tenant = authorize_workflow_read(
&state,
&headers,
&path,
raw_query.as_deref(),
workflow_id,
false,
)
.await?;
let mut rows = state
.db
.list_workflow_runs_page(
@@ -171,7 +223,7 @@ pub async fn run_approvals(
headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let path = format!("/workflows/{workflow_id}/runs/{run_id}/approvals");
let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id).await?;
let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, false).await?;
let run = state
.db

580
crates/buzz-relay/src/handlers/command_executor.rs
View File

@@ -92,11 +92,11 @@ enum PersistResult {
/// If the event is a duplicate (ON CONFLICT DO NOTHING), the transaction is
/// rolled back and `PersistResult::Duplicate` is returned — no mutations needed.
///
/// Most domain mutations still execute on the connection pool rather than this
/// transaction, so their handlers rely on idempotency if event commit fails.
/// Workflow definition ingest is the exception: it materializes the workflow
/// revision on this same transaction so the signed event and revision pointer
/// commit or roll back together.
/// NOTE: Most domain mutations still execute on the connection pool rather
/// than in this transaction. Workflow-definition ingestion is the exception:
/// its materialized workflow row and exact signed revision are written through
/// this transaction so the event and revision binding commit atomically.
/// Other operations remain idempotent but not strictly atomic.
#[datastore_span(name = "persist_command_event", system = "postgresql")]
async fn persist_command_event(
db: &buzz_db::Db,
@@ -736,7 +736,8 @@ async fn handle_workflow_def(
.map_err(|e| IngestError::Internal(format!("error: json serialize: {e}")))?;
let hash = compute_definition_hash(&definition_json_final);
// Persist the signed definition and materialized revision atomically.
// Persist the command event — returns the transaction that will also own
// the materialized workflow revision update.
let mut tx = match persist_command_event(&state.db, tenant, event, Some(channel_id)).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
@@ -787,11 +788,14 @@ async fn handle_workflow_def(
other => IngestError::Internal(format!("error: db upsert_workflow: {other}")),
})?;
// Commit before cache invalidation so a concurrent refill can observe the new revision.
// Commit the event transaction after the idempotent workflow upsert succeeds.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
// Invalidate only after commit. Invalidating while the new row is still
// invisible lets a concurrent trigger refill the cache with the old
// definition and retain it until TTL expiry.
state
.workflow_engine
.invalidate_channel_workflows(community_id, channel_id);
@@ -811,6 +815,101 @@ async fn handle_workflow_def(
})
}
async fn caller_controls_workflow(
state: &Arc<AppState>,
community_id: CommunityId,
workflow_owner: &[u8],
caller: &[u8],
) -> Result<bool, IngestError> {
if workflow_owner == caller {
return Ok(true);
}
state
.db
.is_agent_owner(community_id, workflow_owner, caller)
.await
.map_err(|e| IngestError::Internal(format!("error: workflow owner check: {e}")))
}
fn exact_tag_value<'a>(event: &'a Event, name: &str) -> Option<&'a str> {
let mut values = event.tags.iter().filter_map(|tag| {
(tag.kind().to_string() == name)
.then(|| tag.content())
.flatten()
});
let value = values.next()?;
values.next().is_none().then_some(value)
}
async fn verify_workflow_revision(
state: &Arc<AppState>,
mut tx: Option<&mut sqlx::Transaction<'_, sqlx::Postgres>>,
community_id: CommunityId,
workflow: &buzz_db::workflow::WorkflowRecord,
requested_revision: &[u8],
) -> Result<(), IngestError> {
let Some(persisted_revision) = workflow.definition_event_id.as_deref() else {
return Err(IngestError::Rejected(
"invalid: owner-signed workflow revision is unavailable".into(),
));
};
if persisted_revision != requested_revision {
return Err(IngestError::Rejected(
"conflict: workflow revision does not match current definition".into(),
));
}
let stored = match tx.as_mut() {
Some(tx) => {
state
.db
.get_event_by_id_in_transaction(tx, community_id, persisted_revision)
.await
}
None => {
state
.db
.get_event_by_id(community_id, persisted_revision)
.await
}
}
.map_err(|e| IngestError::Internal(format!("error: workflow revision lookup: {e}")))?
.ok_or_else(|| IngestError::Rejected("invalid: signed workflow revision not found".into()))?;
let definition_event = &stored.event;
let workflow_id = workflow.id.to_string();
let workflow_channel_id = workflow.channel_id.map(|id| id.to_string());
if definition_event.id.as_bytes() != persisted_revision
|| !definition_event.verify_id()
|| !definition_event.verify_signature()
|| definition_event.kind.as_u16() as u32 != KIND_WORKFLOW_DEF
|| definition_event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey
|| exact_tag_value(definition_event, "d") != Some(workflow_id.as_str())
|| workflow_channel_id.is_none()
|| exact_tag_value(definition_event, "h") != workflow_channel_id.as_deref()
|| stored.channel_id != workflow.channel_id
{
return Err(IngestError::Rejected(
"invalid: signed workflow revision binding mismatch".into(),
));
}
let (_, signed_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition_event.content)
.map_err(|_| {
IngestError::Rejected("invalid: signed workflow revision is malformed".into())
})?;
let signed_definition: serde_json::Value =
serde_json::from_str(&signed_json).map_err(|_| {
IngestError::Rejected("invalid: signed workflow revision is malformed".into())
})?;
if signed_definition != webhook_secret::strip_secret(&workflow.definition) {
return Err(IngestError::Rejected(
"invalid: signed workflow revision differs from materialized definition".into(),
));
}
Ok(())
}
async fn handle_workflow_trigger(
tenant: &TenantContext,
state: &Arc<AppState>,
@@ -819,13 +918,21 @@ async fn handle_workflow_trigger(
) -> Result<IngestResult, IngestError> {
let self_bytes = auth.pubkey().to_bytes().to_vec();
// 1. Extract workflow reference from `d` tag or `e` tag
let workflow_id_str = extract_d_tag(event)
.or_else(|| extract_e_tag(event))
.ok_or_else(|| {
IngestError::Rejected("invalid: missing workflow reference (d or e tag)".into())
})?;
let workflow_id = Uuid::parse_str(&workflow_id_str)
// 1. Bind the command to both the workflow UUID and one exact signed revision.
let workflow_id_str = exact_tag_value(event, "d").ok_or_else(|| {
IngestError::Rejected("invalid: expected exactly one workflow d tag".into())
})?;
let revision_hex = exact_tag_value(event, "e").ok_or_else(|| {
IngestError::Rejected("invalid: expected exactly one workflow revision e tag".into())
})?;
let requested_revision = hex::decode(revision_hex)
.map_err(|_| IngestError::Rejected("invalid: bad workflow revision event id".into()))?;
if requested_revision.len() != 32 {
return Err(IngestError::Rejected(
"invalid: bad workflow revision event id".into(),
));
}
let workflow_id = Uuid::parse_str(workflow_id_str)
.map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?;
// 2. Validate workflow exists — scoped to the caller's community. The same
@@ -839,14 +946,20 @@ async fn handle_workflow_trigger(
.await
.map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?;
// 3. Manual triggers execute with the workflow owner's authority, so only
// the owner may start them. Channel membership alone is insufficient: a
// member could otherwise invoke another user's webhook or message actions.
if workflow.owner_pubkey != self_bytes {
// 3. Manual triggers execute with the workflow owner's authority. Permit
// that principal and, for a managed agent, its immutable human owner.
// Channel membership alone remains insufficient.
if !caller_controls_workflow(state, community_id, &workflow.owner_pubkey, &self_bytes).await? {
return Err(IngestError::Rejected(
"forbidden: not authorized to trigger this workflow".into(),
));
}
// Managed-agent ownership is immutable. Carry the authorized workflow
// principal across the transaction boundary so no pool-backed ownership
// lookup is attempted while the command transaction holds its connection.
let authorized_workflow_owner = workflow.owner_pubkey.clone();
verify_workflow_revision(state, None, community_id, &workflow, &requested_revision).await?;
// SEC-006: manual triggers must honor the workflow's lifecycle state and
// recheck the owner's *current* channel authority before creating a run.
@@ -876,7 +989,7 @@ async fn handle_workflow_trigger(
// Persist the command event under the workflow channel even though the
// trigger event itself only carries the workflow UUID. Storing channel
// triggers as global events leaks workflow IDs to unrelated relay members.
let tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? {
let mut tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -887,6 +1000,57 @@ async fn handle_workflow_trigger(
PersistResult::Inserted(tx) => tx,
};
// Serialize the final authority check and run commit with channel
// membership writers. If revocation commits first we observe no role; if
// this lock wins, revocation cannot commit until this run is durable.
buzz_db::channel::acquire_channel_membership_lock(&mut tx, community_id, wf_channel_id)
.await
.map_err(|e| IngestError::Internal(format!("error: membership lock: {e}")))?;
// Re-read the workflow under a row lock on the same transaction that will
// commit the trigger event and run. Definition replacement updates this row,
// so it cannot commit between this exact-revision check and our commit. A
// replacement that won first is observed here and rejected as stale.
let workflow = state
.db
.get_workflow_for_share_in_transaction(&mut tx, community_id, workflow_id)
.await
.map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?;
if workflow.owner_pubkey != authorized_workflow_owner {
return Err(IngestError::Rejected(
"conflict: workflow owner changed while trigger was being processed".into(),
));
}
verify_workflow_revision(
state,
Some(&mut tx),
community_id,
&workflow,
&requested_revision,
)
.await?;
if !workflow.enabled || workflow.status != buzz_db::workflow::WorkflowStatus::Active {
return Err(IngestError::Rejected(
"forbidden: workflow is disabled or inactive".into(),
));
}
let role = buzz_db::channel::get_member_role_in_transaction(
&mut tx,
community_id,
wf_channel_id,
&workflow.owner_pubkey,
)
.await
.map_err(|e| IngestError::Internal(format!("error: owner authority lookup: {e}")))?;
if !matches!(
(role.as_deref(), def.requires_elevated_authority()),
(Some(_), false) | (Some("owner" | "admin"), true)
) {
return Err(IngestError::Rejected(
"forbidden: not authorized to trigger this workflow".into(),
));
}
// 4. Execute: create workflow run
let mut trigger_ctx = TriggerContext {
channel_id: workflow
@@ -912,10 +1076,11 @@ async fn handle_workflow_trigger(
let event_id_bytes = event.id.as_bytes().to_vec();
let run_id = state
.db
.create_workflow_run(
.create_workflow_run_in_transaction(
&mut tx,
community_id,
workflow_id,
workflow.definition_event_id.as_deref(),
&requested_revision,
Some(&event_id_bytes),
trigger_ctx_json.as_ref(),
)
@@ -929,44 +1094,23 @@ async fn handle_workflow_trigger(
// 5. Spawn workflow execution
let engine = Arc::clone(&state.workflow_engine);
let db = state.db.clone();
let def_value = workflow.definition.clone();
let trigger_ctx_clone = trigger_ctx.clone();
tokio::spawn(async move {
let def: buzz_workflow::WorkflowDef = match serde_json::from_value(def_value) {
Ok(d) => d,
Err(e) => {
tracing::error!("workflow_trigger: failed to parse definition: {e}");
if let Err(db_err) = db
.update_workflow_run(
community_id,
run_id,
RunStatus::Failed,
0,
&serde_json::json!([]),
Some(buzz_db::workflow::WorkflowRunFailure {
code: "invalid_definition",
message: &format!("definition parse error: {e}"),
}),
)
.await
{
tracing::error!("workflow_trigger: failed to mark run as failed: {db_err}");
}
return;
let result = match engine.load_run_definition(community_id, run_id).await {
Ok((_, definition)) => {
buzz_workflow::executor::execute_from_step(
&engine,
community_id,
run_id,
&definition,
&trigger_ctx_clone,
0,
None,
)
.await
}
Err(error) => Err((error, buzz_workflow::error::PartialProgress::default())),
};
let result = buzz_workflow::executor::execute_from_step(
&engine,
community_id,
run_id,
&def,
&trigger_ctx_clone,
0,
None,
)
.await;
engine
.finalize_run(community_id, run_id, result, None)
.await;
@@ -1375,7 +1519,10 @@ mod postgres_tests {
let url = std::env::var("BUZZ_TEST_DATABASE_URL")
.or_else(|_| std::env::var("DATABASE_URL"))
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); // sadscan:disable np.postgres.1 -- local test-only credentials
let pool = sqlx::PgPool::connect(&url)
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(1)
.acquire_timeout(std::time::Duration::from_secs(1))
.connect(&url)
.await
.expect("connect workflow persistence test database");
let db = buzz_db::Db::from_pool(pool);
@@ -1393,6 +1540,180 @@ mod postgres_tests {
(db, TenantContext::resolved(community, host))
}
async fn manual_trigger_test_context() -> (Arc<AppState>, TenantContext, Keys, Keys, Uuid, Event)
{
use buzz_core::channel::{ChannelType, ChannelVisibility};
let url = std::env::var("BUZZ_TEST_DATABASE_URL")
.or_else(|_| std::env::var("DATABASE_URL"))
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string());
let setup_pool = sqlx::PgPool::connect(&url)
.await
.expect("connect workflow trigger setup database");
let setup_db = buzz_db::Db::from_pool(setup_pool.clone());
setup_db
.migrate()
.await
.expect("migrate workflow trigger test database");
let host = format!("workflow-trigger-{}.example", Uuid::new_v4().simple());
let community = setup_db
.ensure_configured_community(&host)
.await
.expect("create workflow trigger test community")
.id;
let tenant = TenantContext::resolved(community, host.clone());
let human = Keys::generate();
let agent = Keys::generate();
let human_bytes = human.public_key().to_bytes();
let agent_bytes = agent.public_key().to_bytes();
setup_db
.ensure_user(community, &human_bytes)
.await
.expect("ensure human owner");
setup_db
.ensure_user(community, &agent_bytes)
.await
.expect("ensure managed agent");
assert!(setup_db
.set_agent_owner(community, &agent_bytes, &human_bytes)
.await
.expect("set immutable agent owner"));
let channel = setup_db
.create_channel(
community,
"manual-trigger-pool",
ChannelType::Stream,
ChannelVisibility::Open,
None,
&agent_bytes,
None,
)
.await
.expect("create workflow channel");
let workflow_id = Uuid::new_v4();
let definition = EventBuilder::new(
Kind::Custom(KIND_WORKFLOW_DEF as u16),
concat!(
"name: manual-trigger-pool\n",
"trigger:\n on: message_posted\n",
"steps:\n - id: send\n action: send_message\n text: done\n",
),
)
.tags(vec![
Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"),
Tag::parse(["h", channel.id.to_string().as_str()]).expect("h tag"),
])
.sign_with_keys(&agent)
.expect("sign workflow definition");
let (_, definition_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition.content)
.expect("parse signed workflow definition");
let definition_hash = compute_definition_hash(&definition_json);
let mut tx = setup_db
.begin_transaction()
.await
.expect("begin workflow seed");
buzz_db::event::insert_event_in_transaction(
&mut tx,
community,
&definition,
Some(channel.id),
)
.await
.expect("persist signed workflow definition");
setup_db
.upsert_workflow(
&mut tx,
community,
workflow_id,
Some(channel.id),
&agent_bytes,
"manual-trigger-pool",
&definition_json,
&definition_hash,
definition.id.as_bytes(),
)
.await
.expect("materialize signed workflow");
tx.commit().await.expect("commit signed workflow");
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(1)
.acquire_timeout(std::time::Duration::from_secs(1))
.connect(&url)
.await
.expect("connect one-connection workflow trigger pool");
let db = buzz_db::Db::from_pool(pool.clone());
let mut config = crate::config::Config::from_env().expect("config from env");
config.database_url = url;
config.redis_url = "redis://127.0.0.1:1".to_string();
config.relay_url = format!("wss://{host}");
config.require_relay_membership = false;
let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
.create_pool(Some(deadpool_redis::Runtime::Tokio1))
.expect("redis pool config");
let pubsub = Arc::new(
buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
.await
.expect("pubsub manager"),
);
let audit = buzz_audit::AuditService::new(pool.clone());
let auth = buzz_auth::AuthService::new(config.auth.clone());
let search = buzz_search::SearchService::new(pool);
let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new(
db.clone(),
buzz_workflow::WorkflowConfig::default(),
));
let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage");
let (state, _audit_shutdown) = AppState::new(
config,
db,
redis_pool,
audit,
pubsub,
auth,
search,
workflow_engine,
Keys::generate(),
media_storage,
);
setup_pool.close().await;
(
Arc::new(state),
tenant,
human,
agent,
workflow_id,
definition,
)
}
fn workflow_trigger_event_for_revision(
keys: &Keys,
workflow_id: Uuid,
revision: &str,
) -> Event {
EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "")
.tags(vec![
Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"),
Tag::parse(["e", revision]).expect("revision tag"),
])
.sign_with_keys(keys)
.expect("sign workflow trigger")
}
fn workflow_trigger_event(keys: &Keys, workflow_id: Uuid, revision: &Event) -> Event {
workflow_trigger_event_for_revision(keys, workflow_id, &revision.id.to_hex())
}
fn http_auth(keys: &Keys) -> IngestAuth {
IngestAuth::Http {
pubkey: keys.public_key(),
scopes: vec![buzz_auth::Scope::MessagesWrite],
auth_method: super::super::ingest::HttpAuthMethod::Nip98,
}
}
fn workflow_event(
keys: &Keys,
workflow_id: Uuid,
@@ -1428,6 +1749,134 @@ mod postgres_tests {
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn human_owner_manual_trigger_completes_with_one_connection() {
let (state, tenant, human, agent, workflow_id, revision) =
manual_trigger_test_context().await;
let trigger = workflow_trigger_event(&human, workflow_id, &revision);
let result = tokio::time::timeout(
std::time::Duration::from_secs(3),
handle_workflow_trigger(&tenant, &state, &trigger, &http_auth(&human)),
)
.await
.expect("human-owner trigger must not wait for a second pool connection")
.expect("human-owner trigger must succeed");
let run_id = Uuid::parse_str(
serde_json::from_str::<serde_json::Value>(
result
.message
.strip_prefix("response:")
.expect("workflow trigger response prefix"),
)
.expect("workflow trigger response JSON")["run_id"]
.as_str()
.expect("workflow trigger run id"),
)
.expect("workflow trigger run UUID");
let (_, loaded_definition) = state
.workflow_engine
.load_run_definition(tenant.community(), run_id)
.await
.expect("manual execution must load its exact signed revision");
assert_eq!(loaded_definition.name, "manual-trigger-pool");
let agent_trigger = workflow_trigger_event(&agent, workflow_id, &revision);
let agent_result =
handle_workflow_trigger(&tenant, &state, &agent_trigger, &http_auth(&agent))
.await
.expect("workflow principal must be able to trigger its own workflow");
assert!(agent_result.message.contains("\"run_id\""));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn manual_trigger_rejects_non_owner_and_stale_or_missing_revision() {
let (state, tenant, human, _agent, workflow_id, revision) =
manual_trigger_test_context().await;
let stranger = Keys::generate();
let unauthorized = workflow_trigger_event(&stranger, workflow_id, &revision);
let unauthorized_error =
match handle_workflow_trigger(&tenant, &state, &unauthorized, &http_auth(&stranger))
.await
{
Err(error) => error,
Ok(_) => panic!("channel membership must not grant manual trigger authority"),
};
assert!(matches!(
unauthorized_error,
IngestError::Rejected(ref message)
if message == "forbidden: not authorized to trigger this workflow"
));
let stale = workflow_trigger_event_for_revision(&human, workflow_id, &"42".repeat(32));
let stale_error =
match handle_workflow_trigger(&tenant, &state, &stale, &http_auth(&human)).await {
Err(error) => error,
Ok(_) => panic!("a stale signed revision must be rejected"),
};
assert!(matches!(
stale_error,
IngestError::Rejected(ref message)
if message == "conflict: workflow revision does not match current definition"
));
let missing = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "")
.tag(Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"))
.sign_with_keys(&human)
.expect("sign revision-less trigger");
let missing_error =
match handle_workflow_trigger(&tenant, &state, &missing, &http_auth(&human)).await {
Err(error) => error,
Ok(_) => panic!("a revision-less trigger must fail closed"),
};
assert!(matches!(
missing_error,
IngestError::Rejected(ref message)
if message == "invalid: expected exactly one workflow revision e tag"
));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn concurrent_human_owner_manual_triggers_do_not_starve_one_connection_pool() {
let (state, tenant, human, _agent, workflow_id, revision) =
manual_trigger_test_context().await;
let triggers = (0..8)
.map(|_| workflow_trigger_event(&human, workflow_id, &revision))
.collect::<Vec<_>>();
let results = tokio::time::timeout(std::time::Duration::from_secs(8), async {
let mut tasks = tokio::task::JoinSet::new();
for trigger in triggers {
let state = Arc::clone(&state);
let tenant = tenant.clone();
let auth = http_auth(&human);
tasks.spawn(async move {
handle_workflow_trigger(&tenant, &state, &trigger, &auth).await
});
}
let mut results = Vec::new();
while let Some(result) = tasks.join_next().await {
results.push(result.expect("trigger task must not panic"));
}
results
})
.await
.expect("concurrent triggers must drain rather than pool-starve");
assert_eq!(results.len(), 8);
for result in results {
assert!(
result
.expect("concurrent human-owner trigger must succeed")
.accepted,
"manual trigger should be accepted"
);
}
}
#[test]
fn workflow_revision_parser_accepts_create_and_valid_update() {
let revision = [0x42; 32];
@@ -1476,6 +1925,10 @@ mod postgres_tests {
let workflow_id = Uuid::new_v4();
let created_at = Timestamp::now().as_secs();
let create = workflow_event(&keys, workflow_id, created_at, None, "create");
let channel_id = Uuid::parse_str(
exact_tag_value(&create, "h").expect("workflow definition channel tag"),
)
.expect("workflow definition channel UUID");
let missing_revision = hex::encode([0x24; 32]);
let missing_revision_update = workflow_event(
@@ -1496,15 +1949,22 @@ mod postgres_tests {
if message == "conflict: workflow revision does not exist"
));
let PersistResult::Inserted(tx) = persist_command_event(&db, &tenant, &create, None)
.await
.expect("persist create")
let PersistResult::Inserted(tx) =
persist_command_event(&db, &tenant, &create, Some(channel_id))
.await
.expect("persist create")
else {
panic!("first create must insert");
};
tx.commit().await.expect("commit create");
let stored_create = db
.get_event_by_id(tenant.community(), create.id.as_bytes())
.await
.expect("load persisted workflow definition")
.expect("persisted workflow definition");
assert_eq!(stored_create.channel_id, Some(channel_id));
assert!(matches!(
persist_command_event(&db, &tenant, &create, None)
persist_command_event(&db, &tenant, &create, Some(channel_id))
.await
.expect("replay create"),
PersistResult::Duplicate

4
crates/buzz-relay/src/router.rs
View File

@@ -76,6 +76,10 @@ pub fn build_router(state: Arc<AppState>) -> Router {
// Relay-owned third-party GIF metadata proxy (NIP-98 auth).
.route(api::gifs::SEARCH_PATH, post(api::gifs::search))
.route(api::gifs::SHARE_PATH, post(api::gifs::share))
.route(
"/workflows/{workflow_id}/revision",
get(api::workflows::workflow_revision),
)
.route(
"/workflows/{workflow_id}/runs",
get(api::workflows::workflow_runs),

17
crates/buzz-sdk/src/builders.rs
View File

@@ -1658,9 +1658,16 @@ pub fn build_workflow_delete(
build_delete_addressable(KIND_WORKFLOW_DEF, author_pubkey, &workflow_id.to_string())
}
/// Build a workflow trigger event (kind 46020).
pub fn build_workflow_trigger(workflow_id: Uuid) -> Result<EventBuilder, SdkError> {
let tags = vec![tag(&["d", &workflow_id.to_string()])?];
/// Build a workflow trigger event (kind 46020) bound to an exact signed revision.
pub fn build_workflow_trigger(
workflow_id: Uuid,
definition_event_id: &str,
) -> Result<EventBuilder, SdkError> {
let revision = check_hex_exact(definition_event_id, 64, "definition_event_id")?;
let tags = vec![
tag(&["d", &workflow_id.to_string()])?,
tag(&["e", &revision])?,
];
Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "").tags(tags))
}
@@ -4113,9 +4120,11 @@ mod tests {
#[test]
fn workflow_trigger_happy_path() {
let wid = uuid();
let ev = sign(build_workflow_trigger(wid).unwrap());
let ev = sign(build_workflow_trigger(wid, &"ab".repeat(32)).unwrap());
assert_eq!(ev.kind.as_u16(), 46020);
assert!(has_tag(&ev, "d", &wid.to_string()));
assert!(has_tag(&ev, "e", &"ab".repeat(32)));
assert!(build_workflow_trigger(wid, "not-an-event-id").is_err());
}
#[test]

60
crates/buzz-workflow/src/lib.rs
View File

@@ -44,7 +44,9 @@ use std::collections::HashMap;
use std::sync::Arc;
use std::sync::OnceLock;
use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION};
use buzz_core::kind::{
event_kind_u32, is_workflow_execution_kind, KIND_REACTION, KIND_WORKFLOW_DEF,
};
use buzz_core::tenant::CommunityId;
use buzz_db::workflow::RunStatus;
use buzz_db::Db;
@@ -122,6 +124,62 @@ impl WorkflowEngine {
}
}
/// Load and verify the exact owner-signed definition bound to a run.
///
/// The mutable `workflows` row supplies only immutable identity/channel
/// binding. Definition content always comes from the run's signed event;
/// legacy runs without a revision fail closed.
pub async fn load_run_definition(
&self,
community_id: CommunityId,
run_id: Uuid,
) -> Result<(buzz_db::workflow::WorkflowRunRecord, WorkflowDef), WorkflowError> {
let run = self.db.get_workflow_run(community_id, run_id).await?;
let revision = run.definition_event_id.as_deref().ok_or_else(|| {
WorkflowError::InvalidDefinition(
"workflow run has no owner-signed definition revision".into(),
)
})?;
let workflow = self.db.get_workflow(community_id, run.workflow_id).await?;
let stored = self
.db
.get_event_by_id_including_deleted(community_id, revision)
.await?
.ok_or_else(|| {
WorkflowError::InvalidDefinition(
"workflow run definition event is unavailable".into(),
)
})?;
let event = &stored.event;
let workflow_id = run.workflow_id.to_string();
let channel_id = workflow.channel_id.map(|id| id.to_string());
let exact_tag = |name: &str| {
let mut values = event.tags.iter().filter_map(|tag| {
(tag.kind().to_string() == name)
.then(|| tag.content())
.flatten()
});
let value = values.next();
value.filter(|_| values.next().is_none())
};
if event.id.as_bytes() != revision
|| !event.verify_id()
|| !event.verify_signature()
|| event_kind_u32(event) != KIND_WORKFLOW_DEF
|| event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey
|| exact_tag("d") != Some(workflow_id.as_str())
|| channel_id.is_none()
|| exact_tag("h") != channel_id.as_deref()
|| stored.channel_id != workflow.channel_id
{
return Err(WorkflowError::InvalidDefinition(
"workflow run definition event binding mismatch".into(),
));
}
let (definition, _) = Self::parse_yaml(&event.content)?;
Ok((run, definition))
}
/// Drop the cached enabled-workflow list for a channel.
///
/// Must be called after any write to a workflow's trigger eligibility or

8
desktop/src-tauri/src/commands/workflows.rs
View File

@@ -323,7 +323,13 @@ pub async fn trigger_workflow(
workflow_id: String,
state: State<'_, AppState>,
) -> Result<WorkflowTriggerWire, String> {
let builder = events::build_workflow_trigger(&workflow_id)?;
// Resolve the current signed definition at trigger time. The relay binds
// authorization and execution to this exact revision and rejects a stale
// result if an update races this command.
let revision: nostr::Event =
get_relay_json(&state, &format!("/workflows/{workflow_id}/revision")).await?;
let definition_event_id = revision.id.to_hex();
let builder = events::build_workflow_trigger(&workflow_id, &definition_event_id)?;
let result = submit_event(builder, &state).await?;
trigger_wire_from_message(workflow_id, &result.message)
}

13
desktop/src-tauri/src/events/workflows.rs
View File

@@ -31,9 +31,16 @@ pub fn build_workflow_delete(
Ok(EventBuilder::new(Kind::Custom(5), "").tags(tags))
}
/// Kind 46020 — trigger a workflow run by id.
pub fn build_workflow_trigger(workflow_id: &str) -> Result<EventBuilder, String> {
let tags = vec![tag(vec!["d", workflow_id])?];
/// Kind 46020 — trigger a workflow run by id, bound to one exact definition revision.
pub fn build_workflow_trigger(
workflow_id: &str,
definition_event_id: &str,
) -> Result<EventBuilder, String> {
EventId::from_hex(definition_event_id).map_err(|_| "invalid workflow revision".to_string())?;
let tags = vec![
tag(vec!["d", workflow_id])?,
tag(vec!["e", definition_event_id])?,
];
Ok(EventBuilder::new(Kind::Custom(46020), "").tags(tags))
}