Fence legacy workflow revision reconciliation

Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
This commit is contained in:
Logan Johnson
2026-08-28 10:15:59 -04:00
parent dbe2b67af8
commit b064beb5c3
2 changed files with 55 additions and 14 deletions

View File

@@ -424,13 +424,14 @@ pub async fn bind_legacy_workflow_revision(
let result = sqlx::query(
r#"
UPDATE workflows
SET definition_event_id = $7
SET definition_event_id = $8
WHERE community_id = $1
AND id = $2
AND owner_pubkey = $3
AND channel_id IS NOT DISTINCT FROM $4
AND definition = $5
AND definition_hash = $6
AND updated_at = $7
AND definition_event_id IS NULL
"#,
)
@@ -440,6 +441,7 @@ pub async fn bind_legacy_workflow_revision(
.bind(workflow.channel_id)
.bind(&workflow.definition)
.bind(&workflow.definition_hash)
.bind(workflow.updated_at)
.bind(definition_event_id)
.execute(pool)
.await?;
@@ -1997,6 +1999,43 @@ mod tests {
assert_eq!(current.definition_hash, rewritten_hash);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn legacy_revision_binding_rejects_an_equal_definition_rewrite() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
let (workflow_id, _) = make_workflow_in(&pool, community).await;
let snapshot = get_workflow(&pool, community, workflow_id)
.await
.expect("read legacy workflow snapshot");
sqlx::query(
r#"
UPDATE workflows
SET definition = definition,
definition_hash = definition_hash,
updated_at = updated_at + INTERVAL '1 microsecond'
WHERE community_id = $1 AND id = $2
"#,
)
.bind(community.as_uuid())
.bind(workflow_id)
.execute(&pool)
.await
.expect("simulate equal-materialization legacy rewrite");
assert!(
!bind_legacy_workflow_revision(&pool, &snapshot, &[0x42; 32])
.await
.expect("reject stale generation")
);
assert!(get_workflow(&pool, community, workflow_id)
.await
.expect("read rewritten workflow")
.definition_event_id
.is_none());
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn revision_capture_is_additive_and_transactional() {

View File

@@ -328,16 +328,12 @@ fn legacy_definition_matches(workflow: &WorkflowRecord, event: &Event) -> bool {
// Webhook secrets were generated by the relay and never existed in the
// owner-signed event. Reuse the stored secret before comparing the exact
// materialized JSON and its legacy hash.
let materialized_json =
if let Some(secret) = webhook_secret::extract_secret(&workflow.definition) {
webhook_secret::inject_secret(&mut definition, &secret);
let Ok(json) = serde_json::to_string(&definition) else {
return false;
};
json
} else {
canonical_json
};
if let Some(secret) = webhook_secret::extract_secret(&workflow.definition) {
webhook_secret::inject_secret(&mut definition, &secret);
}
let Ok(materialized_json) = serde_json::to_string(&definition) else {
return false;
};
definition == workflow.definition
&& compute_definition_hash(&materialized_json) == workflow.definition_hash
@@ -1532,6 +1528,8 @@ mod tests {
.expect("canonical workflow definition");
let definition: serde_json::Value =
serde_json::from_str(&canonical_json).expect("definition json");
let materialized_json =
serde_json::to_string(&definition).expect("production materialized json");
WorkflowRecord {
id: Uuid::parse_str(extract_d_tag(event).as_deref().expect("d tag"))
.expect("workflow UUID"),
@@ -1539,7 +1537,7 @@ mod tests {
name: "legacy".to_string(),
owner_pubkey: event.pubkey.to_bytes().to_vec(),
channel_id: Some(channel_id),
definition_hash: compute_definition_hash(&canonical_json),
definition_hash: compute_definition_hash(&materialized_json),
definition,
definition_event_id: None,
status: buzz_db::workflow::WorkflowStatus::Active,
@@ -1626,14 +1624,18 @@ mod tests {
let content = "name: legacy\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: hello\n";
let (_, canonical_json) = buzz_workflow::WorkflowEngine::parse_yaml(content)
.expect("canonical workflow definition");
let definition: serde_json::Value =
serde_json::from_str(&canonical_json).expect("production definition value");
let definition_json_final =
serde_json::to_string(&definition).expect("production materialized definition");
let workflow_id = db
.create_workflow(
tenant.community(),
Some(channel.id),
owner.as_slice(),
"legacy",
&canonical_json,
&compute_definition_hash(&canonical_json),
&definition_json_final,
&compute_definition_hash(&definition_json_final),
)
.await
.expect("create legacy workflow");