-- pgqueue — Postgres-native queue primitive for Scobleclaw.
--
-- One engine, three jobs. This is the queue primitive. Lives in `pgqueue`
-- schema so it can be installed in any database (pool_manager, homeschool, etc.)
-- without polluting the public schema.
--
-- Design (2026-06-01):
--   - SKIP LOCKED for atomic claim (Postgres does the work)
--   - Heartbeat-based lease: `lease_until` column, janitor resets stale claims
--   - LISTEN/NOTIFY trigger for sub-second wakeup
--   - Idempotency key column: dedupes retried enqueues
--   - Per-queue isolation: claim from one queue doesn't block another
--
-- Why not pgmq: it would mean adding a new extension. The visibility-timeout
-- primitive pgmq buys you is a 30-line janitor + a `lease_until` column. Same
-- semantics, no extension.
--
-- Reference: skills/data-stack-conventions.md (Postgres as the data stack).

BEGIN;

CREATE SCHEMA IF NOT EXISTS pgqueue;

-- ── jobs table ──────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS pgqueue.jobs (
    id              bigserial PRIMARY KEY,
    queue           text        NOT NULL,
    payload         jsonb       NOT NULL,
    idempotency_key text,
    state           text        NOT NULL DEFAULT 'ready'
                      CHECK (state IN ('ready', 'claimed', 'done', 'dead')),
    priority        int         NOT NULL DEFAULT 0,  -- higher = sooner
    run_after       timestamptz NOT NULL DEFAULT now(),
    attempts        int         NOT NULL DEFAULT 0,
    max_attempts    int         NOT NULL DEFAULT 5,
    last_error      text,
    claimed_by      text,
    claimed_at      timestamptz,
    lease_until     timestamptz,  -- heartbeat; janitor resets if expired
    created_at      timestamptz NOT NULL DEFAULT now(),
    updated_at      timestamptz NOT NULL DEFAULT now(),
    finished_at     timestamptz
);

-- Index for claim query: order by (priority desc, created_at) on ready jobs
CREATE INDEX IF NOT EXISTS jobs_claim_idx
    ON pgqueue.jobs (queue, priority DESC, created_at)
    WHERE state = 'ready';

-- Idempotency: same (queue, idempotency_key) returns the same job id
CREATE UNIQUE INDEX IF NOT EXISTS jobs_idem_idx
    ON pgqueue.jobs (queue, idempotency_key)
    WHERE idempotency_key IS NOT NULL;

-- Index for janitor: scan claimed jobs whose lease has expired
CREATE INDEX IF NOT EXISTS jobs_lease_idx
    ON pgqueue.jobs (lease_until)
    WHERE state = 'claimed';

-- Index for dashboard
CREATE INDEX IF NOT EXISTS jobs_state_idx
    ON pgqueue.jobs (queue, state, created_at DESC);

-- ── enqueue ─────────────────────────────────────────────────────────
CREATE OR REPLACE FUNCTION pgqueue.enqueue(
    p_queue           text,
    p_payload         jsonb,
    p_idempotency_key text DEFAULT NULL,
    p_priority        int  DEFAULT 0,
    p_run_after       timestamptz DEFAULT now()
)
RETURNS bigint
LANGUAGE plpgsql
AS $$
DECLARE
    v_id bigint;
BEGIN
    -- If idempotency_key provided and the row already exists, return its id.
    -- This dedupes retries from upstream (e.g., a cron enqueuing twice).
    IF p_idempotency_key IS NOT NULL THEN
        SELECT id INTO v_id
        FROM pgqueue.jobs
        WHERE queue = p_queue AND idempotency_key = p_idempotency_key;
        IF v_id IS NOT NULL THEN
            RETURN v_id;
        END IF;
    END IF;

    INSERT INTO pgqueue.jobs (queue, payload, idempotency_key, priority, run_after)
    VALUES (p_queue, p_payload, p_idempotency_key, p_priority, p_run_after)
    RETURNING id INTO v_id;

    -- Wake up any worker LISTENing on this queue
    PERFORM pg_notify('pgqueue:' || p_queue, v_id::text);

    RETURN v_id;
