-- =============================================================================
-- Durable Work in Postgres — schema migration (inbox reference implementation)
-- =============================================================================
-- Established patterns (EIP, Transactional Outbox, etc.) implemented in Postgres.
-- Shapes: inbox (Competing Consumers), outbox (Transactional Outbox), projection, webhooks.
-- Brokers at boundaries are transport — not a substitute for this claim loop.
--   • inbox table (reference shape — same columns work for outbox / checkpoints)
--   • workers table (heartbeats + live worker registry)
--   • consistent hash via partition_key + partition_bucket
--   • partition_bucket  (stable hash at enqueue for SQL claim filter — not "owner at INSERT time")
--   • lease_generation for fencing tokens
--   • NOTIFY trigger for optional worker wake-up
--
-- Pattern also applies to outbox relays, projections, and saga steps.
-- Run: psql $DATABASE_URL -f schema.sql
-- UUIDv7: use built-in/extension uuidv7() when available, or pass app-generated UUIDv7 IDs.
-- =============================================================================

BEGIN;

-- ── Workers registry ──────────────────────────────────────────────────────────

CREATE TABLE IF NOT EXISTS workers (
  id           TEXT PRIMARY KEY,              -- e.g. hostname-pid
  status       TEXT NOT NULL DEFAULT 'alive'
               CHECK (status IN ('alive', 'draining', 'dead')),
  last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  started_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
  metadata     JSONB NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX IF NOT EXISTS idx_workers_alive
  ON workers (last_seen_at)
  WHERE status = 'alive';

-- ── Inbox queue ───────────────────────────────────────────────────────────────

DO $$ BEGIN
  CREATE TYPE inbox_status AS ENUM (
    'pending', 'processing', 'completed', 'failed', 'dead_letter'
  );
EXCEPTION
  WHEN duplicate_object THEN NULL;
END $$;

CREATE TABLE IF NOT EXISTS inbox (
  id                UUID PRIMARY KEY DEFAULT uuidv7(), -- or generate UUIDv7 in the app
  partition_key     TEXT NOT NULL,            -- consistent hash input
  partition_bucket  INT NOT NULL CHECK (partition_bucket >= 0 AND partition_bucket < 1024),
  payload           JSONB NOT NULL,
  status            inbox_status NOT NULL DEFAULT 'pending',
  attempts          INT NOT NULL DEFAULT 0,
  max_attempts      INT NOT NULL DEFAULT 5,
  claimed_by        TEXT REFERENCES workers(id) ON DELETE SET NULL,
  claimed_at        TIMESTAMPTZ,
  lease_expires_at  TIMESTAMPTZ,
  lease_generation  BIGINT NOT NULL DEFAULT 0, -- fencing token; bumped on each claim
  available_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
  completed_at      TIMESTAMPTZ,
  last_error        TEXT,
  idempotency_key   TEXT,
  created_at        TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Pending rows ready to claim
CREATE INDEX IF NOT EXISTS idx_inbox_claim
  ON inbox (available_at, created_at, id)
  WHERE status = 'pending';

-- Index-friendly ownership filter via partition_bucket
CREATE INDEX IF NOT EXISTS idx_inbox_bucket_claim
  ON inbox (partition_bucket, available_at, created_at, id)
  WHERE status = 'pending';

-- Stale processing / reaper scans
CREATE INDEX IF NOT EXISTS idx_inbox_stale_processing
  ON inbox (lease_expires_at, id)
  WHERE status = 'processing';

-- Per-key ordering guard: NOT EXISTS on processing siblings
CREATE INDEX IF NOT EXISTS idx_inbox_partition_processing
  ON inbox (partition_key)
  WHERE status = 'processing';

-- Monitoring: queue depth by status
CREATE INDEX IF NOT EXISTS idx_inbox_status_created
  ON inbox (status, created_at);

-- Producer-side dedupe, without indexing rows that do not use enqueue idempotency
CREATE UNIQUE INDEX IF NOT EXISTS idx_inbox_idempotency_key
  ON inbox (idempotency_key)
  WHERE idempotency_key IS NOT NULL;

-- ── NOTIFY on enqueue (optional worker wake-up) ───────────────────────────────

CREATE OR REPLACE FUNCTION notify_inbox_insert()
RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify('inbox_new', NEW.partition_key);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS inbox_notify ON inbox;
CREATE TRIGGER inbox_notify
  AFTER INSERT ON inbox
  FOR EACH ROW EXECUTE FUNCTION notify_inbox_insert();

COMMIT;

-- =============================================================================
-- OPERATIONS (run from workers / schedulers — not part of DDL transaction)
-- =============================================================================

-- ── Heartbeat (every 10s per worker) ──────────────────────────────────────────
/*
INSERT INTO workers (id, last_seen_at, metadata)
VALUES ('worker-host1-4821', now(), '{"version":"1.2.0"}')
ON CONFLICT (id) DO UPDATE SET
  last_seen_at = EXCLUDED.last_seen_at,
  status       = 'alive',
  metadata     = EXCLUDED.metadata;
*/

-- ── Mark stale workers dead (inside maybeRunHousekeeping + advisory lock) ───
/*
UPDATE workers
SET status = 'dead'
WHERE status IN ('alive', 'draining')
  AND last_seen_at < now() - interval '30 seconds';
*/

-- ── Housekeeping entry point (each worker calls ~every 30s) ───────────────────
/*
BEGIN;
SELECT pg_try_advisory_xact_lock(847291) AS acquired;
-- if acquired: run stale-worker UPDATE + lease cleanup UPDATEs, then COMMIT
COMMIT;
*/

-- ── Graceful drain (on SIGTERM) ───────────────────────────────────────────────
/*
UPDATE workers SET status = 'draining' WHERE id = $1;
*/

-- ── Live workers for ring building ────────────────────────────────────────────
/*
SELECT id FROM workers
WHERE status = 'alive'
  AND last_seen_at > now() - interval '30 seconds'
ORDER BY id;
*/

-- ── Enqueue ───────────────────────────────────────────────────────────────────
/*
INSERT INTO inbox (partition_key, partition_bucket, payload, idempotency_key)
VALUES (
  'user:4821',
  142,  -- application: hash('user:4821') % 1024
  '{"event":"send_welcome_email","userId":4821}',
  'welcome-4821-v1'
)
ON CONFLICT (idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING
RETURNING id;
*/

-- ── Claim batch (filter partition_bucket in app or via ANY()) ─────────────────
/*
WITH candidates AS (
  SELECT i.id
  FROM inbox i
  WHERE i.status = 'pending'
    AND i.available_at <= now()
    AND i.partition_bucket = ANY($1::int[])
    AND NOT EXISTS (
      SELECT 1 FROM inbox busy
      WHERE busy.partition_key = i.partition_key
        AND busy.status = 'processing'
    )
  ORDER BY i.partition_key, i.created_at, i.id
  LIMIT 100
  FOR UPDATE SKIP LOCKED
)
UPDATE inbox i
SET
  status           = 'processing',
  claimed_by       = $2,
  claimed_at       = now(),
  lease_expires_at = now() + interval '90 seconds',
  lease_generation = i.lease_generation + 1,
  attempts         = i.attempts + 1
FROM candidates c
WHERE i.id = c.id
RETURNING i.*;
*/

-- ── Handler idempotency (Layer 2) ─────────────────────────────────────────────
/*
CREATE TABLE IF NOT EXISTS processed_effects (
  idempotency_key TEXT PRIMARY KEY,
  processed_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- In handler, before side-effect:
INSERT INTO processed_effects (idempotency_key)
VALUES ($1)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING idempotency_key;
-- If no row returned → already processed → skip side-effect
*/

-- ── Claim batch (simple — no ordering guard) ──────────────────────────────────
/*
WITH candidates AS (
  SELECT id
  FROM inbox
  WHERE status = 'pending'
    AND available_at <= now()
    AND partition_bucket = ANY($1::int[])  -- owned buckets for this worker
  ORDER BY created_at, id
  LIMIT 100
  FOR UPDATE SKIP LOCKED
)
UPDATE inbox i
SET
  status           = 'processing',
  claimed_by       = $2,
  claimed_at       = now(),
  lease_expires_at = now() + interval '90 seconds',
  lease_generation = i.lease_generation + 1,
  attempts         = i.attempts + 1
FROM candidates c
WHERE i.id = c.id
RETURNING i.*;
*/

-- ── Lease renewal (long-running jobs, every 30s) ──────────────────────────────
/*
UPDATE inbox
SET lease_expires_at = now() + interval '90 seconds'
WHERE id = $1
  AND claimed_by = $2
  AND lease_generation = $3
  AND status = 'processing'
  AND lease_expires_at > now();
*/

-- ── Complete ──────────────────────────────────────────────────────────────────
/*
UPDATE inbox
SET status = 'completed', completed_at = now()
WHERE id = $1
  AND claimed_by = $2
  AND lease_generation = $3
  AND status = 'processing'
  AND lease_expires_at > now();
*/

-- ── Retryable failure ─────────────────────────────────────────────────────────
/*
UPDATE inbox
SET
  status           = 'pending',
  claimed_by       = NULL,
  claimed_at       = NULL,
  lease_expires_at = NULL,
  last_error       = $3,
  available_at     = now() + (LEAST(POWER(2, attempts)::int, 3600) * interval '1 second')
WHERE id = $1
  AND claimed_by = $2
  AND lease_generation = $3
  AND status = 'processing';
*/

-- ── Lease cleanup: release expired leases (inside maybeRunHousekeeping) ──────
/*
BEGIN;
SELECT pg_try_advisory_xact_lock(847291) AS acquired;
-- if acquired:

UPDATE inbox
SET
  status           = 'pending',
  claimed_by       = NULL,
  claimed_at       = NULL,
  lease_expires_at = NULL,
  available_at     = now() + (LEAST(POWER(2, attempts)::int, 3600) * interval '1 second')
WHERE status = 'processing'
  AND lease_expires_at < now()
  AND attempts < max_attempts;

UPDATE inbox
SET status = 'dead_letter', last_error = COALESCE(last_error, 'max attempts via reaper')
WHERE status = 'processing'
  AND lease_expires_at < now()
  AND attempts >= max_attempts;

COMMIT;
*/

-- ── Archive completed rows (keep inbox hot — run daily or in housekeeping) ───
/*
CREATE TABLE IF NOT EXISTS inbox_archive (
  id                UUID NOT NULL,
  partition_key     TEXT NOT NULL,
  partition_bucket  INT NOT NULL,
  payload           JSONB NOT NULL,
  status            inbox_status NOT NULL,
  attempts          INT NOT NULL,
  max_attempts      INT NOT NULL,
  completed_at      TIMESTAMPTZ,
  last_error        TEXT,
  idempotency_key   TEXT,
  created_at        TIMESTAMPTZ NOT NULL,
  archived_at       TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Batch delete to avoid long locks on large tables:
WITH moved AS (
  DELETE FROM inbox
  WHERE id IN (
    SELECT id FROM inbox
    WHERE status = 'completed'
      AND completed_at < now() - interval '7 days'
    LIMIT 10000
  )
  RETURNING *
)
INSERT INTO inbox_archive SELECT * FROM moved;
*/

-- ── Performance tuning quick reference ────────────────────────────────────────
/*
  Heartbeat:     every 10s          (worker dead TTL = 30s = 3× miss)
  Lease:         90s default        (≥ 3× p99 handler; renew every 30s if longer)
  Claim batch:   25 start           (p99 claim < 50ms; cap ~50–100)
  Idle poll:     1–2s + jitter      (LISTEN inbox_new for wake-up)
  Active poll:   50–100ms           (when pending depth > 0)
  Ring cache:    5–10s              (don't rebuild every loop)
  Buckets:       360–1024           (≥ 10× worker count)
  Payload:       < 2 KB JSONB       (store S3 keys / ids, not blobs)
  Handler pool:  4–16 concurrent    (per worker process, I/O-bound)
*/

-- ── Monitoring ────────────────────────────────────────────────────────────────
/*
SELECT status, count(*), min(created_at) AS oldest
FROM inbox GROUP BY status;

SELECT claimed_by, count(*), max(now() - claimed_at) AS max_age
FROM inbox WHERE status = 'processing' GROUP BY claimed_by;

SELECT id, status, now() - last_seen_at AS stale_for
FROM workers ORDER BY last_seen_at DESC;
*/
