719 lines
21 KiB
JavaScript
719 lines
21 KiB
JavaScript
// Database-backed background task queue.
|
|
|
|
function normalizeText(value) {
|
|
return String(value || '').trim();
|
|
}
|
|
|
|
function toIsoDate(value) {
|
|
if (!value) {
|
|
return '';
|
|
}
|
|
|
|
const date = value instanceof Date ? value : new Date(value);
|
|
return Number.isNaN(date.getTime()) ? '' : date.toISOString();
|
|
}
|
|
|
|
function normalizeIntervalMs(value, unit) {
|
|
const numericValue = Math.max(1, Number(value) || 0);
|
|
const normalizedUnit = String(unit || 'minutes').trim().toLowerCase();
|
|
if (normalizedUnit === 'seconds') {
|
|
return numericValue * 1000;
|
|
}
|
|
return numericValue * 60 * 1000;
|
|
}
|
|
|
|
function createBackgroundTaskQueue(options) {
|
|
const pool = options && options.pool;
|
|
const maxConcurrent = Math.max(1, Number(options && options.maxConcurrent) || 1);
|
|
const taskHandlers = new Map();
|
|
const recurringJobsByKey = new Map();
|
|
let activeCount = 0;
|
|
let drainScheduled = false;
|
|
let initializationPromise = null;
|
|
|
|
if (!pool) {
|
|
throw new Error('createBackgroundTaskQueue requires a database pool.');
|
|
}
|
|
|
|
function buildSnapshot(task) {
|
|
return {
|
|
id: Number(task && task.id) || 0,
|
|
key: String(task && task.key || '').trim(),
|
|
taskType: String(task && task.taskType || '').trim(),
|
|
title: String(task && task.title || 'Background task').trim() || 'Background task',
|
|
category: String(task && task.category || 'general').trim() || 'general',
|
|
status: String(task && task.status || 'queued').trim() || 'queued',
|
|
createdAt: String(task && task.createdAt || ''),
|
|
startedAt: String(task && task.startedAt || ''),
|
|
finishedAt: String(task && task.finishedAt || ''),
|
|
errorMessage: String(task && task.errorMessage || ''),
|
|
attempts: Math.max(0, Number(task && task.attempts) || 0),
|
|
metadata: task && task.metadata ? task.metadata : {},
|
|
payload: task && task.payload !== undefined ? task.payload : null
|
|
};
|
|
}
|
|
|
|
function parseJsonValue(value, fallback) {
|
|
if (value === null || value === undefined || value === '') {
|
|
return fallback;
|
|
}
|
|
|
|
if (typeof value === 'object') {
|
|
return value;
|
|
}
|
|
|
|
try {
|
|
return JSON.parse(String(value));
|
|
} catch (_error) {
|
|
return fallback;
|
|
}
|
|
}
|
|
|
|
function stringifyJsonValue(value) {
|
|
if (value === undefined || value === null) {
|
|
return null;
|
|
}
|
|
|
|
return JSON.stringify(value);
|
|
}
|
|
|
|
function hydrateTaskRow(row) {
|
|
if (!row) {
|
|
return null;
|
|
}
|
|
|
|
return {
|
|
id: Number(row.id),
|
|
key: String(row.task_key || '').trim(),
|
|
taskType: String(row.task_type || '').trim(),
|
|
title: String(row.title || 'Background task').trim() || 'Background task',
|
|
category: String(row.category || 'general').trim() || 'general',
|
|
status: String(row.status || 'queued').trim() || 'queued',
|
|
createdAt: row.created_at ? new Date(row.created_at).toISOString() : '',
|
|
startedAt: row.started_at ? new Date(row.started_at).toISOString() : '',
|
|
finishedAt: row.finished_at ? new Date(row.finished_at).toISOString() : '',
|
|
errorMessage: String(row.error_message || ''),
|
|
attempts: Math.max(0, Number(row.attempts) || 0),
|
|
metadata: parseJsonValue(row.metadata_json, {}),
|
|
payload: parseJsonValue(row.payload_json, null)
|
|
};
|
|
}
|
|
|
|
function buildTaskRecord(task) {
|
|
return {
|
|
task_key: task.key || null,
|
|
task_type: task.taskType || null,
|
|
title: task.title,
|
|
category: task.category || 'general',
|
|
status: task.status,
|
|
payload_json: stringifyJsonValue(task.payload),
|
|
metadata_json: stringifyJsonValue(task.metadata),
|
|
attempts: Math.max(0, Number(task.attempts) || 0),
|
|
created_at: task.createdAt ? new Date(task.createdAt) : new Date(),
|
|
started_at: task.startedAt ? new Date(task.startedAt) : null,
|
|
finished_at: task.finishedAt ? new Date(task.finishedAt) : null,
|
|
error_message: task.errorMessage || null
|
|
};
|
|
}
|
|
|
|
async function persistTaskInsert(task) {
|
|
const normalizedTaskType = normalizeText(task && task.taskType);
|
|
if (!normalizedTaskType) {
|
|
throw new Error('Background tasks require a task type.');
|
|
}
|
|
|
|
const record = buildTaskRecord(Object.assign({}, task, {
|
|
taskType: normalizedTaskType
|
|
}));
|
|
const [result] = await pool.query(
|
|
`INSERT INTO o_background_tasks (
|
|
task_key,
|
|
task_type,
|
|
title,
|
|
category,
|
|
status,
|
|
payload_json,
|
|
metadata_json,
|
|
attempts,
|
|
created_at,
|
|
started_at,
|
|
finished_at,
|
|
error_message
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
[
|
|
record.task_key,
|
|
record.task_type,
|
|
record.title,
|
|
record.category,
|
|
record.status,
|
|
record.payload_json,
|
|
record.metadata_json,
|
|
record.attempts,
|
|
record.created_at,
|
|
record.started_at,
|
|
record.finished_at,
|
|
record.error_message
|
|
]
|
|
);
|
|
|
|
task.id = Number(result.insertId);
|
|
return task;
|
|
}
|
|
|
|
async function persistTaskUpdate(task) {
|
|
if (!task || !Number.isInteger(Number(task.id)) || Number(task.id) <= 0) {
|
|
return;
|
|
}
|
|
|
|
const record = buildTaskRecord(task);
|
|
await pool.query(
|
|
`UPDATE o_background_tasks
|
|
SET task_key = ?, task_type = ?, title = ?, category = ?, status = ?, payload_json = ?, metadata_json = ?, attempts = ?, started_at = ?, finished_at = ?, error_message = ?
|
|
WHERE id = ?`,
|
|
[
|
|
record.task_key,
|
|
record.task_type,
|
|
record.title,
|
|
record.category,
|
|
record.status,
|
|
record.payload_json,
|
|
record.metadata_json,
|
|
record.attempts,
|
|
record.started_at,
|
|
record.finished_at,
|
|
record.error_message,
|
|
Number(task.id)
|
|
]
|
|
);
|
|
}
|
|
|
|
async function persistTaskDelete(taskId) {
|
|
const numericTaskId = Number(taskId);
|
|
if (!Number.isFinite(numericTaskId) || numericTaskId <= 0) {
|
|
return;
|
|
}
|
|
|
|
await pool.query('DELETE FROM o_background_tasks WHERE id = ?', [numericTaskId]);
|
|
}
|
|
|
|
async function fetchTaskById(taskId) {
|
|
const numericTaskId = Number(taskId);
|
|
if (!Number.isFinite(numericTaskId) || numericTaskId <= 0) {
|
|
return null;
|
|
}
|
|
|
|
const [rows] = await pool.query(
|
|
`SELECT id, task_key, task_type, title, category, status, payload_json, metadata_json, attempts, created_at, started_at, finished_at, error_message
|
|
FROM o_background_tasks
|
|
WHERE id = ?
|
|
LIMIT 1`,
|
|
[numericTaskId]
|
|
);
|
|
|
|
return hydrateTaskRow(rows && rows[0]) || null;
|
|
}
|
|
|
|
async function fetchQueuedTaskCount() {
|
|
const [rows] = await pool.query(`SELECT COUNT(*) AS count FROM o_background_tasks WHERE status = 'queued'`);
|
|
return Number(rows && rows[0] && rows[0].count) || 0;
|
|
}
|
|
|
|
async function claimNextTask() {
|
|
const connection = await pool.getConnection();
|
|
try {
|
|
await connection.beginTransaction();
|
|
const [rows] = await connection.query(
|
|
`SELECT id, task_key, task_type, title, category, status, payload_json, metadata_json, attempts, created_at, started_at, finished_at, error_message
|
|
FROM o_background_tasks
|
|
WHERE status = 'queued'
|
|
ORDER BY created_at ASC, id ASC
|
|
LIMIT 1
|
|
FOR UPDATE SKIP LOCKED`
|
|
);
|
|
|
|
const row = rows && rows[0] ? rows[0] : null;
|
|
if (!row) {
|
|
await connection.commit();
|
|
return null;
|
|
}
|
|
|
|
const nextAttempts = Math.max(0, Number(row.attempts) || 0) + 1;
|
|
await connection.query(
|
|
`UPDATE o_background_tasks
|
|
SET status = 'running', started_at = NOW(), attempts = ?, error_message = NULL
|
|
WHERE id = ?`,
|
|
[nextAttempts, row.id]
|
|
);
|
|
await connection.commit();
|
|
|
|
return hydrateTaskRow(Object.assign({}, row, {
|
|
status: 'running',
|
|
started_at: new Date(),
|
|
attempts: nextAttempts,
|
|
error_message: null
|
|
}));
|
|
} catch (error) {
|
|
try {
|
|
await connection.rollback();
|
|
} catch (_rollbackError) {
|
|
// Ignore rollback failures and surface the original error.
|
|
}
|
|
throw error;
|
|
} finally {
|
|
connection.release();
|
|
}
|
|
}
|
|
|
|
async function syncRecurringTaskState(task, status, errorMessage) {
|
|
const recurringKey = String(task && task.metadata && task.metadata.recurringKey || '').trim();
|
|
if (!recurringKey) {
|
|
return;
|
|
}
|
|
|
|
const job = recurringJobsByKey.get(recurringKey);
|
|
if (!job) {
|
|
return;
|
|
}
|
|
|
|
job.lastRunAt = toIsoDate(new Date());
|
|
job.lastStatus = String(status || '');
|
|
job.lastError = errorMessage ? String(errorMessage) : '';
|
|
}
|
|
|
|
function setTaskHandler(taskType, handler) {
|
|
const normalizedTaskType = normalizeText(taskType);
|
|
if (!normalizedTaskType || typeof handler !== 'function') {
|
|
return false;
|
|
}
|
|
|
|
taskHandlers.set(normalizedTaskType, handler);
|
|
return true;
|
|
}
|
|
|
|
setTaskHandler('recurring-run', async function (task) {
|
|
const recurringKey = String(task && task.metadata && task.metadata.recurringKey || '').trim();
|
|
if (!recurringKey) {
|
|
throw new Error('Recurring task key is required.');
|
|
}
|
|
|
|
const job = recurringJobsByKey.get(recurringKey);
|
|
if (!job || typeof job.run !== 'function') {
|
|
throw new Error('Recurring task not found.');
|
|
}
|
|
|
|
return job.run(task);
|
|
});
|
|
|
|
function scheduleDrain() {
|
|
if (drainScheduled) {
|
|
return;
|
|
}
|
|
|
|
drainScheduled = true;
|
|
setTimeout(function () {
|
|
drainScheduled = false;
|
|
processQueue().catch(function (error) {
|
|
console.error(error);
|
|
});
|
|
}, 0);
|
|
}
|
|
|
|
async function initialize() {
|
|
if (initializationPromise) {
|
|
return initializationPromise;
|
|
}
|
|
|
|
initializationPromise = (async function () {
|
|
await pool.query(
|
|
`UPDATE o_background_tasks
|
|
SET status = 'queued', started_at = NULL, finished_at = NULL, error_message = NULL
|
|
WHERE status = 'running'`
|
|
);
|
|
|
|
if (await fetchQueuedTaskCount() > 0) {
|
|
scheduleDrain();
|
|
}
|
|
})();
|
|
|
|
return initializationPromise;
|
|
}
|
|
|
|
async function processQueue() {
|
|
while (activeCount < maxConcurrent) {
|
|
const task = await claimNextTask();
|
|
if (!task) {
|
|
break;
|
|
}
|
|
|
|
activeCount += 1;
|
|
|
|
(async function () {
|
|
try {
|
|
const handler = taskHandlers.get(task.taskType);
|
|
if (!handler) {
|
|
throw new Error('No handler registered for task type ' + task.taskType + '.');
|
|
}
|
|
|
|
await handler(task);
|
|
task.status = 'completed';
|
|
task.finishedAt = toIsoDate(new Date());
|
|
task.errorMessage = '';
|
|
await persistTaskUpdate(task);
|
|
await syncRecurringTaskState(task, task.status, '');
|
|
} catch (error) {
|
|
task.status = 'failed';
|
|
task.errorMessage = String(error && error.message ? error.message : 'Background task failed.');
|
|
task.finishedAt = toIsoDate(new Date());
|
|
try {
|
|
await persistTaskUpdate(task);
|
|
} catch (persistError) {
|
|
console.error(persistError);
|
|
}
|
|
await syncRecurringTaskState(task, task.status, task.errorMessage);
|
|
} finally {
|
|
activeCount = Math.max(0, activeCount - 1);
|
|
scheduleDrain();
|
|
}
|
|
})().catch(function (error) {
|
|
console.error(error);
|
|
});
|
|
}
|
|
}
|
|
|
|
async function enqueueTask(definition) {
|
|
const normalizedKey = normalizeText(definition && definition.key);
|
|
const normalizedTitle = normalizeText(definition && definition.title) || 'Background task';
|
|
const normalizedTaskType = normalizeText(definition && definition.taskType);
|
|
if (!normalizedTaskType) {
|
|
throw new Error('Background tasks require a task type.');
|
|
}
|
|
|
|
const task = {
|
|
id: 0,
|
|
key: normalizedKey,
|
|
taskType: normalizedTaskType,
|
|
title: normalizedTitle,
|
|
category: normalizeText(definition && definition.category) || 'general',
|
|
status: 'queued',
|
|
createdAt: toIsoDate(new Date()),
|
|
startedAt: '',
|
|
finishedAt: '',
|
|
errorMessage: '',
|
|
attempts: 0,
|
|
metadata: definition && definition.metadata ? definition.metadata : {},
|
|
payload: definition && definition.payload !== undefined ? definition.payload : null
|
|
};
|
|
|
|
await persistTaskInsert(task);
|
|
scheduleDrain();
|
|
return buildSnapshot(task);
|
|
}
|
|
|
|
async function enqueueTaskAndWait(definition) {
|
|
const snapshot = await enqueueTask(definition);
|
|
const taskId = snapshot && snapshot.id ? snapshot.id : null;
|
|
if (!taskId) {
|
|
return snapshot;
|
|
}
|
|
|
|
while (true) {
|
|
const currentTask = await fetchTaskById(taskId);
|
|
if (!currentTask) {
|
|
return snapshot;
|
|
}
|
|
|
|
if (currentTask.status !== 'queued' && currentTask.status !== 'running') {
|
|
return buildSnapshot(currentTask);
|
|
}
|
|
|
|
await new Promise(function (resolve) {
|
|
setTimeout(resolve, 250);
|
|
});
|
|
}
|
|
}
|
|
|
|
function clearRecurringTimer(job) {
|
|
if (job && job.timerId) {
|
|
clearTimeout(job.timerId);
|
|
job.timerId = null;
|
|
}
|
|
}
|
|
|
|
function scheduleRecurringRun(job, delayMs) {
|
|
if (!job || job.enabled === false) {
|
|
return;
|
|
}
|
|
|
|
clearRecurringTimer(job);
|
|
const safeDelay = Math.max(1, Number(delayMs) || job.intervalMs || 0);
|
|
job.nextRunAt = toIsoDate(new Date(Date.now() + safeDelay));
|
|
job.timerId = setTimeout(function () {
|
|
job.timerId = null;
|
|
triggerRecurringJob(job.key).catch(function (error) {
|
|
console.error(error);
|
|
});
|
|
}, safeDelay);
|
|
}
|
|
|
|
async function triggerRecurringJob(recurringKey) {
|
|
const job = recurringJobsByKey.get(recurringKey);
|
|
if (!job || job.enabled === false) {
|
|
return false;
|
|
}
|
|
|
|
await enqueueTask({
|
|
key: `${job.key}:${Date.now()}`,
|
|
title: job.title,
|
|
category: job.category,
|
|
taskType: 'recurring-run',
|
|
metadata: Object.assign({}, job.metadata || {}, {
|
|
recurringKey: job.key,
|
|
recurringTitle: job.title
|
|
}),
|
|
payload: Object.assign({}, job.payload || {}, {
|
|
recurringKey: job.key
|
|
})
|
|
});
|
|
|
|
scheduleRecurringRun(job, job.intervalMs);
|
|
return true;
|
|
}
|
|
|
|
async function runRecurringTask(recurringKey) {
|
|
const normalizedKey = normalizeText(recurringKey);
|
|
if (!normalizedKey || !recurringJobsByKey.has(normalizedKey)) {
|
|
return false;
|
|
}
|
|
|
|
return triggerRecurringJob(normalizedKey);
|
|
}
|
|
|
|
function registerRecurringTask(definition) {
|
|
const normalizedKey = normalizeText(definition && definition.key);
|
|
if (!normalizedKey) {
|
|
throw new Error('Recurring tasks require a key.');
|
|
}
|
|
|
|
const intervalMs = Math.max(1000, Number(definition && definition.intervalMs) || 0);
|
|
if (!Number.isFinite(intervalMs) || intervalMs < 1000) {
|
|
throw new Error('Recurring tasks require a valid interval.');
|
|
}
|
|
|
|
const job = recurringJobsByKey.get(normalizedKey) || {
|
|
key: normalizedKey,
|
|
lastRunAt: '',
|
|
lastStatus: '',
|
|
lastError: '',
|
|
nextRunAt: '',
|
|
timerId: null,
|
|
enabled: true
|
|
};
|
|
|
|
clearRecurringTimer(job);
|
|
job.title = normalizeText(definition && definition.title) || 'Background task';
|
|
job.category = normalizeText(definition && definition.category) || 'general';
|
|
job.intervalMs = intervalMs;
|
|
job.metadata = definition && definition.metadata ? definition.metadata : {};
|
|
job.payload = definition && definition.payload ? definition.payload : null;
|
|
job.run = typeof definition.run === 'function' ? definition.run : function () {
|
|
return Promise.resolve();
|
|
};
|
|
job.enabled = definition && definition.enabled === false ? false : true;
|
|
recurringJobsByKey.set(normalizedKey, job);
|
|
|
|
if (job.enabled) {
|
|
scheduleRecurringRun(job, intervalMs);
|
|
}
|
|
|
|
return buildRecurringSnapshot(job);
|
|
}
|
|
|
|
function removeRecurringTask(recurringKey) {
|
|
const normalizedKey = normalizeText(recurringKey);
|
|
const job = recurringJobsByKey.get(normalizedKey);
|
|
if (!job) {
|
|
return false;
|
|
}
|
|
|
|
clearRecurringTimer(job);
|
|
recurringJobsByKey.delete(normalizedKey);
|
|
return true;
|
|
}
|
|
|
|
function buildRecurringSnapshot(job, state) {
|
|
const latestState = state || null;
|
|
const activeTaskId = latestState && (latestState.status === 'queued' || latestState.status === 'running')
|
|
? latestState.id
|
|
: null;
|
|
const lastRunAt = latestState
|
|
? latestState.finishedAt || latestState.startedAt || latestState.createdAt || ''
|
|
: job.lastRunAt || '';
|
|
|
|
return {
|
|
key: job.key,
|
|
title: job.title,
|
|
category: job.category,
|
|
intervalMs: job.intervalMs,
|
|
enabled: job.enabled !== false,
|
|
activeTaskId: activeTaskId,
|
|
createdAt: latestState ? latestState.createdAt || '' : job.createdAt || '',
|
|
nextRunAt: job.nextRunAt || '',
|
|
lastRunAt: lastRunAt || '',
|
|
lastStatus: latestState ? latestState.status || '' : job.lastStatus || '',
|
|
lastError: latestState ? latestState.errorMessage || '' : job.lastError || '',
|
|
metadata: job.metadata || {}
|
|
};
|
|
}
|
|
|
|
async function fetchRecurringTaskStates() {
|
|
const [rows] = await pool.query(
|
|
`SELECT id, status, created_at, started_at, finished_at, error_message, metadata_json
|
|
FROM o_background_tasks
|
|
WHERE JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.recurringKey')) IS NOT NULL
|
|
ORDER BY created_at DESC, id DESC`
|
|
);
|
|
|
|
const statesByKey = new Map();
|
|
for (const row of rows || []) {
|
|
const metadata = parseJsonValue(row.metadata_json, {});
|
|
const recurringKey = normalizeText(metadata && metadata.recurringKey);
|
|
if (!recurringKey || statesByKey.has(recurringKey)) {
|
|
continue;
|
|
}
|
|
|
|
statesByKey.set(recurringKey, {
|
|
id: Number(row.id),
|
|
status: String(row.status || ''),
|
|
createdAt: row.created_at ? new Date(row.created_at).toISOString() : '',
|
|
startedAt: row.started_at ? new Date(row.started_at).toISOString() : '',
|
|
finishedAt: row.finished_at ? new Date(row.finished_at).toISOString() : '',
|
|
errorMessage: String(row.error_message || '')
|
|
});
|
|
}
|
|
|
|
return statesByKey;
|
|
}
|
|
|
|
async function listTasks() {
|
|
const [rows] = await pool.query(
|
|
`SELECT id, task_key, task_type, title, category, status, payload_json, metadata_json, attempts, created_at, started_at, finished_at, error_message
|
|
FROM o_background_tasks
|
|
ORDER BY
|
|
CASE status
|
|
WHEN 'running' THEN 0
|
|
WHEN 'queued' THEN 1
|
|
WHEN 'failed' THEN 2
|
|
WHEN 'completed' THEN 3
|
|
WHEN 'canceled' THEN 4
|
|
ELSE 5
|
|
END,
|
|
COALESCE(started_at, created_at) DESC,
|
|
id DESC`
|
|
);
|
|
|
|
return (rows || []).map(hydrateTaskRow);
|
|
}
|
|
|
|
async function listRecurringTasks() {
|
|
const statesByKey = await fetchRecurringTaskStates();
|
|
return Array.from(recurringJobsByKey.values())
|
|
.slice()
|
|
.sort(function (left, right) {
|
|
return left.key.localeCompare(right.key);
|
|
})
|
|
.map(function (job) {
|
|
return buildRecurringSnapshot(job, statesByKey.get(job.key) || null);
|
|
});
|
|
}
|
|
|
|
async function clearFinishedTasks() {
|
|
const [result] = await pool.query(
|
|
`DELETE FROM o_background_tasks
|
|
WHERE status IN ('completed', 'failed', 'canceled')`
|
|
);
|
|
return Number(result && result.affectedRows) || 0;
|
|
}
|
|
|
|
async function cancelTask(taskId) {
|
|
const numericTaskId = Number(taskId);
|
|
if (!Number.isFinite(numericTaskId) || numericTaskId <= 0) {
|
|
return false;
|
|
}
|
|
|
|
const [result] = await pool.query(
|
|
`UPDATE o_background_tasks
|
|
SET status = 'canceled',
|
|
finished_at = NOW(),
|
|
error_message = 'Task canceled.'
|
|
WHERE id = ?
|
|
AND status = 'queued'`,
|
|
[numericTaskId]
|
|
);
|
|
return Number(result && result.affectedRows) > 0;
|
|
}
|
|
|
|
async function retryTask(taskId) {
|
|
const task = await fetchTaskById(taskId);
|
|
if (!task || task.status !== 'failed') {
|
|
return null;
|
|
}
|
|
|
|
return enqueueTask({
|
|
key: task.key,
|
|
title: task.title,
|
|
category: task.category,
|
|
taskType: task.taskType,
|
|
metadata: task.metadata,
|
|
payload: task.payload
|
|
});
|
|
}
|
|
|
|
async function getSummary() {
|
|
const [rows] = await pool.query(
|
|
`SELECT
|
|
SUM(status = 'queued') AS queued,
|
|
SUM(status = 'running') AS running,
|
|
SUM(status = 'completed') AS completed,
|
|
SUM(status = 'failed') AS failed,
|
|
SUM(status = 'canceled') AS canceled,
|
|
COUNT(*) AS total
|
|
FROM o_background_tasks`
|
|
);
|
|
const row = rows && rows[0] ? rows[0] : {};
|
|
|
|
return {
|
|
activeCount: activeCount,
|
|
counts: {
|
|
queued: Number(row.queued) || 0,
|
|
running: Number(row.running) || 0,
|
|
completed: Number(row.completed) || 0,
|
|
failed: Number(row.failed) || 0,
|
|
canceled: Number(row.canceled) || 0
|
|
},
|
|
scheduledCount: recurringJobsByKey.size,
|
|
total: Number(row.total) || 0
|
|
};
|
|
}
|
|
|
|
return {
|
|
enqueueTask: enqueueTask,
|
|
enqueueTaskAndWait: enqueueTaskAndWait,
|
|
initialize: initialize,
|
|
setTaskHandler: setTaskHandler,
|
|
registerRecurringTask: registerRecurringTask,
|
|
removeRecurringTask: removeRecurringTask,
|
|
runRecurringTask: runRecurringTask,
|
|
listTasks: listTasks,
|
|
listRecurringTasks: listRecurringTasks,
|
|
getTaskById: fetchTaskById,
|
|
getSummary: getSummary,
|
|
cancelTask: cancelTask,
|
|
retryTask: retryTask,
|
|
clearFinishedTasks: clearFinishedTasks
|
|
};
|
|
}
|
|
|
|
module.exports = {
|
|
createBackgroundTaskQueue: createBackgroundTaskQueue,
|
|
normalizeIntervalMs: normalizeIntervalMs
|
|
}; |