aboutsummaryrefslogtreecommitdiffstats
path: root/packages/plugins-queue-liteque/src/index.ts
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/src/index.ts
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/src/index.ts')
-rw-r--r--packages/plugins-queue-liteque/src/index.ts137
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;
- }
-}