feat: 接入QQBot插件定时任务调度桥

This commit is contained in:
sunlei 2026-06-17 09:28:27 +08:00
parent 7775fd580d
commit 62161bac8f
11 changed files with 848 additions and 26 deletions

View File

@ -20,6 +20,9 @@ import type {
QqbotPluginRuntimeEvent as QqbotPluginWorkerRuntimeEvent,
QqbotPluginWorkerRuntime,
} from '../infrastructure/integration/runtime';
import { QqbotPluginTaskManifestSynchronizer } from './task/qqbot-plugin-task-manifest.synchronizer';
import { QqbotPluginTaskSchedulerService } from './task/qqbot-plugin-task-scheduler.service';
import type { QqbotPluginTaskTriggerType } from './task/qqbot-plugin-task.types';
import { QqbotPluginArgumentParserService } from './argument/qqbot-plugin-argument-parser.service';
import { QqbotBuiltinPluginPackageLoaderService } from '../infrastructure/integration/package/builtin-plugin-package-loader.service';
import { QqbotPluginPackageReaderService } from '../infrastructure/integration/package/plugin-package-reader.service';
@ -56,6 +59,7 @@ export type QqbotPluginRuntimeFactory = {
| 'dispose'
| 'drainRuntimeEvents'
| 'executeOperation'
| 'executeTask'
| 'handleEvent'
| 'health'
| 'load'
@ -159,6 +163,10 @@ export class QqbotPluginPlatformService
private readonly packageReader?: QqbotPluginPackageReaderService,
@Optional()
private readonly builtinPluginLoader?: QqbotBuiltinPluginPackageLoaderService,
@Optional()
private readonly taskSynchronizer?: QqbotPluginTaskManifestSynchronizer,
@Optional()
private readonly taskScheduler?: QqbotPluginTaskSchedulerService,
) {}
async onModuleInit() {
@ -254,13 +262,15 @@ export class QqbotPluginPlatformService
await this.persistManifestCapabilities(plugin.id, manifest);
return this.installationRepository.save({
const installation = await this.installationRepository.save({
installedPath: pluginPackage.packagePath,
pluginId: plugin.id,
runtimeStatus: 'stopped',
status: 'installed',
versionId: version.id,
});
await this.syncManifestTasksForInstallation(installation, manifest, false);
return installation;
}
async enableInstallation(body: InstallationActionBody) {
@ -281,6 +291,7 @@ export class QqbotPluginPlatformService
async disableInstallation(body: InstallationActionBody) {
const installation = await this.requireInstallation(body);
await this.stopWorkersForInstallation(installation);
await this.taskScheduler?.removeSchedulersForInstallation(installation.id);
await this.refreshActiveRegistries(installation, false);
await this.updateInstallationRuntime(installation, 'disabled', 'stopped');
await this.recordRuntimeEvent(installation, 'disable-finished');
@ -329,6 +340,7 @@ export class QqbotPluginPlatformService
}
await this.refreshActiveRegistries(installation, false);
await this.taskScheduler?.removeSchedulersForInstallation(installation.id);
await this.updateInstallationRuntime(
installation,
'uninstalled',
@ -410,6 +422,34 @@ export class QqbotPluginPlatformService
return handled;
}
async executeTask(input: {
input: Record<string, unknown>;
installationId: string;
pluginId: string;
taskHandlerName: string;
taskId: string;
taskKey: string;
timeoutMs: number;
triggerType: QqbotPluginTaskTriggerType;
}) {
const workerContext = this.activeWorkerContexts.get(input.installationId);
if (!workerContext) {
throwVbenError('插件运行时未启用');
}
try {
return await workerContext.worker.executeTask({
input: input.input,
taskHandlerName: input.taskHandlerName,
taskId: input.taskId,
taskKey: input.taskKey,
timeoutMs: input.timeoutMs,
triggerType: input.triggerType,
});
} finally {
await this.flushWorkerRuntimeEvents(workerContext);
}
}
async listActiveOperations() {
const workerOperations = this.listActiveWorkerOperations();
if (workerOperations.length > 0) return workerOperations;
@ -708,6 +748,27 @@ export class QqbotPluginPlatformService
this.activeWorkerPluginAliases.set(alias, manifest.pluginKey);
this.activeWorkersByPluginKey.set(alias, workerContext);
}
await this.syncManifestTasksForInstallation(installation, manifest, true);
}
private async syncManifestTasksForInstallation(
installation: QqbotPluginInstallation,
manifest: QqbotPluginManifest,
scheduleEnabledTasks: boolean,
) {
if (!this.taskSynchronizer || !manifest.tasks.length) return [];
const tasks = await this.taskSynchronizer.syncManifestTasks({
installationId: installation.id,
manifestTasks: manifest.tasks,
pluginId: installation.pluginId,
});
if (scheduleEnabledTasks && this.taskScheduler) {
for (const task of tasks) {
await this.taskScheduler.syncTaskScheduler(task);
}
}
return tasks;
}
private async stopExistingWorkersForManifest(manifest: QqbotPluginManifest) {

View File

@ -1,4 +1,6 @@
export * from './qqbot-plugin-task-cron.validator';
export * from './qqbot-plugin-task-manifest.synchronizer';
export * from './qqbot-plugin-task-scheduler.service';
export * from './qqbot-plugin-task.service';
export * from './qqbot-plugin-task.types';
export * from './qqbot-plugin-task-worker.processor';

View File

@ -0,0 +1,199 @@
import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { InjectRepository } from '@nestjs/typeorm';
import { Queue, type ConnectionOptions } from 'bullmq';
import { Repository } from 'typeorm';
import { QqbotPluginTask } from '../../infrastructure/persistence';
export type QqbotPluginTaskJobData = {
input?: Record<string, unknown>;
taskId: string;
triggerType: 'manual' | 'schedule';
};
@Injectable()
export class QqbotPluginTaskSchedulerService
implements OnModuleDestroy, OnModuleInit
{
private readonly queue: Queue<QqbotPluginTaskJobData>;
constructor(
configService: ConfigService,
@InjectRepository(QqbotPluginTask)
private readonly taskRepository: Repository<QqbotPluginTask>,
) {
this.queue = new Queue(QQBOT_PLUGIN_TASK_QUEUE_NAME, {
connection: resolveQqbotPluginTaskQueueConnection(configService),
prefix: readQqbotPluginTaskQueuePrefix(configService),
});
}
async onModuleInit() {
await this.queue.waitUntilReady();
await this.resyncEnabledTasks();
}
async onModuleDestroy() {
await this.queue.close();
}
async resyncEnabledTasks() {
const tasks = await this.taskRepository.find({ where: { enabled: true } });
for (const task of tasks) {
await this.syncTaskScheduler(task);
}
}
async syncTaskScheduler(
task: Pick<
QqbotPluginTask,
'cronExpression' | 'enabled' | 'id' | 'installationId' | 'taskKey' | 'timeoutMs'
>,
) {
const schedulerId = this.buildSchedulerId(task.id);
if (!task.enabled) {
await this.removeTaskScheduler(task.id);
await this.taskRepository.update(
{ id: task.id },
{ nextRunAt: null, runtimeStatus: 'disabled' },
);
return;
}
await this.queue.upsertJobScheduler(
schedulerId,
{ pattern: task.cronExpression },
{
data: {
taskId: task.id,
triggerType: 'schedule',
},
name: QQBOT_PLUGIN_TASK_JOB_NAME,
opts: {
attempts: 1,
removeOnComplete: true,
removeOnFail: 100,
},
},
);
await this.taskRepository.update(
{ id: task.id },
{ runtimeStatus: 'scheduled' },
);
}
async removeTaskScheduler(taskId: string) {
await this.queue.removeJobScheduler(this.buildSchedulerId(taskId));
}
async removeSchedulersForInstallation(installationId: string) {
const tasks = await this.taskRepository.find({ where: { installationId } });
for (const task of tasks) {
await this.removeTaskScheduler(task.id);
}
}
async enqueueManualRun(taskId: string, input: Record<string, unknown>) {
return this.queue.add(
QQBOT_PLUGIN_TASK_JOB_NAME,
{
input,
taskId,
triggerType: 'manual',
},
{
attempts: 1,
removeOnComplete: true,
removeOnFail: 100,
},
);
}
private buildSchedulerId(taskId: string) {
return `plugin-task:${taskId}`;
}
}
export const QQBOT_PLUGIN_TASK_QUEUE_NAME = 'qqbot-plugin-task';
export const QQBOT_PLUGIN_TASK_JOB_NAME = 'execute-plugin-task';
export function readQqbotPluginTaskQueuePrefix(
configService: ConfigService,
) {
return readStringConfig(
configService,
[
'QQBOT_PLUGIN_TASK_QUEUE_REDIS_PREFIX',
'QQBOT_PLUGIN_TASK_QUEUE_PREFIX',
'QQBOT_PLUGIN_QUEUE_REDIS_PREFIX',
],
'kt:qqbot:plugin-task',
);
}
export function resolveQqbotPluginTaskQueueConnection(
configService: ConfigService,
): ConnectionOptions {
const host = readStringConfig(configService, [
'QQBOT_PLUGIN_TASK_QUEUE_REDIS_HOST',
'QQBOT_PLUGIN_QUEUE_REDIS_HOST',
'REDIS_HOST',
]);
if (!host) {
throw new Error('QQBot 插件定时任务队列缺少 Redis 主机配置');
}
const password = readStringConfig(configService, [
'QQBOT_PLUGIN_TASK_QUEUE_REDIS_PASSWORD',
'QQBOT_PLUGIN_QUEUE_REDIS_PASSWORD',
'REDIS_PASSWORD',
]);
return {
db: readNumberConfig(
configService,
[
'QQBOT_PLUGIN_TASK_QUEUE_REDIS_DB',
'QQBOT_PLUGIN_QUEUE_REDIS_DB',
'REDIS_DB',
],
0,
),
host,
password: password || undefined,
port: readNumberConfig(
configService,
[
'QQBOT_PLUGIN_TASK_QUEUE_REDIS_PORT',
'QQBOT_PLUGIN_QUEUE_REDIS_PORT',
'REDIS_PORT',
],
6379,
),
};
}
function readStringConfig(
configService: ConfigService,
keys: string[],
fallback = '',
) {
for (const key of keys) {
const value = configService.get<string | number | undefined>(key);
if (value !== undefined && value !== null && `${value}`.trim()) {
return `${value}`.trim();
}
}
return fallback;
}
function readNumberConfig(
configService: ConfigService,
keys: string[],
fallback: number,
) {
const value = readStringConfig(configService, keys);
if (!value) return fallback;
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : fallback;
}

View File

@ -0,0 +1,261 @@
import {
Injectable,
Logger,
OnModuleDestroy,
OnModuleInit,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { InjectRepository } from '@nestjs/typeorm';
import { Job, Worker } from 'bullmq';
import { Repository } from 'typeorm';
import { QqbotPluginPlatformService } from '../plugin-platform.service';
import {
QqbotPluginTask,
QqbotPluginTaskRun,
type QqbotPluginTaskRunStatus,
type QqbotPluginTaskTriggerType,
} from '../../infrastructure/persistence';
import {
QQBOT_PLUGIN_TASK_JOB_NAME,
QQBOT_PLUGIN_TASK_QUEUE_NAME,
readQqbotPluginTaskQueuePrefix,
resolveQqbotPluginTaskQueueConnection,
type QqbotPluginTaskJobData,
} from './qqbot-plugin-task-scheduler.service';
@Injectable()
export class QqbotPluginTaskWorkerProcessor
implements OnModuleDestroy, OnModuleInit
{
private readonly logger = new Logger(QqbotPluginTaskWorkerProcessor.name);
private worker?: Worker<QqbotPluginTaskJobData>;
constructor(
private readonly configService: ConfigService,
private readonly platformService: QqbotPluginPlatformService,
@InjectRepository(QqbotPluginTask)
private readonly taskRepository: Repository<QqbotPluginTask>,
@InjectRepository(QqbotPluginTaskRun)
private readonly runRepository: Repository<QqbotPluginTaskRun>,
) {}
async onModuleInit() {
this.worker = new Worker<QqbotPluginTaskJobData>(
QQBOT_PLUGIN_TASK_QUEUE_NAME,
async (job) => this.processJob(job),
{
concurrency: 1,
connection: resolveQqbotPluginTaskQueueConnection(this.configService),
prefix: readQqbotPluginTaskQueuePrefix(this.configService),
},
);
this.worker.on('error', (error) => {
this.logger.error(error.message, error.stack);
});
await this.worker.waitUntilReady();
}
async onModuleDestroy() {
await this.worker?.close();
}
private async processJob(job: Job<QqbotPluginTaskJobData>) {
if (job.name && job.name !== QQBOT_PLUGIN_TASK_JOB_NAME) {
return { ok: false, reason: 'unknown-job', skipped: true };
}
const task = await this.taskRepository.findOne({
where: { id: job.data.taskId },
});
if (!task) {
return { ok: false, reason: 'task-not-found', skipped: true };
}
if (job.data.triggerType === 'schedule' && !task.enabled) {
return this.writeSkippedRun(
task,
`${job.id || ''}`,
job.data.triggerType,
'task-disabled',
);
}
const running = await this.runRepository.findOne({
where: { status: 'running', taskId: task.id },
});
if (running) {
return this.writeSkippedRun(
task,
`${job.id || ''}`,
job.data.triggerType,
'previous-run-running',
);
}
return this.executeTaskRun(
task,
`${job.id || ''}`,
job.data.triggerType,
job.data.input || {},
);
}
private async executeTaskRun(
task: QqbotPluginTask,
jobId: string,
triggerType: QqbotPluginTaskTriggerType,
input: Record<string, unknown>,
) {
const startedAt = new Date();
const run = await this.runRepository.save({
installationId: task.installationId,
jobId,
pluginId: task.pluginId,
safeSummary: {
inputKeys: Object.keys(input).sort(),
},
startedAt,
status: 'running',
taskId: task.id,
taskKey: task.taskKey,
triggerType,
});
await this.taskRepository.update(
{ id: task.id },
{
lastRunId: run.id,
runtimeStatus: 'running',
},
);
try {
const output = await this.platformService.executeTask({
input,
installationId: task.installationId,
pluginId: task.pluginId,
taskHandlerName: task.handlerName,
taskId: task.id,
taskKey: task.taskKey,
timeoutMs: task.timeoutMs,
triggerType,
});
const durationMs = Date.now() - startedAt.getTime();
const finishedAt = new Date();
const saved = await this.finishRun(run, {
durationMs,
finishedAt,
safeSummary: {
outputKeys: this.getOutputKeys(output),
},
status: 'success',
});
await this.finishTask(task, saved, {
durationMs,
errorMessage: null,
finishedAt,
status: 'success',
});
return {
ok: true,
runId: saved.id,
status: 'success',
};
} catch (error) {
const durationMs = Date.now() - startedAt.getTime();
const finishedAt = new Date();
const errorMessage =
error instanceof Error ? error.message : `${error || ''}`;
const saved = await this.finishRun(run, {
durationMs,
errorMessage,
finishedAt,
status: 'failed',
});
await this.finishTask(task, saved, {
durationMs,
errorMessage,
finishedAt,
status: 'failed',
});
throw error;
}
}
private async writeSkippedRun(
task: QqbotPluginTask,
jobId: string,
triggerType: QqbotPluginTaskTriggerType,
reason: string,
) {
const now = new Date();
const run = await this.runRepository.save({
durationMs: 0,
finishedAt: now,
installationId: task.installationId,
jobId,
pluginId: task.pluginId,
safeSummary: { reason },
startedAt: now,
status: 'skipped',
taskId: task.id,
taskKey: task.taskKey,
triggerType,
});
await this.finishTask(task, run, {
durationMs: 0,
errorMessage: reason,
finishedAt: now,
status: 'skipped',
});
return {
ok: true,
reason,
runId: run.id,
status: 'skipped',
};
}
private async finishRun(
run: QqbotPluginTaskRun,
patch: Partial<QqbotPluginTaskRun>,
) {
return this.runRepository.save({
...run,
...patch,
});
}
private async finishTask(
task: QqbotPluginTask,
run: QqbotPluginTaskRun,
result: {
durationMs: number;
errorMessage: null | string;
finishedAt: Date;
status: QqbotPluginTaskRunStatus;
},
) {
await this.taskRepository.update(
{ id: task.id },
{
lastDurationMs: result.durationMs,
lastError: result.errorMessage,
lastRunAt: result.finishedAt,
lastRunId: run.id,
lastStatus: result.status,
runtimeStatus:
result.status === 'failed'
? 'failed'
: task.enabled
? 'scheduled'
: 'idle',
},
);
}
private getOutputKeys(output: unknown) {
return output && typeof output === 'object'
? Object.keys(output as Record<string, unknown>).sort()
: [];
}
}

View File

@ -1,4 +1,4 @@
import { Injectable } from '@nestjs/common';
import { Injectable, Optional } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import {
Between,
@ -18,6 +18,7 @@ import type {
QqbotPluginTaskPageQuery,
QqbotPluginTaskRunPageQuery,
} from './qqbot-plugin-task.types';
import { QqbotPluginTaskSchedulerService } from './qqbot-plugin-task-scheduler.service';
@Injectable()
export class QqbotPluginTaskService {
@ -29,6 +30,8 @@ export class QqbotPluginTaskService {
@InjectRepository(QqbotPlugin)
private readonly pluginRepository: Repository<QqbotPlugin>,
private readonly toolsService: ToolsService,
@Optional()
private readonly scheduler?: QqbotPluginTaskSchedulerService,
) {}
async pageTasks(query: QqbotPluginTaskPageQuery) {
@ -57,39 +60,38 @@ export class QqbotPluginTaskService {
}
async enableTask(id: string) {
await this.taskRepository.update(
{ id },
{ enabled: true, runtimeStatus: 'scheduled' },
);
return this.getTaskDetail(id);
const task = await this.getTaskDetail(id);
task.enabled = true;
task.runtimeStatus = 'scheduled';
const saved = await this.taskRepository.save(task);
await this.requireScheduler().syncTaskScheduler(saved);
return saved;
}
async disableTask(id: string) {
await this.taskRepository.update(
{ id },
{ enabled: false, runtimeStatus: 'disabled' },
);
return this.getTaskDetail(id);
const task = await this.getTaskDetail(id);
task.enabled = false;
task.runtimeStatus = 'disabled';
const saved = await this.taskRepository.save(task);
await this.requireScheduler().removeTaskScheduler(id);
return saved;
}
async updateTaskCron(id: string, body: { cronExpression?: string }) {
const cronExpression = requireQqbotPluginTaskCron(body.cronExpression);
await this.taskRepository.update({ id }, { cronExpression });
return this.getTaskDetail(id);
const task = await this.getTaskDetail(id);
task.cronExpression = requireQqbotPluginTaskCron(body.cronExpression);
const saved = await this.taskRepository.save(task);
await this.requireScheduler().syncTaskScheduler(saved);
return saved;
}
async runTaskOnce(id: string, body: { input?: Record<string, unknown> }) {
void body;
const task = await this.getTaskDetail(id);
const run = await this.runRepository.save({
installationId: task.installationId,
pluginId: task.pluginId,
status: 'running',
taskId: task.id,
taskKey: task.taskKey,
triggerType: 'manual',
});
return { jobId: run.jobId || `${run.id || ''}`, taskId: task.id };
await this.getTaskDetail(id);
const job = await this.requireScheduler().enqueueManualRun(
id,
body.input || {},
);
return { jobId: `${job.id || ''}`, taskId: id };
}
async pageTaskRuns(id: string, query: QqbotPluginTaskRunPageQuery) {
@ -135,4 +137,11 @@ export class QqbotPluginTaskService {
}
return {};
}
private requireScheduler() {
if (!this.scheduler) {
throwVbenError('插件定时任务调度器未初始化');
}
return this.scheduler;
}
}

View File

@ -36,6 +36,11 @@ import type { QqbotPluginWorkerRequest } from './worker-runtime.types';
type RuntimeCommandPlugin = QqbotIntegrationPlugin & {
activate?: () => Promise<unknown> | unknown;
dispose?: () => Promise<unknown> | unknown;
tasks?: Array<{
execute(input: Record<string, unknown>): Promise<unknown> | unknown;
handlerName: string;
key: string;
}>;
};
type RuntimeEventPlugin = {
@ -139,6 +144,8 @@ async function handleWorkerRequest(message: QqbotPluginWorkerRequest) {
return health();
case 'executeOperation':
return executeOperation(message);
case 'executeTask':
return executeTask(message);
case 'handleEvent':
return handleEvent(message);
case 'deactivate':
@ -193,6 +200,18 @@ async function executeOperation(message: QqbotPluginWorkerRequest) {
);
}
async function executeTask(message: QqbotPluginWorkerRequest) {
const task = commandPlugin?.tasks?.find(
(item) =>
item.key === message.taskKey ||
item.handlerName === message.taskHandlerName,
);
if (!task) {
throw new Error(`QQBot 插件定时任务不存在:${message.taskKey}`);
}
return task.execute((message.input || {}) as Record<string, unknown>);
}
async function handleEvent(message: QqbotPluginWorkerRequest) {
if (!eventPlugin) return false;
const definition = eventPlugin.getDefinition();

View File

@ -5,6 +5,7 @@ import {
type QqbotPluginRuntimeEvent,
type QqbotPluginRuntimeStatus,
type QqbotPluginSafeInputSummary,
type QqbotPluginTaskRequest,
type QqbotPluginWorkerRequestQueue,
type QqbotPluginWorkerRequest,
type QqbotPluginWorkerRequestType,
@ -135,6 +136,21 @@ export class QqbotPluginWorkerRuntime {
);
}
async executeTask(request: QqbotPluginTaskRequest) {
return this.request(
'executeTask',
{
input: request.input,
safeInputSummary: summarizeInput(request.input),
taskHandlerName: request.taskHandlerName,
taskId: request.taskId,
taskKey: request.taskKey,
triggerType: request.triggerType,
},
request.timeoutMs,
);
}
async health() {
return this.request('health');
}

View File

@ -3,6 +3,7 @@ export type QqbotPluginWorkerRequestType =
| 'deactivate'
| 'dispose'
| 'executeOperation'
| 'executeTask'
| 'handleEvent'
| 'health'
| 'load';
@ -29,7 +30,11 @@ export type QqbotPluginWorkerRequest = {
input?: unknown;
pluginKey: string;
safeInputSummary?: QqbotPluginSafeInputSummary;
taskHandlerName?: string;
taskId?: string;
taskKey?: string;
timeoutMs: number;
triggerType?: 'bootstrap' | 'manual' | 'schedule';
type: QqbotPluginWorkerRequestType;
};
@ -75,3 +80,12 @@ export type QqbotPluginEventRequest = {
eventKey: string;
timeoutMs?: number;
};
export type QqbotPluginTaskRequest = {
input: Record<string, unknown>;
taskHandlerName: string;
taskId: string;
taskKey: string;
timeoutMs?: number;
triggerType: 'bootstrap' | 'manual' | 'schedule';
};

View File

@ -13,7 +13,9 @@ import { QqbotPluginExecutionAdapter } from './application/plugin-execution.adap
import { QqbotPluginPlatformService } from './application/plugin-platform.service';
import {
QqbotPluginTaskManifestSynchronizer,
QqbotPluginTaskSchedulerService,
QqbotPluginTaskService,
QqbotPluginTaskWorkerProcessor,
} from './application/task';
import { QqbotPluginPlatformTaskController } from './contract/plugin-platform-task.controller';
import { QqbotPluginPlatformController } from './contract/plugin-platform.controller';
@ -63,7 +65,9 @@ import { QQBOT_PLUGIN_RUNTIME_FACTORY } from './application/plugin-platform.serv
QqbotBuiltinPluginWorkerRuntimeFactoryService,
QqbotPluginPackageReaderService,
QqbotPluginTaskManifestSynchronizer,
QqbotPluginTaskSchedulerService,
QqbotPluginTaskService,
QqbotPluginTaskWorkerProcessor,
{
provide: QQBOT_PLUGIN_EXECUTION_PORT,
useExisting: QqbotPluginExecutionAdapter,

View File

@ -0,0 +1,208 @@
const createdQueues: any[] = [];
const createdWorkers: any[] = [];
jest.mock('bullmq', () => ({
Queue: class MockQueue {
readonly schedulers = new Map<string, unknown>();
constructor(
public name: string,
public options: unknown,
) {
createdQueues.push(this);
}
async add(name: string, data: unknown, opts?: unknown) {
return { data, id: `${name}-job`, name, opts };
}
async close() {}
async removeJobScheduler(id: string) {
this.schedulers.delete(id);
return 1;
}
async upsertJobScheduler(
id: string,
repeat: unknown,
template: unknown,
) {
this.schedulers.set(id, { repeat, template });
return { id };
}
async waitUntilReady() {}
},
Worker: class MockWorker {
constructor(
public name: string,
public processor: (job: unknown) => Promise<unknown> | unknown,
public options: unknown,
) {
createdWorkers.push(this);
}
on() {
return this;
}
async close() {}
async waitUntilReady() {}
},
}));
import {
QqbotPluginTaskSchedulerService,
QqbotPluginTaskWorkerProcessor,
} from '../../../../src/modules/qqbot/plugin-platform/application/task';
describe('QQBot plugin task scheduler', () => {
beforeEach(() => {
createdQueues.length = 0;
createdWorkers.length = 0;
});
it('registers cron through BullMQ Job Scheduler with a stable scheduler id', async () => {
const taskRepository = createTaskRepository([
{
cronExpression: '0 */6 * * *',
enabled: true,
id: 'task-1',
installationId: 'install-1',
taskKey: 'bangdream.bestdori.sync-main-data',
timeoutMs: 120000,
},
]);
const scheduler = new QqbotPluginTaskSchedulerService(
createConfigService(),
taskRepository as any,
);
await scheduler.syncTaskScheduler({
cronExpression: '0 */6 * * *',
enabled: true,
id: 'task-1',
installationId: 'install-1',
taskKey: 'bangdream.bestdori.sync-main-data',
timeoutMs: 120000,
} as any);
expect(createdQueues[0].schedulers.get('plugin-task:task-1')).toMatchObject(
{
repeat: { pattern: '0 */6 * * *' },
template: {
data: { taskId: 'task-1', triggerType: 'schedule' },
name: 'execute-plugin-task',
},
},
);
expect(taskRepository.update).toHaveBeenCalledWith(
{ id: 'task-1' },
{ runtimeStatus: 'scheduled' },
);
await scheduler.onModuleDestroy();
});
it('executes a task job through the platform worker and stores only safe output keys', async () => {
const task = {
enabled: true,
handlerName: 'syncBestdoriMainData',
id: 'task-1',
installationId: 'install-1',
pluginId: 'plugin-1',
taskKey: 'bangdream.bestdori.sync-main-data',
timeoutMs: 120000,
};
const taskRepository = createTaskRepository([task]);
const runRepository = createRunRepository();
const platformService = {
executeTask: jest.fn(async () => ({
replyText: 'secret text',
syncedKeys: ['songs'],
})),
};
const processor = new QqbotPluginTaskWorkerProcessor(
createConfigService(),
platformService as any,
taskRepository as any,
runRepository as any,
);
await processor.onModuleInit();
const result = await createdWorkers[0].processor({
data: {
input: { force: true, payload: 'secret payload' },
taskId: 'task-1',
triggerType: 'manual',
},
id: 'job-1',
});
expect(platformService.executeTask).toHaveBeenCalledWith({
input: { force: true, payload: 'secret payload' },
installationId: 'install-1',
pluginId: 'plugin-1',
taskHandlerName: 'syncBestdoriMainData',
taskId: 'task-1',
taskKey: 'bangdream.bestdori.sync-main-data',
timeoutMs: 120000,
triggerType: 'manual',
});
expect(runRepository.save).toHaveBeenCalledWith(
expect.objectContaining({
safeSummary: { outputKeys: ['replyText', 'syncedKeys'] },
status: 'success',
}),
);
expect(JSON.stringify(runRepository.save.mock.calls)).not.toContain(
'secret text',
);
expect(result).toEqual({
ok: true,
runId: 'run-1',
status: 'success',
});
await processor.onModuleDestroy();
});
});
function createConfigService(): any {
return {
get: (key: string) =>
({
QQBOT_PLUGIN_QUEUE_REDIS_HOST: 'redis.local',
QQBOT_PLUGIN_TASK_QUEUE_REDIS_PREFIX: 'kt:qqbot:plugin-task',
})[key],
};
}
function createTaskRepository(tasks: any[]) {
return {
find: jest.fn(async () => tasks),
findOne: jest.fn(async ({ where }: any) =>
tasks.find((task) => task.id === where.id) || null,
),
save: jest.fn(async (value) => value),
update: jest.fn(async () => ({ affected: 1 })),
};
}
function createRunRepository() {
let runningRun: any;
return {
findOne: jest.fn(async ({ where }: any) =>
where.status === 'running' ? runningRun || null : null,
),
save: jest.fn(async (value: any) => {
const next = { ...value, id: value.id || 'run-1' };
if (next.status === 'running') {
runningRun = next;
} else {
runningRun = undefined;
}
return next;
}),
};
}

View File

@ -417,6 +417,35 @@ describe('QQBot plugin worker runtime', () => {
expect(driver.disposed).toBe(true);
});
it('sends executeTask RPC with safe input summary and timeout', async () => {
const { driver, runtime } = createRuntime();
driver.responses.set('executeTask', { syncedKeys: ['songs'] });
await expect(
runtime.executeTask({
input: { force: true, fullPayload: 'secret' },
taskHandlerName: 'syncBestdoriMainData',
taskId: 'task-1',
taskKey: 'bangdream.bestdori.sync-main-data',
timeoutMs: 120000,
triggerType: 'manual',
}),
).resolves.toEqual({ syncedKeys: ['songs'] });
expect(driver.requests[0]).toMatchObject({
safeInputSummary: { fieldCount: 2, keys: ['force', 'fullPayload'] },
taskHandlerName: 'syncBestdoriMainData',
taskId: 'task-1',
taskKey: 'bangdream.bestdori.sync-main-data',
timeoutMs: 120000,
triggerType: 'manual',
type: 'executeTask',
});
expect(JSON.stringify(driver.requests[0].safeInputSummary)).not.toContain(
'secret',
);
});
it('isolates worker crashes as plugin runtime events without throwing raw errors', async () => {
const { runtime } = createRuntime(
new RecordingDriver(