Postgres job queue

schema.sql

CREATE TYPE job_status AS ENUM ('pending', 'in_progress', 'done', 'failed');

CREATE TABLE jobs (
    id SERIAL PRIMARY KEY,
    status job_status NOT NULL DEFAULT 'pending',
    payload JSONB,
    created_at TIMESTAMP DEFAULT now(),
    updated_at TIMESTAMP DEFAULT now(),
    visible_at TIMESTAMP DEFAULT now(), -- SQS visibility timeout
    retry_count INT DEFAULT 0
);

-- ENQUEUE
INSERT INTO jobs (status, payload)
VALUES ('pending', '{"task": "sync_data"}'::jsonb);

-- DEQUEUE 
BEGIN;

WITH next_job AS (
    SELECT id
    FROM jobs
    WHERE
        (
            status = 'pending'
            OR (status = 'in_progress' AND visible_at <= now())
        )
    ORDER BY created_at
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE jobs
SET status = 'in_progress',
    updated_at = now(),
    visible_at = now() + interval '30 seconds', -- Timeout maybe configurable.
    retry_count = retry_count + 1
FROM next_job
WHERE jobs.id = next_job.id
RETURNING jobs.*;


--- ACKING
UPDATE jobs
SET status = 'done',
    updated_at = now()
WHERE id = :job_id;