From f6d9f235534733dfa0a46c6adddd89532c8d5975 Mon Sep 17 00:00:00 2001 From: shunli Date: Sun, 8 Mar 2026 13:14:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(P19IM1-1,P19IM1-2):=20=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E9=80=9A=E7=9F=A5=E6=B8=A0=E9=81=93=E9=85=8D=E7=BD=AE=E4=B8=8E?= =?UTF-8?q?Outbox=E6=B4=BE=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/src/constants/notificationTypes.js | 86 ++++ server/src/jobs/integrationOutboxJob.js | 44 +- server/src/routes/notifications.js | 455 ++++++++++++++---- server/src/services/email/index.js | 152 +----- .../src/services/integrations/dispatcher.js | 56 ++- server/src/services/integrations/im/wecom.js | 183 +++++++ server/src/services/notifications/channels.js | 96 ++-- .../src/services/notifications/imEnqueue.js | 199 ++++++++ .../src/services/notifications/imRouting.js | 74 +++ .../unit/integrations.dispatcher.test.js | 270 +++++++++-- server/tests/unit/services/imEnqueue.test.js | 170 +++++++ server/tests/unit/services/imRouting.test.js | 91 ++++ .../services/integrations/dispatcher.test.js | 70 +-- 13 files changed, 1582 insertions(+), 364 deletions(-) create mode 100644 server/src/constants/notificationTypes.js create mode 100644 server/src/services/integrations/im/wecom.js create mode 100644 server/src/services/notifications/imEnqueue.js create mode 100644 server/src/services/notifications/imRouting.js create mode 100644 server/tests/unit/services/imEnqueue.test.js create mode 100644 server/tests/unit/services/imRouting.test.js diff --git a/server/src/constants/notificationTypes.js b/server/src/constants/notificationTypes.js new file mode 100644 index 00000000..35aca5b5 --- /dev/null +++ b/server/src/constants/notificationTypes.js @@ -0,0 +1,86 @@ +const NOTIFICATION_DOMAIN_SCOPES = Object.freeze({ + workflow: 'workflow', + reimbursement: 'reimbursement', + leave: 'leave', + procurement: 'procurement', + asset: 'asset', + inventory: 'inventory', +}); + +const NOTIFICATION_TYPES = Object.freeze({ + workflowTaskAssigned: 'workflow.task_assigned', + reimbursementApproved: 'reimbursement.approved', + reimbursementRejected: 'reimbursement.rejected', + reimbursementReturned: 'reimbursement.returned', + reimbursementPaid: 'reimbursement.paid', + leaveApproved: 'leave.approved', + leaveRejected: 'leave.rejected', + purchaseRequestApproved: 'purchase_request.approved', + purchaseRequestRejected: 'purchase_request.rejected', + assetAssigned: 'asset.assigned', + assetReturned: 'asset.returned', + assetTransferred: 'asset.transferred', + inventoryCheckCompleted: 'inventory_check.completed', +}); + +const IM_SUPPORTED_NOTIFICATION_TYPES = Object.freeze([ + NOTIFICATION_TYPES.workflowTaskAssigned, + NOTIFICATION_TYPES.reimbursementApproved, + NOTIFICATION_TYPES.reimbursementRejected, + NOTIFICATION_TYPES.reimbursementReturned, + NOTIFICATION_TYPES.reimbursementPaid, + NOTIFICATION_TYPES.leaveApproved, + NOTIFICATION_TYPES.leaveRejected, + NOTIFICATION_TYPES.purchaseRequestApproved, + NOTIFICATION_TYPES.purchaseRequestRejected, + NOTIFICATION_TYPES.assetAssigned, + NOTIFICATION_TYPES.assetReturned, + NOTIFICATION_TYPES.assetTransferred, + NOTIFICATION_TYPES.inventoryCheckCompleted, +]); + +const NOTIFICATION_TYPE_TO_DOMAIN_SCOPE = Object.freeze({ + [NOTIFICATION_TYPES.workflowTaskAssigned]: NOTIFICATION_DOMAIN_SCOPES.workflow, + [NOTIFICATION_TYPES.reimbursementApproved]: NOTIFICATION_DOMAIN_SCOPES.reimbursement, + [NOTIFICATION_TYPES.reimbursementRejected]: NOTIFICATION_DOMAIN_SCOPES.reimbursement, + [NOTIFICATION_TYPES.reimbursementReturned]: NOTIFICATION_DOMAIN_SCOPES.reimbursement, + [NOTIFICATION_TYPES.reimbursementPaid]: NOTIFICATION_DOMAIN_SCOPES.reimbursement, + [NOTIFICATION_TYPES.leaveApproved]: NOTIFICATION_DOMAIN_SCOPES.leave, + [NOTIFICATION_TYPES.leaveRejected]: NOTIFICATION_DOMAIN_SCOPES.leave, + [NOTIFICATION_TYPES.purchaseRequestApproved]: NOTIFICATION_DOMAIN_SCOPES.procurement, + [NOTIFICATION_TYPES.purchaseRequestRejected]: NOTIFICATION_DOMAIN_SCOPES.procurement, + [NOTIFICATION_TYPES.assetAssigned]: NOTIFICATION_DOMAIN_SCOPES.asset, + [NOTIFICATION_TYPES.assetReturned]: NOTIFICATION_DOMAIN_SCOPES.asset, + [NOTIFICATION_TYPES.assetTransferred]: NOTIFICATION_DOMAIN_SCOPES.asset, + [NOTIFICATION_TYPES.inventoryCheckCompleted]: NOTIFICATION_DOMAIN_SCOPES.inventory, +}); + +const TARGET_TYPE_TO_DOMAIN_SCOPE = Object.freeze({ + workflow_task: NOTIFICATION_DOMAIN_SCOPES.workflow, + reimbursement_form: NOTIFICATION_DOMAIN_SCOPES.reimbursement, + leave_request: NOTIFICATION_DOMAIN_SCOPES.leave, + purchase_request: NOTIFICATION_DOMAIN_SCOPES.procurement, + asset_card: NOTIFICATION_DOMAIN_SCOPES.asset, + inventory_check: NOTIFICATION_DOMAIN_SCOPES.inventory, +}); + +function resolveNotificationDomainScope(type, targetType = null) { + const normalizedType = type == null ? '' : String(type).trim(); + if (normalizedType && NOTIFICATION_TYPE_TO_DOMAIN_SCOPE[normalizedType]) { + return NOTIFICATION_TYPE_TO_DOMAIN_SCOPE[normalizedType]; + } + + const normalizedTargetType = targetType == null ? '' : String(targetType).trim(); + if (normalizedTargetType && TARGET_TYPE_TO_DOMAIN_SCOPE[normalizedTargetType]) { + return TARGET_TYPE_TO_DOMAIN_SCOPE[normalizedTargetType]; + } + + return null; +} + +module.exports = { + NOTIFICATION_TYPES, + IM_SUPPORTED_NOTIFICATION_TYPES, + NOTIFICATION_DOMAIN_SCOPES, + resolveNotificationDomainScope, +}; diff --git a/server/src/jobs/integrationOutboxJob.js b/server/src/jobs/integrationOutboxJob.js index bef1fbfb..0455f10f 100644 --- a/server/src/jobs/integrationOutboxJob.js +++ b/server/src/jobs/integrationOutboxJob.js @@ -1,12 +1,12 @@ -const cron = require("node-cron"); +const cron = require('node-cron'); const { fetchPendingOutboxEvents, markOutboxFailed, markOutboxProcessing, markOutboxSucceeded, -} = require("../services/integrations/outbox"); -const { dispatchOutboxEvent } = require("../services/integrations/dispatcher"); +} = require('../services/integrations/outbox'); +const { dispatchOutboxEvent } = require('../services/integrations/dispatcher'); async function runIntegrationOutboxOnce(prisma, config, logger) { const items = await fetchPendingOutboxEvents(prisma, { @@ -21,6 +21,7 @@ async function runIntegrationOutboxOnce(prisma, config, logger) { try { await dispatchOutboxEvent({ + prisma, config, eventType: item.event_type, payload: item.payload, @@ -28,7 +29,7 @@ async function runIntegrationOutboxOnce(prisma, config, logger) { await markOutboxSucceeded(prisma, item.id, null); succeeded += 1; } catch (error) { - const retryable = typeof error?.retryable === "boolean" ? error.retryable : true; + const retryable = typeof error?.retryable === 'boolean' ? error.retryable : true; await markOutboxFailed(prisma, config, item, error, null, { retryable }); failed += 1; logger.error( @@ -37,8 +38,11 @@ async function runIntegrationOutboxOnce(prisma, config, logger) { retryable, outboxId: item.id, eventType: item.event_type, + channelId: item.payload?.channelId ?? null, + channelType: item.payload?.channelType ?? null, + notificationType: item.payload?.notificationType ?? null, }, - "Integration outbox dispatch failed", + 'Integration outbox dispatch failed', ); } } @@ -66,16 +70,16 @@ function startIntegrationOutboxJob(fastify) { fastify.log, ); fastify.metrics?.jobRunsTotal?.inc?.({ - job: "integration.outbox.dispatch", - driver: "cron", - outcome: "success", + job: 'integration.outbox.dispatch', + driver: 'cron', + outcome: 'success', }); if (result.succeeded > 0) { fastify.metrics?.jobProcessedTotal?.inc?.( { - job: "integration.outbox.dispatch", - driver: "cron", - kind: "succeeded", + job: 'integration.outbox.dispatch', + driver: 'cron', + kind: 'succeeded', }, result.succeeded, ); @@ -83,26 +87,26 @@ function startIntegrationOutboxJob(fastify) { if (result.failed > 0) { fastify.metrics?.jobProcessedTotal?.inc?.( { - job: "integration.outbox.dispatch", - driver: "cron", - kind: "failed", + job: 'integration.outbox.dispatch', + driver: 'cron', + kind: 'failed', }, result.failed, ); } } catch (error) { - fastify.log.error({ err: error }, "Integration outbox job failed"); + fastify.log.error({ err: error }, 'Integration outbox job failed'); fastify.metrics?.jobRunsTotal?.inc?.({ - job: "integration.outbox.dispatch", - driver: "cron", - outcome: "failure", + job: 'integration.outbox.dispatch', + driver: 'cron', + outcome: 'failure', }); } finally { running = false; } }); - fastify.addHook("onClose", async () => { + fastify.addHook('onClose', async () => { task.stop(); task.destroy(); }); @@ -112,7 +116,7 @@ function startIntegrationOutboxJob(fastify) { cron: fastify.config.integrationOutboxCron, batchSize: fastify.config.integrationOutboxBatchSize, }, - "Integration outbox job scheduled", + 'Integration outbox job scheduled', ); return task; diff --git a/server/src/routes/notifications.js b/server/src/routes/notifications.js index 5a95dca7..26f50c28 100644 --- a/server/src/routes/notifications.js +++ b/server/src/routes/notifications.js @@ -1,31 +1,114 @@ -const { z } = require("zod"); +const { z } = require('zod'); -const { AppError } = require("../utils/AppError"); -const { sendData } = require("../utils/reply"); -const { buildAuditLogData } = require("../utils/audit"); -const { sendInstantMessageByChannel } = require("../services/notifications/channels"); +const { AppError } = require('../utils/AppError'); +const { sendData } = require('../utils/reply'); +const { buildAuditLogData } = require('../utils/audit'); +const { sendInstantMessageByChannel } = require('../services/notifications/channels'); +const { ROLE_PERMISSIONS } = require('../constants/permissions'); +const { + IM_SUPPORTED_NOTIFICATION_TYPES, + NOTIFICATION_DOMAIN_SCOPES, +} = require('../constants/notificationTypes'); const LIST_QUERY_SCHEMA = z.object({ page: z.coerce.number().int().min(1).optional(), pageSize: z.coerce.number().int().min(1).max(100).optional(), - unreadOnly: z.enum(["0", "1", "true", "false"]).optional(), + unreadOnly: z.enum(['0', '1', 'true', 'false']).optional(), }); +const CHANNEL_ID_SCHEMA = z.string().uuid(); +const ROLE_CODE_VALUES = Object.freeze(Object.keys(ROLE_PERMISSIONS)); +const CHANNEL_TYPE_VALUES = Object.freeze(['wecom', 'wechat_work']); +const CHANNEL_MODE_VALUES = Object.freeze(['mock', 'wecom_webhook']); +const DOMAIN_SCOPE_VALUES = Object.freeze(Object.values(NOTIFICATION_DOMAIN_SCOPES)); +const DEFAULT_TEMPLATE_JSON = Object.freeze({ + title: '{{title}}', + body: '{{body}}\n\n查看详情：{{detailUrl}}', + footer: '来自海音达工作台', + deepLinkLabel: '查看详情', +}); + +const ROUTING_SCHEMA = z + .object({ + roleCodes: z.array(z.enum(ROLE_CODE_VALUES)).optional(), + domainScopes: z.array(z.enum(DOMAIN_SCOPE_VALUES)).optional(), + allowedTypes: z.array(z.enum(IM_SUPPORTED_NOTIFICATION_TYPES)).optional(), + }) + .strict(); + +const CONFIG_SCHEMA = z + .object({ + mode: z.enum(CHANNEL_MODE_VALUES), + webhookUrl: z.string().trim().max(2000).nullish(), + secret: z.string().trim().max(500).nullish(), + timeoutMs: z.coerce.number().int().min(100).max(60000).optional(), + routing: ROUTING_SCHEMA.optional(), + }) + .strict() + .superRefine((value, ctx) => { + if (value.mode === 'wecom_webhook' && !String(value.webhookUrl ?? '').trim()) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['webhookUrl'], + message: 'wecom_webhook_requires_webhook_url', + }); + } + }); + +const TEMPLATE_SCHEMA = z + .object({ + title: z.string().trim().min(1).max(200).optional(), + body: z.string().trim().min(1).max(4000).optional(), + footer: z.string().trim().min(1).max(200).optional(), + deepLinkLabel: z.string().trim().min(1).max(50).optional(), + }) + .strict(); + +const CHANNEL_CREATE_SCHEMA = z + .object({ + channelType: z.enum(CHANNEL_TYPE_VALUES), + name: z.string().trim().min(1).max(100), + enabled: z.boolean().optional(), + configJson: CONFIG_SCHEMA.optional(), + templateJson: TEMPLATE_SCHEMA.optional(), + }) + .strict(); + +const CHANNEL_PATCH_SCHEMA = z + .object({ + name: z.string().trim().min(1).max(100).optional(), + enabled: z.boolean().optional(), + configJson: CONFIG_SCHEMA.optional(), + templateJson: TEMPLATE_SCHEMA.optional(), + }) + .strict() + .refine((value) => Object.keys(value).length > 0, { + message: 'empty_patch', + }); + +function buildValidationError(details, statusCode = 400) { + return new AppError({ + code: 'VALID_001', + message: '参数校验失败', + statusCode, + details, + }); +} + +function uniqStrings(values) { + return Array.from(new Set((values ?? []).filter(Boolean))); +} + function parseListQuery(query) { const parsed = LIST_QUERY_SCHEMA.safeParse(query ?? {}); if (!parsed.success) { - throw new AppError({ - code: "VALID_001", - message: "参数校验失败", - statusCode: 400, - details: parsed.error.flatten(), - }); + throw buildValidationError(parsed.error.flatten()); } const page = parsed.data.page ?? 1; const pageSize = parsed.data.pageSize ?? 20; - const unreadOnlyRaw = parsed.data.unreadOnly ?? "0"; - const unreadOnly = unreadOnlyRaw === "1" || unreadOnlyRaw === "true"; + const unreadOnlyRaw = parsed.data.unreadOnly ?? '0'; + const unreadOnly = unreadOnlyRaw === '1' || unreadOnlyRaw === 'true'; return { page, @@ -35,18 +118,49 @@ function parseListQuery(query) { } function parseNotificationId(value) { - const parsed = z.string().uuid().safeParse(value); + const parsed = CHANNEL_ID_SCHEMA.safeParse(value); if (!parsed.success) { - throw new AppError({ - code: "VALID_001", - message: "参数校验失败", - statusCode: 400, - details: { id: "invalid_uuid" }, - }); + throw buildValidationError({ id: 'invalid_uuid' }); } return parsed.data; } +function parseBody(schema, body) { + const parsed = schema.safeParse(body ?? {}); + if (!parsed.success) { + throw buildValidationError(parsed.error.flatten()); + } + return parsed.data; +} + +function normalizeRouting(routing) { + return { + roleCodes: uniqStrings((routing?.roleCodes ?? []).map((item) => String(item).trim())), + domainScopes: uniqStrings((routing?.domainScopes ?? []).map((item) => String(item).trim())), + allowedTypes: uniqStrings((routing?.allowedTypes ?? []).map((item) => String(item).trim())), + }; +} + +function normalizeConfigJson(input = {}) { + return { + mode: input.mode ?? 'mock', + webhookUrl: String(input.webhookUrl ?? '').trim() || null, + secret: String(input.secret ?? '').trim() || null, + timeoutMs: Math.max(100, Number(input.timeoutMs ?? 5000)), + routing: normalizeRouting(input.routing), + }; +} + +function normalizeTemplateJson(input = {}) { + return { + title: String(input.title ?? DEFAULT_TEMPLATE_JSON.title).trim() || DEFAULT_TEMPLATE_JSON.title, + body: String(input.body ?? DEFAULT_TEMPLATE_JSON.body).trim() || DEFAULT_TEMPLATE_JSON.body, + footer: String(input.footer ?? DEFAULT_TEMPLATE_JSON.footer).trim() || DEFAULT_TEMPLATE_JSON.footer, + deepLinkLabel: + String(input.deepLinkLabel ?? DEFAULT_TEMPLATE_JSON.deepLinkLabel).trim() || DEFAULT_TEMPLATE_JSON.deepLinkLabel, + }; +} + function mapNotificationItem(record) { return { id: record.id, @@ -61,21 +175,45 @@ function mapNotificationItem(record) { } function mapNotificationChannel(record) { + const configJson = normalizeConfigJson(record.config_json); + const templateJson = normalizeTemplateJson(record.template_json ?? undefined); + return { id: record.id, channelType: record.channel_type, name: record.name, enabled: record.enabled, - configJson: record.config_json, - templateJson: record.template_json, + configJson, + templateJson, createdAt: record.created_at?.toISOString?.() ?? null, updatedAt: record.updated_at?.toISOString?.() ?? null, }; } +function handleNotificationChannelWriteError(error) { + if (error?.code === 'P2002') { + throw buildValidationError({ name: 'duplicate_channel_name' }, 409); + } + throw error; +} + +async function loadNotificationChannelOrThrow(prisma, channelId) { + const channel = await prisma.notificationChannel.findUnique({ + where: { id: channelId }, + }); + if (!channel) { + throw new AppError({ + code: 'SYS_404', + message: 'Not Found', + statusCode: 404, + }); + } + return channel; +} + async function notificationsRoutes(fastify) { fastify.get( - "/notifications", + '/notifications', { preHandler: fastify.authenticate, }, @@ -91,7 +229,7 @@ async function notificationsRoutes(fastify) { const [items, total, unreadCount] = await Promise.all([ fastify.prisma.notification.findMany({ where, - orderBy: { created_at: "desc" }, + orderBy: { created_at: 'desc' }, skip, take: query.pageSize, select: { @@ -126,7 +264,7 @@ async function notificationsRoutes(fastify) { ); fastify.post( - "/notifications/:id/mark-read", + '/notifications/:id/mark-read', { preHandler: fastify.authenticate, }, @@ -142,8 +280,8 @@ async function notificationsRoutes(fastify) { `; if (!Array.isArray(locked) || locked.length === 0) { throw new AppError({ - code: "SYS_404", - message: "Not Found", + code: 'SYS_404', + message: 'Not Found', statusCode: 404, }); } @@ -165,16 +303,16 @@ async function notificationsRoutes(fastify) { if (!current) { throw new AppError({ - code: "SYS_404", - message: "Not Found", + code: 'SYS_404', + message: 'Not Found', statusCode: 404, }); } if (current.recipient_employee_id !== request.employee.id) { throw new AppError({ - code: "PERM_001", - message: "无权限", + code: 'PERM_001', + message: '无权限', statusCode: 403, }); } @@ -206,13 +344,17 @@ async function notificationsRoutes(fastify) { ); fastify.get( - "/notifications/channels", + '/notifications/channels', { - preHandler: fastify.requirePermission("notification.channel.manage"), + preHandler: fastify.requirePermission('notification.channel.manage'), + schema: { + tags: ['notifications'], + summary: 'List Notification Channels', + }, }, async (request, reply) => { const items = await fastify.prisma.notificationChannel.findMany({ - orderBy: [{ channel_type: "asc" }, { name: "asc" }], + orderBy: [{ channel_type: 'asc' }, { name: 'asc' }], }); return sendData(reply, request, { @@ -222,63 +364,120 @@ async function notificationsRoutes(fastify) { ); fastify.post( - "/notifications/channels/test", + '/notifications/channels', { - preHandler: fastify.requirePermission("notification.channel.manage"), + preHandler: fastify.requirePermission('notification.channel.manage'), + schema: { + tags: ['notifications'], + summary: 'Create Notification Channel', + }, }, async (request, reply) => { - const schema = z - .object({ - channelType: z.string().trim().min(1).max(32), - name: z.string().trim().min(1).max(100), - }) - .strict(); + const input = parseBody(CHANNEL_CREATE_SCHEMA, request.body); - const parsed = schema.safeParse(request.body ?? {}); - if (!parsed.success) { - throw new AppError({ - code: "VALID_001", - message: "参数校验失败", - statusCode: 400, - details: parsed.error.flatten(), - }); - } - - const input = parsed.data; - - const channel = await fastify.prisma.notificationChannel.upsert({ - where: { - channel_type_name: { + try { + const channel = await fastify.prisma.notificationChannel.create({ + data: { channel_type: input.channelType, name: input.name, + enabled: input.enabled ?? true, + config_json: normalizeConfigJson(input.configJson), + template_json: normalizeTemplateJson(input.templateJson), + created_by: request.employee.id, + updated_by: request.employee.id, }, - }, - update: { - enabled: true, - config_json: { - mode: "mock", + }); + + await fastify.prisma.auditLog.create({ + data: buildAuditLogData(request, { + actorId: request.employee.id, + action: 'NOTIFICATION_CHANNEL_CREATED', + targetType: 'notification_channel', + targetId: channel.id, + extra: { + channelType: channel.channel_type, + name: channel.name, + }, + }), + }); + + return sendData(reply, request, { + channel: mapNotificationChannel(channel), + }); + } catch (error) { + handleNotificationChannelWriteError(error); + } + }, + ); + + fastify.patch( + '/notifications/channels/:id', + { + preHandler: fastify.requirePermission('notification.channel.manage'), + schema: { + tags: ['notifications'], + summary: 'Update Notification Channel', + }, + }, + async (request, reply) => { + const channelId = parseNotificationId(request.params?.id); + const input = parseBody(CHANNEL_PATCH_SCHEMA, request.body); + await loadNotificationChannelOrThrow(fastify.prisma, channelId); + + try { + const channel = await fastify.prisma.notificationChannel.update({ + where: { id: channelId }, + data: { + ...(input.name ? { name: input.name } : {}), + ...(typeof input.enabled === 'boolean' ? { enabled: input.enabled } : {}), + ...(input.configJson ? { config_json: normalizeConfigJson(input.configJson) } : {}), + ...(input.templateJson ? { template_json: normalizeTemplateJson(input.templateJson) } : {}), + updated_by: request.employee.id, }, - updated_by: request.employee.id, - }, - create: { - channel_type: input.channelType, - name: input.name, - enabled: true, - config_json: { - mode: "mock", - }, - created_by: request.employee.id, - updated_by: request.employee.id, - }, - }); + }); + + await fastify.prisma.auditLog.create({ + data: buildAuditLogData(request, { + actorId: request.employee.id, + action: 'NOTIFICATION_CHANNEL_UPDATED', + targetType: 'notification_channel', + targetId: channel.id, + extra: { + channelType: channel.channel_type, + name: channel.name, + }, + }), + }); + + return sendData(reply, request, { + channel: mapNotificationChannel(channel), + }); + } catch (error) { + handleNotificationChannelWriteError(error); + } + }, + ); + + fastify.post( + '/notifications/channels/:id/test', + { + preHandler: fastify.requirePermission('notification.channel.manage'), + schema: { + tags: ['notifications'], + summary: 'Test Notification Channel By Id', + }, + }, + async (request, reply) => { + const channelId = parseNotificationId(request.params?.id); + const channel = await loadNotificationChannelOrThrow(fastify.prisma, channelId); const result = await sendInstantMessageByChannel({ config: fastify.config, channel, payload: { - title: "通知渠道测试", - body: `测试人：${request.employee.id}`, - targetType: "notification_channel", + title: '通知渠道测试', + body: `测试人：${request.employee.name}（${request.employee.id}）`, + targetType: 'notification_channel', targetId: channel.id, }, }); @@ -286,8 +485,8 @@ async function notificationsRoutes(fastify) { await fastify.prisma.auditLog.create({ data: buildAuditLogData(request, { actorId: request.employee.id, - action: "NOTIFICATION_CHANNEL_TESTED", - targetType: "notification_channel", + action: 'NOTIFICATION_CHANNEL_TESTED', + targetType: 'notification_channel', targetId: channel.id, extra: { channelType: channel.channel_type, @@ -299,7 +498,99 @@ async function notificationsRoutes(fastify) { return sendData(reply, request, { channel: mapNotificationChannel(channel), - testResult: result, + testResult: { + ok: result.ok, + code: result.code, + previewMessage: result.previewMessage ?? null, + response: result.response ?? null, + skipped: result.skipped ?? false, + }, + }); + }, + ); + + fastify.post( + '/notifications/channels/test', + { + preHandler: fastify.requirePermission('notification.channel.manage'), + schema: { + tags: ['notifications'], + summary: 'Legacy Test Notification Channel', + }, + }, + async (request, reply) => { + const schema = z + .object({ + channelType: z.enum(CHANNEL_TYPE_VALUES), + name: z.string().trim().min(1).max(100), + }) + .strict(); + + const input = parseBody(schema, request.body); + + let channel; + try { + channel = await fastify.prisma.notificationChannel.upsert({ + where: { + channel_type_name: { + channel_type: input.channelType, + name: input.name, + }, + }, + update: { + enabled: true, + config_json: normalizeConfigJson({ mode: 'mock' }), + template_json: normalizeTemplateJson(), + updated_by: request.employee.id, + }, + create: { + channel_type: input.channelType, + name: input.name, + enabled: true, + config_json: normalizeConfigJson({ mode: 'mock' }), + template_json: normalizeTemplateJson(), + created_by: request.employee.id, + updated_by: request.employee.id, + }, + }); + } catch (error) { + handleNotificationChannelWriteError(error); + } + + const result = await sendInstantMessageByChannel({ + config: fastify.config, + channel, + payload: { + title: '通知渠道测试', + body: `测试人：${request.employee.id}`, + targetType: 'notification_channel', + targetId: channel.id, + }, + }); + + await fastify.prisma.auditLog.create({ + data: buildAuditLogData(request, { + actorId: request.employee.id, + action: 'NOTIFICATION_CHANNEL_TESTED', + targetType: 'notification_channel', + targetId: channel.id, + extra: { + channelType: channel.channel_type, + name: channel.name, + resultCode: result.code, + }, + }), + }); + + return sendData(reply, request, { + channel: mapNotificationChannel(channel), + testResult: { + ok: result.ok, + code: result.code, + previewMessage: result.previewMessage ?? null, + response: result.response ?? null, + skipped: result.skipped ?? false, + }, }); }, ); diff --git a/server/src/services/email/index.js b/server/src/services/email/index.js index ed1ff586..e9141121 100644 --- a/server/src/services/email/index.js +++ b/server/src/services/email/index.js @@ -1,19 +1,17 @@ -const nodemailer = require("nodemailer"); - -const { sendInstantMessageByChannel } = require("../notifications/channels"); +const nodemailer = require('nodemailer'); function resolveEmailBaseUrl(config) { - const explicit = String(config.emailBaseUrl ?? "").trim(); + const explicit = String(config.emailBaseUrl ?? '').trim(); if (explicit) { - return explicit.replace(/\/+$/, ""); + return explicit.replace(/\/+$/, ''); } - const origin = String(config.appOrigin ?? "").split(",")[0]?.trim?.() ?? ""; + const origin = String(config.appOrigin ?? '').split(',')[0]?.trim?.() ?? ''; if (origin) { - return origin.replace(/\/+$/, ""); + return origin.replace(/\/+$/, ''); } - return ""; + return ''; } let cachedTransportKey = null; @@ -28,8 +26,8 @@ function getTransporter(config) { return null; } - const authUser = String(config.smtpUser ?? "").trim(); - const authPass = String(config.smtpPass ?? "").trim(); + const authUser = String(config.smtpUser ?? '').trim(); + const authPass = String(config.smtpPass ?? '').trim(); const key = JSON.stringify({ host: config.smtpHost, @@ -59,14 +57,9 @@ function uniq(values) { async function sendNotificationEmailSafe(params) { try { - const [emailResult, imResult] = await Promise.all([ - sendNotificationEmail(params), - sendNotificationIm(params), - ]); - + const emailResult = await sendNotificationEmail(params); return { email: emailResult, - im: imResult, }; } catch (error) { params.logger?.error?.( @@ -76,11 +69,10 @@ async function sendNotificationEmailSafe(params) { notificationType: params.type, recipients: params.recipientEmployeeIds?.length ?? 0, }, - "Failed to send notification", + 'Failed to send notification email', ); return { email: { skipped: false, sent: 0, failed: params.recipientEmployeeIds?.length ?? 0 }, - im: { skipped: false, sent: 0, failed: params.recipientEmployeeIds?.length ?? 0 }, }; } } @@ -93,7 +85,7 @@ async function sendNotificationEmail({ recipientEmployeeIds, subject, body, - type = "notification", + type = 'notification', targetPath, }) { if (!config.emailEnabled) { @@ -102,7 +94,7 @@ async function sendNotificationEmail({ const transporter = getTransporter(config); if (!transporter) { - logger?.warn?.({ requestId }, "Email enabled but transporter not configured"); + logger?.warn?.({ requestId }, 'Email enabled but transporter not configured'); return { skipped: true, sent: 0, failed: 0 }; } @@ -114,7 +106,7 @@ async function sendNotificationEmail({ const employees = await prisma.employee.findMany({ where: { id: { in: recipients }, - status: "active", + status: 'active', }, select: { id: true, @@ -123,16 +115,16 @@ async function sendNotificationEmail({ }, }); - const emailFrom = String(config.emailFrom ?? "").trim(); + const emailFrom = String(config.emailFrom ?? '').trim(); if (!emailFrom) { - logger?.warn?.({ requestId }, "Email enabled but EMAIL_FROM is empty"); + logger?.warn?.({ requestId }, 'Email enabled but EMAIL_FROM is empty'); return { skipped: true, sent: 0, failed: 0 }; } const baseUrl = resolveEmailBaseUrl(config); const link = targetPath && baseUrl ? `${baseUrl}${targetPath}` : null; - const safeSubject = String(subject ?? "").trim().slice(0, 200) || "系统通知"; - const safeBody = String(body ?? "").trim(); + const safeSubject = String(subject ?? '').trim().slice(0, 200) || '系统通知'; + const safeBody = String(body ?? '').trim(); let sent = 0; let failed = 0; @@ -152,8 +144,8 @@ async function sendNotificationEmail({ subject: safeSubject, text, headers: { - "x-hyb-notification-type": type, - "x-request-id": requestId ?? "", + 'x-hyb-notification-type': type, + 'x-request-id': requestId ?? '', }, }); sent += 1; @@ -167,111 +159,7 @@ async function sendNotificationEmail({ email: employee.email, notificationType: type, }, - "Email send failed", - ); - } - } - - return { skipped: false, sent, failed }; -} - -async function sendNotificationIm({ - prisma, - config, - logger, - requestId, - recipientEmployeeIds, - subject, - body, - type = "notification", - targetPath, -}) { - if (!config.imNotifierEnabled) { - return { skipped: true, sent: 0, failed: 0 }; - } - - const notificationType = String(type ?? "notification"); - const allowedTypePrefixes = ["reimbursement."]; - const allowedTypes = new Set(["workflow.task_assigned"]); - - const allowed = - allowedTypes.has(notificationType) || - allowedTypePrefixes.some((prefix) => notificationType.startsWith(prefix)); - - if (!allowed) { - return { skipped: true, sent: 0, failed: 0 }; - } - - const recipients = uniq(recipientEmployeeIds); - if (!recipients.length) { - return { skipped: true, sent: 0, failed: 0 }; - } - - const channels = await prisma.notificationChannel.findMany({ - where: { - enabled: true, - channel_type: { - in: ["dingtalk", "wechat", "wechat_work", "wecom"], - }, - }, - select: { - id: true, - channel_type: true, - name: true, - enabled: true, - config_json: true, - template_json: true, - }, - }); - - if (!channels.length) { - return { skipped: true, sent: 0, failed: 0 }; - } - - const safeSubject = String(subject ?? "").trim().slice(0, 200) || "系统通知"; - const safeBody = String(body ?? "").trim(); - - let sent = 0; - let failed = 0; - - for (const channel of channels) { - try { - const result = await sendInstantMessageByChannel({ - config, - channel, - payload: { - title: safeSubject, - body: safeBody, - targetType: type, - targetId: targetPath ?? null, - recipientCount: recipients.length, - }, - }); - - if (result.ok) { - sent += 1; - } else { - failed += 1; - logger?.warn?.( - { - requestId, - channelId: channel.id, - channelType: channel.channel_type, - resultCode: result.code, - }, - "IM notification skipped", - ); - } - } catch (error) { - failed += 1; - logger?.error?.( - { - err: error, - requestId, - channelId: channel.id, - channelType: channel.channel_type, - }, - "IM notification send failed", + 'Email send failed', ); } } diff --git a/server/src/services/integrations/dispatcher.js b/server/src/services/integrations/dispatcher.js index 3d4fda35..c186a3c7 100644 --- a/server/src/services/integrations/dispatcher.js +++ b/server/src/services/integrations/dispatcher.js @@ -1,11 +1,12 @@ -const { pushPaidReimbursementToErp } = require("./erp"); +const { pushPaidReimbursementToErp } = require('./erp'); +const { sendInstantMessageByChannel } = require('../notifications/channels'); class IntegrationDispatchError extends Error { constructor({ code, message, retryable, statusCode, response }) { - super(message || code || "integration_dispatch_failed"); - this.name = "IntegrationDispatchError"; - this.code = code || "integration_dispatch_failed"; - this.retryable = typeof retryable === "boolean" ? retryable : true; + super(message || code || 'integration_dispatch_failed'); + this.name = 'IntegrationDispatchError'; + this.code = code || 'integration_dispatch_failed'; + this.retryable = typeof retryable === 'boolean' ? retryable : true; this.statusCode = Number.isFinite(statusCode) ? Number(statusCode) : null; this.response = response ?? null; } @@ -21,32 +22,57 @@ function toDispatchError(result) { }); } -async function dispatchOutboxEvent({ config, eventType, payload }) { - if (eventType === "erp.reimbursement.paid") { +async function dispatchOutboxEvent({ prisma, config, eventType, payload }) { + if (eventType === 'erp.reimbursement.paid') { const result = await pushPaidReimbursementToErp(config, payload); if (!result.ok) { throw toDispatchError(result); } return { - channel: "erp", + channel: 'erp', code: result.code, response: result.response ?? null, }; } - if (eventType === "notification.im") { + if (eventType === 'notification.im') { + const channelId = String(payload?.channelId ?? '').trim(); + const channel = channelId + ? await prisma.notificationChannel.findUnique({ + where: { id: channelId }, + select: { + id: true, + channel_type: true, + name: true, + enabled: true, + config_json: true, + template_json: true, + }, + }) + : null; + + const result = await sendInstantMessageByChannel({ + config, + channel, + payload, + }); + + if (!result.ok) { + throw toDispatchError(result); + } + return { - channel: "im", - code: "mock_sent", - response: { - channelType: payload.channelType, - }, + channel: 'im', + code: result.code, + response: result.response ?? null, + skipped: result.skipped ?? false, + previewMessage: result.previewMessage ?? null, }; } throw new IntegrationDispatchError({ - code: "unsupported_outbox_event", + code: 'unsupported_outbox_event', message: `unsupported_outbox_event:${eventType}`, retryable: false, }); diff --git a/server/src/services/integrations/im/wecom.js b/server/src/services/integrations/im/wecom.js new file mode 100644 index 00000000..cb37a4e3 --- /dev/null +++ b/server/src/services/integrations/im/wecom.js @@ -0,0 +1,183 @@ +const crypto = require('node:crypto'); + +function toSafeString(value, maxLength = 2000) { + if (value == null) { + return ''; + } + return String(value).trim().slice(0, maxLength); +} + +function buildWecomMarkdownMessage(message) { + const title = toSafeString(message?.title, 200); + const body = toSafeString(message?.body, 4000); + const detailUrl = toSafeString(message?.detailUrl, 2000); + const footer = toSafeString(message?.footer, 200); + const deepLinkLabel = toSafeString(message?.deepLinkLabel || '查看详情', 50) || '查看详情'; + + const lines = []; + if (title) { + lines.push(`# ${title}`); + } + if (body) { + lines.push(body); + } + if (detailUrl) { + lines.push(`[${deepLinkLabel}](${detailUrl})`); + } + if (footer) { + lines.push(footer); + } + + return lines.filter(Boolean).join('\n\n').trim(); +} + +function buildSignedWebhookUrl(webhookUrl, secret) { + const rawUrl = String(webhookUrl ?? '').trim(); + if (!rawUrl || !secret) { + return rawUrl; + } + + const timestamp = Date.now(); + const signBase = `${timestamp}\n${secret}`; + const sign = encodeURIComponent( + crypto.createHmac('sha256', secret).update(signBase).digest('base64'), + ); + + const separator = rawUrl.includes('?') ? '&' : '?'; + return `${rawUrl}${separator}timestamp=${timestamp}&sign=${sign}`; +} + +async function sendWecomMessage({ channel, message }) { + const configJson = channel?.config_json && typeof channel.config_json === 'object' ? channel.config_json : {}; + const mode = String(configJson.mode ?? 'mock').trim() || 'mock'; + const previewMessage = buildWecomMarkdownMessage(message); + + if (mode === 'mock') { + return { + ok: true, + skipped: false, + code: 'mock_sent', + response: { + channelType: 'wecom', + mode, + }, + previewMessage, + }; + } + + if (mode !== 'wecom_webhook') { + return { + ok: false, + skipped: false, + retryable: false, + code: 'invalid_channel_mode', + message: `Unsupported wecom mode: ${mode}`, + response: null, + previewMessage, + }; + } + + const webhookUrl = String(configJson.webhookUrl ?? '').trim(); + if (!webhookUrl) { + return { + ok: false, + skipped: false, + retryable: false, + code: 'wecom_webhook_missing', + message: 'WeCom webhook url missing', + response: null, + previewMessage, + }; + } + + const timeoutMs = Math.max(100, Number(configJson.timeoutMs ?? 5000)); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + + try { + const response = await fetch(buildSignedWebhookUrl(webhookUrl, configJson.secret ?? null), { + method: 'POST', + headers: { + 'content-type': 'application/json', + }, + body: JSON.stringify({ + msgtype: 'markdown', + markdown: { + content: previewMessage, + }, + }), + signal: controller.signal, + }); + + let responseBody = null; + try { + responseBody = await response.json(); + } catch { + responseBody = null; + } + + if (!response.ok) { + return { + ok: false, + skipped: false, + retryable: response.status >= 500 || response.status === 429, + code: response.status >= 500 ? 'wecom_http_5xx' : response.status === 429 ? 'wecom_http_429' : 'wecom_http_4xx', + message: `WeCom webhook HTTP ${response.status}`, + response: { + status: response.status, + body: responseBody, + }, + previewMessage, + }; + } + + if (responseBody && Number(responseBody.errcode ?? 0) !== 0) { + return { + ok: false, + skipped: false, + retryable: false, + code: 'wecom_api_error', + message: `WeCom API errcode ${responseBody.errcode}`, + response: responseBody, + previewMessage, + }; + } + + return { + ok: true, + skipped: false, + code: 'wecom_sent', + response: responseBody, + previewMessage, + }; + } catch (error) { + if (error?.name === 'AbortError') { + return { + ok: false, + skipped: false, + retryable: true, + code: 'wecom_timeout', + message: `WeCom timeout (${timeoutMs}ms)`, + response: null, + previewMessage, + }; + } + + return { + ok: false, + skipped: false, + retryable: true, + code: 'wecom_network_error', + message: String(error?.message ?? error ?? 'WeCom network error'), + response: null, + previewMessage, + }; + } finally { + clearTimeout(timer); + } +} + +module.exports = { + buildWecomMarkdownMessage, + sendWecomMessage, +}; diff --git a/server/src/services/notifications/channels.js b/server/src/services/notifications/channels.js index 4ac5d967..7e5898aa 100644 --- a/server/src/services/notifications/channels.js +++ b/server/src/services/notifications/channels.js @@ -1,75 +1,93 @@ -const { dispatchOutboxEvent } = require("../integrations/dispatcher"); +const { sendWecomMessage } = require('../integrations/im/wecom'); +const { isImChannelTypeSupported, normalizeChannelType } = require('./imRouting'); function applyTemplate(template, context) { - if (typeof template !== "string") { - return ""; + if (typeof template !== 'string') { + return ''; } return template.replace(/\{\{\s*([a-zA-Z0-9_]+)\s*\}\}/g, (_, key) => { const value = context[key]; - return value == null ? "" : String(value); + return value == null ? '' : String(value); }); } function buildNotificationMessage(channel, payload) { - const context = { - title: String(payload?.title ?? "系统通知").slice(0, 200), - body: String(payload?.body ?? "").slice(0, 2000), - targetType: payload?.targetType ?? "", - targetId: payload?.targetId ?? "", - }; - const template = - channel?.template_json && typeof channel.template_json === "object" + channel?.template_json && typeof channel.template_json === 'object' ? channel.template_json : null; - const templatedTitle = template?.title ? applyTemplate(template.title, context) : ""; - const templatedBody = template?.body ? applyTemplate(template.body, context) : ""; + const context = { + title: String(payload?.title ?? '系统通知').slice(0, 200), + body: String(payload?.body ?? '').slice(0, 4000), + targetType: payload?.targetType ?? '', + targetId: payload?.targetId ?? '', + detailUrl: payload?.detailUrl ?? '', + footer: payload?.footer ?? '来自海音达工作台', + deepLinkLabel: payload?.deepLinkLabel ?? '查看详情', + }; return { - title: (templatedTitle || context.title).slice(0, 200), - body: (templatedBody || context.body).slice(0, 2000), + title: (template?.title ? applyTemplate(template.title, context) : context.title).slice(0, 200), + body: (template?.body ? applyTemplate(template.body, context) : context.body).slice(0, 4000), + detailUrl: context.detailUrl || null, + footer: (template?.footer ? applyTemplate(template.footer, context) : context.footer).slice(0, 200), + deepLinkLabel: String(template?.deepLinkLabel ?? context.deepLinkLabel).slice(0, 50) || '查看详情', targetType: context.targetType || null, targetId: context.targetId || null, }; } async function sendInstantMessageByChannel({ config, channel, payload }) { - if (!config.imNotifierEnabled) { + const normalizedChannelType = normalizeChannelType(channel?.channel_type ?? payload?.channelType ?? ''); + + if (!config?.imNotifierEnabled) { return { - ok: false, - code: "im_notifier_disabled", - message: "IM notifier disabled", + ok: true, + skipped: true, + code: 'im_notifier_disabled', + response: { channelType: normalizedChannelType || null }, + previewMessage: null, }; } - if (!channel?.enabled) { + if (!channel) { return { - ok: false, - code: "channel_disabled", - message: "notification channel disabled", + ok: true, + skipped: true, + code: 'channel_not_found', + response: { channelType: normalizedChannelType || null }, + previewMessage: null, + }; + } + + if (!channel.enabled) { + return { + ok: true, + skipped: true, + code: 'channel_disabled', + response: { channelType: normalizedChannelType || null }, + previewMessage: null, + }; + } + + if (!isImChannelTypeSupported(channel.channel_type)) { + return { + ok: true, + skipped: true, + code: 'unsupported_channel_type', + response: { channelType: normalizedChannelType || null }, + previewMessage: null, }; } const message = buildNotificationMessage(channel, payload); - const result = await dispatchOutboxEvent({ - config, - eventType: "notification.im", - payload: { - channelType: channel.channel_type, - channelName: channel.name, - message, - }, - }); - - return { - ok: true, - code: result.code, - response: result.response, - }; + return sendWecomMessage({ channel, message }); } module.exports = { + applyTemplate, + buildNotificationMessage, sendInstantMessageByChannel, }; diff --git a/server/src/services/notifications/imEnqueue.js b/server/src/services/notifications/imEnqueue.js new file mode 100644 index 00000000..d9721664 --- /dev/null +++ b/server/src/services/notifications/imEnqueue.js @@ -0,0 +1,199 @@ +const { enqueueIntegrationEvent } = require('../integrations/outbox'); +const { + IM_SUPPORTED_NOTIFICATION_TYPES, + resolveNotificationDomainScope, +} = require('../../constants/notificationTypes'); +const { + matchesNotificationChannel, + normalizeChannelType, +} = require('./imRouting'); + +function uniq(values) { + return Array.from(new Set((values ?? []).filter(Boolean))); +} + +function resolveBaseUrl(config) { + const origin = String(config?.appOrigin ?? '').split(',')[0]?.trim?.() ?? ''; + return origin ? origin.replace(/\/+$/, '') : ''; +} + +function buildDetailUrl(config, targetPath) { + const path = String(targetPath ?? '').trim(); + if (!path) { + return null; + } + + const baseUrl = resolveBaseUrl(config); + if (!baseUrl) { + return null; + } + + if (path.startsWith('http://') || path.startsWith('https://')) { + return path; + } + + return `${baseUrl}${path.startsWith('/') ? path : `/${path}`}`; +} + +async function resolveRecipientRoleCodes(tx, recipientEmployeeIds) { + const recipientIds = uniq(recipientEmployeeIds); + if (!recipientIds.length) { + return []; + } + + const rows = await tx.employeeRole.findMany({ + where: { + employee_id: { in: recipientIds }, + employee: { + status: 'active', + }, + }, + select: { + role_code: true, + }, + }); + + return uniq(rows.map((row) => row.role_code)); +} + +function buildImDedupeKey({ notificationType, targetType, targetId, targetPath, channelId }) { + return [ + notificationType, + targetType, + targetId || targetPath || 'no-target', + channelId, + ].join(':'); +} + +async function enqueueImNotifications(tx, params) { + const { + config, + logger, + recipientEmployeeIds, + type, + title, + body, + targetType, + targetId, + targetPath, + entityType, + entityId, + actorId, + } = params; + + if (!config?.imNotifierEnabled) { + return { queued: 0, duplicated: 0, matchedChannels: [], skippedReason: 'im_notifier_disabled' }; + } + + const notificationType = String(type ?? '').trim(); + if (!IM_SUPPORTED_NOTIFICATION_TYPES.includes(notificationType)) { + return { queued: 0, duplicated: 0, matchedChannels: [], skippedReason: 'notification_type_not_supported' }; + } + + const recipientIds = uniq(recipientEmployeeIds); + if (!recipientIds.length) { + return { queued: 0, duplicated: 0, matchedChannels: [], skippedReason: 'no_recipients' }; + } + + const domainScope = resolveNotificationDomainScope(notificationType, targetType ?? null); + if (!domainScope) { + return { queued: 0, duplicated: 0, matchedChannels: [], skippedReason: 'domain_scope_unresolved' }; + } + + const detailUrl = buildDetailUrl(config, targetPath ?? null); + if (!detailUrl && targetPath) { + logger?.warn?.( + { + notificationType, + targetPath, + }, + 'APP_ORIGIN missing, IM detailUrl omitted', + ); + } + + const recipientRoleCodes = await resolveRecipientRoleCodes(tx, recipientIds); + const channels = await tx.notificationChannel.findMany({ + where: { + enabled: true, + channel_type: { + in: ['wecom', 'wechat_work'], + }, + }, + orderBy: [{ channel_type: 'asc' }, { name: 'asc' }], + select: { + id: true, + channel_type: true, + name: true, + enabled: true, + config_json: true, + template_json: true, + }, + }); + + const matchedChannels = channels.filter((channel) => + matchesNotificationChannel(channel, { + notificationType, + domainScope, + recipientRoleCodes, + }), + ); + + if (!matchedChannels.length) { + return { queued: 0, duplicated: 0, matchedChannels: [], skippedReason: 'no_matching_channel' }; + } + + let queued = 0; + let duplicated = 0; + for (const channel of matchedChannels) { + const result = await enqueueIntegrationEvent(tx, { + eventType: 'notification.im', + dedupeKey: buildImDedupeKey({ + notificationType, + targetType, + targetId, + targetPath, + channelId: channel.id, + }), + maxAttempts: config.integrationOutboxMaxAttempts, + actorId: actorId ?? null, + payload: { + channelId: channel.id, + channelType: normalizeChannelType(channel.channel_type), + notificationType, + domainScope, + title: String(title ?? '').slice(0, 200), + body: body == null ? '' : String(body).slice(0, 4000), + targetType: targetType ?? null, + targetId: targetId ?? null, + targetPath: targetPath ?? null, + detailUrl, + recipientEmployeeIds: recipientIds, + recipientRoleCodes, + entityType: entityType ?? targetType ?? null, + entityId: entityId ?? targetId ?? null, + }, + }); + + if (result.duplicated) { + duplicated += 1; + } else { + queued += 1; + } + } + + return { + queued, + duplicated, + matchedChannels: matchedChannels.map((channel) => channel.id), + skippedReason: null, + recipientRoleCodes, + domainScope, + }; +} + +module.exports = { + buildDetailUrl, + buildImDedupeKey, + resolveRecipientRoleCodes, + enqueueImNotifications, +}; diff --git a/server/src/services/notifications/imRouting.js b/server/src/services/notifications/imRouting.js new file mode 100644 index 00000000..cb968c5a --- /dev/null +++ b/server/src/services/notifications/imRouting.js @@ -0,0 +1,74 @@ +const { IM_SUPPORTED_NOTIFICATION_TYPES } = require('../../constants/notificationTypes'); + +const CANONICAL_CHANNEL_TYPE_MAP = Object.freeze({ + wechat_work: 'wecom', + wecom: 'wecom', +}); + +function uniq(values) { + return Array.from(new Set((values ?? []).filter(Boolean))); +} + +function normalizeChannelType(channelType) { + const normalized = String(channelType ?? '').trim().toLowerCase(); + return CANONICAL_CHANNEL_TYPE_MAP[normalized] ?? normalized; +} + +function isImChannelTypeSupported(channelType) { + return normalizeChannelType(channelType) === 'wecom'; +} + +function normalizeRouting(configJson) { + const config = configJson && typeof configJson === 'object' ? configJson : {}; + const routing = config.routing && typeof config.routing === 'object' ? config.routing : {}; + + return { + roleCodes: uniq((Array.isArray(routing.roleCodes) ? routing.roleCodes : []).map((item) => String(item).trim()).filter(Boolean)), + domainScopes: uniq((Array.isArray(routing.domainScopes) ? routing.domainScopes : []).map((item) => String(item).trim()).filter(Boolean)), + allowedTypes: uniq((Array.isArray(routing.allowedTypes) ? routing.allowedTypes : []).map((item) => String(item).trim()).filter(Boolean)), + }; +} + +function hasIntersection(left, right) { + const rightSet = new Set(right ?? []); + return (left ?? []).some((item) => rightSet.has(item)); +} + +function matchesNotificationChannel(channel, context) { + if (!channel?.enabled) { + return false; + } + + if (!isImChannelTypeSupported(channel.channel_type)) { + return false; + } + + const notificationType = String(context?.notificationType ?? '').trim(); + if (!IM_SUPPORTED_NOTIFICATION_TYPES.includes(notificationType)) { + return false; + } + + const routing = normalizeRouting(channel.config_json); + if (routing.allowedTypes.length > 0 && !routing.allowedTypes.includes(notificationType)) { + return false; + } + + const domainScope = String(context?.domainScope ?? '').trim(); + if (routing.domainScopes.length > 0 && !routing.domainScopes.includes(domainScope)) { + return false; + } + + const recipientRoleCodes = uniq((context?.recipientRoleCodes ?? []).map((item) => String(item).trim()).filter(Boolean)); + if (routing.roleCodes.length > 0 && !hasIntersection(routing.roleCodes, recipientRoleCodes)) { + return false; + } + + return true; +} + +module.exports = { + normalizeChannelType, + isImChannelTypeSupported, + normalizeRouting, + matchesNotificationChannel, +}; diff --git a/server/tests/unit/integrations.dispatcher.test.js b/server/tests/unit/integrations.dispatcher.test.js index d7adb89b..25695022 100644 --- a/server/tests/unit/integrations.dispatcher.test.js +++ b/server/tests/unit/integrations.dispatcher.test.js @@ -4,6 +4,13 @@ const { } = require("../../src/services/integrations/dispatcher"); describe("dispatchOutboxEvent", () => { + const originalFetch = global.fetch; + + afterEach(() => { + global.fetch = originalFetch; + vi.restoreAllMocks(); + }); + it("handles mock ERP success", async () => { const result = await dispatchOutboxEvent({ config: { @@ -22,58 +29,235 @@ describe("dispatchOutboxEvent", () => { expect(result.response.externalId).toContain("BX-001"); }); - it("maps ERP disabled as non-retryable dispatch error", async () => { - await expect( - dispatchOutboxEvent({ - config: { - erpSyncEnabled: false, - erpSyncProvider: "mock", - }, - eventType: "erp.reimbursement.paid", - payload: { - reimbursementId: "rb-2", - formNumber: "BX-002", - }, - }), - ).rejects.toMatchObject({ - name: "IntegrationDispatchError", - code: "erp_sync_disabled", - retryable: false, - }); - }); + it("loads IM channel and dispatches mock wecom message", async () => { + const prisma = { + notificationChannel: { + findUnique: vi.fn().mockResolvedValue({ + id: "channel-1", + channel_type: "wecom", + name: "采购经理群", + enabled: true, + config_json: { + mode: "mock", + routing: {}, + }, + template_json: { + title: "{{title}}", + body: "{{body}}", + footer: "来自海音达工作台", + deepLinkLabel: "查看详情", + }, + }), + }, + }; - it("maps invalid ERP provider as non-retryable dispatch error", async () => { - await expect( - dispatchOutboxEvent({ - config: { - erpSyncEnabled: true, - erpSyncProvider: "invalid-provider", - }, - eventType: "erp.reimbursement.paid", - payload: { - reimbursementId: "rb-3", - formNumber: "BX-003", - }, - }), - ).rejects.toMatchObject({ - name: "IntegrationDispatchError", - code: "erp_sync_provider_invalid", - retryable: false, - }); - }); - - it("handles IM notification outbox", async () => { const result = await dispatchOutboxEvent({ - config: {}, + prisma, + config: { + imNotifierEnabled: true, + }, eventType: "notification.im", payload: { + channelId: "channel-1", + channelType: "wecom", + notificationType: "workflow.task_assigned", + title: "【待办】采购申请待审批", + body: "采购申请已流转到你，请及时处理。", + }, + }); + + expect(prisma.notificationChannel.findUnique).toHaveBeenCalledWith({ + where: { id: "channel-1" }, + select: { + id: true, + channel_type: true, + name: true, + enabled: true, + config_json: true, + template_json: true, + }, + }); + expect(result.channel).toBe("im"); + expect(result.code).toBe("mock_sent"); + expect(result.previewMessage).toContain("采购申请待审批"); + }); + + it("safely skips missing IM channel", async () => { + const result = await dispatchOutboxEvent({ + prisma: { + notificationChannel: { + findUnique: vi.fn().mockResolvedValue(null), + }, + }, + config: { + imNotifierEnabled: true, + }, + eventType: "notification.im", + payload: { + channelId: "missing-channel", channelType: "wecom", }, }); expect(result.channel).toBe("im"); - expect(result.code).toBe("mock_sent"); - expect(result.response.channelType).toBe("wecom"); + expect(result.code).toBe("channel_not_found"); + expect(result.skipped).toBe(true); + }); + + it("safely skips disabled IM channel", async () => { + const result = await dispatchOutboxEvent({ + prisma: { + notificationChannel: { + findUnique: vi.fn().mockResolvedValue({ + id: "channel-2", + channel_type: "wecom", + name: "disabled", + enabled: false, + config_json: { mode: "mock", routing: {} }, + template_json: null, + }), + }, + }, + config: { + imNotifierEnabled: true, + }, + eventType: "notification.im", + payload: { + channelId: "channel-2", + channelType: "wecom", + }, + }); + + expect(result.code).toBe("channel_disabled"); + expect(result.skipped).toBe(true); + }); + + it("maps wecom 4xx as non-retryable IM dispatch error", async () => { + global.fetch = vi.fn().mockResolvedValue({ + ok: false, + status: 400, + json: vi.fn().mockResolvedValue({ errcode: 40013, errmsg: "bad request" }), + }); + + await expect( + dispatchOutboxEvent({ + prisma: { + notificationChannel: { + findUnique: vi.fn().mockResolvedValue({ + id: "channel-3", + channel_type: "wecom", + name: "wecom-4xx", + enabled: true, + config_json: { + mode: "wecom_webhook", + webhookUrl: "https://example.com/wecom", + timeoutMs: 500, + routing: {}, + }, + template_json: null, + }), + }, + }, + config: { + imNotifierEnabled: true, + }, + eventType: "notification.im", + payload: { + channelId: "channel-3", + channelType: "wecom", + title: "test", + body: "body", + }, + }), + ).rejects.toMatchObject({ + name: "IntegrationDispatchError", + code: "wecom_http_4xx", + retryable: false, + }); + }); + + it("maps wecom 5xx as retryable IM dispatch error", async () => { + global.fetch = vi.fn().mockResolvedValue({ + ok: false, + status: 503, + json: vi.fn().mockResolvedValue({ errcode: 500, errmsg: "server error" }), + }); + + await expect( + dispatchOutboxEvent({ + prisma: { + notificationChannel: { + findUnique: vi.fn().mockResolvedValue({ + id: "channel-4", + channel_type: "wecom", + name: "wecom-5xx", + enabled: true, + config_json: { + mode: "wecom_webhook", + webhookUrl: "https://example.com/wecom", + timeoutMs: 500, + routing: {}, + }, + template_json: null, + }), + }, + }, + config: { + imNotifierEnabled: true, + }, + eventType: "notification.im", + payload: { + channelId: "channel-4", + channelType: "wecom", + title: "test", + body: "body", + }, + }), + ).rejects.toMatchObject({ + name: "IntegrationDispatchError", + code: "wecom_http_5xx", + retryable: true, + }); + }); + + it("maps wecom timeout as retryable IM dispatch error", async () => { + global.fetch = vi.fn().mockRejectedValue({ name: "AbortError" }); + + await expect( + dispatchOutboxEvent({ + prisma: { + notificationChannel: { + findUnique: vi.fn().mockResolvedValue({ + id: "channel-5", + channel_type: "wecom", + name: "wecom-timeout", + enabled: true, + config_json: { + mode: "wecom_webhook", + webhookUrl: "https://example.com/wecom", + timeoutMs: 500, + routing: {}, + }, + template_json: null, + }), + }, + }, + config: { + imNotifierEnabled: true, + }, + eventType: "notification.im", + payload: { + channelId: "channel-5", + channelType: "wecom", + title: "test", + body: "body", + }, + }), + ).rejects.toMatchObject({ + name: "IntegrationDispatchError", + code: "wecom_timeout", + retryable: true, + }); }); it("throws non-retryable error for unsupported event", async () => { diff --git a/server/tests/unit/services/imEnqueue.test.js b/server/tests/unit/services/imEnqueue.test.js new file mode 100644 index 00000000..1e5625dc --- /dev/null +++ b/server/tests/unit/services/imEnqueue.test.js @@ -0,0 +1,170 @@ +const { + buildDetailUrl, + buildImDedupeKey, + enqueueImNotifications, + resolveRecipientRoleCodes, +} = require("../../../src/services/notifications/imEnqueue"); + +describe("IM enqueue", () => { + it("builds detail url from APP_ORIGIN", () => { + expect(buildDetailUrl({ appOrigin: "https://app.example.com" }, "/workflow/tasks/1")).toBe( + "https://app.example.com/workflow/tasks/1", + ); + expect(buildDetailUrl({ appOrigin: "" }, "/workflow/tasks/1")).toBeNull(); + }); + + it("builds dedupe key with target id first", () => { + expect( + buildImDedupeKey({ + notificationType: "workflow.task_assigned", + targetType: "workflow_task", + targetId: "task-1", + targetPath: "/workflow/tasks/task-1", + channelId: "channel-1", + }), + ).toBe("workflow.task_assigned:workflow_task:task-1:channel-1"); + }); + + it("resolves unique recipient role codes", async () => { + const roles = await resolveRecipientRoleCodes( + { + employeeRole: { + findMany: vi.fn().mockResolvedValue([ + { role_code: "manager" }, + { role_code: "manager" }, + { role_code: "finance_mgr" }, + ]), + }, + }, + ["employee-1", "employee-2"], + ); + + expect(roles).toEqual(["manager", "finance_mgr"]); + }); + + it("enqueues one outbox event per matched channel", async () => { + const tx = { + employeeRole: { + findMany: vi.fn().mockResolvedValue([{ role_code: "manager" }]), + }, + notificationChannel: { + findMany: vi.fn().mockResolvedValue([ + { + id: "channel-1", + channel_type: "wecom", + name: "manager-workflow", + enabled: true, + config_json: { + routing: { + roleCodes: ["manager"], + domainScopes: ["workflow"], + allowedTypes: ["workflow.task_assigned"], + }, + }, + template_json: null, + }, + { + id: "channel-2", + channel_type: "wechat_work", + name: "all-workflow", + enabled: true, + config_json: { + routing: { + roleCodes: [], + domainScopes: ["workflow"], + allowedTypes: [], + }, + }, + template_json: null, + }, + ]), + }, + integrationOutbox: { + findUnique: vi + .fn() + .mockResolvedValueOnce(null) + .mockResolvedValueOnce({ id: "outbox-2", status: "pending" }), + create: vi.fn().mockImplementation(async ({ data }) => ({ id: `${data.dedupe_key}-id`, status: "pending" })), + }, + }; + + const result = await enqueueImNotifications(tx, { + config: { + imNotifierEnabled: true, + integrationOutboxMaxAttempts: 5, + appOrigin: "https://app.example.com", + }, + logger: { warn: vi.fn() }, + recipientEmployeeIds: ["employee-1"], + type: "workflow.task_assigned", + title: "【待办】采购申请待审批", + body: "采购申请已流转到你，请及时处理。", + targetType: "workflow_task", + targetId: "task-1", + targetPath: "/workflow/tasks/task-1", + entityType: "purchase_request", + entityId: "pr-1", + actorId: "actor-1", + }); + + expect(result.queued).toBe(1); + expect(result.duplicated).toBe(1); + expect(result.matchedChannels).toEqual(["channel-1", "channel-2"]); + expect(tx.integrationOutbox.create).toHaveBeenCalledTimes(1); + expect(tx.integrationOutbox.create.mock.calls[0][0].data.payload.detailUrl).toBe( + "https://app.example.com/workflow/tasks/task-1", + ); + }); + + it("safely skips when no channel matches", async () => { + const tx = { + employeeRole: { + findMany: vi.fn().mockResolvedValue([{ role_code: "manager" }]), + }, + notificationChannel: { + findMany: vi.fn().mockResolvedValue([ + { + id: "channel-3", + channel_type: "wecom", + name: "finance-only", + enabled: true, + config_json: { + routing: { + roleCodes: ["finance_mgr"], + domainScopes: ["procurement"], + allowedTypes: ["purchase_request.approved"], + }, + }, + template_json: null, + }, + ]), + }, + integrationOutbox: { + findUnique: vi.fn(), + create: vi.fn(), + }, + }; + + const result = await enqueueImNotifications(tx, { + config: { + imNotifierEnabled: true, + integrationOutboxMaxAttempts: 5, + appOrigin: "https://app.example.com", + }, + logger: { warn: vi.fn() }, + recipientEmployeeIds: ["employee-1"], + type: "workflow.task_assigned", + title: "title", + body: "body", + targetType: "workflow_task", + targetId: "task-1", + targetPath: "/workflow/tasks/task-1", + entityType: "purchase_request", + entityId: "pr-1", + actorId: "actor-1", + }); + + expect(result.skippedReason).toBe("no_matching_channel"); + expect(tx.integrationOutbox.create).not.toHaveBeenCalled(); + }); +}); diff --git a/server/tests/unit/services/imRouting.test.js b/server/tests/unit/services/imRouting.test.js new file mode 100644 index 00000000..e1aac3e7 --- /dev/null +++ b/server/tests/unit/services/imRouting.test.js @@ -0,0 +1,91 @@ +const { + isImChannelTypeSupported, + matchesNotificationChannel, + normalizeChannelType, + normalizeRouting, +} = require("../../../src/services/notifications/imRouting"); + +describe("IM routing", () => { + it("normalizes wechat_work alias to wecom", () => { + expect(normalizeChannelType("wechat_work")).toBe("wecom"); + expect(isImChannelTypeSupported("wecom")).toBe(true); + expect(isImChannelTypeSupported("dingtalk")).toBe(false); + }); + + it("normalizes routing arrays and drops empty values", () => { + expect( + normalizeRouting({ + routing: { + roleCodes: ["manager", "manager", ""], + domainScopes: ["workflow", "workflow"], + allowedTypes: ["workflow.task_assigned", ""], + }, + }), + ).toEqual({ + roleCodes: ["manager"], + domainScopes: ["workflow"], + allowedTypes: ["workflow.task_assigned"], + }); + }); + + it("matches channel when role scope type all hit", () => { + const matched = matchesNotificationChannel( + { + enabled: true, + channel_type: "wecom", + config_json: { + routing: { + roleCodes: ["manager"], + domainScopes: ["workflow"], + allowedTypes: ["workflow.task_assigned"], + }, + }, + }, + { + notificationType: "workflow.task_assigned", + domainScope: "workflow", + recipientRoleCodes: ["manager", "finance_mgr"], + }, + ); + + expect(matched).toBe(true); + }); + + it("rejects channel when any routing dimension does not match", () => { + const baseChannel = { + enabled: true, + channel_type: "wecom", + config_json: { + routing: { + roleCodes: ["manager"], + domainScopes: ["workflow"], + allowedTypes: ["workflow.task_assigned"], + }, + }, + }; + + expect( + matchesNotificationChannel(baseChannel, { + notificationType: "workflow.task_assigned", + domainScope: "procurement", + recipientRoleCodes: ["manager"], + }), + ).toBe(false); + + expect( + matchesNotificationChannel(baseChannel, { + notificationType: "workflow.task_assigned", + domainScope: "workflow", + recipientRoleCodes: ["finance_mgr"], + }), + ).toBe(false); + + expect( + matchesNotificationChannel(baseChannel, { + notificationType: "leave.approved", + domainScope: "workflow", + recipientRoleCodes: ["manager"], + }), + ).toBe(false); + }); +}); diff --git a/server/tests/unit/services/integrations/dispatcher.test.js b/server/tests/unit/services/integrations/dispatcher.test.js index 470f7045..4abec747 100644 --- a/server/tests/unit/services/integrations/dispatcher.test.js +++ b/server/tests/unit/services/integrations/dispatcher.test.js @@ -1,42 +1,46 @@ const dispatcher = require("../../../../src/services/integrations/dispatcher"); describe("Integration Dispatcher", () => { - it("dispatches erp.reimbursement.paid", async () => { - const result = await dispatcher.dispatchOutboxEvent({ - config: { erpSyncEnabled: true, erpSyncProvider: "mock" }, - eventType: "erp.reimbursement.paid", - payload: { reimbursementId: 1 }, - }); - expect(result.channel).toBe("erp"); - expect(result.code).toBe("mock_synced"); + it("dispatches erp.reimbursement.paid", async () => { + const result = await dispatcher.dispatchOutboxEvent({ + config: { erpSyncEnabled: true, erpSyncProvider: "mock" }, + eventType: "erp.reimbursement.paid", + payload: { reimbursementId: "rb-1", formNumber: "BX-001" }, }); - it("throws IntegrationDispatchError if erp fails", async () => { - await expect( - dispatcher.dispatchOutboxEvent({ - config: { erpSyncEnabled: false }, - eventType: "erp.reimbursement.paid", - payload: { reimbursementId: 1 }, - }) - ).rejects.toThrow(dispatcher.IntegrationDispatchError); + expect(result.channel).toBe("erp"); + expect(result.code).toBe("mock_synced"); + }); + + it("dispatches notification.im with mock channel", async () => { + const result = await dispatcher.dispatchOutboxEvent({ + prisma: { + notificationChannel: { + findUnique: vi.fn().mockResolvedValue({ + id: "channel-1", + channel_type: "wecom", + name: "test", + enabled: true, + config_json: { mode: "mock", routing: {} }, + template_json: null, + }), + }, + }, + config: { imNotifierEnabled: true }, + eventType: "notification.im", + payload: { channelId: "channel-1", channelType: "wecom", title: "title", body: "body" }, }); - it("dispatches notification.im", async () => { - const result = await dispatcher.dispatchOutboxEvent({ - config: {}, - eventType: "notification.im", - payload: { channelType: "slack" }, - }); - expect(result.channel).toBe("im"); - expect(result.response.channelType).toBe("slack"); - }); + expect(result.channel).toBe("im"); + expect(result.code).toBe("mock_sent"); + }); - it("throws error for unsupported event", async () => { - await expect( - dispatcher.dispatchOutboxEvent({ - config: {}, - eventType: "unknown.event", - }) - ).rejects.toThrow("unsupported_outbox_event:unknown.event"); - }); + it("throws error for unsupported event", async () => { + await expect( + dispatcher.dispatchOutboxEvent({ + config: {}, + eventType: "unknown.event", + }), + ).rejects.toThrow("unsupported_outbox_event:unknown.event"); + }); });