diff --git a/README.md b/README.md index 9fed5b7..b285d20 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,7 @@ A schema-driven SQLite manager for [Bun](https://bun.sh), featuring automatic sc - [Raw SQL](#raw-sql) - [Query API Reference](#query-api-reference) - [Vector Table Support](#vector-table-support) +- [Queue Tables](#queue-tables) - [TypeScript Type Generation](#typescript-type-generation) - [Default Fields](#default-fields) - [Project Structure](#project-structure) @@ -204,6 +205,7 @@ interface BUN_SQLITE_TableSchemaType { tableNameOld?: string; // rename: set this to the old name to trigger ALTER TABLE RENAME isVector?: boolean; // mark this as a sqlite-vec virtual table vectorType?: string; // virtual table type, defaults to "vec0" + isQueue?: boolean | BUN_SQLITE_QueueOptions; // make this a job queue (see Queue Tables) } ``` @@ -759,6 +761,75 @@ const res = await BunSQLite.select({ --- +## Queue Tables + +Set `isQueue` on a table to use it as a job queue. Your fields are the job payload; the schema sync adds the queue columns and an index on `(status, available_at)`. + +```ts +{ + tableName: "email_jobs", + isQueue: { + maxAttempts: 3, // claims before a failing job is marked "failed" (default 3) + lockTimeout: 300_000, // ms a job can stay "processing" before it is released (default 5 min) + backoff: 1000, // base retry delay in ms, doubled each attempt (default 1000) + deleteOnComplete: false, // delete completed jobs instead of keeping them (default false) + }, + fields: [ + { fieldName: "recipient", dataType: "TEXT" }, + { fieldName: "subject", dataType: "TEXT" }, + ], +} +``` + +`isQueue: true` uses the defaults. A table can't be both a queue and a vector table. + +| Field | Type | Description | +| -------------- | ---------------------------- | ---------------------------------------------------- | +| `status` | `TEXT NOT NULL` | `pending`, `processing`, `completed` or `failed` | +| `attempts` | `INTEGER NOT NULL DEFAULT 0` | Times the job has been claimed | +| `available_at` | `INTEGER NOT NULL DEFAULT 0` | Unix timestamp (ms) from which the job can be claimed | +| `locked_at` | `INTEGER` | Unix timestamp (ms) when the job was claimed | +| `locked_by` | `TEXT` | Worker that claimed the job | +| `last_error` | `TEXT` | Error from the last failed attempt | + +### Working the queue + +```ts +// Enqueue (optionally delayed, in ms) +await BunSQLite.queue.add({ + table: "email_jobs", + data: [{ recipient: "a@example.com", subject: "Hi" }], + delay: 0, +}); + +// Worker loop +const { payload: jobs } = await BunSQLite.queue.claim({ + table: "email_jobs", + worker: "worker-1", + limit: 10, +}); + +for (const job of jobs || []) { + try { + await sendEmail(job); + await BunSQLite.queue.complete({ table: "email_jobs", id: job.id }); + } catch (error: any) { + await BunSQLite.queue.fail({ + table: "email_jobs", + id: job.id, + error: error.message, + }); + } +} +``` + +- `claim` sets jobs to `processing` in a single `UPDATE ... RETURNING` statement. SQLite only allows one writer at a time, so two workers never get the same job, even across processes. +- `fail` puts the job back to `pending` with an exponential backoff, or sets it to `failed` once it has used `maxAttempts`. +- Jobs stuck in `processing` longer than `lockTimeout` (for example, the worker crashed) are released by `claim` before it picks jobs. You can also call `BunSQLite.queue.release_stale({ table })` directly. +- `complete` and `fail` only touch jobs that are `processing`. + +--- + ## TypeScript Type Generation Run the `typedef` command (or pass `--typedef` to `schema`) to generate a `.ts` file containing: diff --git a/dist/functions/grab-sqlite-client.js b/dist/functions/grab-sqlite-client.js index c9a00ef..dd9397c 100644 --- a/dist/functions/grab-sqlite-client.js +++ b/dist/functions/grab-sqlite-client.js @@ -13,6 +13,9 @@ export default function grabSQLiteClient({ config }) { create: true, }); sqliteVec.load(SQLiteClient); + // Wait for other connections' write locks instead of failing with + // SQLITE_BUSY (queue workers in separate processes) + SQLiteClient.run("PRAGMA busy_timeout = 5000;"); if (config.wal_mode) { SQLiteClient.run("PRAGMA journal_mode = WAL;"); } diff --git a/dist/index.d.ts b/dist/index.d.ts index 6228ef6..7dced81 100644 --- a/dist/index.d.ts +++ b/dist/index.d.ts @@ -3,6 +3,7 @@ import DbInsert from "./lib/sqlite/db-insert"; import DbSelect from "./lib/sqlite/db-select"; import DbSQL from "./lib/sqlite/db-sql"; import DbUpdate from "./lib/sqlite/db-update"; +import { QueueAdd, QueueClaim, QueueComplete, QueueFail, QueueReleaseStale } from "./lib/sqlite/db-queue"; import grabDbSchema from "./utils/grab-db-schema"; import grabJoinFieldsFromQueryObject from "./utils/grab-join-fields-from-query-object"; declare const BunSQLite: { @@ -11,6 +12,13 @@ declare const BunSQLite: { readonly update: typeof DbUpdate; readonly delete: typeof DbDelete; readonly sql: typeof DbSQL; + readonly queue: { + readonly add: typeof QueueAdd; + readonly claim: typeof QueueClaim; + readonly complete: typeof QueueComplete; + readonly fail: typeof QueueFail; + readonly release_stale: typeof QueueReleaseStale; + }; readonly utils: { readonly grab_db_schema: typeof grabDbSchema; readonly grab_join_fields_from_query_object: typeof grabJoinFieldsFromQueryObject; @@ -18,4 +26,4 @@ declare const BunSQLite: { }; export default BunSQLite; export type * from "./types"; -export { UsersOmitedFields, MariaDBCollations, TextFieldTypesArray, BUN_SQLITE_DATATYPES, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, } from "./types"; +export { UsersOmitedFields, MariaDBCollations, TextFieldTypesArray, BUN_SQLITE_DATATYPES, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, QueueFields, QueueStatuses, } from "./types"; diff --git a/dist/index.js b/dist/index.js index 65bc39d..18c6964 100644 --- a/dist/index.js +++ b/dist/index.js @@ -4,6 +4,7 @@ import DbInsert from "./lib/sqlite/db-insert"; import DbSelect from "./lib/sqlite/db-select"; import DbSQL from "./lib/sqlite/db-sql"; import DbUpdate from "./lib/sqlite/db-update"; +import { QueueAdd, QueueClaim, QueueComplete, QueueFail, QueueReleaseStale, } from "./lib/sqlite/db-queue"; import grabDbSchema from "./utils/grab-db-schema"; import grabJoinFieldsFromQueryObject from "./utils/grab-join-fields-from-query-object"; /** @@ -22,10 +23,17 @@ const BunSQLite = { update: withInit(DbUpdate), delete: withInit(DbDelete), sql: withInit(DbSQL), + queue: { + add: withInit(QueueAdd), + claim: withInit(QueueClaim), + complete: withInit(QueueComplete), + fail: withInit(QueueFail), + release_stale: withInit(QueueReleaseStale), + }, utils: { grab_db_schema: withInit(grabDbSchema), grab_join_fields_from_query_object: grabJoinFieldsFromQueryObject, }, }; export default BunSQLite; -export { UsersOmitedFields, MariaDBCollations, TextFieldTypesArray, BUN_SQLITE_DATATYPES, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, } from "./types"; +export { UsersOmitedFields, MariaDBCollations, TextFieldTypesArray, BUN_SQLITE_DATATYPES, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, QueueFields, QueueStatuses, } from "./types"; diff --git a/dist/lib/grab-duplicate-safe-insert-sql.js b/dist/lib/grab-duplicate-safe-insert-sql.js index b4c5bc9..7e51ead 100644 --- a/dist/lib/grab-duplicate-safe-insert-sql.js +++ b/dist/lib/grab-duplicate-safe-insert-sql.js @@ -1,5 +1,5 @@ import init from "../functions/init"; -import { sqlIdent, sqlQuoteIdent } from "../utils/sql-safe"; +import { sqlName } from "../utils/sql-safe"; export default async function ({ sql: passed_sql, table, data }) { let sql = passed_sql; const { dbSchema } = init(); @@ -9,7 +9,7 @@ export default async function ({ sql: passed_sql, table, data }) { const set_sql_arr = Object.keys(Array.isArray(data) ? data[0] : data) .filter((field) => field !== "updated_at") .map((field) => { - const column = sqlIdent(field, "column name"); + const column = sqlName(field, "column name"); return `${column} = excluded.${column}`; }); set_sql_arr.push(`updated_at = ${now}`); @@ -19,13 +19,13 @@ export default async function ({ sql: passed_sql, table, data }) { const field = unique_fields[i]; if (!field?.fieldName) continue; - sql += ` ON CONFLICT(${sqlQuoteIdent(field.fieldName)}) DO UPDATE SET ${set_sql}`; + sql += ` ON CONFLICT(${sqlName(field.fieldName, "column name")}) DO UPDATE SET ${set_sql}`; } if (table_schema.uniqueConstraints?.[0]) { for (let i = 0; i < table_schema.uniqueConstraints.length; i++) { const constraint = table_schema.uniqueConstraints[i]; const columns = constraint?.constraintTableFields - ?.map((c) => sqlQuoteIdent(String(c.value))) + ?.map((c) => sqlName(c.value, "column name")) ?.join(", "); if (!columns) continue; diff --git a/dist/lib/schema/create-db-schema.js b/dist/lib/schema/create-db-schema.js index 7194bf6..fd11a20 100644 --- a/dist/lib/schema/create-db-schema.js +++ b/dist/lib/schema/create-db-schema.js @@ -19,6 +19,11 @@ export default function createDBSchema(params) { * Reorder Tables (parents before children with FKs) */ const db_schema = orderDBSchema(params); + for (const table of db_schema.tables) { + if (table.isQueue && table.isVector) { + throw new Error(`Table "${table.tableName}" can't be both a queue and a vector table`); + } + } db.run("PRAGMA foreign_keys = OFF;"); db.run("BEGIN"); try { diff --git a/dist/lib/sqlite/db-delete.js b/dist/lib/sqlite/db-delete.js index 4695486..95a8111 100644 --- a/dist/lib/sqlite/db-delete.js +++ b/dist/lib/sqlite/db-delete.js @@ -1,7 +1,7 @@ import dbHandler from "../db-handler"; import _ from "lodash"; import sqlGenerator from "../../utils/sql-generator"; -import { sqlIdent } from "../../utils/sql-safe"; +import { sqlName } from "../../utils/sql-safe"; export default async function DbDelete({ table, query, targetId, config, }) { let sqlObj = null; try { @@ -13,7 +13,7 @@ export default async function DbDelete({ table, query, targetId, config, }) { }); const whereClause = sqlObj.where.string; if (whereClause) { - let sql = `DELETE FROM ${sqlIdent(table, "table name")} ${whereClause}`; + let sql = `DELETE FROM ${sqlName(table, "table name")} ${whereClause}`; sqlObj.string = sql; sqlObj.values = sqlObj.where.values; const res = dbHandler({ diff --git a/dist/lib/sqlite/db-queue.d.ts b/dist/lib/sqlite/db-queue.d.ts new file mode 100644 index 0000000..efa04c2 --- /dev/null +++ b/dist/lib/sqlite/db-queue.d.ts @@ -0,0 +1,64 @@ +import type { APIResponseObject, BunSQLiteConfig } from "../../types"; +type BaseParams = { + table: Table; + config?: BunSQLiteConfig; +}; +/** + * # Add jobs to a queue table + * @description `delay` (ms) holds the jobs back before they can be claimed + */ +export declare function QueueAdd({ table, data, delay, config, }: BaseParams
& { + data: Schema[]; + delay?: number; +}): Promise>; +/** + * # 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 declare function QueueClaim({ table, worker, limit, config, }: BaseParams
& { + worker?: string; + limit?: number; +}): Promise>; +/** + * # Mark a claimed job as done + * @description Deletes the job instead when `deleteOnComplete` is set + */ +export declare function QueueComplete({ table, id, config, }: BaseParams
& { + id: number | string; +}): Promise>; +/** + * # 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 declare function QueueFail({ table, id, error, config, }: BaseParams
& { + id: number | string; + error?: string; +}): Promise>; +/** + * # Release stale locks + * @description For jobs whose worker died. `claim` also does this. + */ +export declare function QueueReleaseStale({ table, config }: BaseParams
): Promise>; +export {}; diff --git a/dist/lib/sqlite/db-queue.js b/dist/lib/sqlite/db-queue.js new file mode 100644 index 0000000..f39289a --- /dev/null +++ b/dist/lib/sqlite/db-queue.js @@ -0,0 +1,186 @@ +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 } }; + } +} diff --git a/dist/lib/sqlite/db-update.js b/dist/lib/sqlite/db-update.js index d1ad8c6..97fd2f5 100644 --- a/dist/lib/sqlite/db-update.js +++ b/dist/lib/sqlite/db-update.js @@ -3,7 +3,7 @@ import _ from "lodash"; import init from "../../functions/init"; import sqlGenerator from "../../utils/sql-generator"; import { sqlValue } from "../../utils/sql-insert-generator"; -import { sqlIdent } from "../../utils/sql-safe"; +import { sqlName } from "../../utils/sql-safe"; export default async function DbUpdate({ table, data, query, targetId, returning = true, config, }) { let sqlObj = { string: "", @@ -23,7 +23,7 @@ export default async function DbUpdate({ table, data, query, targetId, returning let values = []; const whereClause = sqlQueryObj.where.string; if (whereClause) { - let sql = `UPDATE ${sqlIdent(table, "table name")} SET`; + let sql = `UPDATE ${sqlName(table, "table name")} SET`; const finalData = { updated_at: Date.now(), ...data, @@ -34,7 +34,7 @@ export default async function DbUpdate({ table, data, query, targetId, returning if (!key) continue; const isLast = i == keys.length - 1; - sql += ` ${sqlIdent(key, "column name")}=?`; + sql += ` ${sqlName(key, "column name")}=?`; values.push(sqlValue(finalData[key])); if (!isLast) { sql += `,`; diff --git a/dist/types/index.d.ts b/dist/types/index.d.ts index 12cc939..bc155bf 100644 --- a/dist/types/index.d.ts +++ b/dist/types/index.d.ts @@ -89,6 +89,35 @@ export interface BUN_SQLITE_TableSchemaType { * Type of vector. Defaults to `vec0` */ vectorType?: string; + /** + * Make this a queue table. Adds the `QueueFields` columns and an index + * on `(status, available_at)`. Use `BunSQLite.queue` to work the queue. + */ + isQueue?: boolean | BUN_SQLITE_QueueOptions; +} +/** + * Options for a queue table (`isQueue`). + */ +export interface BUN_SQLITE_QueueOptions { + /** + * Claims allowed before a failing job is marked `failed`. Defaults to `3` + */ + maxAttempts?: number; + /** + * Milliseconds a claimed job can stay `processing` before it is + * released back to the queue. Defaults to `300000` (5 minutes) + */ + lockTimeout?: number; + /** + * Base retry delay in milliseconds, doubled on every attempt. Defaults + * to `1000` + */ + backoff?: number; + /** + * Delete jobs on completion instead of marking them `completed`. + * Defaults to `false` + */ + deleteOnComplete?: boolean; } /** * Reference object used to link a table to one of its child tables. @@ -1420,6 +1449,11 @@ export type SchemaChange = { * Default fields automatically suggested for new tables. */ export declare const DefaultFields: BUN_SQLITE_FieldSchemaType[]; +export declare const QueueStatuses: readonly ["pending", "processing", "completed", "failed"]; +/** + * Fields added to queue tables (`isQueue`) + */ +export declare const QueueFields: BUN_SQLITE_FieldSchemaType[]; export type BunSQLiteQueryFieldValues = { field: F; table?: T; diff --git a/dist/types/index.js b/dist/types/index.js index 2c05bbd..f40ef3f 100644 --- a/dist/types/index.js +++ b/dist/types/index.js @@ -185,3 +185,50 @@ export const DefaultFields = [ fieldDescription: "The time when the record was updated. (Unix Timestamp)", }, ]; +export const QueueStatuses = [ + "pending", + "processing", + "completed", + "failed", +]; +/** + * Fields added to queue tables (`isQueue`) + */ +export const QueueFields = [ + { + fieldName: "status", + dataType: "TEXT", + notNullValue: true, + defaultValue: "pending", + fieldDescription: `The state of the job. (${QueueStatuses.join(" | ")})`, + }, + { + fieldName: "attempts", + dataType: "INTEGER", + notNullValue: true, + defaultValue: 0, + fieldDescription: "The number of times the job has been claimed.", + }, + { + fieldName: "available_at", + dataType: "INTEGER", + notNullValue: true, + defaultValue: 0, + fieldDescription: "The time from which the job can be claimed. (Unix Timestamp)", + }, + { + fieldName: "locked_at", + dataType: "INTEGER", + fieldDescription: "The time when the job was claimed. (Unix Timestamp)", + }, + { + fieldName: "locked_by", + dataType: "TEXT", + fieldDescription: "The worker that claimed the job.", + }, + { + fieldName: "last_error", + dataType: "TEXT", + fieldDescription: "The error from the last failed attempt.", + }, +]; diff --git a/dist/utils/append-default-fields-to-db-schema.js b/dist/utils/append-default-fields-to-db-schema.js index d1b758a..1c8842e 100644 --- a/dist/utils/append-default-fields-to-db-schema.js +++ b/dist/utils/append-default-fields-to-db-schema.js @@ -1,11 +1,22 @@ import _ from "lodash"; -import { DefaultFields } from "../types"; +import { DefaultFields, QueueFields, } from "../types"; +const QueueFieldNames = QueueFields.map((f) => f.fieldName); export default function ({ dbSchema }) { const finaldbSchema = _.cloneDeep(dbSchema); finaldbSchema.tables = finaldbSchema.tables.map((t) => { const newTable = _.cloneDeep(t); newTable.fields = newTable.fields.filter((f) => !f.fieldName?.match(/^(id|created_at|updated_at)$/)); newTable.fields.unshift(...DefaultFields); + if (newTable.isQueue) { + newTable.fields = newTable.fields.filter((f) => !QueueFieldNames.includes(f.fieldName)); + newTable.fields.push(...QueueFields); + const indexName = `idx_${newTable.tableName}_queue`; + newTable.indexes = (newTable.indexes || []).filter((idx) => idx.indexName !== indexName); + newTable.indexes.push({ + indexName, + indexTableFields: ["status", "available_at"], + }); + } return newTable; }); return finaldbSchema; diff --git a/dist/utils/sql-generator-gen-join-str.js b/dist/utils/sql-generator-gen-join-str.js index bfae69e..2185d21 100644 --- a/dist/utils/sql-generator-gen-join-str.js +++ b/dist/utils/sql-generator-gen-join-str.js @@ -1,4 +1,4 @@ -import { sqlIdent, sqlOperator, sqlRawEquality } from "./sql-safe"; +import { sqlName, sqlOperator, sqlRawEquality } from "./sql-safe"; export default function sqlGenGenJoinStr({ join, mtch, table_name }) { let values = []; if (mtch.__batch) { @@ -24,7 +24,10 @@ export default function sqlGenGenJoinStr({ join, mtch, table_name }) { }; } const equality = mtch.raw_equality ? sqlRawEquality(mtch.raw_equality) : "="; - const lhs = `${sqlIdent(typeof mtch.source == "object" ? mtch.source.tableName : table_name, "table name")}.${sqlIdent(typeof mtch.source == "object" ? mtch.source.fieldName : mtch.source, "field name")}`; + // `table_name` is already validated and quoted by `sqlGenerator` + const lhs = `${typeof mtch.source == "object" + ? sqlName(mtch.source.tableName, "table name") + : table_name}.${sqlName(typeof mtch.source == "object" ? mtch.source.fieldName : mtch.source, "field name")}`; if (mtch.between) { values.push(mtch.between.min, mtch.between.max); return { @@ -50,7 +53,7 @@ export default function sqlGenGenJoinStr({ join, mtch, table_name }) { const targetField = typeof mtch.target == "object" ? mtch.target.fieldName : mtch.target; - return `${sqlIdent(targetTable, "table name")}.${sqlIdent(targetField, "field name")}`; + return `${sqlName(targetTable, "table name")}.${sqlName(targetField, "field name")}`; })(); return { str: `${lhs} ${equality} ${rhs}`, diff --git a/dist/utils/sql-generator-gen-query-str.js b/dist/utils/sql-generator-gen-query-str.js index babf571..3491846 100644 --- a/dist/utils/sql-generator-gen-query-str.js +++ b/dist/utils/sql-generator-gen-query-str.js @@ -2,7 +2,7 @@ import { isUndefined } from "lodash"; import sqlGenGrabConcatStr from "./sql-generator-grab-concat-str"; import sqlGenGenJoinStr from "./sql-generator-gen-join-str"; import sqlGenGrabSelectFieldSQL from "./sql-generator-grab-select-field-sql"; -import { sqlIdent, sqlJoinType, sqlOperator } from "./sql-safe"; +import { sqlName, sqlJoinType, sqlOperator } from "./sql-safe"; export default function sqlGenGenQueryStr(params) { let str = "SELECT"; const genObject = params.genObject; @@ -40,9 +40,9 @@ export default function sqlGenGenQueryStr(params) { const countSubQuery = genObject.countSubQueries[i]; if (!countSubQuery) continue; - const countTable = sqlIdent(countSubQuery.table, "table name"); + const countTable = sqlName(countSubQuery.table, "table name"); const tableAlias = countSubQuery.table_alias - ? sqlIdent(countSubQuery.table_alias, "alias") + ? sqlName(countSubQuery.table_alias, "alias") : undefined; let subQStr = `(SELECT COUNT(*)`; subQStr += ` FROM ${countTable}${tableAlias ? ` ${tableAlias}` : ""}`; @@ -51,19 +51,19 @@ export default function sqlGenGenQueryStr(params) { const csqSrc = countSubQuery.srcTrgMap[j]; if (!csqSrc) continue; - subQStr += ` ${tableAlias || countTable}.${sqlIdent(csqSrc.src, "field name")}`; + subQStr += ` ${tableAlias || countTable}.${sqlName(csqSrc.src, "field name")}`; if (typeof csqSrc.trg == "string") { subQStr += ` = ?`; sqlSearhValues.push(csqSrc.trg); } else if (typeof csqSrc.trg == "object") { - subQStr += ` = ${sqlIdent(csqSrc.trg.table, "table name")}.${sqlIdent(csqSrc.trg.field, "field name")}`; + subQStr += ` = ${sqlName(csqSrc.trg.table, "table name")}.${sqlName(csqSrc.trg.field, "field name")}`; } if (j < countSubQuery.srcTrgMap.length - 1) { subQStr += ` AND `; } } - subQStr += ` )) AS ${sqlIdent(countSubQuery.alias, "alias")}`; + subQStr += ` )) AS ${sqlName(countSubQuery.alias, "alias")}`; countSqls.push(subQStr); } str += `, ${countSqls.join(",")}`; @@ -74,13 +74,13 @@ export default function sqlGenGenQueryStr(params) { .flat() .filter((j) => !isUndefined(j)) .map((joinObj) => { - const joinTableName = sqlIdent(joinObj.alias ? joinObj.alias : joinObj.tableName, "join table name"); + const joinTableName = sqlName(joinObj.alias ? joinObj.alias : joinObj.tableName, "join table name"); if (existingJoinTableNames.includes(joinTableName)) return null; existingJoinTableNames.push(joinTableName); if (joinObj.group_concat) { return sqlGenGrabConcatStr({ - field: `${joinTableName}.${sqlIdent(joinObj.group_concat.field, "field name")}`, + field: `${joinTableName}.${sqlName(joinObj.group_concat.field, "field name")}`, alias: joinObj.group_concat.alias, separator: joinObj.group_concat.separator, distinct: joinObj.group_concat.distinct, @@ -90,9 +90,9 @@ export default function sqlGenGenQueryStr(params) { return joinObj.selectFields .map((selectField) => { if (typeof selectField != "object") { - return `${joinTableName}.${sqlIdent(selectField, "field name")}`; + return `${joinTableName}.${sqlName(selectField, "field name")}`; } - const field = `${joinTableName}.${sqlIdent(selectField.field, "field name")}`; + const field = `${joinTableName}.${sqlName(selectField.field, "field name")}`; if (selectField.group_concat && selectField.alias) { return sqlGenGrabConcatStr({ @@ -122,7 +122,7 @@ export default function sqlGenGenQueryStr(params) { aliasSelectField = `DISTINCT ${field}`; } if (selectField.alias) - aliasSelectField += ` AS ${sqlIdent(selectField.alias, "alias")}`; + aliasSelectField += ` AS ${sqlName(selectField.alias, "alias")}`; return aliasSelectField; }) .join(","); @@ -145,11 +145,11 @@ export default function sqlGenGenQueryStr(params) { .flat() .filter((j) => !isUndefined(j)) .map((join) => { - const joinTable = sqlIdent(join.tableName, "table name"); + const joinTable = sqlName(join.tableName, "table name"); return (sqlJoinType(join.joinType) + " " + (join.alias - ? `${joinTable} ${sqlIdent(join.alias, "alias")}` + ? `${joinTable} ${sqlName(join.alias, "alias")}` : joinTable) + " ON " + (() => { diff --git a/dist/utils/sql-generator-gen-search-str.js b/dist/utils/sql-generator-gen-search-str.js index b5119db..d80aea8 100644 --- a/dist/utils/sql-generator-gen-search-str.js +++ b/dist/utils/sql-generator-gen-search-str.js @@ -1,5 +1,5 @@ import sqlGenOperatorGen from "./sql-gen-operator-gen"; -import { sqlField, sqlIdent, sqlOperator, sqlRawEquality } from "./sql-safe"; +import { sqlColumn, sqlName, sqlOperator, sqlRawEquality } from "./sql-safe"; function isPresent(val) { return val !== undefined && val !== null; } @@ -8,10 +8,10 @@ function isValuesObject(val) { } export default function sqlGenGenSearchStr({ queryObj, join, field, table_name, }) { let sqlSearhValues = []; - const safeField = sqlField(field); + const safeField = sqlColumn(field); const finalFieldName = (() => { if (queryObj?.tableName) { - return `${sqlIdent(queryObj.tableName, "table name")}.${safeField}`; + return `${sqlName(queryObj.tableName, "table name")}.${safeField}`; } if (join) { return `${table_name}.${safeField}`; @@ -28,7 +28,7 @@ export default function sqlGenGenSearchStr({ queryObj, join, field, table_name, equality: val.equality || queryObj.equality, fieldName: finalFieldName, value: isFieldValue - ? `${sqlIdent(val.tableName, "table name")}.${sqlIdent(val.fieldName, "field name")}` + ? `${sqlName(val.tableName, "table name")}.${sqlName(val.fieldName, "field name")}` : val.value, isValueFieldValue: isFieldValue, }); diff --git a/dist/utils/sql-generator-grab-concat-str.d.ts b/dist/utils/sql-generator-grab-concat-str.d.ts index 3eb39c7..fa23b97 100644 --- a/dist/utils/sql-generator-grab-concat-str.d.ts +++ b/dist/utils/sql-generator-grab-concat-str.d.ts @@ -1,4 +1,7 @@ type Param = { + /** + * An already validated and quoted column reference + */ field: string; alias: string; separator?: string; diff --git a/dist/utils/sql-generator-grab-concat-str.js b/dist/utils/sql-generator-grab-concat-str.js index cafc548..869885e 100644 --- a/dist/utils/sql-generator-grab-concat-str.js +++ b/dist/utils/sql-generator-grab-concat-str.js @@ -1,14 +1,14 @@ -import { sqlField, sqlIdent, sqlStringLiteral } from "./sql-safe"; +import { sqlName, sqlStringLiteral } from "./sql-safe"; export default function sqlGenGrabConcatStr({ alias, field, separator = ",", distinct, }) { let gc = `GROUP_CONCAT(`; if (distinct) { gc += `DISTINCT `; } - gc += sqlField(field); + gc += field; if (!distinct) { gc += `, ${sqlStringLiteral(String(separator))}`; } gc += `)`; - gc += ` AS ${sqlIdent(alias, "alias")}`; + gc += ` AS ${sqlName(alias, "alias")}`; return gc; } diff --git a/dist/utils/sql-generator-grab-select-field-sql.js b/dist/utils/sql-generator-grab-select-field-sql.js index e164b38..b70c43d 100644 --- a/dist/utils/sql-generator-grab-select-field-sql.js +++ b/dist/utils/sql-generator-grab-select-field-sql.js @@ -1,20 +1,20 @@ import sqlGenGrabConcatStr from "./sql-generator-grab-concat-str"; -import { sqlField, sqlIdent } from "./sql-safe"; +import { sqlColumn, sqlName } from "./sql-safe"; export default function sqlGenGrabSelectFieldSQL({ selectFields, append_table_names, table_name }) { let str = ""; str += ` ${selectFields ?.map((fld) => { if (typeof fld != "object") { - return sqlField(fld); + return sqlColumn(fld); } - const fld_name = sqlField(fld.fieldName); + const fld_name = sqlColumn(fld.fieldName); const final_fld_name = append_table_names && !fld_name.includes(".") ? `${table_name}.${fld_name}` : fld_name; if (fld.group_concat) { return sqlGenGrabConcatStr({ field: final_fld_name, - alias: fld.group_concat.alias || fld.alias || fld_name, + alias: fld.group_concat.alias || fld.alias || String(fld.fieldName), separator: fld.group_concat.separator, distinct: fld.group_concat.distinct, }); @@ -46,7 +46,7 @@ export default function sqlGenGrabSelectFieldSQL({ selectFields, append_table_na fld_str += final_fld_name; } if (alias) { - fld_str += ` AS ${sqlIdent(alias, "alias")}`; + fld_str += ` AS ${sqlName(alias, "alias")}`; } return fld_str; }) diff --git a/dist/utils/sql-generator.js b/dist/utils/sql-generator.js index a589705..15e6786 100644 --- a/dist/utils/sql-generator.js +++ b/dist/utils/sql-generator.js @@ -1,13 +1,13 @@ import sqlGenGenSearchStr from "./sql-generator-gen-search-str"; import sqlGenGenQueryStr from "./sql-generator-gen-query-str"; -import { sqlField, sqlIdent, sqlInt, sqlOperator, sqlOrderStrategy, } from "./sql-safe"; +import { sqlColumn, sqlInt, sqlName, sqlOperator, sqlOrderStrategy, } from "./sql-safe"; /** * # SQL Query Generator * @description Generates a parameterized SQLite SELECT query. All values are * returned in `values` to be bound; identifiers are validated. */ export default function sqlGenerator({ tableName: passedTableName, genObject, count, targetId, }) { - const tableName = sqlIdent(passedTableName, "table name"); + const tableName = sqlName(passedTableName, "table name"); const finalQuery = genObject?.query ? genObject.query : undefined; const queryKeys = finalQuery ? Object.keys(finalQuery) : undefined; const sqlSearhValues = []; @@ -83,10 +83,10 @@ export default function sqlGenerator({ tableName: passedTableName, genObject, co const grabGroupStr = (group) => { if (typeof group == "object") { return group.table - ? `${sqlIdent(group.table, "table name")}.${sqlIdent(group.field, "field name")}` - : sqlField(group.field); + ? `${sqlName(group.table, "table name")}.${sqlName(group.field, "field name")}` + : sqlColumn(group.field); } - return sqlField(group); + return sqlColumn(group); }; const group_by_txt = Array.isArray(genObject.group) ? genObject.group.map(grabGroupStr).join(",") @@ -95,8 +95,8 @@ export default function sqlGenerator({ tableName: passedTableName, genObject, co } function grabOrderString(order) { const field = genObject?.join && !String(order.field).includes(".") - ? `${tableName}.${sqlIdent(order.field, "order field")}` - : sqlField(order.field, "order field"); + ? `${tableName}.${sqlName(order.field, "order field")}` + : sqlColumn(order.field, "order field"); return ` ${field} ${sqlOrderStrategy(order.strategy)}`; } if (genObject?.order) { diff --git a/dist/utils/sql-insert-generator.js b/dist/utils/sql-insert-generator.js index b898307..dd13d47 100644 --- a/dist/utils/sql-insert-generator.js +++ b/dist/utils/sql-insert-generator.js @@ -1,10 +1,10 @@ -import { sqlIdent } from "./sql-safe"; +import { sqlIdent, sqlName } from "./sql-safe"; /** * # SQL Insert Generator */ export default function sqlInsertGenerator({ tableName, data, dbFullName, }) { const finalDbName = dbFullName - ? `${sqlIdent(dbFullName, "database name")}.` + ? `${sqlName(dbFullName, "database name")}.` : ""; if (Array.isArray(data) && data?.[0]) { let insertKeys = []; @@ -32,7 +32,7 @@ export default function sqlInsertGenerator({ tableName, data, dbFullName, }) { }) .join(",")})`); }); - let query = `INSERT INTO ${finalDbName}${sqlIdent(tableName, "table name")} (${insertKeys.join(",")}) VALUES ${queryBatches.join(",")}`; + let query = `INSERT INTO ${finalDbName}${sqlName(tableName, "table name")} (${insertKeys.map((ky) => sqlName(ky)).join(",")}) VALUES ${queryBatches.join(",")}`; return { query: query, values: queryValues, diff --git a/dist/utils/sql-safe.d.ts b/dist/utils/sql-safe.d.ts index bb678df..e7ee96d 100644 --- a/dist/utils/sql-safe.d.ts +++ b/dist/utils/sql-safe.d.ts @@ -11,6 +11,19 @@ export declare function sqlIdent(name: unknown, label?: string): string; * @description Accepts `field`, `table.field` or `table.*` */ export declare function sqlField(name: unknown, label?: string): string; +/** + * # Validate and quote an identifier for queries + * @description Quoting lets reserved words (`to`, `order`, ...) be used as + * names. Backticks, not double quotes: SQLite reads an unknown `"name"` as + * a string literal, so a typo would silently match nothing instead of + * failing with "no such column". + */ +export declare function sqlName(name: unknown, label?: string): string; +/** + * # Validate and quote a column reference + * @description Accepts `field`, `table.field` or `table.*` + */ +export declare function sqlColumn(name: unknown, label?: string): string; /** * # Quote an identifier for DDL statements */ diff --git a/dist/utils/sql-safe.js b/dist/utils/sql-safe.js index 244d125..cf4397a 100644 --- a/dist/utils/sql-safe.js +++ b/dist/utils/sql-safe.js @@ -25,6 +25,26 @@ export function sqlField(name, label = "field name") { } return str; } +/** + * # Validate and quote an identifier for queries + * @description Quoting lets reserved words (`to`, `order`, ...) be used as + * names. Backticks, not double quotes: SQLite reads an unknown `"name"` as + * a string literal, so a typo would silently match nothing instead of + * failing with "no such column". + */ +export function sqlName(name, label = "identifier") { + return `\`${sqlIdent(name, label)}\``; +} +/** + * # Validate and quote a column reference + * @description Accepts `field`, `table.field` or `table.*` + */ +export function sqlColumn(name, label = "field name") { + return sqlField(name, label) + .split(".") + .map((part) => (part === "*" ? part : `\`${part}\``)) + .join("."); +} /** * # Quote an identifier for DDL statements */ diff --git a/package.json b/package.json index 2002195..7313fc4 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@moduletrace/bun-sqlite", - "version": "1.2.1", + "version": "1.3.0", "description": "Schema-driven SQLite manager for Bun", "author": "Benjamin Toby", "type": "module", diff --git a/src/functions/grab-sqlite-client.ts b/src/functions/grab-sqlite-client.ts index 95acb7c..69a3a11 100644 --- a/src/functions/grab-sqlite-client.ts +++ b/src/functions/grab-sqlite-client.ts @@ -24,6 +24,10 @@ export default function grabSQLiteClient({ config }: Params): Database { sqliteVec.load(SQLiteClient); + // Wait for other connections' write locks instead of failing with + // SQLITE_BUSY (queue workers in separate processes) + SQLiteClient.run("PRAGMA busy_timeout = 5000;"); + if (config.wal_mode) { SQLiteClient.run("PRAGMA journal_mode = WAL;"); } diff --git a/src/index.ts b/src/index.ts index b7a3e39..a4a67af 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,6 +4,13 @@ import DbInsert from "./lib/sqlite/db-insert"; import DbSelect from "./lib/sqlite/db-select"; import DbSQL from "./lib/sqlite/db-sql"; import DbUpdate from "./lib/sqlite/db-update"; +import { + QueueAdd, + QueueClaim, + QueueComplete, + QueueFail, + QueueReleaseStale, +} from "./lib/sqlite/db-queue"; import grabDbSchema from "./utils/grab-db-schema"; import grabJoinFieldsFromQueryObject from "./utils/grab-join-fields-from-query-object"; @@ -24,6 +31,13 @@ const BunSQLite = { update: withInit(DbUpdate), delete: withInit(DbDelete), sql: withInit(DbSQL), + queue: { + add: withInit(QueueAdd), + claim: withInit(QueueClaim), + complete: withInit(QueueComplete), + fail: withInit(QueueFail), + release_stale: withInit(QueueReleaseStale), + }, utils: { grab_db_schema: withInit(grabDbSchema), grab_join_fields_from_query_object: grabJoinFieldsFromQueryObject, @@ -48,4 +62,6 @@ export { DockerComposeServices, IndexTypes, DefaultFields, + QueueFields, + QueueStatuses, } from "./types"; diff --git a/src/lib/grab-duplicate-safe-insert-sql.ts b/src/lib/grab-duplicate-safe-insert-sql.ts index 0e3fff7..106691b 100644 --- a/src/lib/grab-duplicate-safe-insert-sql.ts +++ b/src/lib/grab-duplicate-safe-insert-sql.ts @@ -1,5 +1,5 @@ import init from "../functions/init"; -import { sqlIdent, sqlQuoteIdent } from "../utils/sql-safe"; +import { sqlName } from "../utils/sql-safe"; type Params = { sql: string; @@ -17,7 +17,7 @@ export default async function ({ sql: passed_sql, table, data }: Params) { const set_sql_arr = Object.keys(Array.isArray(data) ? data[0] : data) .filter((field) => field !== "updated_at") .map((field) => { - const column = sqlIdent(field, "column name"); + const column = sqlName(field, "column name"); return `${column} = excluded.${column}`; }); @@ -30,14 +30,14 @@ export default async function ({ sql: passed_sql, table, data }: Params) { for (let i = 0; i < unique_fields.length; i++) { const field = unique_fields[i]; if (!field?.fieldName) continue; - sql += ` ON CONFLICT(${sqlQuoteIdent(field.fieldName)}) DO UPDATE SET ${set_sql}`; + sql += ` ON CONFLICT(${sqlName(field.fieldName, "column name")}) DO UPDATE SET ${set_sql}`; } if (table_schema.uniqueConstraints?.[0]) { for (let i = 0; i < table_schema.uniqueConstraints.length; i++) { const constraint = table_schema.uniqueConstraints[i]; const columns = constraint?.constraintTableFields - ?.map((c) => sqlQuoteIdent(String(c.value))) + ?.map((c) => sqlName(c.value, "column name")) ?.join(", "); if (!columns) continue; sql += ` ON CONFLICT(${columns}) DO UPDATE SET ${set_sql}`; diff --git a/src/lib/schema/create-db-schema.ts b/src/lib/schema/create-db-schema.ts index 77619ea..1811ecf 100644 --- a/src/lib/schema/create-db-schema.ts +++ b/src/lib/schema/create-db-schema.ts @@ -23,6 +23,14 @@ export default function createDBSchema(params: CreateDBSchemaParams) { */ const db_schema = orderDBSchema(params); + for (const table of db_schema.tables) { + if (table.isQueue && table.isVector) { + throw new Error( + `Table "${table.tableName}" can't be both a queue and a vector table`, + ); + } + } + db.run("PRAGMA foreign_keys = OFF;"); db.run("BEGIN"); diff --git a/src/lib/sqlite/db-delete.ts b/src/lib/sqlite/db-delete.ts index 5a60a65..e9cc466 100644 --- a/src/lib/sqlite/db-delete.ts +++ b/src/lib/sqlite/db-delete.ts @@ -6,7 +6,7 @@ import type { ServerQueryParam, } from "../../types"; import sqlGenerator from "../../utils/sql-generator"; -import { sqlIdent } from "../../utils/sql-safe"; +import { sqlName } from "../../utils/sql-safe"; type Params< Schema extends { [k: string]: any } = { [k: string]: any }, @@ -41,7 +41,7 @@ export default async function DbDelete< const whereClause = sqlObj.where.string; if (whereClause) { - let sql = `DELETE FROM ${sqlIdent(table, "table name")} ${whereClause}`; + let sql = `DELETE FROM ${sqlName(table, "table name")} ${whereClause}`; sqlObj.string = sql; sqlObj.values = sqlObj.where.values; diff --git a/src/lib/sqlite/db-queue.ts b/src/lib/sqlite/db-queue.ts new file mode 100644 index 0000000..fea2380 --- /dev/null +++ b/src/lib/sqlite/db-queue.ts @@ -0,0 +1,271 @@ +import dbHandler from "../db-handler"; +import init from "../../functions/init"; +import DbInsert from "./db-insert"; +import type { + APIResponseObject, + BunSQLiteConfig, + BUN_SQLITE_QueueOptions, +} from "../../types"; +import { sqlInt, sqlName } from "../../utils/sql-safe"; + +type BaseParams
= { + table: Table; + config?: BunSQLiteConfig; +}; + +const DefaultQueueOptions: Required = { + maxAttempts: 3, + lockTimeout: 300_000, + backoff: 1000, + deleteOnComplete: false, +}; + +/** + * Validate the table name and resolve its queue options from the schema + */ +function resolveQueue(table: string) { + 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: Required = { + ...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: string) { + 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: APIResponseObject, + sql: string, +): APIResponseObject { + 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< + Schema extends { [k: string]: any } = { [k: string]: any }, + Table extends string = string, +>({ + table, + data, + delay = 0, + config, +}: BaseParams
& { + data: Schema[]; + delay?: number; +}): Promise> { + 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: any) { + 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< + Schema extends { [k: string]: any } = { [k: string]: any }, + Table extends string = string, +>({ + table, + worker, + limit = 1, + config, +}: BaseParams
& { + worker?: string; + limit?: number; +}): Promise> { + 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 as APIResponseObject; + } + + 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: any) { + 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< + Schema extends { [k: string]: any } = { [k: string]: any }, + Table extends string = string, +>({ + table, + id, + config, +}: BaseParams
& { + id: number | string; +}): Promise> { + 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: any) { + 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< + Schema extends { [k: string]: any } = { [k: string]: any }, + Table extends string = string, +>({ + table, + id, + error, + config, +}: BaseParams
& { + id: number | string; + error?: string; +}): Promise> { + 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: any) { + 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< + Schema extends { [k: string]: any } = { [k: string]: any }, + Table extends string = string, +>({ table, config }: BaseParams
): Promise> { + 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: any) { + return { success: false, error: error.message, debug: { sql } }; + } +} diff --git a/src/lib/sqlite/db-update.ts b/src/lib/sqlite/db-update.ts index 1ea10d5..c331643 100644 --- a/src/lib/sqlite/db-update.ts +++ b/src/lib/sqlite/db-update.ts @@ -9,7 +9,7 @@ import type { } from "../../types"; import sqlGenerator from "../../utils/sql-generator"; import { sqlValue } from "../../utils/sql-insert-generator"; -import { sqlIdent } from "../../utils/sql-safe"; +import { sqlName } from "../../utils/sql-safe"; type Params< Schema extends { [k: string]: any } = { [k: string]: any }, @@ -65,7 +65,7 @@ export default async function DbUpdate< const whereClause = sqlQueryObj.where.string; if (whereClause) { - let sql = `UPDATE ${sqlIdent(table, "table name")} SET`; + let sql = `UPDATE ${sqlName(table, "table name")} SET`; const finalData: { [k: string]: SQLInsertGenValueType } = { updated_at: Date.now(), @@ -80,7 +80,7 @@ export default async function DbUpdate< const isLast = i == keys.length - 1; - sql += ` ${sqlIdent(key, "column name")}=?`; + sql += ` ${sqlName(key, "column name")}=?`; values.push(sqlValue(finalData[key])); if (!isLast) { diff --git a/src/types/index.ts b/src/types/index.ts index 8b35119..3f94e9d 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -108,6 +108,36 @@ export interface BUN_SQLITE_TableSchemaType { * Type of vector. Defaults to `vec0` */ vectorType?: string; + /** + * Make this a queue table. Adds the `QueueFields` columns and an index + * on `(status, available_at)`. Use `BunSQLite.queue` to work the queue. + */ + isQueue?: boolean | BUN_SQLITE_QueueOptions; +} + +/** + * Options for a queue table (`isQueue`). + */ +export interface BUN_SQLITE_QueueOptions { + /** + * Claims allowed before a failing job is marked `failed`. Defaults to `3` + */ + maxAttempts?: number; + /** + * Milliseconds a claimed job can stay `processing` before it is + * released back to the queue. Defaults to `300000` (5 minutes) + */ + lockTimeout?: number; + /** + * Base retry delay in milliseconds, doubled on every attempt. Defaults + * to `1000` + */ + backoff?: number; + /** + * Delete jobs on completion instead of marking them `completed`. + * Defaults to `false` + */ + deleteOnComplete?: boolean; } /** @@ -1630,6 +1660,56 @@ export const DefaultFields: BUN_SQLITE_FieldSchemaType[] = [ }, ]; +export const QueueStatuses = [ + "pending", + "processing", + "completed", + "failed", +] as const; + +/** + * Fields added to queue tables (`isQueue`) + */ +export const QueueFields: BUN_SQLITE_FieldSchemaType[] = [ + { + fieldName: "status", + dataType: "TEXT", + notNullValue: true, + defaultValue: "pending", + fieldDescription: `The state of the job. (${QueueStatuses.join(" | ")})`, + }, + { + fieldName: "attempts", + dataType: "INTEGER", + notNullValue: true, + defaultValue: 0, + fieldDescription: "The number of times the job has been claimed.", + }, + { + fieldName: "available_at", + dataType: "INTEGER", + notNullValue: true, + defaultValue: 0, + fieldDescription: + "The time from which the job can be claimed. (Unix Timestamp)", + }, + { + fieldName: "locked_at", + dataType: "INTEGER", + fieldDescription: "The time when the job was claimed. (Unix Timestamp)", + }, + { + fieldName: "locked_by", + dataType: "TEXT", + fieldDescription: "The worker that claimed the job.", + }, + { + fieldName: "last_error", + dataType: "TEXT", + fieldDescription: "The error from the last failed attempt.", + }, +]; + export type BunSQLiteQueryFieldValues< F extends string = string, T extends string = string, diff --git a/src/utils/append-default-fields-to-db-schema.ts b/src/utils/append-default-fields-to-db-schema.ts index 7e7c769..cf06717 100644 --- a/src/utils/append-default-fields-to-db-schema.ts +++ b/src/utils/append-default-fields-to-db-schema.ts @@ -1,10 +1,16 @@ import _ from "lodash"; -import { DefaultFields, type BUN_SQLITE_DatabaseSchemaType } from "../types"; +import { + DefaultFields, + QueueFields, + type BUN_SQLITE_DatabaseSchemaType, +} from "../types"; type Params = { dbSchema: BUN_SQLITE_DatabaseSchemaType; }; +const QueueFieldNames = QueueFields.map((f) => f.fieldName); + export default function ({ dbSchema }: Params): BUN_SQLITE_DatabaseSchemaType { const finaldbSchema = _.cloneDeep(dbSchema); finaldbSchema.tables = finaldbSchema.tables.map((t) => { @@ -13,6 +19,23 @@ export default function ({ dbSchema }: Params): BUN_SQLITE_DatabaseSchemaType { (f) => !f.fieldName?.match(/^(id|created_at|updated_at)$/), ); newTable.fields.unshift(...DefaultFields); + + if (newTable.isQueue) { + newTable.fields = newTable.fields.filter( + (f) => !QueueFieldNames.includes(f.fieldName), + ); + newTable.fields.push(...QueueFields); + + const indexName = `idx_${newTable.tableName}_queue`; + newTable.indexes = (newTable.indexes || []).filter( + (idx) => idx.indexName !== indexName, + ); + newTable.indexes.push({ + indexName, + indexTableFields: ["status", "available_at"], + }); + } + return newTable; }); diff --git a/src/utils/sql-generator-gen-join-str.ts b/src/utils/sql-generator-gen-join-str.ts index bbe6513..f4ce546 100644 --- a/src/utils/sql-generator-gen-join-str.ts +++ b/src/utils/sql-generator-gen-join-str.ts @@ -3,7 +3,7 @@ import type { ServerQueryParamsJoinMatchObject, SQLInsertGenValueType, } from "../types"; -import { sqlIdent, sqlOperator, sqlRawEquality } from "./sql-safe"; +import { sqlName, sqlOperator, sqlRawEquality } from "./sql-safe"; type Param = { mtch: ServerQueryParamsJoinMatchObject; @@ -48,10 +48,12 @@ export default function sqlGenGenJoinStr({ join, mtch, table_name }: Param) { const equality = mtch.raw_equality ? sqlRawEquality(mtch.raw_equality) : "="; - const lhs = `${sqlIdent( - typeof mtch.source == "object" ? mtch.source.tableName : table_name, - "table name", - )}.${sqlIdent( + // `table_name` is already validated and quoted by `sqlGenerator` + const lhs = `${ + typeof mtch.source == "object" + ? sqlName(mtch.source.tableName, "table name") + : table_name + }.${sqlName( typeof mtch.source == "object" ? mtch.source.fieldName : mtch.source, "field name", )}`; @@ -90,7 +92,7 @@ export default function sqlGenGenJoinStr({ join, mtch, table_name }: Param) { ? mtch.target.fieldName : mtch.target; - return `${sqlIdent(targetTable, "table name")}.${sqlIdent( + return `${sqlName(targetTable, "table name")}.${sqlName( targetField, "field name", )}`; diff --git a/src/utils/sql-generator-gen-query-str.ts b/src/utils/sql-generator-gen-query-str.ts index 3153366..cd1c886 100644 --- a/src/utils/sql-generator-gen-query-str.ts +++ b/src/utils/sql-generator-gen-query-str.ts @@ -3,7 +3,7 @@ import type { ServerQueryParam, TableSelectFieldsObject } from "../types"; import sqlGenGrabConcatStr from "./sql-generator-grab-concat-str"; import sqlGenGenJoinStr from "./sql-generator-gen-join-str"; import sqlGenGrabSelectFieldSQL from "./sql-generator-grab-select-field-sql"; -import { sqlIdent, sqlJoinType, sqlOperator } from "./sql-safe"; +import { sqlName, sqlJoinType, sqlOperator } from "./sql-safe"; type Param = { genObject?: ServerQueryParam; @@ -52,9 +52,9 @@ export default function sqlGenGenQueryStr< const countSubQuery = genObject.countSubQueries[i]; if (!countSubQuery) continue; - const countTable = sqlIdent(countSubQuery.table, "table name"); + const countTable = sqlName(countSubQuery.table, "table name"); const tableAlias = countSubQuery.table_alias - ? sqlIdent(countSubQuery.table_alias, "alias") + ? sqlName(countSubQuery.table_alias, "alias") : undefined; let subQStr = `(SELECT COUNT(*)`; @@ -67,7 +67,7 @@ export default function sqlGenGenQueryStr< const csqSrc = countSubQuery.srcTrgMap[j]; if (!csqSrc) continue; - subQStr += ` ${tableAlias || countTable}.${sqlIdent( + subQStr += ` ${tableAlias || countTable}.${sqlName( csqSrc.src, "field name", )}`; @@ -76,7 +76,7 @@ export default function sqlGenGenQueryStr< subQStr += ` = ?`; sqlSearhValues.push(csqSrc.trg); } else if (typeof csqSrc.trg == "object") { - subQStr += ` = ${sqlIdent(csqSrc.trg.table, "table name")}.${sqlIdent(csqSrc.trg.field, "field name")}`; + subQStr += ` = ${sqlName(csqSrc.trg.table, "table name")}.${sqlName(csqSrc.trg.field, "field name")}`; } if (j < countSubQuery.srcTrgMap.length - 1) { @@ -84,7 +84,7 @@ export default function sqlGenGenQueryStr< } } - subQStr += ` )) AS ${sqlIdent(countSubQuery.alias, "alias")}`; + subQStr += ` )) AS ${sqlName(countSubQuery.alias, "alias")}`; countSqls.push(subQStr); } @@ -98,7 +98,7 @@ export default function sqlGenGenQueryStr< .flat() .filter((j) => !isUndefined(j)) .map((joinObj) => { - const joinTableName = sqlIdent( + const joinTableName = sqlName( joinObj.alias ? joinObj.alias : joinObj.tableName, "join table name", ); @@ -109,7 +109,7 @@ export default function sqlGenGenQueryStr< if (joinObj.group_concat) { return sqlGenGrabConcatStr({ - field: `${joinTableName}.${sqlIdent(joinObj.group_concat.field, "field name")}`, + field: `${joinTableName}.${sqlName(joinObj.group_concat.field, "field name")}`, alias: joinObj.group_concat.alias, separator: joinObj.group_concat.separator, distinct: joinObj.group_concat.distinct, @@ -118,10 +118,10 @@ export default function sqlGenGenQueryStr< return joinObj.selectFields .map((selectField) => { if (typeof selectField != "object") { - return `${joinTableName}.${sqlIdent(selectField, "field name")}`; + return `${joinTableName}.${sqlName(selectField, "field name")}`; } - const field = `${joinTableName}.${sqlIdent(selectField.field, "field name")}`; + const field = `${joinTableName}.${sqlName(selectField.field, "field name")}`; if ( selectField.group_concat && @@ -154,7 +154,7 @@ export default function sqlGenGenQueryStr< } if (selectField.alias) - aliasSelectField += ` AS ${sqlIdent(selectField.alias, "alias")}`; + aliasSelectField += ` AS ${sqlName(selectField.alias, "alias")}`; return aliasSelectField; }) .join(","); @@ -179,13 +179,13 @@ export default function sqlGenGenQueryStr< .flat() .filter((j) => !isUndefined(j)) .map((join) => { - const joinTable = sqlIdent(join.tableName, "table name"); + const joinTable = sqlName(join.tableName, "table name"); return ( sqlJoinType(join.joinType) + " " + (join.alias - ? `${joinTable} ${sqlIdent(join.alias, "alias")}` + ? `${joinTable} ${sqlName(join.alias, "alias")}` : joinTable) + " ON " + (() => { diff --git a/src/utils/sql-generator-gen-search-str.ts b/src/utils/sql-generator-gen-search-str.ts index be94db3..624bc66 100644 --- a/src/utils/sql-generator-gen-search-str.ts +++ b/src/utils/sql-generator-gen-search-str.ts @@ -5,7 +5,7 @@ import type { SQLInsertGenValueType, } from "../types"; import sqlGenOperatorGen from "./sql-gen-operator-gen"; -import { sqlField, sqlIdent, sqlOperator, sqlRawEquality } from "./sql-safe"; +import { sqlColumn, sqlName, sqlOperator, sqlRawEquality } from "./sql-safe"; type Param = { queryObj: ServerQueryQueryObject[string]; @@ -37,11 +37,11 @@ export default function sqlGenGenSearchStr({ }: Param) { let sqlSearhValues: SQLInsertGenValueType[] = []; - const safeField = sqlField(field); + const safeField = sqlColumn(field); const finalFieldName = (() => { if (queryObj?.tableName) { - return `${sqlIdent(queryObj.tableName, "table name")}.${safeField}`; + return `${sqlName(queryObj.tableName, "table name")}.${safeField}`; } if (join) { return `${table_name}.${safeField}`; @@ -60,7 +60,7 @@ export default function sqlGenGenSearchStr({ equality: val.equality || queryObj.equality, fieldName: finalFieldName, value: isFieldValue - ? `${sqlIdent(val.tableName, "table name")}.${sqlIdent(val.fieldName, "field name")}` + ? `${sqlName(val.tableName, "table name")}.${sqlName(val.fieldName, "field name")}` : val.value, isValueFieldValue: isFieldValue, }); diff --git a/src/utils/sql-generator-grab-concat-str.ts b/src/utils/sql-generator-grab-concat-str.ts index ba14481..6dde60a 100644 --- a/src/utils/sql-generator-grab-concat-str.ts +++ b/src/utils/sql-generator-grab-concat-str.ts @@ -1,6 +1,9 @@ -import { sqlField, sqlIdent, sqlStringLiteral } from "./sql-safe"; +import { sqlName, sqlStringLiteral } from "./sql-safe"; type Param = { + /** + * An already validated and quoted column reference + */ field: string; alias: string; separator?: string; @@ -19,14 +22,14 @@ export default function sqlGenGrabConcatStr({ gc += `DISTINCT `; } - gc += sqlField(field); + gc += field; if (!distinct) { gc += `, ${sqlStringLiteral(String(separator))}`; } gc += `)`; - gc += ` AS ${sqlIdent(alias, "alias")}`; + gc += ` AS ${sqlName(alias, "alias")}`; return gc; } diff --git a/src/utils/sql-generator-grab-select-field-sql.ts b/src/utils/sql-generator-grab-select-field-sql.ts index 85ac9cd..80fa540 100644 --- a/src/utils/sql-generator-grab-select-field-sql.ts +++ b/src/utils/sql-generator-grab-select-field-sql.ts @@ -1,6 +1,6 @@ import type { TableSelectFieldsObject } from "../types"; import sqlGenGrabConcatStr from "./sql-generator-grab-concat-str"; -import { sqlField, sqlIdent } from "./sql-safe"; +import { sqlColumn, sqlName } from "./sql-safe"; type Param = { selectFields: (keyof T | TableSelectFieldsObject)[]; @@ -16,10 +16,10 @@ export default function sqlGenGrabSelectFieldSQL< str += ` ${selectFields ?.map((fld) => { if (typeof fld != "object") { - return sqlField(fld); + return sqlColumn(fld); } - const fld_name = sqlField(fld.fieldName); + const fld_name = sqlColumn(fld.fieldName); const final_fld_name = append_table_names && !fld_name.includes(".") @@ -29,7 +29,7 @@ export default function sqlGenGrabSelectFieldSQL< if (fld.group_concat) { return sqlGenGrabConcatStr({ field: final_fld_name, - alias: fld.group_concat.alias || fld.alias || fld_name, + alias: fld.group_concat.alias || fld.alias || String(fld.fieldName), separator: fld.group_concat.separator, distinct: fld.group_concat.distinct, }); @@ -58,7 +58,7 @@ export default function sqlGenGrabSelectFieldSQL< } if (alias) { - fld_str += ` AS ${sqlIdent(alias, "alias")}`; + fld_str += ` AS ${sqlName(alias, "alias")}`; } return fld_str; diff --git a/src/utils/sql-generator.ts b/src/utils/sql-generator.ts index 376cca0..a0a1ac3 100644 --- a/src/utils/sql-generator.ts +++ b/src/utils/sql-generator.ts @@ -6,9 +6,9 @@ import type { import sqlGenGenSearchStr from "./sql-generator-gen-search-str"; import sqlGenGenQueryStr from "./sql-generator-gen-query-str"; import { - sqlField, - sqlIdent, + sqlColumn, sqlInt, + sqlName, sqlOperator, sqlOrderStrategy, } from "./sql-safe"; @@ -51,7 +51,7 @@ export default function sqlGenerator< count, targetId, }: Param): Return { - const tableName = sqlIdent(passedTableName, "table name"); + const tableName = sqlName(passedTableName, "table name"); const finalQuery = genObject?.query ? genObject.query : undefined; const queryKeys = finalQuery ? Object.keys(finalQuery) : undefined; @@ -160,11 +160,11 @@ export default function sqlGenerator< ) => { if (typeof group == "object") { return group.table - ? `${sqlIdent(group.table, "table name")}.${sqlIdent(group.field, "field name")}` - : sqlField(group.field); + ? `${sqlName(group.table, "table name")}.${sqlName(group.field, "field name")}` + : sqlColumn(group.field); } - return sqlField(group); + return sqlColumn(group); }; const group_by_txt = Array.isArray(genObject.group) @@ -177,8 +177,8 @@ export default function sqlGenerator< function grabOrderString(order: ServerQueryParamOrder) { const field = genObject?.join && !String(order.field).includes(".") - ? `${tableName}.${sqlIdent(order.field, "order field")}` - : sqlField(order.field, "order field"); + ? `${tableName}.${sqlName(order.field, "order field")}` + : sqlColumn(order.field, "order field"); return ` ${field} ${sqlOrderStrategy(order.strategy)}`; } diff --git a/src/utils/sql-insert-generator.ts b/src/utils/sql-insert-generator.ts index 780d4be..f40c818 100644 --- a/src/utils/sql-insert-generator.ts +++ b/src/utils/sql-insert-generator.ts @@ -3,7 +3,7 @@ import type { SQLInsertGenReturn, SQLInsertGenValueType, } from "../types"; -import { sqlIdent } from "./sql-safe"; +import { sqlIdent, sqlName } from "./sql-safe"; /** * # SQL Insert Generator @@ -14,7 +14,7 @@ export default function sqlInsertGenerator({ dbFullName, }: SQLInsertGenParams): SQLInsertGenReturn | undefined { const finalDbName = dbFullName - ? `${sqlIdent(dbFullName, "database name")}.` + ? `${sqlName(dbFullName, "database name")}.` : ""; if (Array.isArray(data) && data?.[0]) { @@ -51,10 +51,10 @@ export default function sqlInsertGenerator({ ); }); - let query = `INSERT INTO ${finalDbName}${sqlIdent( + let query = `INSERT INTO ${finalDbName}${sqlName( tableName, "table name", - )} (${insertKeys.join(",")}) VALUES ${queryBatches.join(",")}`; + )} (${insertKeys.map((ky) => sqlName(ky)).join(",")}) VALUES ${queryBatches.join(",")}`; return { query: query, diff --git a/src/utils/sql-safe.ts b/src/utils/sql-safe.ts index cded203..9af36df 100644 --- a/src/utils/sql-safe.ts +++ b/src/utils/sql-safe.ts @@ -33,6 +33,28 @@ export function sqlField(name: unknown, label = "field name"): string { return str; } +/** + * # Validate and quote an identifier for queries + * @description Quoting lets reserved words (`to`, `order`, ...) be used as + * names. Backticks, not double quotes: SQLite reads an unknown `"name"` as + * a string literal, so a typo would silently match nothing instead of + * failing with "no such column". + */ +export function sqlName(name: unknown, label = "identifier"): string { + return `\`${sqlIdent(name, label)}\``; +} + +/** + * # Validate and quote a column reference + * @description Accepts `field`, `table.field` or `table.*` + */ +export function sqlColumn(name: unknown, label = "field name"): string { + return sqlField(name, label) + .split(".") + .map((part) => (part === "*" ? part : `\`${part}\``)) + .join("."); +} + /** * # Quote an identifier for DDL statements */ diff --git a/tests/fixtures/queue-worker.ts b/tests/fixtures/queue-worker.ts new file mode 100644 index 0000000..941d145 --- /dev/null +++ b/tests/fixtures/queue-worker.ts @@ -0,0 +1,42 @@ +/** + * Queue worker for the multi-process claim test. Claims jobs one at a time + * until the queue is empty and prints the claimed ids as JSON. + */ +import fs from "fs"; +import path from "path"; +import grabSQLiteClient from "../../src/functions/grab-sqlite-client"; +import { QueueClaim, QueueComplete } from "../../src/lib/sqlite/db-queue"; +import type { BunSQLiteConfig, BUN_SQLITE_DatabaseSchemaType } from "../../src/types"; + +const config: BunSQLiteConfig = JSON.parse(process.env.QUEUE_TEST_CONFIG!); +const dbSchema: BUN_SQLITE_DatabaseSchemaType = JSON.parse( + process.env.QUEUE_TEST_SCHEMA!, +); +const worker = process.env.QUEUE_TEST_WORKER!; + +global.CONFIG = config; +global.DB_SCHEMA = dbSchema; +global.SQLITE_CLIENT = grabSQLiteClient({ config }); + +// Signal ready, then wait for the test to start every worker at once +fs.writeFileSync(path.join(config.db_dir!, `ready-${worker}`), ""); +while (!fs.existsSync(path.join(config.db_dir!, "go"))) {} + +const ids: number[] = []; + +while (true) { + const res = await QueueClaim({ table: "jobs", worker }); + + if (!res.success) { + console.error(res.error); + process.exit(1); + } + + const job = res.singleRes; + if (!job) break; + + ids.push(job.id); + await QueueComplete({ table: "jobs", id: job.id }); +} + +console.log(JSON.stringify(ids)); diff --git a/tests/queue.test.ts b/tests/queue.test.ts new file mode 100644 index 0000000..c4d0748 --- /dev/null +++ b/tests/queue.test.ts @@ -0,0 +1,462 @@ +import { describe, expect, beforeAll, beforeEach, test } from "bun:test"; +import { Database } from "bun:sqlite"; +import fs from "fs"; +import os from "os"; +import path from "path"; +import createDBSchema from "../src/lib/schema/create-db-schema"; +import appendDefaultFieldsToDbSchema from "../src/utils/append-default-fields-to-db-schema"; +import DbInsert from "../src/lib/sqlite/db-insert"; +import DbSelect from "../src/lib/sqlite/db-select"; +import DbUpdate from "../src/lib/sqlite/db-update"; +import DbDelete from "../src/lib/sqlite/db-delete"; +import { + QueueAdd, + QueueClaim, + QueueComplete, + QueueFail, + QueueReleaseStale, +} from "../src/lib/sqlite/db-queue"; +import type { + BunSQLiteConfig, + BUN_SQLITE_DatabaseSchemaType, +} from "../src/types"; + +const db = new Database(":memory:"); + +const dbSchema: BUN_SQLITE_DatabaseSchemaType = { + dbName: "test", + tables: [ + { + tableName: "emails", + isQueue: { maxAttempts: 2, backoff: 1000, lockTimeout: 60_000 }, + fields: [{ fieldName: "recipient", dataType: "TEXT" }], + }, + { + tableName: "pings", + isQueue: { deleteOnComplete: true }, + fields: [{ fieldName: "url", dataType: "TEXT" }], + }, + { + tableName: "users", + fields: [{ fieldName: "name", dataType: "TEXT" }], + }, + ], +}; + +global.CONFIG = { db_name: "test" }; +global.DB_SCHEMA = dbSchema; +global.SQLITE_CLIENT = db; + +beforeAll(() => { + createDBSchema({ + db, + db_schema: appendDefaultFieldsToDbSchema({ dbSchema }), + }); +}); + +beforeEach(() => { + db.run(`DELETE FROM emails`); + db.run(`DELETE FROM pings`); +}); + +describe("queue schema", () => { + test("adds the queue columns and index", () => { + const columns = ( + db.query(`PRAGMA table_info(emails)`).all() as { name: string }[] + ).map((c) => c.name); + + expect(columns).toEqual([ + "id", + "created_at", + "updated_at", + "recipient", + "status", + "attempts", + "available_at", + "locked_at", + "locked_by", + "last_error", + ]); + + const indexes = ( + db.query(`PRAGMA index_list(emails)`).all() as { name: string }[] + ).map((i) => i.name); + + expect(indexes).toContain("idx_emails_queue"); + }); + + test("leaves non-queue tables alone", () => { + const columns = ( + db.query(`PRAGMA table_info(users)`).all() as { name: string }[] + ).map((c) => c.name); + + expect(columns).not.toContain("status"); + }); + + test("rejects a table that is both a queue and a vector table", () => { + expect(() => + createDBSchema({ + db: new Database(":memory:"), + db_schema: { + tables: [ + { + tableName: "bad", + isQueue: true, + isVector: true, + fields: [], + }, + ], + }, + }), + ).toThrow("can't be both a queue and a vector table"); + }); + + test("queue ops reject non-queue tables", async () => { + const res = await QueueClaim({ table: "users" }); + + expect(res.success).toBe(false); + expect(res.error).toContain("not a queue table"); + }); +}); + +describe("queue ops", () => { + test("add, claim, complete", async () => { + const added = await QueueAdd({ + table: "emails", + data: [{ recipient: "a@example.com" }], + }); + + expect(added.success).toBe(true); + expect(added.singleRes?.status).toBe("pending"); + + const claimed = await QueueClaim({ table: "emails", worker: "w1" }); + + expect(claimed.payload?.length).toBe(1); + expect(claimed.singleRes?.status).toBe("processing"); + expect(claimed.singleRes?.attempts).toBe(1); + expect(claimed.singleRes?.locked_by).toBe("w1"); + + const empty = await QueueClaim({ table: "emails" }); + + expect(empty.success).toBe(true); + expect(empty.payload?.length).toBe(0); + + const done = await QueueComplete({ + table: "emails", + id: claimed.singleRes!.id, + }); + + expect(done.success).toBe(true); + expect(done.singleRes?.status).toBe("completed"); + expect(done.singleRes?.locked_by).toBeNull(); + }); + + test("claims in FIFO order and respects limit", async () => { + await QueueAdd({ + table: "emails", + data: [{ recipient: "1" }, { recipient: "2" }, { recipient: "3" }], + }); + + const first = await QueueClaim({ table: "emails", limit: 2 }); + const second = await QueueClaim({ table: "emails", limit: 2 }); + + expect(first.payload?.map((r) => r.recipient)).toEqual(["1", "2"]); + expect(second.payload?.map((r) => r.recipient)).toEqual(["3"]); + }); + + test("concurrent claims never return the same job", async () => { + await QueueAdd({ + table: "emails", + data: Array.from({ length: 20 }, (_, i) => ({ recipient: String(i) })), + }); + + const results = await Promise.all( + Array.from({ length: 10 }, (_, i) => + QueueClaim({ table: "emails", worker: `w${i}`, limit: 3 }), + ), + ); + + const ids = results.flatMap((r) => r.payload!.map((row) => row.id)); + + expect(ids.length).toBe(20); + expect(new Set(ids).size).toBe(20); + }); + + test("delayed jobs are not claimed early", async () => { + await QueueAdd({ + table: "emails", + data: [{ recipient: "later" }], + delay: 60_000, + }); + + const res = await QueueClaim({ table: "emails" }); + + expect(res.payload?.length).toBe(0); + }); + + test("fail retries with backoff, then marks failed", async () => { + await QueueAdd({ table: "emails", data: [{ recipient: "flaky" }] }); + + const first = await QueueClaim({ table: "emails" }); + const id = first.singleRes!.id; + + const before = Date.now(); + const retry = await QueueFail({ table: "emails", id, error: "boom" }); + + expect(retry.singleRes?.status).toBe("pending"); + expect(retry.singleRes?.last_error).toBe("boom"); + expect(retry.singleRes?.available_at).toBeGreaterThanOrEqual( + before + 1000, + ); + + // Skip the backoff + db.run(`UPDATE emails SET available_at = 0 WHERE id = ?`, [id]); + + const second = await QueueClaim({ table: "emails" }); + expect(second.singleRes?.attempts).toBe(2); + + const dead = await QueueFail({ table: "emails", id, error: "boom 2" }); + + expect(dead.singleRes?.status).toBe("failed"); + expect(dead.singleRes?.last_error).toBe("boom 2"); + }); + + test("complete and fail only touch processing jobs", async () => { + const added = await QueueAdd({ table: "emails", data: [{ recipient: "x" }] }); + const id = added.singleRes!.id; + + expect((await QueueComplete({ table: "emails", id })).success).toBe( + false, + ); + expect((await QueueFail({ table: "emails", id })).success).toBe(false); + }); + + test("stale locks are released", async () => { + await QueueAdd({ table: "emails", data: [{ recipient: "stuck" }] }); + const claimed = await QueueClaim({ table: "emails" }); + const id = claimed.singleRes!.id; + + db.run(`UPDATE emails SET locked_at = ? WHERE id = ?`, [ + Date.now() - 120_000, + id, + ]); + + const released = await QueueReleaseStale({ table: "emails" }); + + expect(released.singleRes?.id).toBe(id); + expect(released.singleRes?.status).toBe("pending"); + expect(released.singleRes?.locked_at).toBeNull(); + }); + + test("stale locks out of attempts are marked failed", async () => { + await QueueAdd({ table: "emails", data: [{ recipient: "stuck" }] }); + const claimed = await QueueClaim({ table: "emails" }); + const id = claimed.singleRes!.id; + + db.run(`UPDATE emails SET attempts = 2, locked_at = 0 WHERE id = ?`, [ + id, + ]); + + // claim releases stale locks before claiming + const res = await QueueClaim({ table: "emails" }); + expect(res.payload?.length).toBe(0); + + const row = db.query(`SELECT * FROM emails WHERE id = ?`).get(id) as any; + expect(row.status).toBe("failed"); + expect(row.last_error).toBe("Lock timed out"); + }); + + test("deleteOnComplete removes the job", async () => { + await QueueAdd({ table: "pings", data: [{ url: "https://x" }] }); + const claimed = await QueueClaim({ table: "pings" }); + + const done = await QueueComplete({ + table: "pings", + id: claimed.singleRes!.id, + }); + + expect(done.success).toBe(true); + expect( + (db.query(`SELECT COUNT(*) AS n FROM pings`).get() as any).n, + ).toBe(0); + }); +}); + +describe("queue across processes", () => { + test("workers in separate processes never claim the same job", async () => { + const db_dir = fs.mkdtempSync(path.join(os.tmpdir(), "bun-sqlite-queue-")); + + try { + const config: BunSQLiteConfig = { + db_name: "queue.db", + db_dir, + wal_mode: true, + }; + const schema: BUN_SQLITE_DatabaseSchemaType = { + tables: [ + { + tableName: "jobs", + isQueue: true, + fields: [{ fieldName: "n", dataType: "INTEGER" }], + }, + ], + }; + + const fileDb = new Database(path.join(db_dir, "queue.db")); + fileDb.run("PRAGMA journal_mode = WAL;"); + createDBSchema({ + db: fileDb, + db_schema: appendDefaultFieldsToDbSchema({ dbSchema: schema }), + }); + + const jobCount = 500; + const insert = fileDb.prepare( + "INSERT INTO jobs (n, status, attempts, available_at) VALUES (?, 'pending', 0, 0)", + ); + fileDb.transaction(() => { + for (let i = 0; i < jobCount; i++) insert.run(i); + })(); + fileDb.close(); + + const workers = Array.from({ length: 4 }, (_, i) => + Bun.spawn( + ["bun", path.join(import.meta.dir, "fixtures/queue-worker.ts")], + { + env: { + ...process.env, + QUEUE_TEST_CONFIG: JSON.stringify(config), + QUEUE_TEST_SCHEMA: JSON.stringify(schema), + QUEUE_TEST_WORKER: `w${i}`, + }, + stdout: "pipe", + stderr: "pipe", + }, + ), + ); + + // Wait until every worker is up, then start them together + while ( + fs.readdirSync(db_dir).filter((f) => f.startsWith("ready-")) + .length < workers.length + ) { + await Bun.sleep(10); + } + fs.writeFileSync(path.join(db_dir, "go"), ""); + + const outputs = await Promise.all( + workers.map(async (proc) => { + const [stdout, stderr, code] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + proc.exited, + ]); + + expect(stderr).toBe(""); + expect(code).toBe(0); + + return JSON.parse(stdout.trim().split("\n").pop()!) as number[]; + }), + ); + + const ids = outputs.flat(); + + expect(ids.length).toBe(jobCount); + expect(new Set(ids).size).toBe(jobCount); + // More than one worker got jobs, so the claims really did + // interleave (the split is uneven: a worker holding the write + // lock tends to keep it) + expect(outputs.filter((o) => o.length > 0).length).toBeGreaterThan(1); + + const check = new Database(path.join(db_dir, "queue.db")); + const counts = check + .query("SELECT status, COUNT(*) AS n FROM jobs GROUP BY status") + .all(); + check.close(); + + expect(counts).toEqual([{ status: "completed", n: jobCount }]); + } finally { + fs.rmSync(db_dir, { recursive: true, force: true }); + } + }, 30_000); +}); + +describe("reserved-word names", () => { + test("insert, select, update, upsert and delete work on `to` / `order`", async () => { + db.run( + "CREATE TABLE `order` (id INTEGER PRIMARY KEY AUTOINCREMENT, `to` TEXT UNIQUE, `group` TEXT, created_at INTEGER, updated_at INTEGER)", + ); + + global.DB_SCHEMA = { + ...dbSchema, + tables: [ + ...dbSchema.tables, + { + tableName: "order", + fields: [ + { fieldName: "to", dataType: "TEXT", unique: true }, + { fieldName: "group", dataType: "TEXT" }, + ], + }, + ], + }; + + try { + const inserted = await DbInsert({ + table: "order", + data: [{ to: "a@example.com", group: "x" }], + }); + expect(inserted.success).toBe(true); + expect(inserted.singleRes?.to).toBe("a@example.com"); + + const upserted = await DbInsert({ + table: "order", + data: [{ to: "a@example.com", group: "y" }], + update_on_duplicate: true, + }); + expect(upserted.success).toBe(true); + expect(upserted.singleRes?.group).toBe("y"); + + const selected = await DbSelect({ + table: "order", + query: { + query: { to: { value: "a@example.com" } }, + selectFields: ["to", "group"], + order: { field: "group", strategy: "DESC" }, + group: "to", + }, + }); + expect(selected.success).toBe(true); + expect(selected.payload).toEqual([ + { to: "a@example.com", group: "y" }, + ]); + + const updated = await DbUpdate({ + table: "order", + data: { group: "z" }, + query: { query: { to: { value: "a@example.com" } } }, + }); + expect(updated.singleRes?.group).toBe("z"); + + const deleted = await DbDelete({ + table: "order", + query: { query: { to: { value: "a@example.com" } } }, + }); + expect(deleted.success).toBe(true); + expect( + (db.query("SELECT COUNT(*) AS n FROM `order`").get() as any).n, + ).toBe(0); + } finally { + global.DB_SCHEMA = dbSchema; + } + }); + + test("unknown columns fail instead of matching nothing", async () => { + const res = await DbSelect({ + table: "emails", + query: { query: { nope: { value: "x" } } }, + }); + + expect(res.success).toBe(false); + expect(res.error).toContain("no such column"); + }); +}); diff --git a/tests/sql-generator.test.ts b/tests/sql-generator.test.ts index 6bc3ca5..7002c52 100644 --- a/tests/sql-generator.test.ts +++ b/tests/sql-generator.test.ts @@ -7,7 +7,7 @@ const gen = (genObject: any, extra: Record = {}) => describe("identifiers", () => { test("binds values and keeps identifiers", () => { const { string, values } = gen({ query: { email: { value: "a@b.c" } } }); - expect(string).toBe("SELECT * FROM users WHERE (email = ?)"); + expect(string).toBe("SELECT * FROM `users` WHERE (`email` = ?)"); expect(values).toEqual(["a@b.c"]); }); @@ -51,14 +51,14 @@ describe("targetId", () => { { targetId: 5 }, ); expect(string).toBe( - "SELECT * FROM users WHERE (a = ? OR b = ?) AND id = ?", + "SELECT * FROM `users` WHERE (`a` = ? OR `b` = ?) AND id = ?", ); expect(values).toEqual([1, 2, 5]); }); test("accepts numeric strings", () => { const { string, values } = gen({}, { targetId: "7" }); - expect(string).toBe("SELECT * FROM users WHERE id = ?"); + expect(string).toBe("SELECT * FROM `users` WHERE id = ?"); expect(values).toEqual([7]); }); @@ -79,13 +79,13 @@ describe("targetId", () => { }, { targetId: 3 }, ); - expect(string).toEndWith("WHERE users.id = ?"); + expect(string).toEndWith("WHERE `users`.id = ?"); }); test("is part of the where clause used by update / delete", () => { const { where } = gen({ query: { a: { value: 1 } } }, { targetId: 2 }); expect(where).toEqual({ - string: "WHERE (a = ?) AND id = ?", + string: "WHERE (`a` = ?) AND id = ?", values: [1, 2], }); }); @@ -100,17 +100,17 @@ describe("values", () => { const { string, values } = gen({ query: { id: { value: [1, 2], equality: "IN" } }, }); - expect(string).toBe("SELECT * FROM users WHERE (id IN (?, ?))"); + expect(string).toBe("SELECT * FROM `users` WHERE (`id` IN (?, ?))"); expect(values).toEqual([1, 2]); }); test("empty IN matches nothing, empty NOT IN matches everything", () => { expect(gen({ query: { id: { value: [], equality: "IN" } } }).string).toBe( - "SELECT * FROM users WHERE (1 = 0)", + "SELECT * FROM `users` WHERE (1 = 0)", ); expect( gen({ query: { id: { value: [], equality: "NOT IN" } } }).string, - ).toBe("SELECT * FROM users WHERE (1 = 1)"); + ).toBe("SELECT * FROM `users` WHERE (1 = 1)"); }); test("BETWEEN needs exactly two values", () => { @@ -123,7 +123,7 @@ describe("values", () => { describe("pagination", () => { test("page is converted to an offset", () => { expect(gen({ limit: 10, page: 3 }).string).toBe( - "SELECT * FROM users LIMIT 10 OFFSET 20", + "SELECT * FROM `users` LIMIT 10 OFFSET 20", ); }); diff --git a/tests/sql-values.test.ts b/tests/sql-values.test.ts index f4a0a72..4907d23 100644 --- a/tests/sql-values.test.ts +++ b/tests/sql-values.test.ts @@ -27,7 +27,7 @@ describe("sqlInsertGenerator", () => { ], }); expect(res?.query).toBe( - "INSERT INTO items (name,active,qty,note) VALUES (?,?,?,?),(?,?,?,?)", + "INSERT INTO `items` (`name`,`active`,`qty`,`note`) VALUES (?,?,?,?),(?,?,?,?)", ); expect(res?.values).toEqual(["a", 0, 0, null, "b", null, null, ""]); }); @@ -46,7 +46,7 @@ describe("sqlInsertGenerator", () => { ], }); expect(calls).toBe(1); - expect(res?.query).toBe("INSERT INTO items (emb) VALUES (vec_f32(?))"); + expect(res?.query).toBe("INSERT INTO `items` (`emb`) VALUES (vec_f32(?))"); expect(res?.values).toEqual(["[1,2]"]); });