From 62161bac8fe5c31942d6b2d2aa523ed05f771d58 Mon Sep 17 00:00:00 2001 From: sunlei Date: Wed, 17 Jun 2026 09:28:27 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E5=85=A5QQBot=E6=8F=92?= =?UTF-8?q?=E4=BB=B6=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=E8=B0=83=E5=BA=A6?= =?UTF-8?q?=E6=A1=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../application/plugin-platform.service.ts | 63 ++++- .../plugin-platform/application/task/index.ts | 2 + .../qqbot-plugin-task-scheduler.service.ts | 199 +++++++++++++ .../qqbot-plugin-task-worker.processor.ts | 261 ++++++++++++++++++ .../task/qqbot-plugin-task.service.ts | 59 ++-- .../runtime/builtin-plugin-worker.thread.ts | 19 ++ .../integration/runtime/worker-runtime.ts | 16 ++ .../runtime/worker-runtime.types.ts | 14 + .../plugin-platform/plugin-platform.module.ts | 4 + .../plugin-task-scheduler.spec.ts | 208 ++++++++++++++ .../plugin-platform/worker-runtime.spec.ts | 29 ++ 11 files changed, 848 insertions(+), 26 deletions(-) create mode 100644 src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-scheduler.service.ts create mode 100644 src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-worker.processor.ts create mode 100644 test/modules/qqbot/plugin-platform/plugin-task-scheduler.spec.ts diff --git a/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts b/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts index 7e8a5bc..3f8992e 100644 --- a/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts +++ b/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts @@ -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; + 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) { diff --git a/src/modules/qqbot/plugin-platform/application/task/index.ts b/src/modules/qqbot/plugin-platform/application/task/index.ts index b7b71f8..dba78ee 100644 --- a/src/modules/qqbot/plugin-platform/application/task/index.ts +++ b/src/modules/qqbot/plugin-platform/application/task/index.ts @@ -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'; diff --git a/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-scheduler.service.ts b/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-scheduler.service.ts new file mode 100644 index 0000000..4bd34bc --- /dev/null +++ b/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-scheduler.service.ts @@ -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; + taskId: string; + triggerType: 'manual' | 'schedule'; +}; + +@Injectable() +export class QqbotPluginTaskSchedulerService + implements OnModuleDestroy, OnModuleInit +{ + private readonly queue: Queue; + + constructor( + configService: ConfigService, + @InjectRepository(QqbotPluginTask) + private readonly taskRepository: Repository, + ) { + 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) { + 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(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; +} diff --git a/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-worker.processor.ts b/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-worker.processor.ts new file mode 100644 index 0000000..837b80f --- /dev/null +++ b/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task-worker.processor.ts @@ -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; + + constructor( + private readonly configService: ConfigService, + private readonly platformService: QqbotPluginPlatformService, + @InjectRepository(QqbotPluginTask) + private readonly taskRepository: Repository, + @InjectRepository(QqbotPluginTaskRun) + private readonly runRepository: Repository, + ) {} + + async onModuleInit() { + this.worker = new Worker( + 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) { + 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, + ) { + 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, + ) { + 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).sort() + : []; + } +} diff --git a/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task.service.ts b/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task.service.ts index 1ec67b7..b3e41ab 100644 --- a/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task.service.ts +++ b/src/modules/qqbot/plugin-platform/application/task/qqbot-plugin-task.service.ts @@ -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, 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 }) { - 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; + } } diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker.thread.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker.thread.ts index 1a2bf32..07af9b9 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker.thread.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker.thread.ts @@ -36,6 +36,11 @@ import type { QqbotPluginWorkerRequest } from './worker-runtime.types'; type RuntimeCommandPlugin = QqbotIntegrationPlugin & { activate?: () => Promise | unknown; dispose?: () => Promise | unknown; + tasks?: Array<{ + execute(input: Record): Promise | 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); +} + async function handleEvent(message: QqbotPluginWorkerRequest) { if (!eventPlugin) return false; const definition = eventPlugin.getDefinition(); diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts index 7237acf..cf740af 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts @@ -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'); } diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts index e632251..b6ef3df 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts @@ -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; + taskHandlerName: string; + taskId: string; + taskKey: string; + timeoutMs?: number; + triggerType: 'bootstrap' | 'manual' | 'schedule'; +}; diff --git a/src/modules/qqbot/plugin-platform/plugin-platform.module.ts b/src/modules/qqbot/plugin-platform/plugin-platform.module.ts index 5bdee06..363d0f4 100644 --- a/src/modules/qqbot/plugin-platform/plugin-platform.module.ts +++ b/src/modules/qqbot/plugin-platform/plugin-platform.module.ts @@ -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, diff --git a/test/modules/qqbot/plugin-platform/plugin-task-scheduler.spec.ts b/test/modules/qqbot/plugin-platform/plugin-task-scheduler.spec.ts new file mode 100644 index 0000000..dce3b31 --- /dev/null +++ b/test/modules/qqbot/plugin-platform/plugin-task-scheduler.spec.ts @@ -0,0 +1,208 @@ +const createdQueues: any[] = []; +const createdWorkers: any[] = []; + +jest.mock('bullmq', () => ({ + Queue: class MockQueue { + readonly schedulers = new Map(); + + 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, + 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; + }), + }; +} diff --git a/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts b/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts index 2addde9..6e8a74f 100644 --- a/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts +++ b/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts @@ -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(