END;
$$;

-- ── claim next job (the heart of SKIP LOCKED) ──────────────────────
-- Atomically claim one ready job. Returns NULL if no job available.
-- p_worker_id: identifies the worker (hostname, pod name, etc.)
-- p_lease_for: how long the lease is valid (e.g., '5 minutes'::interval)
--              — worker must heartbeat before this expires or the janitor
--              will reset the claim.
CREATE OR REPLACE FUNCTION pgqueue.claim_next(
    p_queue     text,
    p_worker_id text,
    p_lease_for interval DEFAULT '5 minutes'::interval
)
RETURNS TABLE (
    id              bigint,
    queue           text,
    payload         jsonb,
    priority        int,
    attempts        int,
    max_attempts    int,
    claimed_at      timestamptz,
    lease_until     timestamptz
)
LANGUAGE plpgsql
AS $$
BEGIN
    RETURN QUERY
    WITH next_job AS (
        SELECT j.id
        FROM pgqueue.jobs j
        WHERE j.queue = p_queue
          AND j.state = 'ready'
          AND j.run_after <= now()
        ORDER BY j.priority DESC, j.created_at
        FOR UPDATE SKIP LOCKED
        LIMIT 1
    )
    UPDATE pgqueue.jobs j
    SET state       = 'claimed',
        claimed_by  = p_worker_id,
        claimed_at  = now(),
        lease_until = now() + p_lease_for,
        attempts    = j.attempts + 1,
        updated_at  = now()
    FROM next_job
    WHERE j.id = next_job.id
    RETURNING j.id, j.queue, j.payload, j.priority, j.attempts, j.max_attempts,
              j.claimed_at, j.lease_until;
END;
$$;

-- ── heartbeat: extend the lease on a job you still hold ───────────
CREATE OR REPLACE FUNCTION pgqueue.heartbeat(
    p_job_id    bigint,
    p_worker_id text,
    p_extend_by interval DEFAULT '5 minutes'::interval
)
RETURNS boolean
LANGUAGE plpgsql
AS $$
DECLARE
    v_updated int;
BEGIN
    UPDATE pgqueue.jobs
    SET lease_until = now() + p_extend_by,
        updated_at  = now()
    WHERE id = p_job_id
      AND state = 'claimed'
      AND claimed_by = p_worker_id;
    GET DIAGNOSTICS v_updated = ROW_COUNT;
    RETURN v_updated = 1;
END;
$$;

-- ── ack: mark job done ─────────────────────────────────────────────
CREATE OR REPLACE FUNCTION pgqueue.ack(
    p_job_id    bigint,
    p_worker_id text
)
RETURNS boolean
LANGUAGE plpgsql
AS $$
DECLARE
    v_updated int;
BEGIN
    UPDATE pgqueue.jobs
    SET state       = 'done',
        finished_at = now(),
        updated_at  = now()
    WHERE id = p_job_id
      AND state = 'claimed'
      AND claimed_by = p_worker_id;
    GET DIAGNOSTICS v_updated = ROW_COUNT;
    RETURN v_updated = 1;
END;
$$;

-- ── nack: requeue or dead-letter ──────────────────────────────────
-- If attempts < max_attempts: requeue with backoff
-- Else: mark dead, requires manual intervention (or admin purges)
CREATE OR REPLACE FUNCTION pgqueue.nack(
    p_job_id    bigint,
    p_worker_id text,
    p_error     text DEFAULT NULL,
    p_requeue_after interval DEFAULT '30 seconds'::interval
)
RETURNS text  -- 'requeued' | 'dead' | 'lost'
LANGUAGE plpgsql
AS $$
DECLARE
    v_state    text;
    v_attempts int;
    v_max      int;
