bun-sqlite/dist/lib/sqlite/db-queue.js

import dbHandler from "../db-handler";
import init from "../../functions/init";
import DbInsert from "./db-insert";
import { sqlInt, sqlName } from "../../utils/sql-safe";
const DefaultQueueOptions = {
    maxAttempts: 3,
    lockTimeout: 300_000,
    backoff: 1000,
    deleteOnComplete: false,
};
/**
 * Validate the table name and resolve its queue options from the schema
 */
function resolveQueue(table) {
    const { dbSchema } = init();
    const tableSchema = dbSchema.tables.find((t) => t.tableName === table);
    if (!tableSchema?.isQueue) {
        throw new Error(`Table "${table}" is not a queue table (isQueue)`);
    }
    const options = {
        ...DefaultQueueOptions,
        ...(typeof tableSchema.isQueue === "object" ? tableSchema.isQueue : {}),
    };
    return { tableName: sqlName(table, "table name"), options };
}
/**
 * Move jobs that have been `processing` past `lockTimeout` back to
 * `pending`, or to `failed` once they are out of attempts
 */
function releaseStaleSql(tableName) {
    return `UPDATE ${tableName} SET status = CASE WHEN attempts >= ? THEN 'failed' ELSE 'pending' END, last_error = CASE WHEN attempts >= ? THEN 'Lock timed out' ELSE last_error END, locked_at = NULL, locked_by = NULL, updated_at = ? WHERE status = 'processing' AND locked_at < ? RETURNING *`;
}
function rowsResponse(res, sql) {
    if (!res.success) {
        return { ...res, debug: { sql } };
    }
    const rows = res.payload || [];
    return {
        success: true,
        payload: rows,
        singleRes: rows[0] ?? null,
        debug: { sql },
    };
}
/**
 * # Add jobs to a queue table
 * @description `delay` (ms) holds the jobs back before they can be claimed
 */
export async function QueueAdd({ table, data, delay = 0, config, }) {
    try {
        resolveQueue(table);
        const available_at = Date.now() + sqlInt(delay, "delay");
        return await DbInsert({
            table,
            data: data.map((d) => ({
                ...d,
                status: "pending",
                attempts: 0,
                available_at,
            })),
            config,
        });
    }
    catch (error) {
        return { success: false, error: error.message };
    }
}
/**
 * # Claim jobs from a queue table
 * @description Claimed jobs are set to `processing` in one statement, so
 * two workers can never claim the same job. Stale locks are released first.
 */
export async function QueueClaim({ table, worker, limit = 1, config, }) {
    let sql = "";
    try {
        const { tableName, options } = resolveQueue(table);
        const now = Date.now();
        const released = dbHandler({
            query: releaseStaleSql(tableName),
            values: [
                options.maxAttempts,
                options.maxAttempts,
                now,
                now - options.lockTimeout,
            ],
            config,
        });
        if (!released.success) {
            return released;
        }
        sql = `UPDATE ${tableName} SET status = 'processing', attempts = attempts + 1, locked_at = ?, locked_by = ?, updated_at = ? WHERE id IN (SELECT id FROM ${tableName} WHERE status = 'pending' AND available_at <= ? ORDER BY id LIMIT ?) RETURNING *`;
        const res = dbHandler({
            query: sql,
            values: [now, worker ?? null, now, now, sqlInt(limit, "limit")],
            config,
        });
        return rowsResponse(res, sql);
    }
    catch (error) {
        return { success: false, error: error.message, debug: { sql } };
    }
}
/**
 * # Mark a claimed job as done
 * @description Deletes the job instead when `deleteOnComplete` is set
 */
export async function QueueComplete({ table, id, config, }) {
    let sql = "";
    try {
        const { tableName, options } = resolveQueue(table);
        sql = options.deleteOnComplete
            ? `DELETE FROM ${tableName} WHERE id = ? AND status = 'processing' RETURNING *`
            : `UPDATE ${tableName} SET status = 'completed', locked_at = NULL, locked_by = NULL, updated_at = ? WHERE id = ? AND status = 'processing' RETURNING *`;
        const res = dbHandler({
            query: sql,
            values: options.deleteOnComplete ? [id] : [Date.now(), id],
            config,
        });
        const out = rowsResponse(res, sql);
        return out.success && !out.payload?.length
            ? { ...out, success: false, msg: "No processing job with this id" }
            : out;
    }
    catch (error) {
        return { success: false, error: error.message, debug: { sql } };
    }
}
/**
 * # Mark a claimed job as failed
 * @description The job goes back to `pending` with an exponential
 * `backoff` delay, or to `failed` once it has used `maxAttempts`
 */
export async function QueueFail({ table, id, error, config, }) {
    let sql = "";
    try {
        const { tableName, options } = resolveQueue(table);
        const now = Date.now();
        sql = `UPDATE ${tableName} SET status = CASE WHEN attempts >= ? THEN 'failed' ELSE 'pending' END, available_at = CASE WHEN attempts >= ? THEN available_at ELSE ? + ? * (1 << MIN(MAX(attempts - 1, 0), 30)) END, last_error = ?, locked_at = NULL, locked_by = NULL, updated_at = ? WHERE id = ? AND status = 'processing' RETURNING *`;
        const res = dbHandler({
            query: sql,
            values: [
                options.maxAttempts,
                options.maxAttempts,
                now,
                options.backoff,
                error ?? null,
                now,
                id,
            ],
            config,
        });
        const out = rowsResponse(res, sql);
        return out.success && !out.payload?.length
            ? { ...out, success: false, msg: "No processing job with this id" }
            : out;
    }
    catch (error) {
        return { success: false, error: error.message, debug: { sql } };
    }
}
/**
 * # Release stale locks
 * @description For jobs whose worker died. `claim` also does this.
 */
export async function QueueReleaseStale({ table, config }) {
    let sql = "";
    try {
        const { tableName, options } = resolveQueue(table);
        const now = Date.now();
        sql = releaseStaleSql(tableName);
        const res = dbHandler({
            query: sql,
            values: [
                options.maxAttempts,
                options.maxAttempts,
                now,
                now - options.lockTimeout,
            ],
            config,
        });
        return rowsResponse(res, sql);
    }
    catch (error) {
        return { success: false, error: error.message, debug: { sql } };
    }
}