aboutsummaryrefslogtreecommitdiffstats
path: root/packages/plugins-queue-liteque
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
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')
-rw-r--r--packages/plugins-queue-liteque/package.json5
-rw-r--r--packages/plugins-queue-liteque/src/index.ts9
2 files changed, 9 insertions, 5 deletions
diff --git a/packages/plugins-queue-liteque/package.json b/packages/plugins-queue-liteque/package.json
index a31c9707..bb4b1aac 100644
--- a/packages/plugins-queue-liteque/package.json
+++ b/packages/plugins-queue-liteque/package.json
@@ -6,8 +6,8 @@
"type": "module",
"scripts": {
"typecheck": "tsc --noEmit",
- "format": "prettier . --ignore-path ../../.prettierignore",
- "format:fix": "prettier . --write --ignore-path ../../.prettierignore",
+ "format": "prettier . --cache --ignore-path ../../.prettierignore --check",
+ "format:fix": "prettier . --cache --ignore-path ../../.prettierignore --write",
"lint": "oxlint .",
"lint:fix": "oxlint . --fix",
"test": "vitest"
@@ -24,4 +24,3 @@
},
"prettier": "@karakeep/prettier-config"
}
-
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;
}