fix: 修复QQBot插件队列线上启动

This commit is contained in:
sunlei 2026-06-16 16:07:20 +08:00
parent 5498e2e52f
commit aaeea9102b
4 changed files with 120 additions and 16 deletions

View File

@ -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

View File

@ -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() {

View File

@ -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);
});
});

View File

@ -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 () => {