import { FastifyRequest, FastifyReply } from 'fastify'; import { logger } from '../../../common/logging/index.js'; import { PrismaClient } from '@prisma/client'; import { RedcapAdapter } from '../adapters/RedcapAdapter.js'; import { jobQueue } from '../../../common/jobs/index.js'; /** * REDCap DET Webhook请求体 * * REDCap发送的POST请求包含以下字段: */ interface RedcapWebhookPayload { /** 项目ID */ project_id: string; /** 记录ID */ record: string; /** 表单名称 */ instrument: string; /** 事件名称(纵向研究,可选) */ redcap_event_name?: string; /** 重复实例(可选) */ redcap_repeat_instance?: string; /** 重复表单(可选) */ redcap_repeat_instrument?: string; /** REDCap版本 */ redcap_version?: string; /** REDCap URL */ redcap_url?: string; /** 项目URL */ project_url?: string; } /** * Webhook控制器 * * 职责: * 1. 接收REDCap DET触发的Webhook请求 * 2. 极速响应(<100ms)避免REDCap超时 * 3. 异步处理:拉取完整数据、推送质控队列 * * 性能要求: * - 同步返回200 OK: <100ms * - 数据拉取: <2s * - 企业微信通知: <5s(整体流程) */ export class WebhookController { private prisma: PrismaClient; constructor() { this.prisma = new PrismaClient(); } /** * 处理REDCap Webhook请求 * * 关键设计: * - 立即返回200 OK(<100ms) * - 使用setImmediate异步处理真实业务 * - 防重复:5分钟内同一record+instrument不重复处理 * * @param request Fastify请求 * @param reply Fastify响应 */ async handleWebhook( request: FastifyRequest<{ Body: RedcapWebhookPayload }>, reply: FastifyReply ): Promise { const payload = request.body; // 验证必填参数 if (!payload.project_id || !payload.record || !payload.instrument) { logger.warn('Invalid webhook payload: missing required fields', { payload }); return reply.code(400).send({ error: 'Missing required fields: project_id, record, instrument' }); } logger.info('REDCap Webhook received', { project_id: payload.project_id, record: payload.record, instrument: payload.instrument, event: payload.redcap_event_name }); // 🚀 立即返回200 OK(避免REDCap超时) reply.code(200).send({ status: 'received' }); // 🔄 异步处理真实业务(不阻塞响应) setImmediate(() => { this.processWebhookAsync(payload).catch((error) => { logger.error('Webhook async processing failed', { error: error.message, payload }); }); }); } /** * 异步处理Webhook(真实业务逻辑) * * 流程: * 1. 查找项目配置 * 2. 防重复检查(5分钟幂等窗口) * 3. 拉取完整记录数据 * 4. 推送到质控队列 * 5. 记录审计日志 * * @param payload Webhook负载 */ private async processWebhookAsync(payload: RedcapWebhookPayload): Promise { const startTime = Date.now(); try { // ============================================= // 1. 查找项目配置 // ============================================= const projectConfig = await this.prisma.iitProject.findFirst({ where: { redcapProjectId: String(payload.project_id) } }); if (!projectConfig) { logger.warn('Project not found in IIT system', { project_id: payload.project_id }); return; } // 验证项目状态 if (projectConfig.status !== 'active') { logger.info('Project not active, skipping webhook', { project_id: payload.project_id, status: projectConfig.status }); return; } // ============================================= // 2. 防重复检查(幂等性保证) // ============================================= const isDuplicate = await this.checkDuplicate( projectConfig.id, payload.record, payload.instrument ); if (isDuplicate) { logger.info('Duplicate webhook detected, skipping', { project_id: payload.project_id, record: payload.record, instrument: payload.instrument }); return; } // ============================================= // 3. 拉取完整记录数据 // ============================================= const adapter = new RedcapAdapter( projectConfig.redcapUrl, projectConfig.redcapApiToken ); const records = await adapter.exportRecords({ records: [payload.record] }); if (!records || records.length === 0) { logger.warn('No data returned from REDCap', { project_id: payload.project_id, record: payload.record }); return; } logger.info('Record data fetched from REDCap', { project_id: payload.project_id, record: payload.record, recordCount: records.length }); // ============================================= // 4. 推送到质控队列(pg-boss) // ============================================= await jobQueue.push('iit_quality_check', { projectId: projectConfig.id, redcapProjectId: parseInt(payload.project_id), recordId: payload.record, instrument: payload.instrument, event: payload.redcap_event_name, records: records, triggeredBy: 'webhook', triggeredAt: new Date().toISOString() }); logger.info('Quality check job queued', { projectId: projectConfig.id, recordId: payload.record, instrument: payload.instrument }); // ============================================= // 5. 记录审计日志 // ============================================= await this.prisma.iitAuditLog.create({ data: { projectId: projectConfig.id, userId: 'system', actionType: 'WEBHOOK_RECEIVED', entityType: 'RECORD', entityId: payload.record, details: { source: 'redcap_det', project_id: payload.project_id, record: payload.record, instrument: payload.instrument, event: payload.redcap_event_name }, traceId: `webhook-${Date.now()}`, createdAt: new Date() } }); const totalDuration = Date.now() - startTime; logger.info('Webhook processing completed', { project_id: payload.project_id, record: payload.record, duration: `${totalDuration}ms` }); } catch (error: any) { logger.error('Webhook processing error', { error: error.message, stack: error.stack, payload }); // 记录失败的审计日志 try { await this.prisma.iitAuditLog.create({ data: { projectId: 'unknown', userId: 'system', actionType: 'WEBHOOK_ERROR', entityType: 'WEBHOOK', entityId: payload.record || 'unknown', details: { error: error.message, payload: JSON.parse(JSON.stringify(payload)) }, traceId: `webhook-error-${Date.now()}`, createdAt: new Date() } }); } catch (auditError) { logger.error('Failed to create audit log', { error: auditError }); } } } /** * 防重复检查(幂等性保证) * * 场景: * - REDCap可能重复发送Webhook * - 网络重试可能导致重复 * - CRC快速保存多次 * * 策略:5分钟内同一record+instrument不重复处理 * * @param projectId IIT项目ID * @param recordId REDCap记录ID * @param instrument 表单名称 * @returns 是否重复 */ private async checkDuplicate( projectId: string, recordId: string, instrument: string ): Promise { const fiveMinutesAgo = new Date(Date.now() - 5 * 60 * 1000); const existingLog = await this.prisma.iitAuditLog.findFirst({ where: { projectId: projectId, actionType: 'WEBHOOK_RECEIVED', entityId: recordId, createdAt: { gte: fiveMinutesAgo } }, orderBy: { createdAt: 'desc' } }); // 如果找到了,还需要检查instrument是否匹配 if (existingLog) { const detail = existingLog.details as any; if (detail?.instrument === instrument) { return true; } } return false; } /** * 健康检查端点 * * 用途:验证Webhook服务是否正常运行 */ async healthCheck( request: FastifyRequest, reply: FastifyReply ): Promise { return reply.code(200).send({ status: 'ok', service: 'IIT Manager Webhook', timestamp: new Date().toISOString() }); } }