diff --git a/k8s/prod/api.yaml b/k8s/prod/api.yaml index 266b2b1..6e81507 100644 --- a/k8s/prod/api.yaml +++ b/k8s/prod/api.yaml @@ -105,7 +105,7 @@ spec: spec: containers: - name: redis - image: redis:7.4-alpine + image: k3d-kt-registry.localhost:5000/redis:7.4-alpine imagePullPolicy: IfNotPresent args: - redis-server diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts index 56cec68..1686131 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts @@ -47,9 +47,7 @@ type QqbotWorkerQueueResult = ok: false; }; -export class QqbotBullmqPluginWorkerRequestQueue - implements QqbotPluginWorkerRequestQueue -{ +export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerRequestQueue { private readonly queue: Queue< QqbotWorkerQueueJobData, QqbotWorkerQueueResult, @@ -159,7 +157,9 @@ export class QqbotBullmqPluginWorkerRequestQueue message.type, { expiresAt: - Date.now() + message.timeoutMs + this.options.waitUntilFinishedBufferMs, + Date.now() + + message.timeoutMs + + this.options.waitUntilFinishedBufferMs, generation: this.generation, message, workerInstanceId: this.options.workerInstanceId, @@ -280,7 +280,7 @@ export function resolveQqbotPluginQueueConnection( function buildWorkerQueueName(pluginKey: string, installationId: string) { const safePluginKey = pluginKey.replace(/[^a-zA-Z0-9_-]/g, '-'); const safeInstallationId = installationId.replace(/[^a-zA-Z0-9_-]/g, '-'); - return `qqbot-plugin-worker:${safePluginKey}:${safeInstallationId}`; + return `qqbot-plugin-worker-${safePluginKey}-${safeInstallationId}`; } function createWorkerInstanceId() { diff --git a/test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts b/test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts new file mode 100644 index 0000000..0b24482 --- /dev/null +++ b/test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts @@ -0,0 +1,99 @@ +const mockBullmqCreations: Array<{ + name: string; + options: { prefix?: string }; + type: 'Queue' | 'QueueEvents' | 'Worker'; +}> = []; + +jest.mock('bullmq', () => { + class MockQueue { + constructor(name: string, options: { prefix?: string }) { + if (name.includes(':')) { + throw new Error('Queue name cannot contain :'); + } + mockBullmqCreations.push({ name, options, type: 'Queue' }); + } + + on() { + return this; + } + + waitUntilReady() { + return Promise.resolve(); + } + + close() { + return Promise.resolve(); + } + } + + class MockQueueEvents extends MockQueue { + constructor(name: string, options: { prefix?: string }) { + super(name, options); + mockBullmqCreations[mockBullmqCreations.length - 1].type = 'QueueEvents'; + } + } + + class MockWorker extends MockQueue { + constructor( + name: string, + _processor: unknown, + options: { prefix?: string }, + ) { + super(name, options); + mockBullmqCreations[mockBullmqCreations.length - 1].type = 'Worker'; + } + } + + return { + Job: class MockJob {}, + Queue: MockQueue, + QueueEvents: MockQueueEvents, + Worker: MockWorker, + }; +}); + +import { QqbotBullmqPluginWorkerRequestQueue } from '../../../../src/modules/qqbot/plugin-platform/infrastructure/integration/runtime'; +import type { QqbotPluginWorkerDriver } from '../../../../src/modules/qqbot/plugin-platform/infrastructure/integration/runtime'; + +describe('QQBot BullMQ plugin worker request queue', () => { + beforeEach(() => { + mockBullmqCreations.length = 0; + }); + + it('keeps Redis prefix separate from a colon-free BullMQ queue name', async () => { + const driver: QqbotPluginWorkerDriver = { + dispose: jest.fn(async () => undefined), + request: jest.fn(async () => ({ ok: true })), + }; + + const queue = new QqbotBullmqPluginWorkerRequestQueue(driver, { + connection: { + host: 'redis.local', + port: 6379, + }, + installationId: 'install:1', + pluginKey: 'bang:dream', + prefix: 'kt:qqbot:plugin-worker', + removeOnFailCount: 100, + waitUntilFinishedBufferMs: 5_000, + workerInstanceId: 'worker-1', + }); + + await queue.close(); + + expect(mockBullmqCreations).toHaveLength(3); + expect(mockBullmqCreations.map((creation) => creation.name)).toEqual([ + 'qqbot-plugin-worker-bang-dream-install-1', + 'qqbot-plugin-worker-bang-dream-install-1', + 'qqbot-plugin-worker-bang-dream-install-1', + ]); + expect( + mockBullmqCreations.every((creation) => !creation.name.includes(':')), + ).toBe(true); + expect( + mockBullmqCreations.every( + (creation) => creation.options.prefix === 'kt:qqbot:plugin-worker', + ), + ).toBe(true); + }); +}); diff --git a/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts b/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts index 8fddfda..fc7d135 100644 --- a/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts +++ b/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts @@ -66,9 +66,7 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { it('uses BullMQ queues to serialize plugin worker requests instead of ad hoc in-memory chaining', () => { const source = [ - readSource( - 'src/modules/qqbot/plugin-platform/plugin-platform.module.ts', - ), + readSource('src/modules/qqbot/plugin-platform/plugin-platform.module.ts'), readSource( 'src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts', ), @@ -77,7 +75,7 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { ), ].join('\n'); - expect(source).toContain("@nestjs/bullmq"); + expect(source).toContain('@nestjs/bullmq'); expect(source).toContain("from 'bullmq'"); expect(source).toContain('new Queue('); expect(source).toContain('new Worker('); @@ -103,6 +101,15 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { expect(source).not.toContain('maxUnavailable: 0'); }); + it('pulls the plugin Redis runtime image from the local registry', () => { + const source = readSource('k8s/prod/api.yaml'); + + expect(source).toContain( + 'image: k3d-kt-registry.localhost:5000/redis:7.4-alpine', + ); + expect(source).not.toContain('image: redis:7.4-alpine'); + }); + it('uses dedicated lifecycle use cases instead of direct status flips', () => { const controller = readSource( 'src/modules/qqbot/plugin-platform/contract/plugin-platform.controller.ts', @@ -465,12 +472,10 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { }); expect(worker.dispose).toHaveBeenCalled(); - expect( - (service as any).activeWorkers.has(installation.id), - ).toBe(false); - expect( - (service as any).activeWorkerContexts.has(installation.id), - ).toBe(false); + expect((service as any).activeWorkers.has(installation.id)).toBe(false); + expect((service as any).activeWorkerContexts.has(installation.id)).toBe( + false, + ); }); it('routes enabled command and message executions through active worker runtimes', async () => {