107 lines
3.4 KiB
TypeScript
107 lines
3.4 KiB
TypeScript
import { getDb, hasDb, registerDb, type Db } from "@wrnexus/db";
|
|
import { connectFromConfig } from "@wrnexus/db/connect";
|
|
import { memoryQueueStore, type QueueStore } from "./durable.ts";
|
|
import { installSqliteQueueSchema, sqliteQueueStore } from "./sqlite.ts";
|
|
import { databaseQueueStore, installDatabaseQueueSchema } from "./database.ts";
|
|
|
|
export type QueueStorage = "sqlite" | "database" | "memory";
|
|
|
|
export interface QueueStorageConfig {
|
|
/** Durable SQLite is the default. Use memory only for disposable/test queues. */
|
|
storage?: QueueStorage;
|
|
/** Named `databases` connection, or `default` for the app's `db`. */
|
|
databaseName?: string;
|
|
table?: string;
|
|
/** Used internally to resolve the default `.wrnexus/queue.sqlite` path. */
|
|
appRoot?: string;
|
|
}
|
|
|
|
let configured: QueueStorageConfig = { storage: "sqlite" };
|
|
|
|
export function configureQueueStorage(config: QueueStorageConfig = {}): void {
|
|
configured = { storage: "sqlite", ...config };
|
|
}
|
|
|
|
export function queueStorageConfig(): Readonly<QueueStorageConfig> {
|
|
return { ...configured };
|
|
}
|
|
|
|
function lazyStore(resolve: () => Promise<QueueStore>): QueueStore {
|
|
let pending: Promise<QueueStore> | undefined;
|
|
const ready = () => (pending ??= resolve());
|
|
return {
|
|
async put(job) {
|
|
return (await ready()).put(job);
|
|
},
|
|
async get(id) {
|
|
return (await ready()).get(id);
|
|
},
|
|
async remove(id) {
|
|
return (await ready()).remove(id);
|
|
},
|
|
async due(now, limit) {
|
|
return (await ready()).due(now, limit);
|
|
},
|
|
async list(name) {
|
|
return (await ready()).list(name);
|
|
},
|
|
async findByIdempotencyKey(name, key) {
|
|
return (await ready()).findByIdempotencyKey(name, key);
|
|
},
|
|
async size() {
|
|
return (await ready()).size?.() ?? (await (await ready()).list()).length;
|
|
},
|
|
async claim(id, worker, leaseUntil, now) {
|
|
return (await ready()).claim?.(id, worker, leaseUntil, now) ?? true;
|
|
},
|
|
async release(id, worker) {
|
|
await (await ready()).release?.(id, worker);
|
|
},
|
|
async archive(record) {
|
|
await (await ready()).archive?.(record);
|
|
},
|
|
async history(id) {
|
|
return (await ready()).history?.(id) ?? [];
|
|
},
|
|
};
|
|
}
|
|
|
|
async function resolveConfiguredStore(): Promise<QueueStore> {
|
|
const config = configured;
|
|
if (config.storage === "memory") return memoryQueueStore();
|
|
|
|
const databaseName = config.databaseName ?? "default";
|
|
let db: Db;
|
|
if (config.storage === "database") {
|
|
if (!hasDb(databaseName)) {
|
|
throw new Error(
|
|
`WRN-QUEUE-DATABASE: database '${databaseName}' is not configured. ` +
|
|
"Add it to db/databases or choose storage: 'sqlite'.",
|
|
);
|
|
}
|
|
db = getDb(databaseName);
|
|
} else {
|
|
if (hasDb("__wrnexus_queue")) db = getDb("__wrnexus_queue");
|
|
else {
|
|
db = connectFromConfig(
|
|
{ driver: "sqlite", url: "file:./.wrnexus/queue.sqlite" },
|
|
config.appRoot ?? process.cwd(),
|
|
);
|
|
registerDb("__wrnexus_queue", db);
|
|
}
|
|
}
|
|
|
|
const table = config.table ?? "wrnexus_jobs";
|
|
if (db.driver.dialect === "sqlite") {
|
|
await installSqliteQueueSchema(db, table);
|
|
return sqliteQueueStore(db, table);
|
|
}
|
|
await installDatabaseQueueSchema(db, table);
|
|
return databaseQueueStore(db, table);
|
|
}
|
|
|
|
/** Lazy global store used by defineQueue; runtime config is read on first operation. */
|
|
export function configuredQueueStore(): QueueStore {
|
|
return lazyStore(resolveConfiguredStore);
|
|
}
|