70 lines
2.4 KiB
JavaScript
70 lines
2.4 KiB
JavaScript
import _ from "lodash";
|
|
import { readLiveSchema } from "../../functions/live-schema";
|
|
import { QueueProtectedFields, } from "../../types";
|
|
/**
|
|
* Current time in ms from the database clock. Leases are set and checked
|
|
* with this so workers on hosts with drifting clocks agree on expiry.
|
|
*/
|
|
export const QUEUE_NOW_SQL = "CAST(UNIX_TIMESTAMP(NOW(3)) * 1000 AS SIGNED)";
|
|
/**
|
|
* Extra WHERE condition for generic update/delete on queue tables: rows a
|
|
* worker has claimed are only changed through the queue API.
|
|
*/
|
|
export const QUEUE_NOT_CLAIMED_SQL = "`queue_status` <> 'processing'";
|
|
const QUEUE_TABLE_DEFAULTS = {
|
|
max_attempts: 5,
|
|
lease_ms: 30_000,
|
|
keep_completed: false,
|
|
};
|
|
/**
|
|
* A table's `queue_table` options with defaults filled in (looked up
|
|
* directly or through its parent table), or `undefined` if it is not a
|
|
* queue table.
|
|
*/
|
|
export function grabQueueTableOptions(tableName) {
|
|
const dbSchema = global.DB_SCHEMA || readLiveSchema();
|
|
const table = dbSchema?.tables?.find((t) => t.tableName === tableName);
|
|
if (!table)
|
|
return undefined;
|
|
if (table.queue_table === undefined) {
|
|
return table.parentTableName
|
|
? grabQueueTableOptions(table.parentTableName)
|
|
: undefined;
|
|
}
|
|
if (!table.queue_table)
|
|
return undefined;
|
|
return {
|
|
...QUEUE_TABLE_DEFAULTS,
|
|
...(typeof table.queue_table === "object"
|
|
? _.omitBy(table.queue_table, _.isUndefined)
|
|
: {}),
|
|
};
|
|
}
|
|
/**
|
|
* Whether a table is declared with `queue_table` (directly or through its
|
|
* parent table).
|
|
*/
|
|
export function isQueueTable(tableName) {
|
|
return Boolean(grabQueueTableOptions(tableName));
|
|
}
|
|
/**
|
|
* Queue options for a table the queue API was called on. Throws if the table
|
|
* is not a queue table.
|
|
*/
|
|
export function requireQueueTable(tableName) {
|
|
const options = grabQueueTableOptions(tableName);
|
|
if (!options) {
|
|
throw new Error(`\`${tableName}\` is not a queue table`);
|
|
}
|
|
return options;
|
|
}
|
|
/**
|
|
* Throw if `data` sets queue state the queue API owns.
|
|
*/
|
|
export function assertNoProtectedQueueFields(table, data) {
|
|
const protectedFields = QueueProtectedFields.filter((field) => data.some((d) => d && d[field] !== undefined));
|
|
if (protectedFields.length > 0) {
|
|
throw new Error(`\`${table}\` is a queue table: ${protectedFields.join(", ")} can only be changed through BunMariaDB.queue`);
|
|
}
|
|
}
|