PostgreSQL persistence workflow practice: build a task orchestration system in 30 minutes

--
CREATE TABLE workflows (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workflow_type VARCHAR(100) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    input_data JSONB,
    current_step INTEGER DEFAULT 0,
    max_retries INTEGER DEFAULT 3,
    error_message TEXT,
    created_at TIMESTAMPTZ DEFAULT now(),
    updated_at TIMESTAMPTZ DEFAULT now()
);

-- （）
CREATE TABLE workflow_steps (
    id SERIAL PRIMARY KEY,
    workflow_id UUID REFERENCES workflows(id),
    step_number INTEGER NOT NULL,
    step_name VARCHAR(200),
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    output_data JSONB,
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    error_message TEXT,
    UNIQUE(workflow_id, step_number)
);

-- Worker
CREATE INDEX idx_workflows_status ON workflows(status, created_at);
CREATE INDEX idx_workflow_steps_lookup ON workflow_steps(workflow_id, step_number);

import psycopg2
import json

def dequeue_workflow(conn):
 """ Postgres """
    with conn.cursor() as cur:
        cur.execute("""
            SELECT id, workflow_type, input_data, current_step
            FROM workflows
            WHERE status = 'pending'
            ORDER BY created_at
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        """)
        row = cur.fetchone()
        if row:
            cur.execute(
                "UPDATE workflows SET status = 'running', updated_at = now() WHERE id = %s",
                (row[0],)
            )
            conn.commit()
            return {
                "id": row[0],
                "type": row[1],
                "input": row[2],
                "step": row[3]
            }
    conn.rollback()
    return None

def execute_step(conn, workflow_id, step_number, step_fn, step_name=""):
 """，"""

# （）
    with conn.cursor() as cur:
        cur.execute(
            """SELECT status, output_data FROM workflow_steps
               WHERE workflow_id = %s AND step_number = %s""",
            (workflow_id, step_number)
        )
        existing = cur.fetchone()
        if existing and existing[0] == 'completed':
 return existing[1] # ，

# （ON CONFLICT DO NOTHING ）
    with conn.cursor() as cur:
        cur.execute(
            """INSERT INTO workflow_steps (workflow_id, step_number, step_name, status, started_at)
               VALUES (%s, %s, %s, 'running', now())
               ON CONFLICT (workflow_id, step_number) DO NOTHING""",
            (workflow_id, step_number, step_name)
        )
        conn.commit()

    try:
        result = step_fn()
        with conn.cursor() as cur:
            cur.execute(
                """UPDATE workflow_steps
                   SET status = 'completed', output_data = %s, completed_at = now()
                   WHERE workflow_id = %s AND step_number = %s""",
                (json.dumps(result), workflow_id, step_number)
            )
            cur.execute(
                "UPDATE workflows SET current_step = %s, updated_at = now() WHERE id = %s",
                (step_number, workflow_id)
            )
        conn.commit()
        return result
    except Exception as e:
        with conn.cursor() as cur:
            cur.execute(
                """UPDATE workflow_steps
                   SET status = 'failed', error_message = %s, completed_at = now()
                   WHERE workflow_id = %s AND step_number = %s""",
                (str(e), workflow_id, step_number)
            )
        conn.commit()
        raise

-- 24
SELECT
    workflow_type,
    COUNT(*) AS total,
    COUNT(*) FILTER (WHERE status = 'completed') AS completed,
    ROUND(100.0 * COUNT(*) FILTER (WHERE status = 'completed') / COUNT(*), 1) AS success_pct
FROM workflows
WHERE created_at > now() - INTERVAL '24 hours'
GROUP BY workflow_type;

-- 5 （ Worker ）
SELECT id, workflow_type, current_step, updated_at
FROM workflows
WHERE status = 'running'
  AND updated_at < now() - INTERVAL '5 minutes';

-- ，
SELECT
    w.workflow_type,
    ws.step_name,
    ROUND(AVG(EXTRACT(EPOCH FROM (ws.completed_at - ws.started_at)))::numeric, 2) AS avg_seconds
FROM workflow_steps ws
JOIN workflows w ON w.id = ws.workflow_id
WHERE ws.status = 'completed'
GROUP BY w.workflow_type, ws.step_name
ORDER BY avg_seconds DESC;