From 8d32055485858210252096483bb20533dc8bdf60 Mon Sep 17 00:00:00 2001 From: Mohamed Bassem Date: Sun, 14 Sep 2025 18:16:40 +0000 Subject: refactor: Move callsites to liteque to be behind a plugin --- packages/plugins-queue-liteque/.oxlintrc.json | 19 ++++ packages/plugins-queue-liteque/index.ts | 10 ++ packages/plugins-queue-liteque/package.json | 27 ++++++ packages/plugins-queue-liteque/src/index.ts | 133 ++++++++++++++++++++++++++ packages/plugins-queue-liteque/tsconfig.json | 10 ++ 5 files changed, 199 insertions(+) create mode 100644 packages/plugins-queue-liteque/.oxlintrc.json create mode 100644 packages/plugins-queue-liteque/index.ts create mode 100644 packages/plugins-queue-liteque/package.json create mode 100644 packages/plugins-queue-liteque/src/index.ts create mode 100644 packages/plugins-queue-liteque/tsconfig.json (limited to 'packages/plugins-queue-liteque') diff --git a/packages/plugins-queue-liteque/.oxlintrc.json b/packages/plugins-queue-liteque/.oxlintrc.json new file mode 100644 index 00000000..79ba0255 --- /dev/null +++ b/packages/plugins-queue-liteque/.oxlintrc.json @@ -0,0 +1,19 @@ +{ + "$schema": "../../node_modules/oxlint/configuration_schema.json", + "extends": [ + "../../tooling/oxlint/oxlint-base.json" + ], + "env": { + "builtin": true, + "commonjs": true + }, + "ignorePatterns": [ + "**/*.config.js", + "**/*.config.cjs", + "**/.eslintrc.cjs", + "**/.next", + "**/dist", + "**/build", + "**/pnpm-lock.yaml" + ] +} 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/package.json b/packages/plugins-queue-liteque/package.json new file mode 100644 index 00000000..0a3916c1 --- /dev/null +++ b/packages/plugins-queue-liteque/package.json @@ -0,0 +1,27 @@ +{ + "$schema": "https://json.schemastore.org/package.json", + "name": "@karakeep/plugins-queue-liteque", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "typecheck": "tsc --noEmit", + "format": "prettier . --ignore-path ../../.prettierignore", + "format:fix": "prettier . --write --ignore-path ../../.prettierignore", + "lint": "oxlint .", + "lint:fix": "oxlint . --fix", + "test": "vitest" + }, + "dependencies": { + "@karakeep/shared": "workspace:*", + "liteque": "^0.6.0" + }, + "devDependencies": { + "@karakeep/prettier-config": "workspace:^0.1.0", + "@karakeep/tsconfig": "workspace:^0.1.0", + "vite-tsconfig-paths": "^4.3.1", + "vitest": "^3.2.4" + }, + "prettier": "@karakeep/prettier-config" +} + diff --git a/packages/plugins-queue-liteque/src/index.ts b/packages/plugins-queue-liteque/src/index.ts new file mode 100644 index 00000000..3da161d8 --- /dev/null +++ b/packages/plugins-queue-liteque/src/index.ts @@ -0,0 +1,133 @@ +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 implements Queue { + constructor( + private readonly _name: string, + private readonly lq: LQ, + ) {} + + name(): string { + return this._name; + } + + async enqueue( + payload: T, + options?: EnqueueOptions, + ): Promise { + 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 { + return this.lq.cancelAllNonRunning(); + } + + // Internal accessor for runner + get _impl(): LQ { + return this.lq; + } +} + +class LitequeQueueClient implements QueueClient { + private db = buildDBClient(path.join(serverConfig.dataDir, "queue.db"), { + walEnabled: serverConfig.database.walMode, + }); + + private queues = new Map>(); + + async init(): Promise { + migrateDB(this.db); + } + + createQueue(name: string, options: QueueOptions): Queue { + if (this.queues.has(name)) { + throw new Error(`Queue ${name} already exists`); + } + const lq = new LQ(name, this.db, { + defaultJobArgs: { numRetries: options.defaultJobArgs.numRetries }, + keepFailedJobs: options.keepFailedJobs, + }); + const wrapper = new LitequeQueueWrapper(name, lq); + this.queues.set(name, wrapper); + return wrapper; + } + + createRunner( + queue: Queue, + funcs: RunnerFuncs, + opts: RunnerOptions, + ): Runner { + const name = queue.name(); + let wrapper = this.queues.get(name); + if (!wrapper) { + throw new Error(`Queue ${name} not found`); + } + + const runner = new LQRunner( + wrapper._impl, + { + run: funcs.run, + onComplete: funcs.onComplete as + | ((job: DequeuedJob) => Promise) + | undefined, + onError: funcs.onError as + | ((job: DequeuedJobError) => Promise) + | 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 { + // No-op for sqlite + } +} + +export class LitequeQueueProvider implements PluginProvider { + private client: QueueClient | null = null; + + async getClient(): Promise { + if (!this.client) { + const client = new LitequeQueueClient(); + await client.init(); + this.client = client; + } + return this.client; + } +} diff --git a/packages/plugins-queue-liteque/tsconfig.json b/packages/plugins-queue-liteque/tsconfig.json new file mode 100644 index 00000000..3bfa695c --- /dev/null +++ b/packages/plugins-queue-liteque/tsconfig.json @@ -0,0 +1,10 @@ +{ + "$schema": "https://json.schemastore.org/tsconfig", + "extends": "@karakeep/tsconfig/node.json", + "include": ["**/*.ts"], + "exclude": ["node_modules"], + "compilerOptions": { + "tsBuildInfoFile": "node_modules/.cache/tsbuildinfo.json" + } +} + -- cgit v1.2.3-70-g09d2