From 8ac764e3b888fbd225b359fd757ca525f5f54588 Mon Sep 17 00:00:00 2001 From: Arnas Donauskas Date: Fri, 22 May 2026 15:35:42 +0300 Subject: [PATCH 1/3] fix(flows): redact customer text from reply_received event payload The `reply_received` event was persisting `text: message.text` verbatim into flow_run_events.payload. A customer prompted "what's your card number?" by a collect_input node would leave the PAN sitting in the events table forever, surfaced via the runs viewer to anyone with access. Switch to `text_length` only. The `node_entered` event already records `captured_length` for the same reason after a successful capture; this makes the asymmetry symmetric. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/lib/flows/engine.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/lib/flows/engine.ts b/src/lib/flows/engine.ts index 069c98a..612709d 100644 --- a/src/lib/flows/engine.ts +++ b/src/lib/flows/engine.ts @@ -822,11 +822,18 @@ async function handleReplyForActiveRun( run: FlowRunRow, message: ParsedInbound, ): Promise { + // Note: we intentionally do NOT persist the raw customer text. A + // `collect_input` prompt that asks "what's your card number?" would + // otherwise leave the PAN sitting in flow_run_events.payload forever, + // visible to anyone with access to the runs viewer or the events + // table. Length is enough for "did they actually reply?" debugging; + // for the captured value itself, the `node_entered` event already + // records `captured_key` + `captured_length` after the var is stored. await logEvent(db, run.id, "reply_received", run.current_node_key, { meta_message_id: message.meta_message_id, reply_kind: message.kind, reply_id: message.kind === "interactive_reply" ? message.reply_id : null, - text: message.kind === "text" ? message.text : null, + text_length: message.kind === "text" ? message.text.length : null, }); if (!run.current_node_key) { From e83538678a6de578c55d0889efa24f8acf050501 Mon Sep 17 00:00:00 2001 From: Arnas Donauskas Date: Fri, 22 May 2026 15:39:00 +0300 Subject: [PATCH 2/3] fix(flows): atomic execution_count increment via SECURITY DEFINER RPC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit startNewRun read flows.execution_count from a cached row and wrote `cached + 1`. Two concurrent webhooks starting runs on the same flow for different contacts both read N and both write N+1 — one count permanently lost. Mirrors the automations engine's solution (migration 007): a tiny SECURITY DEFINER function that does the increment server-side, locked to the service role. The engine's call becomes a single RPC. The counter is best-effort: if the RPC fails we log + continue rather than failing the run for a stat error. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/lib/flows/engine.ts | 20 +++++++---- .../012_flows_increment_counter.sql | 36 +++++++++++++++++++ 2 files changed, 49 insertions(+), 7 deletions(-) create mode 100644 supabase/migrations/012_flows_increment_counter.sql diff --git a/src/lib/flows/engine.ts b/src/lib/flows/engine.ts index 069c98a..99052e0 100644 --- a/src/lib/flows/engine.ts +++ b/src/lib/flows/engine.ts @@ -1004,13 +1004,19 @@ async function startNewRun( }); // Bump the flow's execution counter — used by the builder UI to // surface "X runs since activation" on the flow card. - await db - .from("flows") - .update({ - execution_count: flow.execution_count + 1, - last_executed_at: new Date().toISOString(), - }) - .eq("id", flow.id); + // + // Atomic RPC (migration 012) rather than read-modify-write: two + // concurrent webhooks starting runs for different contacts on the + // same flow would otherwise both read N and both write N+1, losing + // a count. Mirrors the automations engine's use of + // `increment_automation_execution_count` (migration 007). + const { error: incErr } = await db.rpc("increment_flow_execution_count", { + p_flow_id: flow.id, + }); + if (incErr) { + // Non-fatal — the run itself succeeded; only the counter is off. + console.error("[flows] execution_count rpc error:", incErr.message); + } // Run the advance loop starting from the entry node. const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!); diff --git a/supabase/migrations/012_flows_increment_counter.sql b/supabase/migrations/012_flows_increment_counter.sql new file mode 100644 index 0000000..25eb4cf --- /dev/null +++ b/supabase/migrations/012_flows_increment_counter.sql @@ -0,0 +1,36 @@ +-- ============================================================ +-- 012_flows_increment_counter.sql +-- +-- Atomic increment of flows.execution_count + refresh of +-- last_executed_at. Called via PostgREST RPC from the engine. +-- +-- Before this, startNewRun did a read-modify-write: +-- UPDATE flows SET execution_count = WHERE id = ... +-- so two concurrent dispatches (e.g. two webhooks for the same flow +-- starting runs for different contacts in the same second) could both +-- read N and both write N+1, permanently losing one count. +-- +-- Mirrors migration 007 for automations — same shape, same security +-- posture. Idempotent: safe to re-run. +-- ============================================================ + +CREATE OR REPLACE FUNCTION increment_flow_execution_count(p_flow_id UUID) +RETURNS VOID +LANGUAGE sql +SECURITY DEFINER +SET search_path = public +AS $$ + UPDATE flows + SET + execution_count = execution_count + 1, + last_executed_at = NOW() + WHERE id = p_flow_id; +$$; + +-- Only the service role needs to call this (engine uses the +-- service-role client). Explicitly lock anon / authenticated out so +-- an authenticated user can't juice someone else's counter via RPC. +REVOKE ALL ON FUNCTION increment_flow_execution_count(UUID) FROM PUBLIC; +REVOKE ALL ON FUNCTION increment_flow_execution_count(UUID) FROM anon; +REVOKE ALL ON FUNCTION increment_flow_execution_count(UUID) FROM authenticated; +GRANT EXECUTE ON FUNCTION increment_flow_execution_count(UUID) TO service_role; From dc58f71925df656734d4f191f94fe8e34601df18 Mon Sep 17 00:00:00 2001 From: Arnas Donauskas Date: Fri, 22 May 2026 15:44:40 +0300 Subject: [PATCH 3/3] perf(flows): preload all flow_nodes once per dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The advance loop called loadNode() on every iteration — one DB round trip per node visited. A flow with 5 chained auto-advance nodes (send_message → condition → set_tag → send_message → end) cost 5 sequential round trips synchronous in the webhook hot path. At ~50ms each that's 250ms before the customer sees anything. Switch to a single upfront SELECT into a Map keyed on the flow_id, passed through advanceFromNodeKey and handleReplyForActiveRun. The advance loop is now in-memory and the hot path costs one round trip regardless of chain length. node_not_found semantics are preserved: missing keys still end the run with that reason, same as the old per-node lookup. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/lib/flows/engine.ts | 47 ++++++++++++++++++++++++++++------------- 1 file changed, 32 insertions(+), 15 deletions(-) diff --git a/src/lib/flows/engine.ts b/src/lib/flows/engine.ts index 069c98a..7f79b5d 100644 --- a/src/lib/flows/engine.ts +++ b/src/lib/flows/engine.ts @@ -203,22 +203,32 @@ async function loadFlow( return (data as FlowRow | null) ?? null; } -async function loadNode( +/** + * Load every node of a flow in one round trip and key them by + * `node_key`. The advance loop is then in-memory — a 5-node + * auto-advancing chain costs one SELECT, not five. + * + * Returns an empty map on error so the caller can still dispatch + * cleanly (every subsequent .get() returns undefined → the run + * fails with node_not_found, same as the old per-node lookup). + */ +async function loadAllNodes( db: AdminClient, flowId: string, - nodeKey: string, -): Promise { +): Promise> { const { data, error } = await db .from("flow_nodes") .select("*") - .eq("flow_id", flowId) - .eq("node_key", nodeKey) - .maybeSingle(); + .eq("flow_id", flowId); if (error) { - console.error("[flows] loadNode error:", error.message); - return null; + console.error("[flows] loadAllNodes error:", error.message); + return new Map(); } - return (data as FlowNodeRow | null) ?? null; + const map = new Map(); + for (const row of (data ?? []) as FlowNodeRow[]) { + map.set(row.node_key, row); + } + return map; } async function logEvent( @@ -522,6 +532,7 @@ async function advanceFromNodeKey( db: AdminClient, run: FlowRunRow, startNodeKey: string, + nodes: Map, ): Promise<{ outcome: "advanced" | "completed" | "handed_off" }> { let currentKey: string | null = startNodeKey; // Defensive cap — if a flow has a cycle (which the validator @@ -534,7 +545,7 @@ async function advanceFromNodeKey( await endRun(db, run.id, "failed", "missing_next_node"); return { outcome: "completed" }; } - const node = await loadNode(db, run.flow_id, currentKey); + const node: FlowNodeRow | null = nodes.get(currentKey) ?? null; if (!node) { await logEvent(db, run.id, "error", currentKey, { reason: "node_not_found", @@ -794,7 +805,10 @@ export async function dispatchInboundToFlows( outcome: "duplicate_inbound_ignored", }; } - return handleReplyForActiveRun(db, activeRun, input.message); + // One SELECT for the whole flow's nodes — advance loop is now + // in-memory. See loadAllNodes. + const nodes = await loadAllNodes(db, activeRun.flow_id); + return handleReplyForActiveRun(db, activeRun, input.message, nodes); } // No active run → look for a flow whose entry trigger matches. @@ -807,7 +821,8 @@ export async function dispatchInboundToFlows( if (!flow || !flow.entry_node_id) { return { consumed: false, outcome: "no_match" }; } - return startNewRun(db, flow, input); + const nodes = await loadAllNodes(db, flow.id); + return startNewRun(db, flow, input, nodes); } catch (err) { console.error( "[flows] dispatchInboundToFlows threw:", @@ -821,6 +836,7 @@ async function handleReplyForActiveRun( db: AdminClient, run: FlowRunRow, message: ParsedInbound, + nodes: Map, ): Promise { await logEvent(db, run.id, "reply_received", run.current_node_key, { meta_message_id: message.meta_message_id, @@ -840,7 +856,7 @@ async function handleReplyForActiveRun( }; } - const currentNode = await loadNode(db, run.flow_id, run.current_node_key); + const currentNode = nodes.get(run.current_node_key) ?? null; if (!currentNode) { await endRun(db, run.id, "failed", "current_node_not_found"); return { consumed: true, flow_run_id: run.id, outcome: "no_match" }; @@ -896,7 +912,7 @@ async function handleReplyForActiveRun( .eq("id", run.id) .maybeSingle(); const next = (fresh.data as FlowRunRow | null) ?? run; - const outcome = await advanceFromNodeKey(db, next, matched); + const outcome = await advanceFromNodeKey(db, next, matched, nodes); return { consumed: true, flow_run_id: run.id, @@ -971,6 +987,7 @@ async function startNewRun( db: AdminClient, flow: FlowRow, input: DispatchInboundInput, + nodes: Map, ): Promise { // INSERT — partial unique index `idx_one_active_run_per_contact` // catches concurrent inserts with 23505. We catch and return as @@ -1013,7 +1030,7 @@ async function startNewRun( .eq("id", flow.id); // Run the advance loop starting from the entry node. - const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!); + const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!, nodes); return { consumed: true, flow_run_id: run.id,