External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflow): scope workflow execution and approvals to their community

Browse Source 
`workflows`, `workflow_runs`, and `workflow_approvals` are all keyed
`(community_id, id|token)`, so the same UUID/token is structurally allowed
in two communities — exactly like channels and events. But the execution
and approval spine still fetched, listed, mutated, and posted by bare id,
so a webhook/manual trigger or NIP-09 deletion in community B could load,
drive, or erase community A's colliding workflow, and workflow side effects
were published under the deployment/default tenant instead of the run's own
community. This threads the owning community through every request-scoped
and run-scoped path so each lookup, write, and side effect is confined to
its tenant.

- `ActionSink::send_message` now takes the run's `community_id` as its first
  parameter. `RelayActionSink` drops `bind_deployment_community(relay_url)` —
  the Issue-4 root cause — and instead resolves the run community's host via
  `lookup_community_host` to form a complete `TenantContext::resolved`, fail
  closed if the community is unmapped. A workflow in B now posts into B.

- Executor (`dispatch_action`, `execute_run`, `execute_from_step`,
  `execute_steps`) and engine (`finalize_run`, `on_event`) carry the run's
  community; every `get_workflow_run` / `get_workflow` / `send_message` and
  the post-store `on_event` call (from `dispatch_persistent_event`, which has
  the bound `tenant`) are scoped. The interval `last_fired` DashMap is keyed
  `(CommunityId, Uuid)` so duplicate workflow UUIDs across communities cannot
  cross-suppress in memory.

- Webhook `/hooks/{id}` now binds its community from the request Host before
  any lookup (`bind_community`), then `get_workflow(community, id)`. The host
  — not the workflow row — determines the tenant, so a request to A's host
  can only reach A's workflows; unmapped host and not-found both fail closed
  with the same generic 404.

- WS manual trigger and `create_workflow` use `tenant.community()` as the
  authoritative owner. `create_workflow` no longer resolves the community via
  the ambiguous `community_of_channel(channel_id)`; it verifies the channel
  exists *inside* the bound community via scoped `get_channel` (the same
  guarantee the composite FK enforces, surfaced as a clean rejection).

- Approval grant/deny/resume handlers and the `buzz-db` approval methods
  (`get_approval`, `get_approval_by_stored_hash`, `get_run_approvals`,
  `update_approval`, `update_approval_by_stored_hash`, `create_approval`)
  are scoped by community; `create_approval`'s INSERT now includes the
  `community_id` NOT-NULL column it previously omitted. NIP-09 a-tag workflow
  deletion (`delete_workflow`, `find_workflow_by_owner_and_name`) is scoped
  to the request tenant.

