mirror of
https://github.com/Luchexx/wacrm.git
synced 2026-11-02 09:19:51 +00:00
Merge pull request #124 from ArnasDon/feat/flows-atomic-execution-count
fix(flows): atomic execution_count increment via SECURITY DEFINER RPC
This commit is contained in:
commit
4b368139f3
@ -1011,13 +1011,19 @@ async function startNewRun(
|
|||||||
});
|
});
|
||||||
// Bump the flow's execution counter — used by the builder UI to
|
// Bump the flow's execution counter — used by the builder UI to
|
||||||
// surface "X runs since activation" on the flow card.
|
// surface "X runs since activation" on the flow card.
|
||||||
await db
|
//
|
||||||
.from("flows")
|
// Atomic RPC (migration 012) rather than read-modify-write: two
|
||||||
.update({
|
// concurrent webhooks starting runs for different contacts on the
|
||||||
execution_count: flow.execution_count + 1,
|
// same flow would otherwise both read N and both write N+1, losing
|
||||||
last_executed_at: new Date().toISOString(),
|
// a count. Mirrors the automations engine's use of
|
||||||
})
|
// `increment_automation_execution_count` (migration 007).
|
||||||
.eq("id", flow.id);
|
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.
|
// 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!);
|
||||||
|
|||||||
36
supabase/migrations/012_flows_increment_counter.sql
Normal file
36
supabase/migrations/012_flows_increment_counter.sql
Normal file
@ -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;
|
||||||
Loading…
Reference in New Issue
Block a user