Skip to content
Closed
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 2 additions & 0 deletions crates/buzz-db/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1088,6 +1088,7 @@ impl Db {
/// Create a new workflow.
pub async fn create_workflow(
&self,
id: Uuid,
channel_id: Option<Uuid>,
owner_pubkey: &[u8],
name: &str,
Expand All @@ -1096,6 +1097,7 @@ impl Db {
) -> Result<Uuid> {
workflow::create_workflow(
&self.pool,
id,
channel_id,
owner_pubkey,
name,
Expand Down
3 changes: 1 addition & 2 deletions crates/buzz-db/src/workflow.rs
Original file line number Diff line number Diff line change
Expand Up @@ -244,14 +244,13 @@ pub struct ApprovalRecord {
/// New workflows start as `active` and `enabled = TRUE`.
pub async fn create_workflow(
pool: &PgPool,
id: Uuid,
channel_id: Option<Uuid>,
owner_pubkey: &[u8],
name: &str,
definition_json: &str,
definition_hash: &[u8],
) -> Result<Uuid> {
let id = Uuid::new_v4();

sqlx::query(
r#"
INSERT INTO workflows
Expand Down
81 changes: 65 additions & 16 deletions crates/buzz-relay/src/handlers/command_executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -568,6 +568,13 @@ async fn handle_workflow_def(
IngestError::Rejected("invalid: missing workflow name (name or d tag)".into())
})?;

// Parse the d-tag as the workflow_id UUID
let d_tag = extract_d_tag(event)
.ok_or_else(|| IngestError::Rejected("invalid: missing d tag (workflow ID)".into()))?;
let workflow_id = Uuid::parse_str(&d_tag).map_err(|_| {
IngestError::Rejected("invalid: workflow ID in d tag must be a valid UUID".into())
})?;

// 2. Validate caller has channel access (minimum: is a member)
let is_member = state
.is_member_cached(channel_id, &self_bytes)
Expand All @@ -579,6 +586,31 @@ async fn handle_workflow_def(
));
}

// Check if workflow already exists to perform update or create checks
let existing = state.db.get_workflow(workflow_id).await;
Comment thread
AaronGoldsmith marked this conversation as resolved.
Outdated
let existing_record = match existing {
Ok(record) => {
if record.owner_pubkey != self_bytes {
return Err(IngestError::Rejected(
"forbidden: cannot update a workflow owned by another user".into(),
));
}
if record.channel_id != Some(channel_id) {
Comment thread
AaronGoldsmith marked this conversation as resolved.
Outdated
return Err(IngestError::Rejected(
"forbidden: cannot change the channel of an existing workflow".into(),
));
}
Some(record)
}
Err(buzz_db::DbError::NotFound(_)) => None,
Err(e) => {
return Err(IngestError::Internal(format!(
"error: db lookup workflow: {e}"
)));
}
};
let is_update = existing_record.is_some();

// 3. Parse YAML from event.content
let (def, definition_json_str) = buzz_workflow::WorkflowEngine::parse_yaml(&event.content)
.map_err(|e| IngestError::Rejected(format!("invalid: workflow YAML parse error: {e}")))?;
Expand All @@ -588,9 +620,17 @@ async fn handle_workflow_def(

// Generate webhook secret if this workflow uses a Webhook trigger
let webhook_secret = if matches!(def.trigger, buzz_workflow::TriggerDef::Webhook) {
let secret = webhook_secret::generate_webhook_secret();
webhook_secret::inject_secret(&mut definition_json, &secret);
Some(secret)
let existing_secret = existing_record
Comment thread
AaronGoldsmith marked this conversation as resolved.
Outdated
.as_ref()
.and_then(|r| crate::webhook_secret::extract_secret(&r.definition));
if let Some(secret) = existing_secret {
webhook_secret::inject_secret(&mut definition_json, &secret);
None
} else {
let secret = webhook_secret::generate_webhook_secret();
webhook_secret::inject_secret(&mut definition_json, &secret);
Some(secret)
}
} else {
None
};
Expand All @@ -612,20 +652,29 @@ async fn handle_workflow_def(
PersistResult::Inserted(tx) => tx,
};

// 4. Execute: create_workflow
let workflow_id = state
.db
.create_workflow(
Some(channel_id),
&self_bytes,
&workflow_name,
&definition_json_final,
&hash,
)
.await
.map_err(|e| IngestError::Internal(format!("error: db create_workflow: {e}")))?;
// 4. Execute: update_workflow or create_workflow
if is_update {
state
.db
.update_workflow(workflow_id, &workflow_name, &definition_json_final, &hash)
.await
.map_err(|e| IngestError::Internal(format!("error: db update_workflow: {e}")))?;
Comment thread
AaronGoldsmith marked this conversation as resolved.
Outdated
} else {
state
.db
.create_workflow(
workflow_id,
Some(channel_id),
&self_bytes,
&workflow_name,
&definition_json_final,
&hash,
)
.await
.map_err(|e| IngestError::Internal(format!("error: db create_workflow: {e}")))?;
}

// Commit: event + workflow creation succeeded atomically.
// Commit: event + workflow creation/update succeeded atomically.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
Expand Down
24 changes: 18 additions & 6 deletions crates/buzz-relay/src/handlers/side_effects.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1586,12 +1586,24 @@ async fn handle_a_tag_deletion(event: &Event, state: &Arc<AppState>) -> anyhow::
buzz_core::kind::KIND_WORKFLOW_DEF => {
// Try UUID first (workflow_id); fall back to name-based lookup.
if let Ok(wf_id) = uuid::Uuid::parse_str(d_tag) {
state
.db
.delete_workflow(wf_id)
.await
.map_err(|e| anyhow::anyhow!("failed to delete workflow {wf_id}: {e}"))?;
tracing::info!(workflow_id = %wf_id, "Workflow deleted via NIP-09 a-tag (UUID)");
let owner_bytes = hex::decode(pubkey_hex).unwrap_or_default();
match state.db.get_workflow(wf_id).await {
Ok(wf) => {
if wf.owner_pubkey != owner_bytes {
return Err(anyhow::anyhow!("forbidden: deletion owner mismatch"));
Comment thread
AaronGoldsmith marked this conversation as resolved.
Outdated
}
state.db.delete_workflow(wf_id).await.map_err(|e| {
anyhow::anyhow!("failed to delete workflow {wf_id}: {e}")
})?;
tracing::info!(workflow_id = %wf_id, "Workflow deleted via NIP-09 a-tag (UUID)");
}
Err(buzz_db::DbError::NotFound(_)) => {
tracing::warn!("NIP-09 a-tag deletion: no workflow '{wf_id}' found");
}
Err(e) => {
return Err(anyhow::anyhow!("failed to lookup workflow for delete: {e}"));
}
}
} else {
// Name-based lookup
let owner_bytes = hex::decode(pubkey_hex).unwrap_or_default();
Expand Down
2 changes: 2 additions & 0 deletions crates/buzz-test-client/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ url = { workspace = true }
rustls = "0.23"

[dev-dependencies]
buzz-db = { workspace = true }
buzz-sdk = { workspace = true }
tracing-subscriber = { workspace = true }
uuid = { workspace = true }
futures-util = { workspace = true }
Expand Down
Loading