aboutsummaryrefslogtreecommitdiffstats
path: root/packages/plugins/queue-liteque
diff options
context:
space:
mode:
authorMohamed Bassem <me@mbassem.com>2025-11-08 14:50:00 +0000
committerGitHub <noreply@github.com>2025-11-08 14:50:00 +0000
commit99413db0e79a156a1b87eacd3c6a7b83e9df946e (patch)
tree73f0a5fceb507f75f662a109b00beeb3fa6b16fb /packages/plugins/queue-liteque
parent737b03172c2e063ba311c23d6552418bd2ab1955 (diff)
downloadkarakeep-99413db0e79a156a1b87eacd3c6a7b83e9df946e.tar.zst
refactor: consolidate multiple karakeep plugins into one package (#2101)
* refactor: consolidate plugin packages into single plugins directory - Create new `packages/plugins` directory with consolidated package.json - Move queue-liteque, queue-restate, and search-meilisearch to subdirectories - Update imports in packages/shared-server/src/plugins.ts - Remove individual plugin package directories - Update shared-server dependency to use @karakeep/plugins This reduces overhead of maintaining multiple separate packages for plugins. * refactor: consolidate plugin config files to root level - Move .oxlintrc.json to packages/plugins root - Move vitest.config.ts to packages/plugins root - Update vitest config paths to work from root - Remove individual config files from plugin subdirectories This reduces configuration duplication across plugin subdirectories. --------- Co-authored-by: Claude <noreply@anthropic.com>
Diffstat (limited to 'packages/plugins/queue-liteque')
-rw-r--r--packages/plugins/queue-liteque/index.ts10
-rw-r--r--packages/plugins/queue-liteque/src/index.ts137
2 files changed, 147 insertions, 0 deletions
diff --git a/packages/plugins/queue-liteque/index.ts b/packages/plugins/queue-liteque/index.ts
new file mode 100644
index 00000000..c3f7f03b
--- /dev/null
+++ b/packages/plugins/queue-liteque/index.ts
@@ -0,0 +1,10 @@
+// Auto-register the Liteque queue provider when this package is imported
+import { PluginManager, PluginType } from "@karakeep/shared/plugins";
+
+import { LitequeQueueProvider } from "./src";
+
+PluginManager.register({
+ type: PluginType.Queue,
+ name: "Liteque",
+ provider: new LitequeQueueProvider(),
+});
diff --git a/packages/plugins/queue-liteque/src/index.ts b/packages/plugins/queue-liteque/src/index.ts
new file mode 100644
index 00000000..ddc2181c
--- /dev/null
+++ b/packages/plugins/queue-liteque/src/index.ts
@@ -0,0 +1,137 @@
+import path from "node:path";
+import {
+ buildDBClient,
+ SqliteQueue as LQ,
+ Runner as LQRunner,
+ migrateDB,
+} from "liteque";
+
+import type { PluginProvider } from "@karakeep/shared/plugins";
+import type {
+ DequeuedJob,
+ DequeuedJobError,
+ EnqueueOptions,
+ Queue,
+ QueueClient,
+ QueueOptions,
+ Runner,
+ RunnerFuncs,
+ RunnerOptions,
+} from "@karakeep/shared/queueing";
+import serverConfig from "@karakeep/shared/config";
+
+class LitequeQueueWrapper<T> implements Queue<T> {
+ constructor(
+ private readonly _name: string,
+ private readonly lq: LQ<T>,
+ public readonly opts: QueueOptions,
+ ) {}
+
+ name(): string {
+ return this._name;
+ }
+
+ async enqueue(
+ payload: T,
+ options?: EnqueueOptions,
+ ): Promise<string | undefined> {
+ const job = await this.lq.enqueue(payload, options);
+ // liteque returns a Job with numeric id
+ return job ? String(job.id) : undefined;
+ }
+
+ async stats() {
+ return this.lq.stats();
+ }
+
+ async cancelAllNonRunning(): Promise<number> {
+ return this.lq.cancelAllNonRunning();
+ }
+
+ // Internal accessor for runner
+ get _impl(): LQ<T> {
+ return this.lq;
+ }
+}
+
+class LitequeQueueClient implements QueueClient {
+ private db = buildDBClient(path.join(serverConfig.dataDir, "queue.db"), {
+ walEnabled: serverConfig.database.walMode,
+ });
+
+ private queues = new Map<string, LitequeQueueWrapper<unknown>>();
+
+ 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`);
+ }
+ const lq = new LQ<T>(name, this.db, {
+ defaultJobArgs: { numRetries: options.defaultJobArgs.numRetries },
+ keepFailedJobs: options.keepFailedJobs,
+ });
+ const wrapper = new LitequeQueueWrapper<T>(name, lq, options);
+ this.queues.set(name, wrapper);
+ return wrapper;
+ }
+
+ createRunner<T>(
+ queue: Queue<T>,
+ funcs: RunnerFuncs<T>,
+ opts: RunnerOptions<T>,
+ ): Runner<T> {
+ const name = queue.name();
+ let wrapper = this.queues.get(name);
+ if (!wrapper) {
+ throw new Error(`Queue ${name} not found`);
+ }
+
+ const runner = new LQRunner<T>(
+ wrapper._impl,
+ {
+ run: funcs.run,
+ onComplete: funcs.onComplete as
+ | ((job: DequeuedJob<T>) => Promise<void>)
+ | undefined,
+ onError: funcs.onError as
+ | ((job: DequeuedJobError<T>) => Promise<void>)
+ | undefined,
+ },
+ {
+ pollIntervalMs: opts.pollIntervalMs ?? 1000,
+ timeoutSecs: opts.timeoutSecs,
+ concurrency: opts.concurrency,
+ validator: opts.validator,
+ },
+ );
+
+ return {
+ run: () => runner.run(),
+ stop: () => runner.stop(),
+ runUntilEmpty: () => runner.runUntilEmpty(),
+ };
+ }
+
+ async shutdown(): Promise<void> {
+ // No-op for sqlite
+ }
+}
+
+export class LitequeQueueProvider implements PluginProvider<QueueClient> {
+ private client: QueueClient | null = null;
+
+ async getClient(): Promise<QueueClient | null> {
+ if (!this.client) {
+ const client = new LitequeQueueClient();
+ this.client = client;
+ }
+ return this.client;
+ }
+}