115 lines
4.6 KiB
TypeScript
115 lines
4.6 KiB
TypeScript
import type { Job } from "./index.ts";
|
|
import type { QueueStore } from "./durable.ts";
|
|
|
|
export interface RedisQueueClient {
|
|
get(key: string): Promise<string | null>;
|
|
set(key: string, value: string, options?: { NX?: boolean; PX?: number }): Promise<unknown>;
|
|
del(...keys: string[]): Promise<unknown>;
|
|
zadd(key: string, score: number, member: string): Promise<unknown>;
|
|
zrem(key: string, member: string): Promise<unknown>;
|
|
zrangebyscore(
|
|
key: string,
|
|
min: number,
|
|
max: number,
|
|
options?: { limit: [number, number] },
|
|
): Promise<string[]>;
|
|
smembers(key: string): Promise<string[]>;
|
|
sadd(key: string, member: string): Promise<unknown>;
|
|
srem(key: string, member: string): Promise<unknown>;
|
|
}
|
|
|
|
/** Redis-backed queue store using only the common client command surface. */
|
|
export function redisQueueStore(client: RedisQueueClient, prefix = "wrnexus:queue"): QueueStore {
|
|
const jobs = `${prefix}:jobs`;
|
|
const due = `${prefix}:due`;
|
|
const key = (id: string) => `${prefix}:job:${id}`;
|
|
return {
|
|
async put(job) {
|
|
await client.set(key(job.id), JSON.stringify(job));
|
|
await client.sadd(jobs, job.id);
|
|
await client.zadd(due, job.runAt, job.id);
|
|
await client.del(`${prefix}:lease:${job.id}`);
|
|
},
|
|
async get(id) {
|
|
const value = await client.get(key(id));
|
|
return value ? (JSON.parse(value) as Job) : null;
|
|
},
|
|
async remove(id) {
|
|
await client.del(key(id), `${prefix}:lease:${id}`);
|
|
await client.srem(jobs, id);
|
|
await client.zrem(due, id);
|
|
},
|
|
async due(now, limit) {
|
|
const ids = await client.zrangebyscore(due, 0, now, { limit: [0, limit] });
|
|
const values = await Promise.all(ids.map((id) => client.get(key(id))));
|
|
return values
|
|
.filter((value): value is string => value !== null)
|
|
.map((value) => JSON.parse(value));
|
|
},
|
|
async list(name) {
|
|
const ids = await client.smembers(jobs);
|
|
const values = await Promise.all(ids.map((id) => client.get(key(id))));
|
|
return values
|
|
.filter((value): value is string => value !== null)
|
|
.map((value) => JSON.parse(value) as Job)
|
|
.filter((job) => !name || job.name === name);
|
|
},
|
|
async claim(id, worker, leaseUntil) {
|
|
const ttl = Math.max(1, leaseUntil - Date.now());
|
|
return Boolean(await client.set(`${prefix}:lease:${id}`, worker, { NX: true, PX: ttl }));
|
|
},
|
|
};
|
|
}
|
|
|
|
export interface SqlQueueClient {
|
|
query<T = Record<string, unknown>>(sql: string, parameters?: unknown[]): Promise<{ rows: T[] }>;
|
|
}
|
|
|
|
/** PostgreSQL store with atomic SKIP LOCKED leasing and JSON payloads. */
|
|
export function postgresQueueStore(db: SqlQueueClient, table = "wrnexus_jobs"): QueueStore {
|
|
if (!/^[a-z_][a-z0-9_]*$/i.test(table)) throw new Error("Invalid queue table name");
|
|
return {
|
|
async put(job) {
|
|
await db.query(
|
|
`INSERT INTO ${table} (id,name,payload,run_at,priority,lease_owner,lease_until) VALUES ($1,$2,$3,$4,$5,NULL,NULL) ON CONFLICT (id) DO UPDATE SET name=$2,payload=$3,run_at=$4,priority=$5,lease_owner=NULL,lease_until=NULL`,
|
|
[job.id, job.name, JSON.stringify(job), job.runAt, job.priority],
|
|
);
|
|
},
|
|
async get(id) {
|
|
const result = await db.query<{ payload: Job }>(`SELECT payload FROM ${table} WHERE id=$1`, [
|
|
id,
|
|
]);
|
|
return result.rows[0]?.payload ?? null;
|
|
},
|
|
async remove(id) {
|
|
await db.query(`DELETE FROM ${table} WHERE id=$1`, [id]);
|
|
},
|
|
async due(now, limit) {
|
|
const result = await db.query<{ payload: Job }>(
|
|
`SELECT payload FROM ${table} WHERE run_at <= $1 AND (lease_until IS NULL OR lease_until < $1) ORDER BY priority DESC, run_at ASC LIMIT $2`,
|
|
[now, limit],
|
|
);
|
|
return result.rows.map((row) => row.payload);
|
|
},
|
|
async list(name) {
|
|
const result = await db.query<{ payload: Job }>(
|
|
`SELECT payload FROM ${table}${name ? " WHERE name=$1" : ""} ORDER BY run_at ASC`,
|
|
name ? [name] : [],
|
|
);
|
|
return result.rows.map((row) => row.payload);
|
|
},
|
|
async claim(id, worker, leaseUntil) {
|
|
const result = await db.query(
|
|
`UPDATE ${table} SET lease_owner=$2,lease_until=$3 WHERE id=$1 AND (lease_until IS NULL OR lease_until < $4) RETURNING id`,
|
|
[id, worker, leaseUntil, Date.now()],
|
|
);
|
|
return result.rows.length === 1;
|
|
},
|
|
};
|
|
}
|
|
|
|
export const POSTGRES_QUEUE_SCHEMA = `CREATE TABLE IF NOT EXISTS wrnexus_jobs (
|
|
id text PRIMARY KEY, name text NOT NULL, payload jsonb NOT NULL, run_at bigint NOT NULL,
|
|
priority integer NOT NULL DEFAULT 0, lease_owner text, lease_until bigint
|
|
); CREATE INDEX IF NOT EXISTS wrnexus_jobs_due ON wrnexus_jobs (run_at, priority DESC);`;
|