mirror of
https://github.com/block/buzz.git
synced 2026-11-04 00:12:02 +00:00
feat(workflows): sign delivery verification receipts
Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
@@ -440,6 +440,11 @@ pub const KIND_WINDOW_BOUNDS: u32 = 39006;
|
||||
|
||||
/// Ephemeral, relay-authored hint that a durable workflow delivery is ready.
|
||||
pub const KIND_WORKFLOW_AGENT_WAKE: u32 = 24620;
|
||||
/// Relay-signed proof binding a durable workflow delivery to its public inputs.
|
||||
///
|
||||
/// Returned by the authenticated delivery API; not accepted as client-authored
|
||||
/// state and not stored in the event log.
|
||||
pub const KIND_WORKFLOW_DELIVERY_RECEIPT: u32 = 24621;
|
||||
|
||||
/// Workflow definition (parameterized replaceable, d=workflow_uuid).
|
||||
pub const KIND_WORKFLOW_DEF: u32 = 30620;
|
||||
@@ -696,6 +701,7 @@ pub const ALL_KINDS: &[u32] = &[
|
||||
KIND_THREAD_SUMMARY,
|
||||
KIND_WINDOW_BOUNDS,
|
||||
KIND_WORKFLOW_AGENT_WAKE,
|
||||
KIND_WORKFLOW_DELIVERY_RECEIPT,
|
||||
KIND_PRESENCE_UPDATE,
|
||||
KIND_TYPING_INDICATOR,
|
||||
KIND_HUDDLE_REACTION,
|
||||
|
||||
@@ -6,11 +6,15 @@
|
||||
|
||||
use std::fmt;
|
||||
|
||||
use nostr::{Event, EventBuilder, EventId, Kind, PublicKey, Tag};
|
||||
use nostr::{Event, EventBuilder, EventId, Keys, Kind, PublicKey, Tag};
|
||||
use sha2::{Digest, Sha256};
|
||||
use thiserror::Error;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{kind::KIND_WORKFLOW_AGENT_WAKE, tenant::CommunityId};
|
||||
use crate::{
|
||||
kind::{KIND_WORKFLOW_AGENT_WAKE, KIND_WORKFLOW_DELIVERY_RECEIPT},
|
||||
tenant::CommunityId,
|
||||
};
|
||||
|
||||
/// The target class admitted by the durable workflow delivery protocol.
|
||||
pub const WORKFLOW_DELIVERY_TARGET: &str = "message-v1";
|
||||
@@ -201,6 +205,135 @@ pub fn message_v1_targets(event: &Event) -> Result<Vec<PublicKey>, WorkflowDeliv
|
||||
Ok(targets)
|
||||
}
|
||||
|
||||
/// Relay-signed proof that the relay produced one public workflow delivery.
|
||||
///
|
||||
/// The receipt contains references and hashes only. Trigger context, webhook
|
||||
/// fields, step outputs, execution traces, and secrets are deliberately absent.
|
||||
/// Its timestamp is copied from the visible message, making signing
|
||||
/// deterministic for one binding/message/key tuple.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct WorkflowDeliveryReceipt {
|
||||
delivery_id: WorkflowDeliveryId,
|
||||
binding: WorkflowDeliveryBinding,
|
||||
message_content_sha256: [u8; 32],
|
||||
}
|
||||
|
||||
impl WorkflowDeliveryReceipt {
|
||||
/// Construct a receipt after validating the visible event against the binding.
|
||||
pub fn new(
|
||||
delivery_id: WorkflowDeliveryId,
|
||||
binding: WorkflowDeliveryBinding,
|
||||
message: &Event,
|
||||
) -> Result<Self, WorkflowDeliveryError> {
|
||||
binding.validate_message_event(message)?;
|
||||
Ok(Self {
|
||||
delivery_id,
|
||||
binding,
|
||||
message_content_sha256: Sha256::digest(message.content.as_bytes()).into(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Return the durable delivery identity.
|
||||
pub const fn delivery_id(&self) -> WorkflowDeliveryId {
|
||||
self.delivery_id
|
||||
}
|
||||
|
||||
/// Return the immutable delivery binding proved by this receipt.
|
||||
pub const fn binding(&self) -> &WorkflowDeliveryBinding {
|
||||
&self.binding
|
||||
}
|
||||
|
||||
/// Return the SHA-256 identity of the visible message content.
|
||||
pub const fn message_content_sha256(&self) -> [u8; 32] {
|
||||
self.message_content_sha256
|
||||
}
|
||||
|
||||
/// Sign the canonical receipt with the relay identity.
|
||||
pub fn sign(&self, relay_keys: &Keys, message: &Event) -> Result<Event, WorkflowDeliveryError> {
|
||||
self.binding.validate_message_event(message)?;
|
||||
if <[u8; 32]>::from(Sha256::digest(message.content.as_bytes()))
|
||||
!= self.message_content_sha256
|
||||
{
|
||||
return Err(WorkflowDeliveryError::MessageContentHashMismatch);
|
||||
}
|
||||
EventBuilder::new(Kind::Custom(KIND_WORKFLOW_DELIVERY_RECEIPT as u16), "")
|
||||
.tags(self.canonical_tags()?)
|
||||
.custom_created_at(message.created_at)
|
||||
.sign_with_keys(relay_keys)
|
||||
.map_err(|error| WorkflowDeliveryError::ReceiptSigning(error.to_string()))
|
||||
}
|
||||
|
||||
/// Verify a receipt's relay signature, canonical shape, binding, and public message.
|
||||
pub fn verify(
|
||||
event: &Event,
|
||||
relay_pubkey: PublicKey,
|
||||
delivery_id: WorkflowDeliveryId,
|
||||
binding: &WorkflowDeliveryBinding,
|
||||
message: &Event,
|
||||
) -> Result<Self, WorkflowDeliveryError> {
|
||||
if event.kind.as_u16() != KIND_WORKFLOW_DELIVERY_RECEIPT as u16 {
|
||||
return Err(WorkflowDeliveryError::WrongReceiptKind(event.kind.as_u16()));
|
||||
}
|
||||
if event.pubkey != relay_pubkey || event.verify().is_err() {
|
||||
return Err(WorkflowDeliveryError::InvalidReceiptSignature);
|
||||
}
|
||||
if !event.content.is_empty() {
|
||||
return Err(WorkflowDeliveryError::ReceiptHasContent);
|
||||
}
|
||||
binding.validate_message_event(message)?;
|
||||
if event.created_at != message.created_at {
|
||||
return Err(WorkflowDeliveryError::NonCanonicalReceipt);
|
||||
}
|
||||
let expected = Self::new(delivery_id, binding.clone(), message)?;
|
||||
let expected_tags = expected.canonical_tags()?;
|
||||
if event.tags.as_slice() != expected_tags.as_slice() {
|
||||
return Err(WorkflowDeliveryError::ReceiptBindingMismatch);
|
||||
}
|
||||
Ok(expected)
|
||||
}
|
||||
|
||||
fn canonical_tags(&self) -> Result<Vec<Tag>, WorkflowDeliveryError> {
|
||||
let mut tags = vec![
|
||||
parse_tag(["delivery", &self.delivery_id.to_string()])?,
|
||||
parse_tag(["community", &self.binding.community_id().to_string()])?,
|
||||
parse_tag(["workflow", &self.binding.workflow_id().to_string()])?,
|
||||
parse_tag(["run", &self.binding.run_id().to_string()])?,
|
||||
parse_tag(["step", self.binding.step_id()])?,
|
||||
parse_tag(["p", &self.binding.target_pubkey().to_hex()])?,
|
||||
parse_tag(["definition", &self.binding.definition_event_id().to_hex()])?,
|
||||
parse_tag([
|
||||
"message",
|
||||
&self.binding.message_event_id().to_hex(),
|
||||
&sha256_hex(&self.message_content_sha256),
|
||||
])?,
|
||||
];
|
||||
tags.push(cause_tag(self.binding.cause())?);
|
||||
Ok(tags)
|
||||
}
|
||||
}
|
||||
|
||||
fn sha256_hex(digest: &[u8; 32]) -> String {
|
||||
use std::fmt::Write as _;
|
||||
|
||||
let mut encoded = String::with_capacity(64);
|
||||
for byte in digest {
|
||||
let _ = write!(encoded, "{byte:02x}");
|
||||
}
|
||||
encoded
|
||||
}
|
||||
|
||||
fn cause_tag(cause: &WorkflowDeliveryCause) -> Result<Tag, WorkflowDeliveryError> {
|
||||
match cause {
|
||||
WorkflowDeliveryCause::Event(event_id) => parse_tag(["cause", "event", &event_id.to_hex()]),
|
||||
WorkflowDeliveryCause::Schedule {
|
||||
scheduled_for_unix_seconds,
|
||||
} => parse_tag(["cause", "schedule", &scheduled_for_unix_seconds.to_string()]),
|
||||
WorkflowDeliveryCause::Webhook { invocation_id } => {
|
||||
parse_tag(["cause", "webhook", &invocation_id.to_string()])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// An ephemeral identifier-only wake hint for a durable delivery.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct WorkflowDeliveryWake {
|
||||
@@ -300,6 +433,27 @@ pub enum WorkflowDeliveryError {
|
||||
/// A `message-v1` recipient appeared more than once.
|
||||
#[error("workflow delivery message has duplicate message-v1 target")]
|
||||
DuplicateMessageV1Target,
|
||||
/// A receipt was not kind 24621.
|
||||
#[error("workflow delivery receipt must be kind 24621, got {0}")]
|
||||
WrongReceiptKind(u16),
|
||||
/// A receipt was not signed by the expected relay or had an invalid signature.
|
||||
#[error("workflow delivery receipt signature is invalid")]
|
||||
InvalidReceiptSignature,
|
||||
/// A receipt carried content rather than references only.
|
||||
#[error("workflow delivery receipt content must be empty")]
|
||||
ReceiptHasContent,
|
||||
/// A receipt did not use the message timestamp.
|
||||
#[error("workflow delivery receipt is not canonical")]
|
||||
NonCanonicalReceipt,
|
||||
/// A receipt's exact canonical tags disagreed with the expected binding.
|
||||
#[error("workflow delivery receipt binding does not match")]
|
||||
ReceiptBindingMismatch,
|
||||
/// The visible message content did not match the receipt hash.
|
||||
#[error("workflow delivery receipt message content hash does not match")]
|
||||
MessageContentHashMismatch,
|
||||
/// Nostr could not sign the receipt.
|
||||
#[error("workflow delivery receipt signing failed: {0}")]
|
||||
ReceiptSigning(String),
|
||||
/// A wake was not kind 24620.
|
||||
#[error("workflow delivery wake must be kind 24620, got {0}")]
|
||||
WrongWakeKind(u16),
|
||||
@@ -643,6 +797,109 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn receipt_binds_every_identity_and_contains_no_private_inputs() {
|
||||
let relay = Keys::generate();
|
||||
let original = binding();
|
||||
let message = EventBuilder::new(Kind::Custom(9), "rendered secret result")
|
||||
.tags([original.message_v1_target_tag().unwrap()])
|
||||
.sign_with_keys(&relay)
|
||||
.unwrap();
|
||||
let binding = WorkflowDeliveryBinding::new(
|
||||
original.community_id(),
|
||||
original.workflow_id(),
|
||||
original.run_id(),
|
||||
original.step_id(),
|
||||
original.target_pubkey(),
|
||||
original.definition_event_id(),
|
||||
message.id,
|
||||
original.cause().clone(),
|
||||
)
|
||||
.unwrap();
|
||||
let delivery_id = WorkflowDeliveryId::from_uuid(Uuid::new_v4());
|
||||
let receipt = WorkflowDeliveryReceipt::new(delivery_id, binding.clone(), &message).unwrap();
|
||||
let signed = receipt.sign(&relay, &message).unwrap();
|
||||
|
||||
assert!(signed.content.is_empty());
|
||||
let wire = serde_json::to_string(&signed).unwrap();
|
||||
assert!(!wire.contains("rendered secret result"));
|
||||
assert!(!wire.contains("trigger_context"));
|
||||
assert!(!wire.contains("execution_trace"));
|
||||
assert_eq!(
|
||||
WorkflowDeliveryReceipt::verify(
|
||||
&signed,
|
||||
relay.public_key(),
|
||||
delivery_id,
|
||||
&binding,
|
||||
&message,
|
||||
)
|
||||
.unwrap(),
|
||||
receipt
|
||||
);
|
||||
|
||||
let other_target = Keys::generate().public_key();
|
||||
let tampered = WorkflowDeliveryBinding::new(
|
||||
binding.community_id(),
|
||||
binding.workflow_id(),
|
||||
binding.run_id(),
|
||||
binding.step_id(),
|
||||
other_target,
|
||||
binding.definition_event_id(),
|
||||
binding.message_event_id(),
|
||||
binding.cause().clone(),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
WorkflowDeliveryReceipt::verify(
|
||||
&signed,
|
||||
relay.public_key(),
|
||||
delivery_id,
|
||||
&tampered,
|
||||
&message,
|
||||
),
|
||||
Err(WorkflowDeliveryError::MissingMessageV1Target)
|
||||
);
|
||||
assert_eq!(
|
||||
WorkflowDeliveryReceipt::verify(
|
||||
&signed,
|
||||
Keys::generate().public_key(),
|
||||
delivery_id,
|
||||
&binding,
|
||||
&message,
|
||||
),
|
||||
Err(WorkflowDeliveryError::InvalidReceiptSignature)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn receipt_rejects_message_content_and_signature_tampering() {
|
||||
let relay = Keys::generate();
|
||||
let original = binding();
|
||||
let message = EventBuilder::new(Kind::Custom(9), "visible")
|
||||
.tags([original.message_v1_target_tag().unwrap()])
|
||||
.sign_with_keys(&relay)
|
||||
.unwrap();
|
||||
let binding = WorkflowDeliveryBinding::new(
|
||||
original.community_id(),
|
||||
original.workflow_id(),
|
||||
original.run_id(),
|
||||
original.step_id(),
|
||||
original.target_pubkey(),
|
||||
original.definition_event_id(),
|
||||
message.id,
|
||||
original.cause().clone(),
|
||||
)
|
||||
.unwrap();
|
||||
let id = WorkflowDeliveryId::from_uuid(Uuid::new_v4());
|
||||
let receipt = WorkflowDeliveryReceipt::new(id, binding.clone(), &message).unwrap();
|
||||
let mut signed = receipt.sign(&relay, &message).unwrap();
|
||||
signed.content = "private input".to_owned();
|
||||
assert_eq!(
|
||||
WorkflowDeliveryReceipt::verify(&signed, relay.public_key(), id, &binding, &message),
|
||||
Err(WorkflowDeliveryError::InvalidReceiptSignature)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wake_round_trips_and_contains_no_authority_beyond_identifier_and_target() {
|
||||
let target = Keys::generate().public_key();
|
||||
|
||||
@@ -3437,6 +3437,23 @@ impl Db {
|
||||
workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await
|
||||
}
|
||||
|
||||
/// Read the run linked by one durable scheduled-fire claim.
|
||||
#[datastore_span(name = "get_scheduled_workflow_fire_run", system = "postgresql")]
|
||||
pub async fn get_scheduled_workflow_fire_run(
|
||||
&self,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
scheduled_for: chrono::DateTime<chrono::Utc>,
|
||||
) -> Result<Option<Uuid>> {
|
||||
workflow::get_scheduled_workflow_fire_run(
|
||||
&self.pool,
|
||||
community_id,
|
||||
workflow_id,
|
||||
scheduled_for,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Attach the workflow run id created from a won scheduled-fire claim.
|
||||
#[datastore_span(name = "attach_scheduled_workflow_run", system = "postgresql")]
|
||||
pub async fn attach_scheduled_workflow_run(
|
||||
@@ -3456,6 +3473,56 @@ impl Db {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Persist a payload-free webhook invocation authority.
|
||||
pub async fn create_workflow_webhook_invocation(
|
||||
&self,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
invocation_id: Uuid,
|
||||
) -> Result<()> {
|
||||
workflow::create_workflow_webhook_invocation(
|
||||
&self.pool,
|
||||
community_id,
|
||||
workflow_id,
|
||||
invocation_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Link a webhook invocation authority to the run it caused.
|
||||
pub async fn attach_workflow_webhook_invocation_run(
|
||||
&self,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
invocation_id: Uuid,
|
||||
workflow_run_id: Uuid,
|
||||
) -> Result<bool> {
|
||||
workflow::attach_workflow_webhook_invocation_run(
|
||||
&self.pool,
|
||||
community_id,
|
||||
workflow_id,
|
||||
invocation_id,
|
||||
workflow_run_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Read the run linked by one opaque webhook invocation.
|
||||
pub async fn get_workflow_webhook_invocation_run(
|
||||
&self,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
invocation_id: Uuid,
|
||||
) -> Result<Option<Uuid>> {
|
||||
workflow::get_workflow_webhook_invocation_run(
|
||||
&self.pool,
|
||||
community_id,
|
||||
workflow_id,
|
||||
invocation_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Delete old scheduled workflow fire claims before a retention cutoff.
|
||||
#[datastore_span(name = "prune_scheduled_workflow_fires_before", system = "postgresql")]
|
||||
pub async fn prune_scheduled_workflow_fires_before(
|
||||
|
||||
@@ -645,7 +645,7 @@ mod tests {
|
||||
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
|
||||
migrations.sort_by_key(|migration| migration.version);
|
||||
|
||||
assert_eq!(migrations.len(), 35);
|
||||
assert_eq!(migrations.len(), 36);
|
||||
assert_eq!(migrations[0].version, 1);
|
||||
assert_eq!(&*migrations[0].description, "initial schema");
|
||||
assert!(migrations[0]
|
||||
@@ -1106,6 +1106,16 @@ mod tests {
|
||||
desired_schema.contains("attach_community_write_fence('workflow_agent_deliveries')")
|
||||
);
|
||||
|
||||
// Payload-free durable authority binds each opaque webhook invocation
|
||||
// to its tenant, workflow, and eventual run.
|
||||
assert_eq!(migrations[35].version, 36);
|
||||
let webhook_invocations = migrations[35].sql.as_str();
|
||||
assert!(webhook_invocations.contains("CREATE TABLE workflow_webhook_invocations"));
|
||||
assert!(webhook_invocations.contains("PRIMARY KEY (community_id, invocation_id)"));
|
||||
assert!(webhook_invocations
|
||||
.contains("attach_community_write_fence('workflow_webhook_invocations')"));
|
||||
assert!(desired_schema.contains("CREATE TABLE workflow_webhook_invocations"));
|
||||
|
||||
// Fresh desired-state bootstrap must install the identical executable
|
||||
// fence as migration 0032. CI and isolated relay startup use schema.sql
|
||||
// without running migrations, so drift reopens rolling-deploy races.
|
||||
@@ -1584,6 +1594,9 @@ mod tests {
|
||||
// Migration 0035 attaches the fence to the durable managed-agent
|
||||
// delivery inbox after 0029; the desired-state schema carries it inline.
|
||||
expected_fences.insert("workflow_agent_deliveries".to_owned());
|
||||
// Migration 0036 likewise adds the payload-free webhook invocation
|
||||
// authority after 0029.
|
||||
expected_fences.insert("workflow_webhook_invocations".to_owned());
|
||||
assert_eq!(
|
||||
expected_fences, schema.fence_attachments,
|
||||
"write-fence attachment targets differ after recovery policy"
|
||||
|
||||
@@ -599,6 +599,32 @@ pub async fn latest_scheduled_workflow_fire(
|
||||
row.try_get("scheduled_for").map_err(Into::into)
|
||||
}
|
||||
|
||||
/// Read the durable authority binding for one scheduled workflow fire.
|
||||
///
|
||||
/// Returns the linked run only from the tenant/workflow/slot primary key; a
|
||||
/// missing or not-yet-attached claim cannot authorize delivery.
|
||||
pub async fn get_scheduled_workflow_fire_run(
|
||||
pool: &PgPool,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
scheduled_for: DateTime<Utc>,
|
||||
) -> Result<Option<Uuid>> {
|
||||
sqlx::query_scalar(
|
||||
r#"
|
||||
SELECT workflow_run_id
|
||||
FROM scheduled_workflow_fires
|
||||
WHERE community_id = $1 AND workflow_id = $2 AND scheduled_for = $3
|
||||
"#,
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
.bind(workflow_id)
|
||||
.bind(scheduled_for)
|
||||
.fetch_optional(pool)
|
||||
.await
|
||||
.map(|value| value.flatten())
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
/// Link a won scheduled-fire claim to the workflow run it created.
|
||||
///
|
||||
/// This is for ops/audit forensics only; the claim row remains the dedupe
|
||||
@@ -632,6 +658,78 @@ pub async fn attach_scheduled_workflow_run(
|
||||
Ok(result.rows_affected() == 1)
|
||||
}
|
||||
|
||||
/// Persist a payload-free webhook invocation authority before run creation.
|
||||
pub async fn create_workflow_webhook_invocation(
|
||||
pool: &PgPool,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
invocation_id: Uuid,
|
||||
) -> Result<()> {
|
||||
sqlx::query(
|
||||
r#"
|
||||
INSERT INTO workflow_webhook_invocations
|
||||
(community_id, invocation_id, workflow_id)
|
||||
SELECT community_id, $3, id
|
||||
FROM workflows
|
||||
WHERE community_id = $1 AND id = $2
|
||||
"#,
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
.bind(workflow_id)
|
||||
.bind(invocation_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Link a durable webhook invocation to the run it caused.
|
||||
pub async fn attach_workflow_webhook_invocation_run(
|
||||
pool: &PgPool,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
invocation_id: Uuid,
|
||||
workflow_run_id: Uuid,
|
||||
) -> Result<bool> {
|
||||
let result = sqlx::query(
|
||||
r#"
|
||||
UPDATE workflow_webhook_invocations
|
||||
SET workflow_run_id = $4
|
||||
WHERE community_id = $1 AND workflow_id = $2 AND invocation_id = $3
|
||||
AND workflow_run_id IS NULL
|
||||
"#,
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
.bind(workflow_id)
|
||||
.bind(invocation_id)
|
||||
.bind(workflow_run_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(result.rows_affected() == 1)
|
||||
}
|
||||
|
||||
/// Read a webhook invocation's durable tenant/workflow/run binding.
|
||||
pub async fn get_workflow_webhook_invocation_run(
|
||||
pool: &PgPool,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
invocation_id: Uuid,
|
||||
) -> Result<Option<Uuid>> {
|
||||
sqlx::query_scalar(
|
||||
r#"
|
||||
SELECT workflow_run_id
|
||||
FROM workflow_webhook_invocations
|
||||
WHERE community_id = $1 AND workflow_id = $2 AND invocation_id = $3
|
||||
"#,
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
.bind(workflow_id)
|
||||
.bind(invocation_id)
|
||||
.fetch_optional(pool)
|
||||
.await
|
||||
.map(|value| value.flatten())
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
/// Delete old scheduled workflow fire claims for retention.
|
||||
///
|
||||
/// Schedule claim rows are correctness metadata, but they grow with every fire.
|
||||
|
||||
@@ -2111,6 +2111,16 @@ pub async fn workflow_webhook(
|
||||
.await
|
||||
.map_err(|_| not_found("workflow not found"))?;
|
||||
|
||||
// Mint durable opaque authority before run creation. The record contains
|
||||
// no request body or secret; a NULL run link remains honest if insertion
|
||||
// of the run later fails.
|
||||
let webhook_invocation_id = uuid::Uuid::new_v4();
|
||||
state
|
||||
.db
|
||||
.create_workflow_webhook_invocation(community_id, id, webhook_invocation_id)
|
||||
.await
|
||||
.map_err(|e| super::internal_error(&format!("persist webhook invocation: {e}")))?;
|
||||
|
||||
let definition_event_id = workflow
|
||||
.definition_event_id
|
||||
.as_deref()
|
||||
@@ -2127,6 +2137,17 @@ pub async fn workflow_webhook(
|
||||
.await
|
||||
.map_err(|e| super::internal_error(&format!("db error: {e}")))?;
|
||||
|
||||
let attached = state
|
||||
.db
|
||||
.attach_workflow_webhook_invocation_run(community_id, id, webhook_invocation_id, run_id)
|
||||
.await
|
||||
.map_err(|e| super::internal_error(&format!("attach webhook invocation: {e}")))?;
|
||||
if !attached {
|
||||
return Err(super::internal_error(
|
||||
"webhook invocation authority was not attachable",
|
||||
));
|
||||
}
|
||||
|
||||
// Spawn workflow execution asynchronously.
|
||||
let engine = Arc::clone(&state.workflow_engine);
|
||||
let trigger_ctx_clone = trigger_ctx.clone();
|
||||
|
||||
@@ -17,7 +17,9 @@ use serde_json::Value;
|
||||
use uuid::Uuid;
|
||||
|
||||
use buzz_core::{
|
||||
workflow_delivery::{WorkflowDeliveryBinding, WorkflowDeliveryCause, WorkflowDeliveryId},
|
||||
workflow_delivery::{
|
||||
WorkflowDeliveryBinding, WorkflowDeliveryCause, WorkflowDeliveryId, WorkflowDeliveryReceipt,
|
||||
},
|
||||
TenantContext,
|
||||
};
|
||||
|
||||
@@ -700,6 +702,66 @@ fn terminal_matches(
|
||||
)
|
||||
}
|
||||
|
||||
async fn validate_delivery_cause_authority(
|
||||
state: &Arc<AppState>,
|
||||
tenant: &TenantContext,
|
||||
binding: &WorkflowDeliveryBinding,
|
||||
) -> Result<(), (StatusCode, Json<Value>)> {
|
||||
let authoritative_run = match binding.cause() {
|
||||
WorkflowDeliveryCause::Event(event_id) => {
|
||||
let event = state
|
||||
.db
|
||||
.get_event_by_id(tenant.community(), event_id.as_bytes())
|
||||
.await
|
||||
.map_err(|error| internal_error(&format!("delivery cause lookup: {error}")))?
|
||||
.ok_or_else(|| {
|
||||
api_error(StatusCode::CONFLICT, "delivery authority is unavailable")
|
||||
})?;
|
||||
if event.event.verify().is_err() {
|
||||
return Err(api_error(
|
||||
StatusCode::CONFLICT,
|
||||
"delivery authority is unavailable",
|
||||
));
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
WorkflowDeliveryCause::Schedule {
|
||||
scheduled_for_unix_seconds,
|
||||
} => {
|
||||
let scheduled_for =
|
||||
chrono::DateTime::<Utc>::from_timestamp(*scheduled_for_unix_seconds, 0)
|
||||
.ok_or_else(|| {
|
||||
api_error(StatusCode::CONFLICT, "delivery authority is unavailable")
|
||||
})?;
|
||||
state
|
||||
.db
|
||||
.get_scheduled_workflow_fire_run(
|
||||
tenant.community(),
|
||||
binding.workflow_id(),
|
||||
scheduled_for,
|
||||
)
|
||||
.await
|
||||
.map_err(|error| internal_error(&format!("schedule authority lookup: {error}")))?
|
||||
}
|
||||
WorkflowDeliveryCause::Webhook { invocation_id } => state
|
||||
.db
|
||||
.get_workflow_webhook_invocation_run(
|
||||
tenant.community(),
|
||||
binding.workflow_id(),
|
||||
*invocation_id,
|
||||
)
|
||||
.await
|
||||
.map_err(|error| internal_error(&format!("webhook authority lookup: {error}")))?,
|
||||
};
|
||||
if authoritative_run != Some(binding.run_id()) {
|
||||
return Err(api_error(
|
||||
StatusCode::CONFLICT,
|
||||
"delivery authority is unavailable",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delivery_response(
|
||||
state: &Arc<AppState>,
|
||||
tenant: &TenantContext,
|
||||
@@ -746,10 +808,16 @@ async fn delivery_response(
|
||||
.await
|
||||
.map_err(|error| internal_error(&format!("delivery message lookup: {error}")))?
|
||||
.ok_or_else(|| api_error(StatusCode::CONFLICT, "delivery binding is unavailable"))?;
|
||||
validate_delivery_cause_authority(state, tenant, &delivery.binding).await?;
|
||||
let receipt =
|
||||
WorkflowDeliveryReceipt::new(delivery.id, delivery.binding.clone(), &message.event)
|
||||
.and_then(|receipt| receipt.sign(&state.relay_keypair, &message.event))
|
||||
.map_err(|error| internal_error(&format!("sign delivery receipt: {error}")))?;
|
||||
Ok(Json(serde_json::json!({
|
||||
"delivery": delivery_json(delivery, lease),
|
||||
"definition_event": definition.event,
|
||||
"message_event": message.event,
|
||||
"receipt_event": receipt,
|
||||
})))
|
||||
}
|
||||
|
||||
@@ -1154,6 +1222,16 @@ mod tests {
|
||||
"private execution input"
|
||||
);
|
||||
assert!(claimed.get("execution_trace").is_none());
|
||||
assert!(claimed.get("trigger_context").is_none());
|
||||
assert!(claimed.get("webhook_fields").is_none());
|
||||
let receipt: nostr::Event =
|
||||
serde_json::from_value(claimed["receipt_event"].clone()).expect("signed receipt event");
|
||||
assert_eq!(receipt.pubkey, fixture.state.relay_keypair.public_key());
|
||||
assert!(receipt.verify().is_ok());
|
||||
assert!(receipt.content.is_empty());
|
||||
assert!(!claimed["receipt_event"]
|
||||
.to_string()
|
||||
.contains("private execution input"));
|
||||
|
||||
let lease = serde_json::json!({
|
||||
"lease_generation": 1,
|
||||
|
||||
20
migrations/0036_workflow_webhook_invocations.sql
Normal file
20
migrations/0036_workflow_webhook_invocations.sql
Normal file
@@ -0,0 +1,20 @@
|
||||
-- Durable, payload-free authority for webhook-triggered workflow runs.
|
||||
-- The body and secret stay in transient trigger context; this table retains
|
||||
-- only an opaque invocation identity and its tenant/workflow/run binding.
|
||||
CREATE TABLE workflow_webhook_invocations (
|
||||
community_id UUID NOT NULL REFERENCES communities(id),
|
||||
invocation_id UUID NOT NULL,
|
||||
workflow_id UUID NOT NULL,
|
||||
workflow_run_id UUID,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
PRIMARY KEY (community_id, invocation_id),
|
||||
FOREIGN KEY (community_id, workflow_id)
|
||||
REFERENCES workflows (community_id, id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (community_id, workflow_run_id)
|
||||
REFERENCES workflow_runs (community_id, id) ON DELETE NO ACTION
|
||||
);
|
||||
|
||||
CREATE INDEX idx_workflow_webhook_invocations_created_at
|
||||
ON workflow_webhook_invocations (created_at);
|
||||
|
||||
SELECT attach_community_write_fence('workflow_webhook_invocations');
|
||||
@@ -536,6 +536,24 @@ CREATE INDEX idx_workflow_agent_deliveries_lease
|
||||
ON workflow_agent_deliveries (lease_until)
|
||||
WHERE status = 'claimed';
|
||||
|
||||
-- ── Workflow webhook invocations ─────────────────────────────────────────────
|
||||
-- Payload-free durable authority linking an opaque webhook call to its run.
|
||||
CREATE TABLE workflow_webhook_invocations (
|
||||
community_id UUID NOT NULL REFERENCES communities(id),
|
||||
invocation_id UUID NOT NULL,
|
||||
workflow_id UUID NOT NULL,
|
||||
workflow_run_id UUID,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
PRIMARY KEY (community_id, invocation_id),
|
||||
FOREIGN KEY (community_id, workflow_id)
|
||||
REFERENCES workflows (community_id, id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (community_id, workflow_run_id)
|
||||
REFERENCES workflow_runs (community_id, id) ON DELETE NO ACTION
|
||||
);
|
||||
|
||||
CREATE INDEX idx_workflow_webhook_invocations_created_at
|
||||
ON workflow_webhook_invocations (created_at);
|
||||
|
||||
-- ── API tokens ────────────────────────────────────────────────────────────────
|
||||
-- Conformance: "API tokens and NIP-98 replay". token_hash uniqueness scoped to
|
||||
-- (community_id, token_hash); channel claims reference channels in same community.
|
||||
@@ -1810,6 +1828,7 @@ SELECT attach_community_write_fence('reactions');
|
||||
SELECT attach_community_write_fence('relay_invites');
|
||||
SELECT attach_community_write_fence('relay_members');
|
||||
SELECT attach_community_write_fence('scheduled_workflow_fires');
|
||||
SELECT attach_community_write_fence('workflow_webhook_invocations');
|
||||
SELECT attach_community_write_fence('subscriptions');
|
||||
SELECT attach_community_write_fence('thread_metadata');
|
||||
SELECT attach_community_write_fence('users');
|
||||
|
||||
Reference in New Issue
Block a user