diff --git a/.env.example b/.env.example index 30dac6e..69b71f9 100644 --- a/.env.example +++ b/.env.example @@ -29,4 +29,6 @@ YYD_NOTIFICATION_PROJECTS_JSON=[{"code":"material_library","name":"素材库","s WORKER_ENABLED=true WORKER_INTERVAL_MS=1000 +# 钉钉发送请求的最小启动间隔,统一限制在 100~5000ms;默认 250ms(最多约 4 条/秒)。 +WORKER_SEND_MIN_INTERVAL_MS=250 READ_STATUS_INTERVAL_MS=5000 diff --git a/README.md b/README.md index f379c0f..c8a7eff 100644 --- a/README.md +++ b/README.md @@ -170,4 +170,4 @@ POST /v1/auth/tickets/redeem - `YYD_NOTIFICATION_PUBLIC_BASE_URL`:通知中心公网 HTTPS 基地址,生产为反向代理后的 `/notification-center` 地址。 - `DINGTALK_LOGIN_TICKET_TTL_SECONDS`:30~60 秒,默认 60 秒。 -新建任务最多投递 5 次,明确的限流等未受理结果按 1、5、15、30 分钟退避。发送调用若在请求已发出后发生网络断开、超时、HTTP 408/5xx、钉钉成功响应缺少 `processQueryKey`,或进程在发送中途崩溃并遗留过期租约,会以 `DELIVERY_OUTCOME_UNKNOWN` 结束并停止自动重试,避免不确定结果造成重复单聊。历史任务保留创建时的最大尝试次数,不通过迁移篡改。 +新建任务最多投递 5 次,钉钉 HTTP 429、`QpsLimit` 和其他明确限流统一归一为 `FLOW_CONTROLLED`,按 1、5、15、30 分钟退避;接入方只观察统一状态和错误码,不解析钉钉厂商错误码。中心默认将发送请求启动间隔限制为至少 250ms(`WORKER_SEND_MIN_INTERVAL_MS` 可在 100~5000ms 内调整),避免整点任务集中冲击下游。发送调用若在请求已发出后发生网络断开、超时、非限流 HTTP 408/5xx、钉钉成功响应缺少 `processQueryKey`,或进程在发送中途崩溃并遗留过期租约,会以 `DELIVERY_OUTCOME_UNKNOWN` 结束并停止自动重试,避免不确定结果造成重复单聊。历史任务保留创建时的最大尝试次数,不通过迁移篡改。 diff --git a/docker-compose.yml b/docker-compose.yml index d8a3d22..bbdf657 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -20,6 +20,7 @@ services: YYD_NOTIFICATION_ADMIN_API_SECRET: ${YYD_NOTIFICATION_ADMIN_API_SECRET} YYD_NOTIFICATION_CONFIG_ENCRYPTION_KEY: ${YYD_NOTIFICATION_CONFIG_ENCRYPTION_KEY} WORKER_ENABLED: ${WORKER_ENABLED:-true} + WORKER_SEND_MIN_INTERVAL_MS: ${WORKER_SEND_MIN_INTERVAL_MS:-250} volumes: - notification_data:/app/data diff --git a/docs/错误与状态规范.md b/docs/错误与状态规范.md index b14b9fd..f6f16ce 100644 --- a/docs/错误与状态规范.md +++ b/docs/错误与状态规范.md @@ -26,7 +26,14 @@ 业务项目一般不需要自己重发 `pending / sending / retrying / sent / unread` 状态。通知中心会处理明确可重试的失败并查询已读状态。 -## 3. 常见错误码 +## 3. 标准边界 + +- 业务项目必须按统一请求契约提交 `recipientUserId` 或已登记的 `recipientAlias`、标题、正文、跳转地址和幂等键;通知中心不识别采购流程、供应商账号、邮箱等项目私有字段。 +- 收件人解析和业务人员配置由接入方负责。姓名、邮箱、外部联系人 ID 和自行拼接的标识不能代替钉钉内部员工 `userId`。 +- 接入方以 HTTP 状态和投递 `status` 作为自动控制流程依据;`lastErrorCode` 用于诊断和人工处置,不得根据未知错误码自行重发。 +- 本规范公布的统一 `lastErrorCode` 语义稳定;尚未归一化的钉钉厂商错误码或附在 `lastErrorMessage` 中的厂商信息只供通知中心维护人排查,不构成对接契约。 + +## 4. 常见错误码 | 错误码 | 原因 | 处理方式 | | --- | --- | --- | @@ -46,10 +53,14 @@ | `IDEMPOTENCY_CONFLICT` | 同一幂等键对应了不同内容 | 修复幂等键或业务数据,不得随机换键规避 | | `NOT_FOUND` | 记录不存在或跨项目访问 | 检查投递 ID 和项目身份 | | `INTERNAL_ERROR` | 通知中心内部异常 | 原幂等键退避重试并通知维护人员 | +| `FLOW_CONTROLLED` | 钉钉明确拒绝本次请求并要求限流 | 接入方等待,通知中心内部自动有限重试 | +| `INVALID_USER` | 钉钉确认收件人内部员工 `userId` 无效或不存在 | 接入方修正人员数据,不得改用邮箱或姓名重试 | +| `FILTERED_USER` | 收件人不在统一机器人可见范围内 | 由通知中心维护人核对应用权限范围 | +| `DELIVERY_OUTCOME_UNKNOWN` | 请求可能已经送达,无法安全确认 | 停止自动重试,人工核验后决定是否补发 | 投递进入 `failed` 时,还应记录返回中的 `lastErrorCode` 和 `lastErrorMessage`,但不要直接把内部错误原文展示给最终用户。 -## 4. 业务端重试原则 +## 5. 业务端重试原则 1. 连接失败或 HTTP `500` 时可以有限重试,必须复用原 `idempotencyKey`,每次生成新 nonce。 2. HTTP `200 / 202` 后不要重新创建通知,改为查询返回的投递 ID。 @@ -57,7 +68,7 @@ 4. 通知中心内部会处理明确未受理的钉钉限流等情况,业务项目不得再叠加无限重试。 5. `DELIVERY_OUTCOME_UNKNOWN` 表示发送结果无法确定。为避免用户收到重复单聊,通知中心会停止自动重试;业务负责人确认后才能人工补发。 -## 5. 建议监控 +## 6. 建议监控 业务项目至少记录: diff --git a/scripts/deploy-material-host.sh b/scripts/deploy-material-host.sh index 63e9cfd..3b86cd8 100644 --- a/scripts/deploy-material-host.sh +++ b/scripts/deploy-material-host.sh @@ -252,6 +252,7 @@ center_lines = [ 'YYD_NOTIFICATION_CONFIG_ENCRYPTION_KEY=' + quoted(config_encryption_key), 'WORKER_ENABLED=true', 'WORKER_INTERVAL_MS=1000', + 'WORKER_SEND_MIN_INTERVAL_MS=250', 'READ_STATUS_INTERVAL_MS=5000', ] target_path.write_text('\n'.join(center_lines) + '\n', encoding='utf-8') diff --git a/src/__tests__/deployment.test.ts b/src/__tests__/deployment.test.ts index 370f4ed..4b1a806 100644 --- a/src/__tests__/deployment.test.ts +++ b/src/__tests__/deployment.test.ts @@ -13,6 +13,7 @@ test('生产发布面透传免登配置并保留已有项目 login', () => { 'DINGTALK_CORP_ID', 'YYD_NOTIFICATION_PUBLIC_BASE_URL', 'DINGTALK_LOGIN_TICKET_TTL_SECONDS', + 'WORKER_SEND_MIN_INTERVAL_MS', ]) { assert.match(compose, new RegExp(key)) assert.match(example, new RegExp(key)) diff --git a/src/__tests__/worker.test.ts b/src/__tests__/worker.test.ts index 4d71d21..c61633b 100644 --- a/src/__tests__/worker.test.ts +++ b/src/__tests__/worker.test.ts @@ -2,7 +2,12 @@ import assert from 'node:assert/strict' import test from 'node:test' import { DingTalkError, sendRobotMessage } from '../dingtalk' import prisma from '../prisma' -import { classifyDeliveryFailure, cleanupDeliveries, quarantineExpiredSendingLeases } from '../worker' +import { + classifyDeliveryFailure, + cleanupDeliveries, + normalizeWorkerSendMinIntervalMs, + quarantineExpiredSendingLeases, +} from '../worker' test('token 失效先刷新重试一次,避免缓存过期导致批量误判终态', async () => { const originalFetch = global.fetch @@ -192,6 +197,96 @@ test('机器人 ActionCard 原样发送统一标题、分段正文和按钮跳 } }) +test('钉钉 HTTP 403 QPS 限流归一为中心标准错误并进入自动重试', async () => { + const originalFetch = global.fetch + const oldAppKey = process.env.DINGTALK_APP_KEY + const oldAppSecret = process.env.DINGTALK_APP_SECRET + const oldRobotCode = process.env.DINGTALK_ROBOT_CODE + process.env.DINGTALK_APP_KEY = 'test-app-key' + process.env.DINGTALK_APP_SECRET = 'test-app-secret' + process.env.DINGTALK_ROBOT_CODE = 'test-robot-code' + global.fetch = (async (input) => { + if (String(input).includes('/gettoken?')) { + return new Response(JSON.stringify({ errcode: 0, access_token: 'qps-test-token', expires_in: 7200 }), { status: 200 }) + } + return new Response(JSON.stringify({ + code: 'Forbidden.AccessDenied.QpsLimitForApi', + message: '当前所有钉钉应用调用该接口次数过多,超出了该接口承受的最大qps,请求被暂时限制了', + }), { status: 403 }) + }) as typeof fetch + + try { + let caught: unknown + try { + await sendRobotMessage({ recipientUserId: 'ding-user-1', title: '测试', content: '测试', actionTitle: '查看', actionUrl: 'https://example.com' }) + } catch (error) { + caught = error + } + assert.ok(caught instanceof DingTalkError) + assert.equal(caught.code, 'FLOW_CONTROLLED') + assert.equal(caught.providerCode, 'Forbidden.AccessDenied.QpsLimitForApi') + assert.equal(caught.deterministic, false) + assert.equal(caught.outcomeUnknown, false) + assert.deepEqual(classifyDeliveryFailure({ error: caught, attemptCount: 1, maxAttempts: 5 }), { + failed: false, + errorCode: 'FLOW_CONTROLLED', + errorMessage: '[Forbidden.AccessDenied.QpsLimitForApi] 当前所有钉钉应用调用该接口次数过多,超出了该接口承受的最大qps,请求被暂时限制了', + }) + } finally { + global.fetch = originalFetch + process.env.DINGTALK_APP_KEY = oldAppKey + process.env.DINGTALK_APP_SECRET = oldAppSecret + process.env.DINGTALK_ROBOT_CODE = oldRobotCode + } +}) + +test('钉钉 staff 不存在归一为收件人无效并保持终态失败', async () => { + const originalFetch = global.fetch + const oldAppKey = process.env.DINGTALK_APP_KEY + const oldAppSecret = process.env.DINGTALK_APP_SECRET + const oldRobotCode = process.env.DINGTALK_ROBOT_CODE + process.env.DINGTALK_APP_KEY = 'test-app-key' + process.env.DINGTALK_APP_SECRET = 'test-app-secret' + process.env.DINGTALK_ROBOT_CODE = 'test-robot-code' + global.fetch = (async (input) => { + if (String(input).includes('/gettoken?')) { + return new Response(JSON.stringify({ errcode: 0, access_token: 'invalid-user-test-token', expires_in: 7200 }), { status: 200 }) + } + return new Response(JSON.stringify({ + code: 'staffId.notExisted', + message: '错误描述: staff 不存在;解决方案:请确认 userIds 是否正确;', + }), { status: 400 }) + }) as typeof fetch + + try { + let caught: unknown + try { + await sendRobotMessage({ recipientUserId: 'ding-user-invalid', title: '测试', content: '测试', actionTitle: '查看', actionUrl: 'https://example.com' }) + } catch (error) { + caught = error + } + assert.ok(caught instanceof DingTalkError) + assert.equal(caught.code, 'INVALID_USER') + assert.equal(caught.providerCode, 'staffId.notExisted') + assert.equal(caught.deterministic, true) + assert.equal(classifyDeliveryFailure({ error: caught, attemptCount: 1, maxAttempts: 5 }).failed, true) + } finally { + global.fetch = originalFetch + process.env.DINGTALK_APP_KEY = oldAppKey + process.env.DINGTALK_APP_SECRET = oldAppSecret + process.env.DINGTALK_ROBOT_CODE = oldRobotCode + } +}) + +test('中心发送间隔使用安全默认值并限制配置边界', () => { + assert.equal(normalizeWorkerSendMinIntervalMs(undefined), 250) + assert.equal(normalizeWorkerSendMinIntervalMs(''), 250) + assert.equal(normalizeWorkerSendMinIntervalMs('invalid'), 250) + assert.equal(normalizeWorkerSendMinIntervalMs('0'), 100) + assert.equal(normalizeWorkerSendMinIntervalMs('375'), 375) + assert.equal(normalizeWorkerSendMinIntervalMs('99999'), 5000) +}) + test('明确限流可安全重试,确定性错误和耗尽次数则停止', () => { assert.equal(classifyDeliveryFailure({ error: new DingTalkError('限流', 'FLOW_CONTROLLED'), attemptCount: 1, maxAttempts: 4, diff --git a/src/dingtalk.ts b/src/dingtalk.ts index b7c05f9..c1e6fa3 100644 --- a/src/dingtalk.ts +++ b/src/dingtalk.ts @@ -8,6 +8,7 @@ export class DingTalkError extends Error { public readonly code: string, public readonly deterministic = false, public readonly outcomeUnknown = false, + public readonly providerCode?: string, ) { super(message) this.name = 'DingTalkError' @@ -32,18 +33,34 @@ const requestJson = async ( }) const data = await response.json().catch(() => ({})) as DingTalkPayload if (!response.ok) { - const code = clean(data.code) || `HTTP_${response.status}` + const providerCode = clean(data.code) || `HTTP_${response.status}` const message = clean(data.message) || `钉钉接口返回 HTTP ${response.status}` - const outcomeUnknown = Boolean(options?.deliveryOutcomeCanBeUnknown && (response.status === 408 || response.status >= 500)) + const code = normalizeDingTalkErrorCode(providerCode, message, response.status) + const rateLimited = code === 'FLOW_CONTROLLED' + const outcomeUnknown = Boolean( + options?.deliveryOutcomeCanBeUnknown && + !rateLimited && + (response.status === 408 || response.status >= 500), + ) throw new DingTalkError( message, code, - response.status >= 400 && response.status < 500 && response.status !== 408 && response.status !== 429, + !rateLimited && response.status >= 400 && response.status < 500 && response.status !== 408 && response.status !== 429, outcomeUnknown, + providerCode !== code ? providerCode : undefined, ) } if (data.errcode !== undefined && Number(data.errcode) !== 0) { - throw new DingTalkError(clean(data.errmsg) || '钉钉接口调用失败', String(data.errcode), isDeterministic(data.errcode, data.errmsg)) + const providerCode = String(data.errcode) + const message = clean(data.errmsg) || '钉钉接口调用失败' + const code = normalizeDingTalkErrorCode(providerCode, message) + throw new DingTalkError( + message, + code, + isDeterministic(code, message), + false, + providerCode !== code ? providerCode : undefined, + ) } return data } catch (error: any) { @@ -64,9 +81,27 @@ const requestJson = async ( const clean = (value: unknown) => String(value || '').trim() const stringList = (value: unknown) => Array.isArray(value) ? value.map(clean).filter(Boolean) : [] -const isDeterministic = (code: unknown, message: unknown) => - ['40014', '40056', '40057', '40078', '41001', '41009', '60111', '60121', '70020'].includes(String(code)) || - /不存在|不在可见范围|无权限|invalid user|forbidden user|userid/i.test(clean(message)) + +const isDingTalkRateLimitError = (code: unknown, message: unknown, status?: number) => { + const normalizedCode = clean(code).toLowerCase() + const normalizedMessage = clean(message) + return status === 429 || + /qps.?limit|rate.?limit|too.?many.?requests|throttl/i.test(normalizedCode) || + /最大\s*qps|qps\s*限制|请求.*(?:过多|限流|暂时限制)|too many requests|rate limit/i.test(normalizedMessage) +} + +const normalizeDingTalkErrorCode = (providerCode: unknown, message: unknown, status?: number) => { + const code = clean(providerCode) + if (isDingTalkRateLimitError(code, message, status)) return 'FLOW_CONTROLLED' + if (code.toLowerCase() === 'staffid.notexisted') return 'INVALID_USER' + return code +} + +const isDeterministic = (code: unknown, message: unknown) => { + if (clean(code) === 'FLOW_CONTROLLED') return false + return ['40014', '40056', '40057', '40078', '41001', '41009', '60111', '60121', '70020', 'INVALID_USER'].includes(String(code)) || + /不存在|不在可见范围|无权限|invalid user|forbidden user|userid/i.test(clean(message)) +} // token 过期属于可恢复错误:先作废本地缓存、用新 token 重试一次;仍失败再按原语义上抛。 const AUTH_EXPIRED_CODES = new Set(['40014', 'HTTP_401', 'InvalidAuthentication']) diff --git a/src/worker.ts b/src/worker.ts index d06615b..2b1d94b 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -6,6 +6,9 @@ const workerId = `${os.hostname()}:${process.pid}:${Math.random().toString(36).s const backoffMinutes = [1, 5, 15, 30] // 单 tick 批量消费,避免积压时每秒只推进一条。 const MAX_DELIVERIES_PER_TICK = 10 +const DEFAULT_SEND_MIN_INTERVAL_MS = 250 +const MIN_SEND_MIN_INTERVAL_MS = 100 +const MAX_SEND_MIN_INTERVAL_MS = 5000 // sent/unread 超过该天数仍无终态则停止回查;终态记录保留该天数后清理。 const READ_STATUS_EXPIRY_DAYS = 14 const DELIVERY_RETENTION_DAYS = 90 @@ -14,6 +17,20 @@ let busy = false let readBusy = false const outcomeUnknownMessage = '钉钉发送结果不确定,请先在钉钉侧人工核验,系统不会自动重试' +export const normalizeWorkerSendMinIntervalMs = (value: unknown) => { + if (value === undefined || value === null || String(value).trim() === '') return DEFAULT_SEND_MIN_INTERVAL_MS + const parsed = Number(value) + if (!Number.isFinite(parsed)) return DEFAULT_SEND_MIN_INTERVAL_MS + return Math.min(MAX_SEND_MIN_INTERVAL_MS, Math.max(MIN_SEND_MIN_INTERVAL_MS, Math.round(parsed))) +} + +const deliveryErrorMessage = (error: unknown) => { + if (error instanceof DingTalkError && error.providerCode) { + return `[${error.providerCode}] ${error.message}`.slice(0, 2000) + } + return String((error as any)?.message || '通知发送失败').slice(0, 2000) +} + export const quarantineExpiredSendingLeases = async (now = new Date()) => prisma.notificationDelivery.updateMany({ where: { status: 'sending', @@ -46,7 +63,7 @@ export const classifyDeliveryFailure = (input: { return { failed: deterministic || exhausted, errorCode: error instanceof DingTalkError ? error.code : 'UNKNOWN', - errorMessage: String((error as any)?.message || '通知发送失败').slice(0, 2000), + errorMessage: deliveryErrorMessage(error), } } @@ -72,10 +89,11 @@ const claimNext = async () => { return prisma.notificationDelivery.findUnique({ where: { id: candidate.id } }) } -const processOne = async (): Promise => { +const processOne = async (beforeSend?: () => Promise): Promise => { const delivery = await claimNext() if (!delivery) return false try { + await beforeSend?.() const processQueryKey = await sendRobotMessage({ recipientUserId: delivery.recipientUserId, title: delivery.renderedTitle, @@ -173,12 +191,19 @@ export const startWorker = () => { if (process.env.WORKER_ENABLED === 'false') return const interval = Math.max(500, Number(process.env.WORKER_INTERVAL_MS) || 1000) const readInterval = Math.max(2000, Number(process.env.READ_STATUS_INTERVAL_MS) || 5000) + const sendMinInterval = normalizeWorkerSendMinIntervalMs(process.env.WORKER_SEND_MIN_INTERVAL_MS) + let lastSendStartedAt = 0 + const waitForSendSlot = async () => { + const waitMs = Math.max(0, sendMinInterval - (Date.now() - lastSendStartedAt)) + if (waitMs > 0) await new Promise((resolve) => setTimeout(resolve, waitMs)) + lastSendStartedAt = Date.now() + } const tick = async () => { if (busy) return busy = true try { for (let processed = 0; processed < MAX_DELIVERIES_PER_TICK; processed++) { - const claimed = await processOne() + const claimed = await processOne(waitForSendSlot) if (!claimed) break } } finally {