diff options
| author | Mohamed Bassem <me@mbassem.com> | 2025-11-08 14:50:00 +0000 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2025-11-08 14:50:00 +0000 |
| commit | 99413db0e79a156a1b87eacd3c6a7b83e9df946e (patch) | |
| tree | 73f0a5fceb507f75f662a109b00beeb3fa6b16fb /packages/plugins/queue-liteque | |
| parent | 737b03172c2e063ba311c23d6552418bd2ab1955 (diff) | |
| download | karakeep-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.ts | 10 | ||||
| -rw-r--r-- | packages/plugins/queue-liteque/src/index.ts | 137 |
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; + } +} |
