62 lines
2.1 KiB
JavaScript
62 lines
2.1 KiB
JavaScript
import dbHandler from "../db-handler";
|
|
import { quoteIdentifier } from "../../utils/sql-identifier";
|
|
import { QUEUE_NOW_SQL, requireQueueTable } from "./queue-utils";
|
|
/**
|
|
* Mark a claimed job as done: deletes it, or sets it to `done` when the
|
|
* table has `keep_completed`.
|
|
*
|
|
* Fails when the lock token no longer matches, i.e. the lease expired and
|
|
* another worker reclaimed the job.
|
|
*/
|
|
export default async function QueueAck({ table, id, lock_token, config, }) {
|
|
try {
|
|
const options = requireQueueTable(table);
|
|
const tableSql = quoteIdentifier(table);
|
|
const claimedWhere = `WHERE id = ? AND queue_status = 'processing' AND queue_lock_token = ?`;
|
|
if (!options.keep_completed) {
|
|
const res = await dbHandler({
|
|
query: `DELETE FROM ${tableSql} ${claimedWhere} RETURNING *`,
|
|
values: [id, lock_token],
|
|
config,
|
|
});
|
|
if (!res.success) {
|
|
return { ...res, success: false };
|
|
}
|
|
if (!res.payload?.length) {
|
|
return {
|
|
...res,
|
|
success: false,
|
|
msg: "Job is not claimed with this lock token",
|
|
};
|
|
}
|
|
return res;
|
|
}
|
|
const res = await dbHandler({
|
|
query: `UPDATE ${tableSql} SET queue_status = 'done', queue_lease_expires_at = NULL, queue_lock_token = NULL, updated_at = ${QUEUE_NOW_SQL} ${claimedWhere}`,
|
|
values: [id, lock_token],
|
|
config,
|
|
});
|
|
if (!res.success) {
|
|
return { ...res, success: false };
|
|
}
|
|
if (!res.db_res?.affectedRows) {
|
|
return {
|
|
...res,
|
|
success: false,
|
|
msg: "Job is not claimed with this lock token",
|
|
};
|
|
}
|
|
return await dbHandler({
|
|
query: `SELECT * FROM ${tableSql} WHERE id = ?`,
|
|
values: [id],
|
|
config,
|
|
});
|
|
}
|
|
catch (error) {
|
|
return {
|
|
success: false,
|
|
error: error.message,
|
|
};
|
|
}
|
|
}
|