From 5fd5588a3f505240a4e06bd2163b98ce5824d255 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 09:08:15 +0000 Subject: [PATCH] feat(db): add dispatcher attempts, consumer modes, and checkpoints E0B needs an additive lease audit trail and a cluster-level off/shadow/live table before any worker can claim. FAV remains constrained to off. Co-authored-by: Cursor --- .../00040_2026-09-15-event-dispatcher.sql | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 deploy/postgres/goose/00040_2026-09-15-event-dispatcher.sql diff --git a/deploy/postgres/goose/00040_2026-09-15-event-dispatcher.sql b/deploy/postgres/goose/00040_2026-09-15-event-dispatcher.sql new file mode 100644 index 00000000..6f528fed --- /dev/null +++ b/deploy/postgres/goose/00040_2026-09-15-event-dispatcher.sql @@ -0,0 +1,65 @@ +-- +goose Up +-- E0B: atomic lease claim, immutable attempts, cluster consumer mode, checkpoints. +-- Dispatcher defaults to off. Do not drop historical envelopes or deliveries. + +CREATE TABLE IF NOT EXISTS event_delivery_attempts ( + id BIGSERIAL PRIMARY KEY, + event_id TEXT NOT NULL, + consumer TEXT NOT NULL, + attempt INT NOT NULL, + lease_owner TEXT NOT NULL, + fencing_token TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + finished_at TIMESTAMPTZ, + outcome TEXT NOT NULL DEFAULT 'leased', + error_code TEXT NOT NULL DEFAULT '', + correlation_id TEXT NOT NULL DEFAULT '', + CONSTRAINT uq_event_delivery_attempt UNIQUE (event_id, consumer, attempt, fencing_token), + CONSTRAINT chk_event_attempt_outcome CHECK (outcome IN ('leased','acked','pending','dead','quarantine','fenced')) +); + +CREATE INDEX IF NOT EXISTS idx_event_delivery_attempts_consumer + ON event_delivery_attempts (consumer, started_at DESC); + +CREATE TABLE IF NOT EXISTS event_consumer_modes ( + environment TEXT NOT NULL, + consumer TEXT NOT NULL, + mode TEXT NOT NULL, + revision BIGINT NOT NULL DEFAULT 1, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (environment, consumer), + CONSTRAINT chk_event_consumer_mode CHECK (mode IN ('off','shadow','live')), + CONSTRAINT chk_event_fav_mode_off CHECK (consumer <> 'fav' OR mode = 'off') +); + +CREATE TABLE IF NOT EXISTS event_consumer_checkpoints ( + consumer TEXT NOT NULL, + aggregate_type TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + last_event_id TEXT NOT NULL, + last_aggregate_version BIGINT, + fencing_token TEXT NOT NULL DEFAULT '', + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (consumer, aggregate_type, aggregate_id) +); + +CREATE TABLE IF NOT EXISTS event_reconciliations ( + event_id TEXT NOT NULL, + consumer TEXT NOT NULL, + mode TEXT NOT NULL, + intended_digest CHAR(64) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (event_id, consumer, mode) +); + +INSERT INTO event_consumer_modes (environment, consumer, mode, revision) +SELECT e.env, c.consumer, 'off', 1 +FROM (VALUES ('dev'), ('test'), ('main')) AS e(env) +CROSS JOIN (VALUES ('notify'), ('projection'), ('workflow_advance'), ('fav')) AS c(consumer) +ON CONFLICT (environment, consumer) DO NOTHING; + +-- +goose Down +DROP TABLE IF EXISTS event_reconciliations; +DROP TABLE IF EXISTS event_consumer_checkpoints; +DROP TABLE IF EXISTS event_consumer_modes; +DROP TABLE IF EXISTS event_delivery_attempts;