aboutsummaryrefslogtreecommitdiffstats
path: root/packages/plugins-queue-liteque/src/index.ts
diff options
context:
space:
mode:
authorMohamed Bassem <me@mbassem.com>2025-10-05 07:04:29 +0100
committerGitHub <noreply@github.com>2025-10-05 07:04:29 +0100
commit74a1f7b6b600d4cb53352dde7def374c3125721a (patch)
tree70b79ebae61456f6ff2cb02a37351fa9817fb342 /packages/plugins-queue-liteque/src/index.ts
parent4a580d713621f99abb8baabc9b847ce039d44842 (diff)
downloadkarakeep-74a1f7b6b600d4cb53352dde7def374c3125721a.tar.zst
feat: Restate-based queue plugin (#2011)
* WIP: Initial restate integration * add retry * add delay + idempotency * implement concurrency limits * add admin stats * add todos * add id provider * handle onComplete failures * add tests * add pub key and fix logging * add priorities * fail call after retries * more fixes * fix retries left * some refactoring * fix package.json * upgrade sdk * some test cleanups
Diffstat (limited to 'packages/plugins-queue-liteque/src/index.ts')
-rw-r--r--packages/plugins-queue-liteque/src/index.ts9
1 files changed, 7 insertions, 2 deletions
diff --git a/packages/plugins-queue-liteque/src/index.ts b/packages/plugins-queue-liteque/src/index.ts
index 16e6e20a..ddc2181c 100644
--- a/packages/plugins-queue-liteque/src/index.ts
+++ b/packages/plugins-queue-liteque/src/index.ts
@@ -24,6 +24,7 @@ class LitequeQueueWrapper<T> implements Queue<T> {
constructor(
private readonly _name: string,
private readonly lq: LQ<T>,
+ public readonly opts: QueueOptions,
) {}
name(): string {
@@ -60,10 +61,14 @@ class LitequeQueueClient implements QueueClient {
private queues = new Map<string, LitequeQueueWrapper<unknown>>();
- async init(): Promise<void> {
+ async prepare(): Promise<void> {
migrateDB(this.db);
}
+ async start(): Promise<void> {
+ // No-op for sqlite
+ }
+
createQueue<T>(name: string, options: QueueOptions): Queue<T> {
if (this.queues.has(name)) {
throw new Error(`Queue ${name} already exists`);
@@ -72,7 +77,7 @@ class LitequeQueueClient implements QueueClient {
defaultJobArgs: { numRetries: options.defaultJobArgs.numRetries },
keepFailedJobs: options.keepFailedJobs,
});
- const wrapper = new LitequeQueueWrapper<T>(name, lq);
+ const wrapper = new LitequeQueueWrapper<T>(name, lq, options);
this.queues.set(name, wrapper);
return wrapper;
}