Merge remote-tracking branch 'origin/main' into feat/flows-drop-reread-and-limit

# Conflicts:
#	src/lib/flows/engine.ts

src/lib/flows/engine.ts

@@ -211,22 +211,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<FlowNodeRow | null> {
+): Promise<Map<string, FlowNodeRow>> {
   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<string, FlowNodeRow>();
+  for (const row of (data ?? []) as FlowNodeRow[]) {
+    map.set(row.node_key, row);
+  }
+  return map;
 }
 
 async function logEvent(
@@ -530,6 +540,7 @@ async function advanceFromNodeKey(
   db: AdminClient,
   run: FlowRunRow,
   startNodeKey: string,
+  nodes: Map<string, FlowNodeRow>,
 ): Promise<{ outcome: "advanced" | "completed" | "handed_off" }> {
   let currentKey: string | null = startNodeKey;
   // Defensive cap — if a flow has a cycle (which the validator
@@ -542,7 +553,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",
@@ -802,7 +813,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.
@@ -815,7 +829,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:",
@@ -829,12 +844,20 @@ async function handleReplyForActiveRun(
   db: AdminClient,
   run: FlowRunRow,
   message: ParsedInbound,
+  nodes: Map<string, FlowNodeRow>,
 ): Promise<DispatchInboundResult> {
+  // 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) {
@@ -848,7 +871,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" };
@@ -912,7 +935,7 @@ async function handleReplyForActiveRun(
         .eq("id", run.id);
       if (!error) run.reprompt_count = 0;
     }
-    const outcome = await advanceFromNodeKey(db, run, matched);
+    const outcome = await advanceFromNodeKey(db, run, matched, nodes);
     return {
       consumed: true,
       flow_run_id: run.id,
@@ -987,6 +1010,7 @@ async function startNewRun(
   db: AdminClient,
   flow: FlowRow,
   input: DispatchInboundInput,
+  nodes: Map<string, FlowNodeRow>,
 ): Promise<DispatchInboundResult> {
   // INSERT — partial unique index `idx_one_active_run_per_contact`
   // catches concurrent inserts with 23505. We catch and return as
@@ -1020,16 +1044,22 @@ 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!);
+  const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!, nodes);
   return {
     consumed: true,
     flow_run_id: run.id,

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 = <cached + 1> 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;