Adds three `#[ignore]` Postgres regressions in `buzz-db::workflow`, each
verified green against live PG and red when the `community_id` predicate is
dropped: `workflow_lookup_is_confined_to_its_community` (dup workflow+channel
UUID in A/B; scoped get/list resolve only the bound community's row, cross
lookup is NotFound), `workflow_delete_is_confined_to_its_community` (deleting
A/id leaves B/id intact), and `approval_is_confined_to_its_community` (same
token in A/B; granting A leaves B pending). Full `cargo test -p buzz-db
-p buzz-workflow -p buzz-relay` green, clippy clean on the trio.

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
This commit is contained in:
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
2026-06-27 14:43:51 -04:00
co-authored by Tyler Longwell
parent ce747ec759
commit c81b893558
10 changed files with 652 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-db/src/lib.rs
+116 -27
View File
@@ -258,6 +258,38 @@ impl Db {
.transpose()
}
/// Returns the normalized host mapped to a community id, if the community
/// exists.
///
/// The reverse of [`lookup_community_by_host`]: used by side-effect
/// producers that already hold a server-resolved `CommunityId` (e.g. the
/// workflow action sink running a run owned by some community) and need a
/// fully-formed [`buzz_core::tenant::TenantContext`] — host included — to
/// fan out under *that* community rather than the deployment default. The
/// community is authoritative; the host is read back for labelling only and
/// is never used to re-derive the community.
pub async fn lookup_community_host(
&self,
community_id: CommunityId,
) -> Result<Option<String>> {
let row = sqlx::query(
r#"
SELECT host
FROM communities
WHERE id = $1
"#,
)
.bind(community_id.as_uuid())
.fetch_optional(&self.pool)
.await?;
row.map(|row| {
let host: String = row.try_get("host")?;
Ok(host)
})
.transpose()
}
/// Ensure a configured community host exists and return its row.
///
/// This is the startup/config seeding path for N=1 deployments. Migrations
@@ -1518,27 +1550,33 @@ impl Db {
.await
}
/// Fetch a single workflow by ID.
pub async fn get_workflow(&self, id: Uuid) -> Result<workflow::WorkflowRecord> {
workflow::get_workflow(&self.pool, id).await
/// Fetch a single workflow by ID, scoped to its community.
pub async fn get_workflow(
&self,
community_id: CommunityId,
id: Uuid,
) -> Result<workflow::WorkflowRecord> {
workflow::get_workflow(&self.pool, community_id, id).await
}
/// List workflows for a channel.
pub async fn list_channel_workflows(
&self,
community_id: CommunityId,
channel_id: Uuid,
limit: Option<i64>,
offset: Option<i64>,
) -> Result<Vec<workflow::WorkflowRecord>> {
workflow::list_channel_workflows(&self.pool, channel_id, limit, offset).await
workflow::list_channel_workflows(&self.pool, community_id, channel_id, limit, offset).await
}
/// List active, enabled workflows for a channel.
pub async fn list_enabled_channel_workflows(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Vec<workflow::WorkflowRecord>> {
workflow::list_enabled_channel_workflows(&self.pool, channel_id).await
workflow::list_enabled_channel_workflows(&self.pool, community_id, channel_id).await
}
/// List all active, enabled schedule-triggered workflows.
@@ -1562,20 +1600,23 @@ impl Db {
/// Fetch the latest claimed schedule instant for interval trigger anchoring.
pub async fn latest_scheduled_workflow_fire(
&self,
community_id: CommunityId,
workflow_id: Uuid,
) -> Result<Option<chrono::DateTime<chrono::Utc>>> {
workflow::latest_scheduled_workflow_fire(&self.pool, workflow_id).await
workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await
}
/// Attach the workflow run id created from a won scheduled-fire claim.
pub async fn attach_scheduled_workflow_run(
&self,
community_id: CommunityId,
workflow_id: Uuid,
scheduled_for: chrono::DateTime<chrono::Utc>,
workflow_run_id: Uuid,
) -> Result<bool> {
workflow::attach_scheduled_workflow_run(
&self.pool,
community_id,
workflow_id,
scheduled_for,
workflow_run_id,
@@ -1594,78 +1635,116 @@ impl Db {
/// Update a workflow's name, definition, and hash.
pub async fn update_workflow(
&self,
community_id: CommunityId,
id: Uuid,
name: &str,
definition_json: &str,
definition_hash: &[u8],
) -> Result<()> {
workflow::update_workflow(&self.pool, id, name, definition_json, definition_hash).await
workflow::update_workflow(
&self.pool,
community_id,
id,
name,
definition_json,
definition_hash,
)
.await
}
/// Update a workflow's status.
pub async fn update_workflow_status(
&self,
community_id: CommunityId,
id: Uuid,
status: workflow::WorkflowStatus,
) -> Result<()> {
workflow::update_workflow_status(&self.pool, id, status).await
workflow::update_workflow_status(&self.pool, community_id, id, status).await
}
/// Enable or disable a workflow.
pub async fn set_workflow_enabled(&self, id: Uuid, enabled: bool) -> Result<()> {
workflow::set_workflow_enabled(&self.pool, id, enabled).await
pub async fn set_workflow_enabled(
&self,
community_id: CommunityId,
id: Uuid,
enabled: bool,
) -> Result<()> {
workflow::set_workflow_enabled(&self.pool, community_id, id, enabled).await
}
/// Delete a workflow and all its runs/approvals.
pub async fn delete_workflow(&self, id: Uuid) -> Result<()> {
workflow::delete_workflow(&self.pool, id).await
pub async fn delete_workflow(&self, community_id: CommunityId, id: Uuid) -> Result<()> {
workflow::delete_workflow(&self.pool, community_id, id).await
}
/// Find a workflow by owner pubkey and name. Used for NIP-09 a-tag deletion
/// where the d-tag is the workflow name (not UUID).
/// Find a workflow by owner pubkey and name within a community. Used for
/// NIP-09 a-tag deletion where the d-tag is the workflow name (not UUID).
pub async fn find_workflow_by_owner_and_name(
&self,
community_id: CommunityId,
owner_pubkey: &[u8],
name: &str,
) -> Result<Option<workflow::WorkflowRecord>> {
workflow::find_by_owner_and_name(&self.pool, owner_pubkey, name).await
workflow::find_by_owner_and_name(&self.pool, community_id, owner_pubkey, name).await
}
/// Create a new workflow run.
pub async fn create_workflow_run(
&self,
community_id: CommunityId,
workflow_id: Uuid,
trigger_event_id: Option<&[u8]>,
trigger_context: Option<&serde_json::Value>,
) -> Result<Uuid> {
workflow::create_workflow_run(&self.pool, workflow_id, trigger_event_id, trigger_context)
.await
workflow::create_workflow_run(
&self.pool,
community_id,
workflow_id,
trigger_event_id,
trigger_context,
)
.await
}
/// Fetch a single workflow run.
pub async fn get_workflow_run(&self, id: Uuid) -> Result<workflow::WorkflowRunRecord> {
workflow::get_workflow_run(&self.pool, id).await
/// Fetch a single workflow run, scoped to its community.
pub async fn get_workflow_run(
&self,
community_id: CommunityId,
id: Uuid,
) -> Result<workflow::WorkflowRunRecord> {
workflow::get_workflow_run(&self.pool, community_id, id).await
}
/// List runs for a workflow.
pub async fn list_workflow_runs(
&self,
community_id: CommunityId,
workflow_id: Uuid,
limit: i64,
) -> Result<Vec<workflow::WorkflowRunRecord>> {
workflow::list_workflow_runs(&self.pool, workflow_id, limit).await
workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await
}
/// Update a workflow run's status.
pub async fn update_workflow_run(
&self,
community_id: CommunityId,
id: Uuid,
status: workflow::RunStatus,
current_step: i32,
trace: &serde_json::Value,
error: Option<&str>,
) -> Result<()> {
workflow::update_workflow_run(&self.pool, id, status, current_step, trace, error).await
workflow::update_workflow_run(
&self.pool,
community_id,
id,
status,
current_step,
trace,
error,
)
.await
}
/// Create an approval request.
@@ -1674,41 +1753,50 @@ impl Db {
}
/// Fetch an approval by raw token.
pub async fn get_approval(&self, token: &str) -> Result<workflow::ApprovalRecord> {
workflow::get_approval(&self.pool, token).await
pub async fn get_approval(
&self,
community_id: CommunityId,
token: &str,
) -> Result<workflow::ApprovalRecord> {
workflow::get_approval(&self.pool, community_id, token).await
}
/// Fetch an approval by its already-hashed token (no re-hashing).
pub async fn get_approval_by_stored_hash(
&self,
community_id: CommunityId,
token_hash: &[u8],
) -> Result<workflow::ApprovalRecord> {
workflow::get_approval_by_stored_hash(&self.pool, token_hash).await
workflow::get_approval_by_stored_hash(&self.pool, community_id, token_hash).await
}
/// Fetch all approvals for a workflow run.
pub async fn get_run_approvals(
&self,
community_id: CommunityId,
workflow_id: uuid::Uuid,
run_id: uuid::Uuid,
) -> Result<Vec<workflow::ApprovalRecord>> {
workflow::get_run_approvals(&self.pool, workflow_id, run_id).await
workflow::get_run_approvals(&self.pool, community_id, workflow_id, run_id).await
}
/// Update an approval's status.
pub async fn update_approval(
&self,
community_id: CommunityId,
token: &str,
status: workflow::ApprovalStatus,
approver_pubkey: Option<&[u8]>,
note: Option<&str>,
) -> Result<bool> {
workflow::update_approval(&self.pool, token, status, approver_pubkey, note).await
workflow::update_approval(&self.pool, community_id, token, status, approver_pubkey, note)
.await
}
/// Update an approval by its already-hashed token (no re-hashing).
pub async fn update_approval_by_stored_hash(
&self,
community_id: CommunityId,
token_hash: &[u8],
status: workflow::ApprovalStatus,
approver_pubkey: Option<&[u8]>,
@@ -1716,6 +1804,7 @@ impl Db {
) -> Result<bool> {
workflow::update_approval_by_stored_hash(
&self.pool,
community_id,
token_hash,
status,
approver_pubkey,
crates/buzz-db/src/workflow.rs
+357 -41
View File
@@ -192,6 +192,13 @@ pub struct WorkflowRecord {
pub struct WorkflowRunRecord {
/// Unique run identifier.
pub id: Uuid,
/// Server-resolved community this run (and its workflow) belongs to.
///
/// `workflow_runs` is keyed `(community_id, id)`; the same run/workflow
/// UUID is allowed across communities, so every run carries its owning
/// community and downstream execution (side-effect sink, scoped lookups)
/// runs under it rather than re-deriving a tenant from the deployment host.
pub community_id: CommunityId,
/// The workflow definition that was executed.
pub workflow_id: Uuid,
/// Current execution status of this run.
@@ -294,16 +301,26 @@ pub async fn create_workflow(
Ok(id)
}
/// Fetch a single workflow by ID. Returns `DbError::InvalidData` if missing.
pub async fn get_workflow(pool: &PgPool, id: Uuid) -> Result<WorkflowRecord> {
/// Fetch a single workflow by ID, scoped to its community.
///
/// `workflows` is keyed `(community_id, id)`; the same workflow UUID can exist
/// in two communities, so a request-scoped lookup must bind both. The caller
/// supplies the server-resolved community (host-bound tenant for request paths,
/// the run's own community for execution paths) — never a client-supplied id.
pub async fn get_workflow(
pool: &PgPool,
community_id: CommunityId,
id: Uuid,
) -> Result<WorkflowRecord> {
let row = sqlx::query(
r#"
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE id = $1
WHERE community_id = $1 AND id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.fetch_optional(pool)
.await?
@@ -318,6 +335,7 @@ pub async fn get_workflow(pool: &PgPool, id: Uuid) -> Result<WorkflowRecord> {
/// `offset` enables pagination (0-based row offset).
pub async fn list_channel_workflows(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
limit: Option<i64>,
offset: Option<i64>,
@@ -330,11 +348,12 @@ pub async fn list_channel_workflows(
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE channel_id = $1
WHERE community_id = $1 AND channel_id = $2
ORDER BY created_at DESC
LIMIT $2 OFFSET $3
LIMIT $3 OFFSET $4
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(limit)
.bind(offset)
@@ -352,6 +371,7 @@ pub async fn list_channel_workflows(
/// an unbounded number of workflows per event.
pub async fn list_enabled_channel_workflows(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Vec<WorkflowRecord>> {
let rows = sqlx::query(
@@ -359,13 +379,15 @@ pub async fn list_enabled_channel_workflows(
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE channel_id = $1
WHERE community_id = $1
AND channel_id = $2
AND status = 'active'
AND enabled = TRUE
ORDER BY created_at DESC
LIMIT $2
LIMIT $3
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(LIST_MAX_LIMIT)
.fetch_all(pool)
@@ -448,15 +470,17 @@ pub async fn claim_scheduled_workflow_fire(
/// row is the source of truth for schedule deduplication.
pub async fn latest_scheduled_workflow_fire(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
) -> Result<Option<DateTime<Utc>>> {
let row = sqlx::query(
r#"
SELECT MAX(scheduled_for) AS scheduled_for
FROM scheduled_workflow_fires
WHERE workflow_id = $1
WHERE community_id = $1 AND workflow_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(workflow_id)
.fetch_one(pool)
.await?;
@@ -472,6 +496,7 @@ pub async fn latest_scheduled_workflow_fire(
/// intentional: the schedule instant was claimed and must not duplicate later.
pub async fn attach_scheduled_workflow_run(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
scheduled_for: DateTime<Utc>,
workflow_run_id: Uuid,
@@ -479,12 +504,14 @@ pub async fn attach_scheduled_workflow_run(
let result = sqlx::query(
r#"
UPDATE scheduled_workflow_fires
SET workflow_run_id = $3
WHERE workflow_id = $1
AND scheduled_for = $2
SET workflow_run_id = $4
WHERE community_id = $1
AND workflow_id = $2
AND scheduled_for = $3
AND workflow_run_id IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(workflow_id)
.bind(scheduled_for)
.bind(workflow_run_id)
@@ -521,6 +548,7 @@ pub async fn prune_scheduled_workflow_fires_before(
/// Update a workflow's name, definition, and definition_hash.
pub async fn update_workflow(
pool: &PgPool,
community_id: CommunityId,
id: Uuid,
name: &str,
definition_json: &str,
@@ -530,12 +558,13 @@ pub async fn update_workflow(
r#"
UPDATE workflows
SET name = $1, definition = $2::jsonb, definition_hash = $3
WHERE id = $4
WHERE community_id = $4 AND id = $5
"#,
)
.bind(name)
.bind(definition_json)
.bind(definition_hash)
.bind(community_id.as_uuid())
.bind(id)
.execute(pool)
.await?
@@ -548,15 +577,21 @@ pub async fn update_workflow(
}
/// Update a workflow's status (active -> disabled -> archived).
pub async fn update_workflow_status(pool: &PgPool, id: Uuid, status: WorkflowStatus) -> Result<()> {
pub async fn update_workflow_status(
pool: &PgPool,
community_id: CommunityId,
id: Uuid,
status: WorkflowStatus,
) -> Result<()> {
let affected = sqlx::query(
r#"
UPDATE workflows
SET status = $1::workflow_status
WHERE id = $2
WHERE community_id = $2 AND id = $3
"#,
)
.bind(status.to_string())
.bind(community_id.as_uuid())
.bind(id)
.execute(pool)
.await?
@@ -569,15 +604,21 @@ pub async fn update_workflow_status(pool: &PgPool, id: Uuid, status: WorkflowSta
}
/// Enable or disable a workflow without changing its status.
pub async fn set_workflow_enabled(pool: &PgPool, id: Uuid, enabled: bool) -> Result<()> {
pub async fn set_workflow_enabled(
pool: &PgPool,
community_id: CommunityId,
id: Uuid,
enabled: bool,
) -> Result<()> {
let affected = sqlx::query(
r#"
UPDATE workflows
SET enabled = $1
WHERE id = $2
WHERE community_id = $2 AND id = $3
"#,
)
.bind(enabled)
.bind(community_id.as_uuid())
.bind(id)
.execute(pool)
.await?
@@ -590,8 +631,9 @@ pub async fn set_workflow_enabled(pool: &PgPool, id: Uuid, enabled: bool) -> Res
}
/// Delete a workflow and all its runs/approvals (CASCADE).
pub async fn delete_workflow(pool: &PgPool, id: Uuid) -> Result<()> {
let affected = sqlx::query("DELETE FROM workflows WHERE id = $1")
pub async fn delete_workflow(pool: &PgPool, community_id: CommunityId, id: Uuid) -> Result<()> {
let affected = sqlx::query("DELETE FROM workflows WHERE community_id = $1 AND id = $2")
.bind(community_id.as_uuid())
.bind(id)
.execute(pool)
.await?
@@ -612,6 +654,7 @@ pub async fn delete_workflow(pool: &PgPool, id: Uuid) -> Result<()> {
/// correctly resolve `{{trigger.*}}` template variables.
pub async fn create_workflow_run(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
trigger_event_id: Option<&[u8]>,
trigger_context: Option<&serde_json::Value>,
@@ -621,10 +664,11 @@ pub async fn create_workflow_run(
sqlx::query(
r#"
INSERT INTO workflow_runs
(id, workflow_id, status, trigger_event_id, current_step, execution_trace, trigger_context)
VALUES ($1, $2, 'pending', $3, 0, '[]', $4)
(community_id, id, workflow_id, status, trigger_event_id, current_step, execution_trace, trigger_context)
VALUES ($1, $2, $3, 'pending', $4, 0, '[]', $5)
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.bind(workflow_id)
.bind(trigger_event_id)
@@ -635,16 +679,21 @@ pub async fn create_workflow_run(
Ok(id)
}
/// Fetch a single workflow run by ID.
pub async fn get_workflow_run(pool: &PgPool, id: Uuid) -> Result<WorkflowRunRecord> {
/// Fetch a single workflow run by ID, scoped to its community.
pub async fn get_workflow_run(
pool: &PgPool,
community_id: CommunityId,
id: Uuid,
) -> Result<WorkflowRunRecord> {
let row = sqlx::query(
r#"
SELECT id, workflow_id, status::text AS status, trigger_event_id, current_step,
SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step,
execution_trace, trigger_context, started_at, completed_at, error_message, created_at
FROM workflow_runs
WHERE id = $1
WHERE community_id = $1 AND id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.fetch_optional(pool)
.await?
@@ -656,20 +705,22 @@ pub async fn get_workflow_run(pool: &PgPool, id: Uuid) -> Result<WorkflowRunReco
/// List runs for a workflow, newest first, up to `limit` rows.
pub async fn list_workflow_runs(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
limit: i64,
) -> Result<Vec<WorkflowRunRecord>> {
let limit = limit.min(1000);
let rows = sqlx::query(
r#"
SELECT id, workflow_id, status::text AS status, trigger_event_id, current_step,
SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step,
execution_trace, trigger_context, started_at, completed_at, error_message, created_at
FROM workflow_runs
WHERE workflow_id = $1
WHERE community_id = $1 AND workflow_id = $2
ORDER BY created_at DESC
LIMIT $2
LIMIT $3
"#,
)
.bind(community_id.as_uuid())
.bind(workflow_id)
.bind(limit)
.fetch_all(pool)
@@ -686,6 +737,7 @@ pub async fn list_workflow_runs(
/// always false. We now check the bind parameter directly.
pub async fn update_workflow_run(
pool: &PgPool,
community_id: CommunityId,
id: Uuid,
status: RunStatus,
current_step: i32,
@@ -704,7 +756,7 @@ pub async fn update_workflow_run(
THEN NOW() ELSE started_at END,
completed_at = CASE WHEN $6 IN ('completed','failed','cancelled')
THEN NOW() ELSE completed_at END
WHERE id = $7
WHERE community_id = $7 AND id = $8
"#,
)
.bind(&status_str)
@@ -713,6 +765,7 @@ pub async fn update_workflow_run(
.bind(error)
.bind(&status_str) // for started_at CASE
.bind(&status_str) // for completed_at CASE
.bind(community_id.as_uuid())
.bind(id)
.execute(pool)
.await?
@@ -728,6 +781,8 @@ pub async fn update_workflow_run(
/// Parameters for creating a new approval request.
pub struct CreateApprovalParams<'a> {
/// Server-resolved community that owns the workflow/run this approval gates.
pub community_id: CommunityId,
/// Raw approval token (will be hashed before storage).
pub token: &'a str,
/// The workflow this approval belongs to.
@@ -750,6 +805,7 @@ pub struct CreateApprovalParams<'a> {
/// SHA-256 before storage so the DB never holds the raw value.
pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) -> Result<()> {
let CreateApprovalParams {
community_id,
token,
workflow_id,
run_id,
@@ -763,10 +819,11 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) ->
sqlx::query(
r#"
INSERT INTO workflow_approvals
(token, workflow_id, run_id, step_id, step_index, approver_spec, status, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, 'pending', $7)
(community_id, token, workflow_id, run_id, step_id, step_index, approver_spec, status, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, 'pending', $8)
"#,
)
.bind(community_id.as_uuid())
.bind(token_hash)
.bind(workflow_id)
.bind(run_id)
@@ -784,17 +841,26 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) ->
///
/// The token is hashed before the DB lookup so plaintext tokens are never
/// sent to the database layer.
pub async fn get_approval(pool: &PgPool, token: &str) -> Result<ApprovalRecord> {
pub async fn get_approval(
pool: &PgPool,
community_id: CommunityId,
token: &str,
) -> Result<ApprovalRecord> {
let token_hash = hash_approval_token(token);
get_approval_by_stored_hash(pool, &token_hash).await
get_approval_by_stored_hash(pool, community_id, &token_hash).await
}
/// Fetch an approval record by its already-hashed token value.
///
/// Use this when you already have the hash stored in the DB (e.g., from
/// `get_run_approvals`). The `token_hash` is used directly without re-hashing.
///
/// `workflow_approvals` is keyed `(community_id, token)`; the same token bytes
/// could in principle collide across communities, so the lookup binds the
/// server-resolved community alongside the token.
pub async fn get_approval_by_stored_hash(
pool: &PgPool,
community_id: CommunityId,
token_hash: &[u8],
) -> Result<ApprovalRecord> {
let row = sqlx::query(
@@ -802,9 +868,10 @@ pub async fn get_approval_by_stored_hash(
SELECT token, workflow_id, run_id, step_id, step_index, approver_spec,
status::text AS status, approver_pubkey, note, expires_at, created_at
FROM workflow_approvals
WHERE token = $1
WHERE community_id = $1 AND token = $2
"#,
)
.bind(community_id.as_uuid())
.bind(token_hash)
.fetch_optional(pool)
.await?
@@ -816,6 +883,7 @@ pub async fn get_approval_by_stored_hash(
/// Fetch all approval records for a given workflow run.
pub async fn get_run_approvals(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
run_id: Uuid,
) -> Result<Vec<ApprovalRecord>> {
@@ -824,10 +892,11 @@ pub async fn get_run_approvals(
SELECT token, workflow_id, run_id, step_id, step_index, approver_spec,
status::text AS status, approver_pubkey, note, expires_at, created_at
FROM workflow_approvals
WHERE run_id = $1 AND workflow_id = $2
WHERE community_id = $1 AND run_id = $2 AND workflow_id = $3
ORDER BY step_index, created_at
"#,
)
.bind(community_id.as_uuid())
.bind(run_id)
.bind(workflow_id)
.fetch_all(pool)
@@ -849,13 +918,15 @@ pub async fn get_run_approvals(
/// returns `Ok(false)`. Callers should treat `false` as a conflict (HTTP 409).
pub async fn update_approval(
pool: &PgPool,
community_id: CommunityId,
token: &str,
status: ApprovalStatus,
approver_pubkey: Option<&[u8]>,
note: Option<&str>,
) -> Result<bool> {
let token_hash = hash_approval_token(token);
update_approval_by_stored_hash(pool, &token_hash, status, approver_pubkey, note).await
update_approval_by_stored_hash(pool, community_id, &token_hash, status, approver_pubkey, note)
.await
}
/// Update an approval by its already-hashed token value.
@@ -863,9 +934,12 @@ pub async fn update_approval(
/// Use this when you already have the hash stored in the DB (e.g., from
/// `get_run_approvals`). The `token_hash` is used directly without re-hashing.
///
/// See [`update_approval`] for TOCTOU safety notes.
/// See [`update_approval`] for TOCTOU safety notes. The predicate binds the
/// server-resolved community alongside the token so an approval action for A/X
/// can never act on B/X.
pub async fn update_approval_by_stored_hash(
pool: &PgPool,
community_id: CommunityId,
token_hash: &[u8],
status: ApprovalStatus,
approver_pubkey: Option<&[u8]>,
@@ -880,7 +954,7 @@ pub async fn update_approval_by_stored_hash(
note = $3,
granted_at = CASE WHEN $4 = 'granted' THEN NOW() ELSE granted_at END,
denied_at = CASE WHEN $5 = 'denied' THEN NOW() ELSE denied_at END
WHERE token = $6 AND status = 'pending'
WHERE community_id = $6 AND token = $7 AND status = 'pending'
"#,
)
.bind(&status_str)
@@ -888,6 +962,7 @@ pub async fn update_approval_by_stored_hash(
.bind(note)
.bind(&status_str) // for granted_at CASE
.bind(&status_str) // for denied_at CASE
.bind(community_id.as_uuid())
.bind(token_hash)
.execute(pool)
.await?
@@ -926,6 +1001,7 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRecord>
fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRunRecord> {
let id: Uuid = row.try_get("id")?;
let community_id: Uuid = row.try_get("community_id")?;
let workflow_id: Uuid = row.try_get("workflow_id")?;
let status_str: String = row.try_get("status")?;
@@ -933,6 +1009,7 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRunRecord> {
Ok(WorkflowRunRecord {
id,
community_id: CommunityId::from_uuid(community_id),
workflow_id,
status,
trigger_event_id: row.try_get("trigger_event_id")?,
@@ -968,9 +1045,11 @@ fn row_to_approval_record(row: sqlx::postgres::PgRow) -> Result<ApprovalRecord>
})
}
/// Find a workflow by owner pubkey and name. Returns the first match (active or not).
/// Find a workflow by owner pubkey and name within a community. Returns the
/// first match (active or not).
pub async fn find_by_owner_and_name(
pool: &PgPool,
community_id: CommunityId,
owner_pubkey: &[u8],
name: &str,
) -> Result<Option<WorkflowRecord>> {
@@ -979,10 +1058,11 @@ pub async fn find_by_owner_and_name(
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE owner_pubkey = $1 AND name = $2
WHERE community_id = $1 AND owner_pubkey = $2 AND name = $3
LIMIT 1
"#,
)
.bind(community_id.as_uuid())
.bind(owner_pubkey)
.bind(name)
.fetch_optional(pool)
@@ -1231,6 +1311,7 @@ mod tests {
let record = WorkflowRunRecord {
id,
community_id: CommunityId::from_uuid(Uuid::new_v4()),
workflow_id,
status: RunStatus::Running,
trigger_event_id: Some(trigger_event_id.clone()),
@@ -1260,6 +1341,7 @@ mod tests {
let now = Utc::now();
let record = WorkflowRunRecord {
id: Uuid::new_v4(),
community_id: CommunityId::from_uuid(Uuid::new_v4()),
workflow_id: Uuid::new_v4(),
status: RunStatus::Pending,
trigger_event_id: None,
@@ -1282,6 +1364,7 @@ mod tests {
let now = Utc::now();
let record = WorkflowRunRecord {
id: Uuid::new_v4(),
community_id: CommunityId::from_uuid(Uuid::new_v4()),
workflow_id: Uuid::new_v4(),
status: RunStatus::Failed,
trigger_event_id: None,
@@ -1312,6 +1395,7 @@ mod tests {
let record = WorkflowRunRecord {
id: Uuid::new_v4(),
community_id: CommunityId::from_uuid(Uuid::new_v4()),
workflow_id: Uuid::new_v4(),
status: RunStatus::Completed,
trigger_event_id: None,
@@ -1333,6 +1417,7 @@ mod tests {
let now = Utc::now();
let record = WorkflowRunRecord {
id: Uuid::new_v4(),
community_id: CommunityId::from_uuid(Uuid::new_v4()),
workflow_id: Uuid::new_v4(),
status: RunStatus::Pending,
trigger_event_id: None,
@@ -1673,7 +1758,7 @@ mod tests {
.await
.expect("backdate ok");
let latest_before = latest_scheduled_workflow_fire(&pool, workflow_id)
let latest_before = latest_scheduled_workflow_fire(&pool, community, workflow_id)
.await
.expect("latest ok");
assert_eq!(
@@ -1693,7 +1778,7 @@ mod tests {
"expected at least one row pruned, got {pruned}"
);
let latest_after = latest_scheduled_workflow_fire(&pool, workflow_id)
let latest_after = latest_scheduled_workflow_fire(&pool, community, workflow_id)
.await
.expect("latest ok");
assert_eq!(
@@ -1702,4 +1787,235 @@ mod tests {
retention cutoff MUST exceed MAX(interval_secs) + safety margin (§5c)",
);
}
// -- Issue 4: workflow / approval community confinement -------------------
/// Insert a workflow under `community` with a caller-chosen `id` and
/// `channel_id`, so two communities can be given the *same* workflow UUID
/// and channel UUID (the PK is `(community_id, id)`, which structurally
/// allows the collision). Returns nothing; callers already hold the ids.
async fn insert_workflow_with_ids(
pool: &PgPool,
community: CommunityId,
id: Uuid,
channel_id: Uuid,
name: &str,
) {
let owner = vec![0xb2; 32];
ensure_user(pool, community, &owner)
.await
.expect("ensure owner");
// The channel must exist first: `workflows.channel_id` is a composite FK
// to `(community_id, channel_id)`.
sqlx::query(
r#"
INSERT INTO channels (id, community_id, name, created_by)
VALUES ($1, $2, $3, $4)
"#,
)
.bind(channel_id)
.bind(community.as_uuid())
.bind(format!("ch-{}", channel_id.simple()))
.bind(&owner)
.execute(pool)
.await
.expect("insert channel");
sqlx::query(
r#"
INSERT INTO workflows
(id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled)
VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE)
"#,
)
.bind(id)
.bind(community.as_uuid())
.bind(name)
.bind(&owner)
.bind(channel_id)
.bind(r#"{"trigger":{"on":"webhook"},"steps":[]}"#)
.bind(&[0u8; 32][..])
.execute(pool)
.await
.expect("insert workflow");
}
/// Issue 4 (workflow identity): the same workflow UUID and channel UUID can
/// exist in communities A and B (PK `(community_id, id)`). A request-scoped
/// `get_workflow` / `list_enabled_channel_workflows` MUST return only the
/// row owned by the bound community — never B's colliding row for an
/// A-scoped lookup. Pre-fix these bound only `id` / `channel_id`, so a
/// B-host request (or a webhook/manual trigger satisfying membership against
/// B's colliding channel) could load and drive A's workflow.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn workflow_lookup_is_confined_to_its_community() {
let pool = setup_pool().await;
let community_a = make_community(&pool).await;
let community_b = make_community(&pool).await;
// Same workflow UUID and channel UUID in both communities.
let shared_workflow_id = Uuid::new_v4();
let shared_channel_id = Uuid::new_v4();
insert_workflow_with_ids(&pool, community_a, shared_workflow_id, shared_channel_id, "wf-A")
.await;
insert_workflow_with_ids(&pool, community_b, shared_workflow_id, shared_channel_id, "wf-B")
.await;
// Scoped get returns each community's own row, never the other's.
let from_a = get_workflow(&pool, community_a, shared_workflow_id)
.await
.expect("A's workflow exists");
let from_b = get_workflow(&pool, community_b, shared_workflow_id)
.await
.expect("B's workflow exists");
assert_eq!(from_a.community_id, community_a, "A lookup must resolve A's row");
assert_eq!(from_a.name, "wf-A");
assert_eq!(from_b.community_id, community_b, "B lookup must resolve B's row");
assert_eq!(from_b.name, "wf-B");
// A workflow that exists ONLY in B must be NotFound under A.
let b_only_id = Uuid::new_v4();
let b_only_channel = Uuid::new_v4();
insert_workflow_with_ids(&pool, community_b, b_only_id, b_only_channel, "wf-B-only").await;
let cross = get_workflow(&pool, community_a, b_only_id).await;
assert!(
matches!(cross, Err(DbError::NotFound(_))),
"A must not see B's workflow by id: {cross:?}"
);
// The channel listing is confined too: A's channel listing yields only
// A's workflow even though B has the same channel UUID.
let listed_a = list_enabled_channel_workflows(&pool, community_a, shared_channel_id)
.await
.expect("list A");
assert_eq!(listed_a.len(), 1, "A's channel listing must contain exactly A's workflow");
assert_eq!(listed_a[0].community_id, community_a);
assert_eq!(listed_a[0].name, "wf-A");
}
/// Issue 4 (workflow lifecycle): deleting `A/id` must not delete `B/id`
/// when both communities hold the same workflow UUID. Pre-fix
/// `delete_workflow` predicated only on `id`, so a NIP-09 a-tag deletion in
/// one community would erase the colliding workflow in every community.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn workflow_delete_is_confined_to_its_community() {
let pool = setup_pool().await;
let community_a = make_community(&pool).await;
let community_b = make_community(&pool).await;
let shared_workflow_id = Uuid::new_v4();
insert_workflow_with_ids(&pool, community_a, shared_workflow_id, Uuid::new_v4(), "wf-A")
.await;
insert_workflow_with_ids(&pool, community_b, shared_workflow_id, Uuid::new_v4(), "wf-B")
.await;
delete_workflow(&pool, community_a, shared_workflow_id)
.await
.expect("delete A's workflow");
// A's row is gone; B's identical-UUID row survives untouched.
assert!(
matches!(
get_workflow(&pool, community_a, shared_workflow_id).await,
Err(DbError::NotFound(_))
),
"A's workflow must be deleted"
);
let surviving_b = get_workflow(&pool, community_b, shared_workflow_id)
.await
.expect("B's workflow must survive A's delete");
assert_eq!(surviving_b.community_id, community_b);
assert_eq!(surviving_b.name, "wf-B");
}
/// Issue 4 (approval path): the same approval token can hash to the same
/// bytes in A and B (PK `(community_id, token)`). A scoped grant/deny acting
/// on `A/token` MUST NOT touch `B/token`. Pre-fix the approval helpers
/// predicated only on `token`, so granting one community's approval would
/// silently resolve another's colliding gate.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn approval_is_confined_to_its_community() {
let pool = setup_pool().await;
let community_a = make_community(&pool).await;
let community_b = make_community(&pool).await;
// Same workflow + run + token in both communities.
let workflow_id = Uuid::new_v4();
let channel_id = Uuid::new_v4();
insert_workflow_with_ids(&pool, community_a, workflow_id, channel_id, "wf-A").await;
insert_workflow_with_ids(&pool, community_b, workflow_id, Uuid::new_v4(), "wf-B").await;
let run_a = create_workflow_run(&pool, community_a, workflow_id, None, None)
.await
.expect("run A");
let run_b = create_workflow_run(&pool, community_b, workflow_id, None, None)
.await
.expect("run B");
let token = "shared-approval-token";
let expires = Utc::now() + chrono::Duration::hours(1);
create_approval(
&pool,
CreateApprovalParams {
community_id: community_a,
token,
workflow_id,
run_id: run_a,
step_id: "gate",
step_index: 0,
approver_spec: "@anyone",
expires_at: expires,
},
)
.await
.expect("create approval A");
create_approval(
&pool,
CreateApprovalParams {
community_id: community_b,
token,
workflow_id,
run_id: run_b,
step_id: "gate",
step_index: 0,
approver_spec: "@anyone",
expires_at: expires,
},
)
.await
.expect("create approval B");
// Scoped read returns each community's own approval (its own run id).
let read_a = get_approval(&pool, community_a, token).await.expect("read A");
let read_b = get_approval(&pool, community_b, token).await.expect("read B");
assert_eq!(read_a.run_id, run_a, "A read must resolve A's approval");
assert_eq!(read_b.run_id, run_b, "B read must resolve B's approval");
// Granting A/token must NOT act on B/token.
let approver = vec![0xc3; 32];
let granted = update_approval(
&pool,
community_a,
token,
ApprovalStatus::Granted,
Some(&approver),
None,
)
.await
.expect("grant A");
assert!(granted, "A's approval must be granted");
let after_a = get_approval(&pool, community_a, token).await.expect("re-read A");
let after_b = get_approval(&pool, community_b, token).await.expect("re-read B");
assert_eq!(after_a.status, ApprovalStatus::Granted, "A is now granted");
assert_eq!(
after_b.status,
ApprovalStatus::Pending,
"B's approval must remain pending after A is granted"
);
}
}
crates/buzz-relay/src/api/bridge.rs
+21 -3
View File
@@ -974,9 +974,25 @@ pub async fn workflow_webhook(
let id = uuid::Uuid::parse_str(&id_str)
.map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid workflow UUID"))?;
// Row zero: bind this webhook to its community from the request host before
// any tenant-scoped lookup or write. The host — not the workflow row —
// determines the tenant: a request for community A's host may only reach
// community A's workflows, even when the same workflow UUID also exists in
// community B. Unmapped host, lookup failure, and a workflow that does not
// exist in *this* community all fail closed with the same generic 404, so a
// caller cannot probe which hosts or workflow ids exist on other tenants.
let raw_host = headers
.get(axum::http::header::HOST)
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let tenant = crate::tenant::bind_community(&state.db, raw_host)
.await
.map_err(|_| not_found("workflow not found"))?;
let community_id = tenant.community();
let workflow = state
.db
.get_workflow(id)
.get_workflow(community_id, id)
.await
.map_err(|_| not_found("workflow not found"))?;
@@ -1045,7 +1061,7 @@ pub async fn workflow_webhook(
let run_id = state
.db
.create_workflow_run(id, None, trigger_ctx_json.as_ref())
.create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref())
.await
.map_err(|e| super::internal_error(&format!("db error: {e}")))?;
@@ -1061,6 +1077,7 @@ pub async fn workflow_webhook(
tracing::error!("webhook: failed to parse definition: {e}");
if let Err(db_err) = db
.update_workflow_run(
community_id,
run_id,
buzz_db::workflow::RunStatus::Failed,
0,
@@ -1077,6 +1094,7 @@ pub async fn workflow_webhook(
let result = buzz_workflow::executor::execute_from_step(
&engine,
community_id,
run_id,
&def,
&trigger_ctx_clone,
@@ -1084,7 +1102,7 @@ pub async fn workflow_webhook(
None,
)
.await;
engine.finalize_run(run_id, result, None).await;
engine.finalize_run(community_id, run_id, result, None).await;
});
Ok((
crates/buzz-relay/src/handlers/command_executor.rs
+42 -18
View File
@@ -18,7 +18,7 @@ use tracing::warn;
use uuid::Uuid;
use buzz_core::kind::*;
use buzz_core::tenant::TenantContext;
use buzz_core::tenant::{CommunityId, TenantContext};
use buzz_db::workflow::{ApprovalStatus, RunStatus};
use buzz_workflow::executor::TriggerContext;
@@ -627,15 +627,21 @@ async fn handle_workflow_def(
PersistResult::Inserted(tx) => tx,
};
// 4. Execute: create_workflow. The workflow's community is resolved from
// the server-owned channel row, not from the client-supplied event. The DB
// also enforces `(community_id, channel_id)` as a composite FK.
let community_id = state
// 4. Execute: create_workflow. The workflow's community is the request's
// server-bound tenant — never re-derived from the (client-supplied) channel
// id. `community_of_channel(channel_id)` is ambiguous when the same channel
// UUID exists in two communities and could mint the workflow under the wrong
// tenant; `tenant.community()` is the authoritative owner. We then verify the
// channel actually exists *inside that community* (scoped `get_channel`),
// which fails closed if the client named a channel that belongs to a
// different community — the same guarantee the `(community_id, channel_id)`
// composite FK enforces on insert, surfaced here as a clean rejection.
let community_id = tenant.community();
state
.db
.community_of_channel(channel_id)
.get_channel(community_id, channel_id)
.await
.map_err(|e| IngestError::Internal(format!("error: db channel community lookup: {e}")))?
.ok_or_else(|| IngestError::Rejected("invalid: workflow channel not found".into()))?;
.map_err(|_| IngestError::Rejected("invalid: workflow channel not found".into()))?;
let workflow_id = state
.db
@@ -687,10 +693,14 @@ async fn handle_workflow_trigger(
let workflow_id = Uuid::parse_str(&workflow_id_str)
.map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?;
// 2. Validate workflow exists
// 2. Validate workflow exists — scoped to the caller's community. The same
// workflow UUID can exist in another community; a bare-id lookup could load
// B's workflow and then satisfy the membership check below against B's
// colliding channel, letting B trigger A's workflow.
let community_id = tenant.community();
let workflow = state
.db
.get_workflow(workflow_id)
.get_workflow(community_id, workflow_id)
.await
.map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?;
@@ -749,6 +759,7 @@ async fn handle_workflow_trigger(
let run_id = state
.db
.create_workflow_run(
community_id,
workflow_id,
Some(&event_id_bytes),
trigger_ctx_json.as_ref(),
@@ -773,6 +784,7 @@ async fn handle_workflow_trigger(
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,
@@ -789,6 +801,7 @@ async fn handle_workflow_trigger(
let result = buzz_workflow::executor::execute_from_step(
&engine,
community_id,
run_id,
&def,
&trigger_ctx_clone,
@@ -796,7 +809,7 @@ async fn handle_workflow_trigger(
None,
)
.await;
engine.finalize_run(run_id, result, None).await;
engine.finalize_run(community_id, run_id, result, None).await;
});
// 6. Return response
@@ -867,7 +880,7 @@ async fn handle_approval_grant(
// 2. Look up the approval record
let approval = state
.db
.get_approval_by_stored_hash(&token_hash)
.get_approval_by_stored_hash(tenant.community(), &token_hash)
.await
.map_err(|_| IngestError::Rejected("invalid: approval not found".into()))?;
@@ -909,6 +922,7 @@ async fn handle_approval_grant(
let updated = state
.db
.update_approval_by_stored_hash(
tenant.community(),
&token_hash,
ApprovalStatus::Granted,
Some(&self_bytes),
@@ -929,6 +943,7 @@ async fn handle_approval_grant(
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
// 6. Resume workflow execution (post-commit, async)
let community_id = tenant.community();
let run_id = approval.run_id;
let workflow_id = approval.workflow_id;
let resume_index = approval.step_index as usize + 1;
@@ -936,7 +951,8 @@ async fn handle_approval_grant(
let db = state.db.clone();
tokio::spawn(async move {
resume_workflow_after_approval(engine, db, run_id, workflow_id, resume_index).await;
resume_workflow_after_approval(engine, db, community_id, run_id, workflow_id, resume_index)
.await;
});
// 7. Return response
@@ -975,7 +991,7 @@ async fn handle_approval_deny(
// 2. Look up the approval record
let approval = state
.db
.get_approval_by_stored_hash(&token_hash)
.get_approval_by_stored_hash(tenant.community(), &token_hash)
.await
.map_err(|_| IngestError::Rejected("invalid: approval not found".into()))?;
@@ -1017,6 +1033,7 @@ async fn handle_approval_deny(
let updated = state
.db
.update_approval_by_stored_hash(
tenant.community(),
&token_hash,
ApprovalStatus::Denied,
Some(&self_bytes),
@@ -1037,12 +1054,13 @@ async fn handle_approval_deny(
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
// 6. Cancel the workflow run (post-commit, async)
let community_id = tenant.community();
let run_id = approval.run_id;
let pubkey_hex = self_hex.clone();
let db = state.db.clone();
tokio::spawn(async move {
let run = match db.get_workflow_run(run_id).await {
let run = match db.get_workflow_run(community_id, run_id).await {
Ok(r) => r,
Err(e) => {
tracing::error!("approval_deny: failed to fetch run {run_id}: {e}");
@@ -1061,6 +1079,7 @@ async fn handle_approval_deny(
let cancel_msg = format!("workflow cancelled: approval denied by {pubkey_hex}");
if let Err(e) = db
.update_workflow_run(
community_id,
run_id,
RunStatus::Cancelled,
run.current_step,
@@ -1091,11 +1110,12 @@ async fn handle_approval_deny(
async fn resume_workflow_after_approval(
engine: Arc<buzz_workflow::WorkflowEngine>,
db: buzz_db::Db,
community_id: CommunityId,
run_id: Uuid,
workflow_id: Uuid,
resume_index: usize,
) {
let run = match db.get_workflow_run(run_id).await {
let run = match db.get_workflow_run(community_id, run_id).await {
Ok(r) => r,
Err(e) => {
tracing::error!("resume_workflow: failed to fetch run {run_id}: {e}");
@@ -1112,7 +1132,7 @@ async fn resume_workflow_after_approval(
return;
}
let workflow = match db.get_workflow(workflow_id).await {
let workflow = match db.get_workflow(community_id, workflow_id).await {
Ok(w) => w,
Err(e) => {
tracing::error!("resume_workflow: failed to fetch workflow {workflow_id}: {e}");
@@ -1127,6 +1147,7 @@ async fn resume_workflow_after_approval(
tracing::error!("resume_workflow: failed to parse workflow definition: {e}");
if let Err(db_err) = db
.update_workflow_run(
community_id,
run_id,
RunStatus::Failed,
run.current_step,
@@ -1166,6 +1187,7 @@ async fn resume_workflow_after_approval(
let existing_trace = run.execution_trace.as_array().cloned();
let result = buzz_workflow::executor::execute_from_step(
&engine,
community_id,
run_id,
&def,
&trigger_ctx,
@@ -1173,5 +1195,7 @@ async fn resume_workflow_after_approval(
Some(initial_outputs),
)
.await;
engine.finalize_run(run_id, result, existing_trace).await;
engine
.finalize_run(community_id, run_id, result, existing_trace)
.await;
}
crates/buzz-relay/src/handlers/event.rs
+10 -1
View File
@@ -375,8 +375,17 @@ pub(crate) async fn dispatch_persistent_event(
let workflow_engine = Arc::clone(&state.workflow_engine);
let workflow_event = stored_event.clone();
let trigger_kind = kind_u32.to_string();
// The event was stored under `tenant.community()`; `StoredEvent` does
// not carry the community, so pass it explicitly. The same channel UUID
// can exist in another community — scoping the workflow lookup to this
// community keeps a colliding channel id in B from triggering A's
// workflows.
let workflow_community = tenant.community();
tokio::spawn(async move {
if let Err(e) = workflow_engine.on_event(&workflow_event).await {
if let Err(e) = workflow_engine
.on_event(workflow_community, &workflow_event)
.await
{
tracing::error!(event_id = ?workflow_event.event.id, "Workflow trigger failed: {e}");
} else {
metrics::counter!("buzz_workflow_runs_total", "trigger" => trigger_kind)
crates/buzz-relay/src/handlers/side_effects.rs
+9 -5
View File
@@ -1747,7 +1747,7 @@ async fn handle_a_tag_deletion(
if let Ok(wf_id) = uuid::Uuid::parse_str(d_tag) {
state
.db
.delete_workflow(wf_id)
.delete_workflow(tenant.community(), 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)");
@@ -1756,13 +1756,17 @@ async fn handle_a_tag_deletion(
let owner_bytes = hex::decode(pubkey_hex).unwrap_or_default();
match state
.db
.find_workflow_by_owner_and_name(&owner_bytes, d_tag)
.find_workflow_by_owner_and_name(tenant.community(), &owner_bytes, d_tag)
.await
{
Ok(Some(wf)) => {
state.db.delete_workflow(wf.id).await.map_err(|e| {
anyhow::anyhow!("failed to delete workflow {}: {e}", wf.id)
})?;
state
.db
.delete_workflow(tenant.community(), wf.id)
.await
.map_err(|e| {
anyhow::anyhow!("failed to delete workflow {}: {e}", wf.id)
})?;
tracing::info!(workflow_id = %wf.id, name = d_tag, "Workflow deleted via NIP-09 a-tag (name)");
}
Ok(None) => {
crates/buzz-relay/src/workflow_sink.rs
+21 -12
View File
@@ -9,6 +9,7 @@ use std::pin::Pin;
use std::sync::{Arc, Weak};
use buzz_core::kind::KIND_STREAM_MESSAGE;
use buzz_core::tenant::CommunityId;
use buzz_workflow::action_sink::{ActionSink, ActionSinkError};
use chrono::Utc;
use nostr::{EventBuilder, Kind, Tag};
@@ -42,6 +43,7 @@ impl RelayActionSink {
impl ActionSink for RelayActionSink {
fn send_message(
&self,
community_id: CommunityId,
channel_id: &str,
text: &str,
author_pubkey: &str,
@@ -57,18 +59,25 @@ impl ActionSink for RelayActionSink {
.upgrade()
.ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?;
// Resolve the deployment's own community from the configured relay
// host — a workflow execution has no inbound connection to bind. The
// relay-signed kind:9 message belongs to that community. Fail closed
// if the host isn't mapped (never a default tenant).
let tenant =
crate::tenant::bind_deployment_community(&state.db, &state.config.relay_url)
.await
.map_err(|e| {
ActionSinkError::Database(format!(
"relay host not mapped to a community: {e:?}"
))
})?;
// The run carries its owning community (`community_id`); the
// relay-signed kind:9 message belongs to *that* community, never the
// deployment default. Re-deriving the tenant from `config.relay_url`
// would post a community-B workflow's output into the deployment/
// default community under N>1. Read the community's host back to
// form a complete TenantContext (host is for labelling only — the
// community is already fixed and is never re-derived from it). Fail
// closed if the community no longer maps to a host.
let host = state
.db
.lookup_community_host(community_id)
.await
.map_err(|e| ActionSinkError::Database(e.to_string()))?
.ok_or_else(|| {
ActionSinkError::Database(format!(
"workflow run community {community_id} is not mapped to a host"
))
})?;
let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host);
// 1. Validate content is not empty/whitespace-only
if text.trim().is_empty() {
crates/buzz-workflow/src/action_sink.rs
+8
View File
@@ -6,6 +6,8 @@
use std::future::Future;
use std::pin::Pin;
use buzz_core::tenant::CommunityId;
/// Errors from action sink operations.
#[derive(Debug, thiserror::Error)]
pub enum ActionSinkError {
@@ -46,6 +48,11 @@ impl From<ActionSinkError> for crate::WorkflowError {
pub trait ActionSink: Send + Sync {
/// Post a message to a channel on behalf of a workflow owner.
///
/// - `community_id`: the server-resolved community that owns the workflow
/// run driving this side effect. The relay-signed message is published
/// under *this* community, never the deployment/default tenant — the run
/// carries its owning community so a workflow in community B posts into B
/// even though the side effect has no inbound connection to bind.
/// - `channel_id`: UUID string of the target channel
/// - `text`: message body (must not be empty/whitespace-only)
/// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for
@@ -54,6 +61,7 @@ pub trait ActionSink: Send + Sync {
/// Returns the event ID hex string on success.
fn send_message(
&self,
community_id: CommunityId,
channel_id: &str,
text: &str,
author_pubkey: &str,
crates/buzz-workflow/src/executor.rs
+26 -11
View File
@@ -11,6 +11,7 @@
use std::collections::HashMap;
use buzz_core::tenant::CommunityId;
use evalexpr::HashMapContext;
use serde_json::Value as JsonValue;
use tracing::{debug, info, warn};
@@ -533,6 +534,7 @@ pub async fn dispatch_action(
step_id: &str,
action: &ActionDef,
engine: &WorkflowEngine,
community_id: CommunityId,
run_id: Uuid,
trigger_ctx: &TriggerContext,
) -> Result<StepResult, WorkflowError> {
@@ -540,15 +542,22 @@ pub async fn dispatch_action(
match action {
SendMessage { text, channel } => {
// Look up workflow metadata for destination validation and attribution.
let wf_run = engine.db.get_workflow_run(run_id).await.map_err(|e| {
WorkflowError::WebhookError(format!(
"SendMessage: failed to load workflow run {run_id}: {e}"
))
})?;
// Look up workflow metadata for destination validation and
// attribution, scoped to the run's community — the same run/workflow
// UUID may exist in another community, so a bare-id lookup could
// load the wrong row and drive a side effect under it.
let wf_run = engine
.db
.get_workflow_run(community_id, run_id)
.await
.map_err(|e| {
WorkflowError::WebhookError(format!(
"SendMessage: failed to load workflow run {run_id}: {e}"
))
})?;
let workflow = engine
.db
.get_workflow(wf_run.workflow_id)
.get_workflow(community_id, wf_run.workflow_id)
.await
.map_err(|e| {
WorkflowError::WebhookError(format!(
@@ -572,7 +581,7 @@ pub async fn dispatch_action(
let event_id = engine
.action_sink()?
.send_message(&channel_id, text, &owner_pubkey_hex)
.send_message(community_id, &channel_id, text, &owner_pubkey_hex)
.await
.map_err(WorkflowError::from)?;
@@ -971,6 +980,7 @@ pub struct ExecutionResult {
/// Transitions the run to `Running` after acquiring a permit.
pub async fn execute_run(
engine: &WorkflowEngine,
community_id: CommunityId,
run_id: Uuid,
def: &WorkflowDef,
trigger_ctx: &TriggerContext,
@@ -986,6 +996,7 @@ pub async fn execute_run(
engine
.db
.update_workflow_run(
community_id,
run_id,
buzz_db::workflow::RunStatus::Running,
0,
@@ -1000,7 +1011,7 @@ pub async fn execute_run(
)
})?;
execute_steps(engine, run_id, def, trigger_ctx, 0, None).await
execute_steps(engine, community_id, run_id, def, trigger_ctx, 0, None).await
}
/// Resume execution from a specific step index (used for approval resume).
@@ -1017,6 +1028,7 @@ pub async fn execute_run(
/// reference `{{steps.PREV_STEP.output.X}}` correctly.
pub async fn execute_from_step(
engine: &WorkflowEngine,
community_id: CommunityId,
run_id: Uuid,
def: &WorkflowDef,
trigger_ctx: &TriggerContext,
@@ -1033,7 +1045,7 @@ pub async fn execute_from_step(
// Mark run as Running now that we have a permit (resume from approval).
// Preserve the existing execution trace from pre-approval steps.
let existing_trace = match engine.db.get_workflow_run(run_id).await {
let existing_trace = match engine.db.get_workflow_run(community_id, run_id).await {
Ok(r) => r.execution_trace,
Err(e) => {
warn!(
@@ -1046,6 +1058,7 @@ pub async fn execute_from_step(
engine
.db
.update_workflow_run(
community_id,
run_id,
buzz_db::workflow::RunStatus::Running,
start_index as i32,
@@ -1062,6 +1075,7 @@ pub async fn execute_from_step(
execute_steps(
engine,
community_id,
run_id,
def,
trigger_ctx,
@@ -1079,6 +1093,7 @@ pub async fn execute_from_step(
/// the trace of steps completed before the failure.
async fn execute_steps(
engine: &WorkflowEngine,
community_id: CommunityId,
run_id: Uuid,
def: &WorkflowDef,
trigger_ctx: &TriggerContext,
@@ -1134,7 +1149,7 @@ async fn execute_steps(
.unwrap_or(engine.config.default_timeout_secs);
let dispatch_result = tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
dispatch_action(&step.id, &resolved_action, engine, run_id, trigger_ctx),
dispatch_action(&step.id, &resolved_action, engine, community_id, run_id, trigger_ctx),
)
.await;
crates/buzz-workflow/src/lib.rs
+42 -14
View File
@@ -21,7 +21,10 @@
//! let (def, json) = WorkflowEngine::parse_yaml(yaml_str)?;
//!
//! // React to an incoming event (called from event handler post-store hook).
//! engine.on_event(&stored_event).await?;
//! // The community is the event's server-resolved tenant, threaded from the
//! // relay's bound `TenantContext` — the same workflow UUID can exist in two
//! // communities, so execution is always scoped to its owner.
//! engine.on_event(community_id, &stored_event).await?;
//!
//! // Run the background scheduler (cron triggers).
//! tokio::spawn(async move { engine.run().await });
@@ -42,6 +45,7 @@ use std::sync::Arc;
use std::sync::OnceLock;
use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION};
use buzz_core::tenant::CommunityId;
use buzz_db::workflow::RunStatus;
use buzz_db::Db;
use chrono::{DateTime, Utc};
@@ -73,10 +77,13 @@ pub struct WorkflowEngine {
pub(crate) config: WorkflowConfig,
/// Semaphore enforcing `config.max_concurrent` simultaneous workflow runs.
pub(crate) run_semaphore: Arc<Semaphore>,
/// Last-fired timestamps for interval-triggered workflows.
/// Last-fired timestamps for interval-triggered workflows, keyed by
/// `(community_id, workflow_id)`. The same workflow UUID can exist in two
/// communities (the PK is `(community_id, id)`); keying by bare id would let
/// one community's interval fire suppress the other's for the interval.
/// In-memory only — lost on restart. Missed fires during downtime are
/// not replayed (acceptable for MVP).
pub(crate) last_fired: DashMap<Uuid, DateTime<Utc>>,
pub(crate) last_fired: DashMap<(CommunityId, Uuid), DateTime<Utc>>,
/// Action sink for executing side-effects (SendMessage, etc.).
/// Late-initialized via [`set_action_sink`] after `AppState` construction.
pub(crate) action_sink: OnceLock<Arc<dyn ActionSink>>,
@@ -138,6 +145,7 @@ impl WorkflowEngine {
/// approval-resume path where pre-approval steps already have trace entries.
pub async fn finalize_run(
&self,
community_id: CommunityId,
run_id: uuid::Uuid,
result: Result<ExecutionResult, (WorkflowError, PartialProgress)>,
existing_trace: Option<Vec<serde_json::Value>>,
@@ -162,6 +170,7 @@ impl WorkflowEngine {
if let Err(e) = self
.db
.update_workflow_run(
community_id,
run_id,
RunStatus::Failed,
step_count,
@@ -180,6 +189,7 @@ impl WorkflowEngine {
if let Err(e) = self
.db
.update_workflow_run(
community_id,
run_id,
RunStatus::Completed,
step_count,
@@ -203,6 +213,7 @@ impl WorkflowEngine {
if let Err(db_err) = self
.db
.update_workflow_run(
community_id,
run_id,
RunStatus::Failed,
progress.step_index as i32,
@@ -225,10 +236,17 @@ impl WorkflowEngine {
/// Checks whether any workflow in the event's channel has a matching trigger.
/// Workflow execution events (kinds 46001–46012) are excluded to prevent loops.
///
/// `community_id` is the server-resolved community the event was stored
/// under — `StoredEvent` does not carry it, and the same channel UUID can
/// exist in two communities, so the workflow lookup/run-creation must be
/// scoped to the caller's tenant or community B could trigger community A's
/// workflow on a colliding channel id.
///
/// The method takes `self: &Arc<Self>` so that the spawned task can hold a
/// clone of the `Arc` without requiring `'static` on `&self`.
pub async fn on_event(
self: &Arc<Self>,
community_id: CommunityId,
event: &buzz_core::StoredEvent,
) -> Result<(), WorkflowError> {
let Some(channel_id) = event.channel_id else {
@@ -249,7 +267,7 @@ impl WorkflowEngine {
let workflows = self
.db
.list_enabled_channel_workflows(channel_id)
.list_enabled_channel_workflows(community_id, channel_id)
.await
.map_err(WorkflowError::from)?;
@@ -288,6 +306,7 @@ impl WorkflowEngine {
let run_id = match self
.db
.create_workflow_run(
community_id,
workflow.id,
Some(&trigger_event_id_bytes),
Some(&trigger_ctx_json),
@@ -312,8 +331,10 @@ impl WorkflowEngine {
let ctx_clone = trigger_ctx.clone();
tokio::spawn(async move {
let result = executor::execute_run(&engine, run_id, &def_clone, &ctx_clone).await;
engine.finalize_run(run_id, result, None).await;
let result =
executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone)
.await;
engine.finalize_run(community_id, run_id, result, None).await;
});
}
@@ -348,6 +369,10 @@ impl WorkflowEngine {
};
for workflow in &workflows {
// The same workflow UUID may exist in another community; carry
// the row's owning community through fire-tracking, run creation,
// and execution so a fire/run never crosses tenants.
let community_id = workflow.community_id;
let def: schema::WorkflowDef =
match serde_json::from_value(workflow.definition.clone()) {
Ok(d) => d,
@@ -387,7 +412,7 @@ impl WorkflowEngine {
interval: Some(dur),
} => {
// Fix 7: delegate to pure helper for testability.
let last = self.last_fired.get(&workflow.id).map(|t| *t);
let last = self.last_fired.get(&(community_id, workflow.id)).map(|t| *t);
(
interval_should_fire(dur, last, now, workflow.id),
"interval",
@@ -421,6 +446,7 @@ impl WorkflowEngine {
let run_id = match self
.db
.create_workflow_run(
community_id,
workflow.id,
None, // no trigger event for cron
trigger_ctx_json.as_ref(),
@@ -442,7 +468,7 @@ impl WorkflowEngine {
// Only needed for interval triggers — cron uses window-based matching
// which already prevents double-fire within the same minute.
if trigger_type == "interval" {
self.last_fired.insert(workflow.id, now);
self.last_fired.insert((community_id, workflow.id), now);
}
// Fix 6: log the specific trigger type (cron vs interval).
@@ -458,17 +484,19 @@ impl WorkflowEngine {
let ctx_clone = trigger_ctx.clone();
tokio::spawn(async move {
let result =
executor::execute_run(&engine, run_id, &def_clone, &ctx_clone).await;
engine.finalize_run(run_id, result, None).await;
executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone)
.await;
engine.finalize_run(community_id, run_id, result, None).await;
});
}
// Fix 1: prune stale last_fired entries for workflows that are no longer
// active/enabled. Without this the DashMap grows monotonically as
// workflows are deleted or disabled.
let active_ids: std::collections::HashSet<Uuid> =
workflows.iter().map(|w| w.id).collect();
self.last_fired.retain(|id, _| active_ids.contains(id));
// workflows are deleted or disabled. Keyed by `(community_id, id)` so
// entries are matched to the same scope they were inserted under.
let active_ids: std::collections::HashSet<(CommunityId, Uuid)> =
workflows.iter().map(|w| (w.community_id, w.id)).collect();
self.last_fired.retain(|key, _| active_ids.contains(key));
}
}
}