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/src/index.ts | |
| 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/src/index.ts')
| -rw-r--r-- | packages/plugins-queue-liteque/src/index.ts | 137 |
1 files changed, 0 insertions, 137 deletions
diff --git a/packages/plugins-queue-liteque/src/index.ts b/packages/plugins-queue-liteque/src/index.ts deleted file mode 100644 index ddc2181c..00000000 --- a/packages/plugins-queue-liteque/src/index.ts +++ /dev/null @@ -1,137 +0,0 @@ -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; - } -} |
