perf(flows): preload all flow_nodes once per dispatch

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<node_key, FlowNodeRow>
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) <noreply@anthropic.com>
This commit is contained in:
Arnas Donauskas 2026-05-22 15:44:40 +03:00
parent 9ff4a8d1ba
commit dc58f71925
No known key found for this signature in database

View File

@ -203,22 +203,32 @@ async function loadFlow(
return (data as FlowRow | null) ?? null; 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, db: AdminClient,
flowId: string, flowId: string,
nodeKey: string, ): Promise<Map<string, FlowNodeRow>> {
): Promise<FlowNodeRow | null> {
const { data, error } = await db const { data, error } = await db
.from("flow_nodes") .from("flow_nodes")
.select("*") .select("*")
.eq("flow_id", flowId) .eq("flow_id", flowId);
.eq("node_key", nodeKey)
.maybeSingle();
if (error) { if (error) {
console.error("[flows] loadNode error:", error.message); console.error("[flows] loadAllNodes error:", error.message);
return null; 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( async function logEvent(
@ -522,6 +532,7 @@ async function advanceFromNodeKey(
db: AdminClient, db: AdminClient,
run: FlowRunRow, run: FlowRunRow,
startNodeKey: string, startNodeKey: string,
nodes: Map<string, FlowNodeRow>,
): Promise<{ outcome: "advanced" | "completed" | "handed_off" }> { ): Promise<{ outcome: "advanced" | "completed" | "handed_off" }> {
let currentKey: string | null = startNodeKey; let currentKey: string | null = startNodeKey;
// Defensive cap — if a flow has a cycle (which the validator // 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"); await endRun(db, run.id, "failed", "missing_next_node");
return { outcome: "completed" }; return { outcome: "completed" };
} }
const node = await loadNode(db, run.flow_id, currentKey); const node: FlowNodeRow | null = nodes.get(currentKey) ?? null;
if (!node) { if (!node) {
await logEvent(db, run.id, "error", currentKey, { await logEvent(db, run.id, "error", currentKey, {
reason: "node_not_found", reason: "node_not_found",
@ -794,7 +805,10 @@ export async function dispatchInboundToFlows(
outcome: "duplicate_inbound_ignored", 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. // 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) { if (!flow || !flow.entry_node_id) {
return { consumed: false, outcome: "no_match" }; 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) { } catch (err) {
console.error( console.error(
"[flows] dispatchInboundToFlows threw:", "[flows] dispatchInboundToFlows threw:",
@ -821,6 +836,7 @@ async function handleReplyForActiveRun(
db: AdminClient, db: AdminClient,
run: FlowRunRow, run: FlowRunRow,
message: ParsedInbound, message: ParsedInbound,
nodes: Map<string, FlowNodeRow>,
): Promise<DispatchInboundResult> { ): Promise<DispatchInboundResult> {
await logEvent(db, run.id, "reply_received", run.current_node_key, { await logEvent(db, run.id, "reply_received", run.current_node_key, {
meta_message_id: message.meta_message_id, 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) { if (!currentNode) {
await endRun(db, run.id, "failed", "current_node_not_found"); await endRun(db, run.id, "failed", "current_node_not_found");
return { consumed: true, flow_run_id: run.id, outcome: "no_match" }; return { consumed: true, flow_run_id: run.id, outcome: "no_match" };
@ -896,7 +912,7 @@ async function handleReplyForActiveRun(
.eq("id", run.id) .eq("id", run.id)
.maybeSingle(); .maybeSingle();
const next = (fresh.data as FlowRunRow | null) ?? run; const next = (fresh.data as FlowRunRow | null) ?? run;
const outcome = await advanceFromNodeKey(db, next, matched); const outcome = await advanceFromNodeKey(db, next, matched, nodes);
return { return {
consumed: true, consumed: true,
flow_run_id: run.id, flow_run_id: run.id,
@ -971,6 +987,7 @@ async function startNewRun(
db: AdminClient, db: AdminClient,
flow: FlowRow, flow: FlowRow,
input: DispatchInboundInput, input: DispatchInboundInput,
nodes: Map<string, FlowNodeRow>,
): Promise<DispatchInboundResult> { ): Promise<DispatchInboundResult> {
// INSERT — partial unique index `idx_one_active_run_per_contact` // INSERT — partial unique index `idx_one_active_run_per_contact`
// catches concurrent inserts with 23505. We catch and return as // catches concurrent inserts with 23505. We catch and return as
@ -1013,7 +1030,7 @@ async function startNewRun(
.eq("id", flow.id); .eq("id", flow.id);
// Run the advance loop starting from the entry node. // 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 { return {
consumed: true, consumed: true,
flow_run_id: run.id, flow_run_id: run.id,