BEGIN
    SELECT state, attempts, max_attempts
      INTO v_state, v_attempts, v_max
    FROM pgqueue.jobs
    WHERE id = p_job_id
    FOR UPDATE;

    IF v_state IS NULL THEN
        RETURN 'lost';  -- job doesn't exist
    END IF;
    IF v_state != 'claimed' THEN
        RETURN 'lost';  -- not in a state we can nack from
    END IF;
    -- Worker id check is enforced by application; we trust here.

    IF v_attempts >= v_max THEN
        UPDATE pgqueue.jobs
        SET state       = 'dead',
            last_error  = p_error,
            finished_at = now(),
            updated_at  = now(),
            claimed_by  = NULL,
            claimed_at  = NULL,
            lease_until = NULL
        WHERE id = p_job_id;
        RETURN 'dead';
    ELSE
        UPDATE pgqueue.jobs
        SET state       = 'ready',
            run_after   = now() + p_requeue_after,
            last_error  = p_error,
            claimed_by  = NULL,
            claimed_at  = NULL,
            lease_until = NULL,
            updated_at  = now()
        WHERE id = p_job_id;

        -- Wake up workers — the job is available again
        PERFORM pg_notify('pgqueue:' || (
            SELECT queue FROM pgqueue.jobs WHERE id = p_job_id
        ), p_job_id::text);
        RETURN 'requeued';
    END IF;
END;
$$;

-- ── janitor: reset stale claims whose lease expired ────────────────
-- Run this every 1-5 minutes via cron. Idempotent and safe to run often.
CREATE OR REPLACE FUNCTION pgqueue.janitor_reset_stale_claims(
    p_grace interval DEFAULT '0 seconds'::interval
)
RETURNS int  -- number of jobs reset
LANGUAGE plpgsql
AS $$
DECLARE
    v_count int;
BEGIN
    WITH stale AS (
        SELECT id
        FROM pgqueue.jobs
        WHERE state = 'claimed'
          AND lease_until IS NOT NULL
          AND lease_until + p_grace < now()
        FOR UPDATE SKIP LOCKED
    )
    UPDATE pgqueue.jobs j
    SET state       = 'ready',
        claimed_by  = NULL,
        claimed_at  = NULL,
        lease_until = NULL,
        last_error  = COALESCE(j.last_error, '') ||
                      CASE WHEN j.last_error IS NOT NULL THEN E'\n' ELSE '' END ||
                      'lease expired at ' || now()::text,
        updated_at  = now()
    FROM stale
    WHERE j.id = stale.id;
    GET DIAGNOSTICS v_count = ROW_COUNT;
    RETURN v_count;
END;
$$;

-- ── view: queue depth per state ───────────────────────────────────
CREATE OR REPLACE VIEW pgqueue.depth AS
SELECT
    queue,
    state,
    count(*)                                        AS count,
    min(created_at)                                 AS oldest,
    max(updated_at)                                 AS newest_update
FROM pgqueue.jobs
GROUP BY queue, state
ORDER BY queue, state;

-- ── helper: notify on insert (in case enqueue() was bypassed) ─────
-- This is a safety net. enqueue() does pg_notify directly.
-- If someone INSERTs into pgqueue.jobs directly, the trigger still wakes
-- listeners. Belt and suspenders.
CREATE OR REPLACE FUNCTION pgqueue.notify_new_job() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('pgqueue:' || NEW.queue, NEW.id::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS jobs_notify_insert ON pgqueue.jobs;
CREATE TRIGGER jobs_notify_insert
    AFTER INSERT ON pgqueue.jobs
    FOR EACH ROW EXECUTE FUNCTION pgqueue.notify_new_job();

-- ── cleanup: drop dead/done jobs older than retention ─────────────
-- For testing; production should run this less often with a longer window.
CREATE OR REPLACE FUNCTION pgqueue.gc(
    p_retain_done interval DEFAULT '7 days'::interval,
    p_retain_dead interval DEFAULT '30 days'::interval
)
RETURNS TABLE (done_deleted int, dead_deleted int)
LANGUAGE plpgsql
AS $$
BEGIN
    DELETE FROM pgqueue.jobs WHERE state = 'done' AND finished_at < now() - p_retain_done;
    GET DIAGNOSTICS done_deleted = ROW_COUNT;

    DELETE FROM pgqueue.jobs WHERE state = 'dead' AND finished_at < now() - p_retain_dead;
    GET DIAGNOSTICS dead_deleted = ROW_COUNT;

    RETURN NEXT;
END;
$$;

COMMIT;