feat(P19IM1-1,P19IM1-2): 新增通知渠道配置与Outbox派发

这个提交包含在:
2026-03-08 13:14:04 +08:00
父节点 43a0af0803
当前提交 f6d9f23553
共修改 13 个文件,包含 1582 行新增和 364 行删除
+20 -132
查看文件
@@ -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',
);
}
}
+41 -15
查看文件
@@ -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,
});
@@ -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,
};
+57 -39
查看文件
@@ -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,
};
@@ -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,
};
@@ -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,
};