"""A tiny job table. SQLite has no SKIP LOCKED; BEGIN IMMEDIATE is the local stand-in.

Production Postgres should claim with:
    SELECT id FROM jobs WHERE status = 'queued' AND run_after <= now()
    ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1
inside a transaction, then mark the row running.
"""

from __future__ import annotations

import sqlite3


def connect(path=":memory:"):
    conn = sqlite3.connect(path, isolation_level=None)
    conn.row_factory = sqlite3.Row
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS jobs (
            id INTEGER PRIMARY KEY,
            idempotency_key TEXT UNIQUE,
            status TEXT NOT NULL,
            payload TEXT NOT NULL,
            attempts INTEGER NOT NULL DEFAULT 0
        )
        """
    )
    return conn


def enqueue(conn, *, idempotency_key: str, payload: str) -> int:
    cur = conn.execute(
        """
        INSERT INTO jobs (idempotency_key, status, payload)
        VALUES (?, 'queued', ?)
        ON CONFLICT(idempotency_key) DO NOTHING
        """,
        (idempotency_key, payload),
    )
    # rowcount is 1 only when this statement inserted. lastrowid survives
    # from the previous insert when ON CONFLICT DO NOTHING skips the row,
    # so a retry can return a different job's id.
    if cur.rowcount == 1:
        return int(cur.lastrowid)
    row = conn.execute(
        "SELECT id FROM jobs WHERE idempotency_key = ?",
        (idempotency_key,),
    ).fetchone()
    return int(row["id"])


def claim(conn):
    conn.execute("BEGIN IMMEDIATE")
    try:
        row = conn.execute(
            "SELECT * FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 1"
        ).fetchone()
        if not row:
            conn.execute("COMMIT")
            return None
        conn.execute(
            "UPDATE jobs SET status = 'running', attempts = attempts + 1 WHERE id = ?",
            (row["id"],),
        )
        # Re-read. The SELECT snapshot still says queued and attempts=0.
        # Returning that snapshot makes a worker log a claim that monitors
        # never count.
        updated = conn.execute("SELECT * FROM jobs WHERE id = ?", (row["id"],)).fetchone()
        conn.execute("COMMIT")
        return dict(updated)
    except Exception:
        conn.execute("ROLLBACK")
        raise


def finish(conn, job_id: int, status: str):
    if status not in {"done", "failed"}:
        raise ValueError("status must be done or failed")
    conn.execute("UPDATE jobs SET status = ? WHERE id = ?", (status, job_id))