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 <cursoragent@cursor.com>
2026-09-15 09:08:15 +00:00
parent 5f3b117c00
commit 5fd5588a3f
1 changed files with 65 additions and 0 deletions
@@ -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;