87 lines
3.2 KiB
TypeScript
87 lines
3.2 KiB
TypeScript
import { resolve } from "node:path";
|
|
import { loadAppConfig } from "@wrnexus/styles";
|
|
import { connectFromConfig } from "@wrnexus/db/connect";
|
|
import { installSqliteQueueSchema } from "@wrnexus/queue";
|
|
|
|
export async function runQueueCommand(root: string, command: string, args: readonly string[]) {
|
|
const appRoot = resolve(root);
|
|
const config = await loadAppConfig(appRoot, process.env.WRNEXUS_PROFILE);
|
|
const selected = config.queue?.databaseName ?? "default";
|
|
const dbConfig = selected === "default" ? config.db : config.databases?.[selected];
|
|
const effective =
|
|
config.queue?.storage === "database"
|
|
? dbConfig
|
|
: { driver: "sqlite" as const, url: "file:./.wrnexus/queue.sqlite" };
|
|
if (!effective || effective.driver === "mongo")
|
|
throw new Error(`Queue database '${selected}' is not configured as SQL.`);
|
|
const db = connectFromConfig(effective, appRoot);
|
|
if (db.driver.dialect !== "sqlite")
|
|
throw new Error("Queue CLI currently requires SQLite configured storage.");
|
|
const table = config.queue?.table ?? "wrnexus_jobs";
|
|
await installSqliteQueueSchema(db, table);
|
|
try {
|
|
const id = args.find((value) => !value.startsWith("--"));
|
|
if (command === "list") {
|
|
console.table(
|
|
await db.all(
|
|
`SELECT id,name,run_at,priority,lease_owner,lease_until FROM ${table} ORDER BY run_at`,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
if (command === "status") {
|
|
const pending = await db.one<{ total: number }>(`SELECT COUNT(*) AS total FROM ${table}`);
|
|
const history = await db.one<{ total: number }>(
|
|
`SELECT COUNT(*) AS total FROM ${table}_history`,
|
|
);
|
|
console.log(
|
|
JSON.stringify(
|
|
{ pending: Number(pending?.total ?? 0), history: Number(history?.total ?? 0) },
|
|
null,
|
|
2,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
if (!id && command !== "clear") throw new Error(`wrnexus queue ${command} requires a job id`);
|
|
if (command === "inspect") {
|
|
console.log(
|
|
JSON.stringify(await db.one(`SELECT * FROM ${table} WHERE id = ?`, [id]), null, 2),
|
|
);
|
|
return;
|
|
}
|
|
if (command === "cancel") {
|
|
await db.exec(`DELETE FROM ${table} WHERE id = ?`, [id]);
|
|
return;
|
|
}
|
|
if (command === "retry") {
|
|
const row = await db.one<{ payload: string }>(
|
|
`SELECT payload FROM ${table}_history WHERE id = ? AND state = 'failed'`,
|
|
[id],
|
|
);
|
|
if (!row) throw new Error(`Failed job '${id}' was not found.`);
|
|
const job = JSON.parse(row.payload) as {
|
|
name: string;
|
|
runAt: number;
|
|
priority: number;
|
|
idempotencyKey?: string;
|
|
};
|
|
job.runAt = Date.now();
|
|
await db.exec(
|
|
`INSERT INTO ${table} (id,name,payload,run_at,priority,idempotency_key) VALUES (?,?,?,?,?,?)`,
|
|
[id, job.name, JSON.stringify(job), job.runAt, job.priority, job.idempotencyKey ?? null],
|
|
);
|
|
return;
|
|
}
|
|
if (command === "clear" && args.includes("--completed")) {
|
|
await db.exec(`DELETE FROM ${table}_history WHERE state = 'completed'`);
|
|
return;
|
|
}
|
|
throw new Error(
|
|
"Use queue list | status | inspect <id> | retry <id> | cancel <id> | clear --completed",
|
|
);
|
|
} finally {
|
|
await db.close();
|
|
}
|
|
}
|