feat(operation-logs): 失败操作日志实时推送徐霞客
- 失败日志(FAILED/RETRY)写入后经壹点通知中心个人推送徐霞客 - 聚水潭固定虚拟库存错误码按 1 小时折叠成一条，其余按 动作+错误码+SPU+SKU 10 分钟合并 - 正文空行分段（钉钉 Markdown 会吞单个换行），纯文字无按钮 - record 挂载异步通知不阻塞业务，NODE_ENV=test 静默 - 新增带 worker 密钥的手动测试入口 /api/internal/operation-logs/failure-notification/run
This commit is contained in:
WuYongJie
parent
8ea81f886a
commit
446c550320
@ -0,0 +1,28 @@
|
||||
import { NextRequest } from 'next/server'
|
||||
import { handleRequest, jsonError, jsonSuccess } from '@/server/core/response'
|
||||
import { ErrorCode } from '@/server/core/errors'
|
||||
import { operationLogFailureNotificationService } from '@/server/services/operation-log-failure-notification.service'
|
||||
|
||||
async function runTestPush() {
|
||||
const result = await operationLogFailureNotificationService.sendTestNotification()
|
||||
return jsonSuccess(result, 200)
|
||||
}
|
||||
|
||||
// 排障与验收入口：携带 worker 密钥触发；管理员（徐霞客）会话触发。
|
||||
export async function POST(request: NextRequest) {
|
||||
const expectedSecret = process.env.BOM_COSTING_WORKER_SECRET || ''
|
||||
const providedSecret = request.headers.get('x-worker-secret') || ''
|
||||
if (expectedSecret && providedSecret === expectedSecret) {
|
||||
return runTestPush()
|
||||
}
|
||||
return handleRequest(
|
||||
request,
|
||||
async (ctx) => {
|
||||
if (ctx.user?.name !== '徐霞客') {
|
||||
return jsonError(ErrorCode.AUTH_PERMISSION_DENIED, '仅徐霞客可触发失败日志通知测试', 403, ctx.requestId)
|
||||
}
|
||||
return runTestPush()
|
||||
},
|
||||
{ requireAuth: true },
|
||||
)
|
||||
}
|
||||
@ -70,6 +70,22 @@ class NotificationDeliveryRepository {
|
||||
})
|
||||
}
|
||||
|
||||
async findRecipientsByNames(names: string[]): Promise<NotificationRecipient[]> {
|
||||
const normalized = [...new Set(names.map((name) => name.trim()).filter(Boolean))]
|
||||
if (normalized.length === 0) return []
|
||||
return prisma.user.findMany({
|
||||
where: { name: { in: normalized } },
|
||||
select: {
|
||||
id: true,
|
||||
name: true,
|
||||
email: true,
|
||||
status: true,
|
||||
businessRoles: true,
|
||||
dingtalkUserId: true,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
findByIdempotencyKey(idempotencyKey: string) {
|
||||
return prisma.notificationDeliveryLog.findUnique({ where: { idempotencyKey } })
|
||||
}
|
||||
|
||||
@ -0,0 +1,284 @@
|
||||
import { OperationLogAction, OperationLogStatus } from '@prisma/client'
|
||||
import {
|
||||
OperationLogFailureNotificationService,
|
||||
buildOperationLogFailureMessage,
|
||||
buildOperationLogFailureNotificationKey,
|
||||
type OperationLogFailureNotificationDeps,
|
||||
type OperationLogFailurePayload,
|
||||
} from '@/server/services/operation-log-failure-notification.service'
|
||||
import type { NotificationRecipient } from '@/server/repositories/notification-delivery.repository'
|
||||
|
||||
// 北京时间 2026-09-17 14:40:14
|
||||
const FIXED_NOW = new Date('2026-09-17T06:40:14Z')
|
||||
|
||||
type SentRecord = { key: string; title: string; content: string; recipientUserId: string; source: string }
|
||||
|
||||
function activeRecipient(overrides: Partial<NotificationRecipient> = {}): NotificationRecipient {
|
||||
return {
|
||||
id: 'user-xu',
|
||||
name: '徐霞客',
|
||||
email: 'xu@example.com',
|
||||
status: 'ACTIVE',
|
||||
businessRoles: [],
|
||||
dingtalkUserId: 'ding-xu',
|
||||
...overrides,
|
||||
} as NotificationRecipient
|
||||
}
|
||||
|
||||
function baseDeps(options: {
|
||||
recipient?: NotificationRecipient | null
|
||||
sendFails?: boolean
|
||||
} = {}) {
|
||||
const sent: SentRecord[] = []
|
||||
const undeliverable: Array<{ code: string; recipientName: string }> = []
|
||||
const recordedKeys = new Set<string>()
|
||||
const recipient = options.recipient === undefined ? activeRecipient() : options.recipient
|
||||
|
||||
const deps: OperationLogFailureNotificationDeps = {
|
||||
findRecipientsByNames: async (names) => (
|
||||
recipient && names.includes(recipient.name) ? [recipient] : []
|
||||
),
|
||||
findByIdempotencyKey: async (key) => (recordedKeys.has(key) ? { id: `delivery-${key}` } : null),
|
||||
sendToEmployee: async (input) => {
|
||||
if (options.sendFails) throw new Error('通知中心连接失败')
|
||||
recordedKeys.add(input.idempotencyKey)
|
||||
sent.push({
|
||||
key: input.idempotencyKey,
|
||||
title: input.title,
|
||||
content: input.content,
|
||||
recipientUserId: input.recipientUserId,
|
||||
source: input.source ?? '',
|
||||
})
|
||||
return { id: `delivery-${sent.length}` }
|
||||
},
|
||||
recordUndeliverable: async (input) => {
|
||||
recordedKeys.add(input.idempotencyKey)
|
||||
undeliverable.push({ code: input.errorCode, recipientName: input.recipient.name })
|
||||
return null
|
||||
},
|
||||
now: () => FIXED_NOW,
|
||||
}
|
||||
|
||||
return { deps, sent, undeliverable, recordedKeys }
|
||||
}
|
||||
|
||||
function serviceWith(deps: OperationLogFailureNotificationDeps) {
|
||||
return new OperationLogFailureNotificationService(deps)
|
||||
}
|
||||
|
||||
function productWriteFailure(overrides: Partial<OperationLogFailurePayload> = {}): OperationLogFailurePayload {
|
||||
return {
|
||||
action: OperationLogAction.INTAKE_PRODUCT_WRITE,
|
||||
status: OperationLogStatus.FAILED,
|
||||
source: 'USER',
|
||||
actorName: '槐米',
|
||||
productSpu: 'LRSF0095',
|
||||
skuCode: null,
|
||||
workspaceCategory: '样包测量',
|
||||
errorCode: 'P2028',
|
||||
message: '产品资料保存失败：Invalid `prisma.intakeNonIntegratedProduct.update()` invocation: Transaction API error',
|
||||
requestId: 'req-1',
|
||||
createdAt: FIXED_NOW,
|
||||
...overrides,
|
||||
}
|
||||
}
|
||||
|
||||
function fixedVirtualInventoryFailure(skuCode: string, at: Date): OperationLogFailurePayload {
|
||||
return {
|
||||
action: OperationLogAction.JUSHUITAN_VIRTUAL_INVENTORY_PUSH,
|
||||
status: OperationLogStatus.FAILED,
|
||||
source: 'SYSTEM',
|
||||
actorName: '聚水潭固定虚拟库存定时任务',
|
||||
skuCode,
|
||||
errorCode: 'JUSHUITAN_FIXED_VIRTUAL_INVENTORY_FAILED',
|
||||
message: 'Jushuitan fixed virtual inventory push failed; retry at 2026-09-17T08:00:00.000Z: 调用太频繁',
|
||||
createdAt: at,
|
||||
}
|
||||
}
|
||||
|
||||
describe('OperationLogFailureNotificationService', () => {
|
||||
afterEach(() => {
|
||||
delete process.env.OPERATION_LOG_NOTIFIER_HOURLY_LIMIT
|
||||
delete process.env.OPERATION_LOG_NOTIFIER_RECIPIENT
|
||||
})
|
||||
|
||||
it('成功日志不推送', async () => {
|
||||
const env = baseDeps()
|
||||
const result = await serviceWith(env.deps).notify({
|
||||
...productWriteFailure(),
|
||||
status: OperationLogStatus.SUCCESS,
|
||||
})
|
||||
|
||||
expect(result.outcome).toBe('ignored')
|
||||
expect(env.sent).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('失败日志推送给徐霞客，且为纯文字（不带按钮字段）', async () => {
|
||||
const env = baseDeps()
|
||||
const result = await serviceWith(env.deps).notify(productWriteFailure())
|
||||
|
||||
expect(result.outcome).toBe('sent')
|
||||
expect(result.recipientName).toBe('徐霞客')
|
||||
expect(env.sent).toHaveLength(1)
|
||||
expect(env.sent[0].recipientUserId).toBe('user-xu')
|
||||
expect(env.sent[0].title).toBe('产品资料保存失败')
|
||||
expect(env.sent[0].content).toBe([
|
||||
'动作：产品资料保存',
|
||||
'时间：2026-09-17 14:40:14',
|
||||
'人员：槐米（人工操作）',
|
||||
'产品：LRSF0095 ／ 节点：样包测量',
|
||||
'错误码：P2028',
|
||||
'原因：Invalid `prisma.intakeNonIntegratedProduct.update()` invocation: Transaction API error',
|
||||
].join('\n\n'))
|
||||
})
|
||||
|
||||
it('系统来源显示为系统任务，且原因剥离动作前缀', async () => {
|
||||
const env = baseDeps()
|
||||
await serviceWith(env.deps).notify(productWriteFailure({ source: 'SYSTEM', actorName: '定时任务' }))
|
||||
|
||||
expect(env.sent[0].content).toContain('人员：定时任务（系统任务）')
|
||||
expect(env.sent[0].content).toContain('原因：Invalid `prisma')
|
||||
})
|
||||
|
||||
it('聚水潭定时任务在同一小时窗内把上千条失败折叠成一条', async () => {
|
||||
const env = baseDeps()
|
||||
const service = serviceWith(env.deps)
|
||||
|
||||
const first = await service.notify(fixedVirtualInventoryFailure('CBG-001', FIXED_NOW))
|
||||
const second = await service.notify(fixedVirtualInventoryFailure('CBG-002', new Date(FIXED_NOW.getTime() + 60_000)))
|
||||
const third = await service.notify(fixedVirtualInventoryFailure('CBG-003', new Date(FIXED_NOW.getTime() + 600_000)))
|
||||
|
||||
expect(first.outcome).toBe('sent')
|
||||
expect(second.outcome).toBe('skipped-duplicate')
|
||||
expect(third.outcome).toBe('skipped-duplicate')
|
||||
expect(env.sent).toHaveLength(1)
|
||||
expect(env.sent[0].content).toContain('说明：该错误码在 60 分钟内多次出现时只通知一次')
|
||||
})
|
||||
|
||||
it('折叠类幂等键不含业务对象，普通类含产品与 SKU', () => {
|
||||
const folded = buildOperationLogFailureNotificationKey({
|
||||
action: OperationLogAction.JUSHUITAN_VIRTUAL_INVENTORY_PUSH,
|
||||
errorCode: 'JUSHUITAN_FIXED_VIRTUAL_INVENTORY_FAILED',
|
||||
productSpu: 'SPU-1',
|
||||
skuCode: 'SKU-1',
|
||||
now: FIXED_NOW,
|
||||
})
|
||||
const scoped = buildOperationLogFailureNotificationKey({
|
||||
action: OperationLogAction.INTAKE_PRODUCT_WRITE,
|
||||
errorCode: 'P2028',
|
||||
productSpu: 'SPU-1',
|
||||
skuCode: 'SKU-1',
|
||||
now: FIXED_NOW,
|
||||
})
|
||||
|
||||
expect(folded).toBe(`oplog-failure:folded:${Math.floor(FIXED_NOW.getTime() / 3_600_000)}:JUSHUITAN_VIRTUAL_INVENTORY_PUSH:JUSHUITAN_FIXED_VIRTUAL_INVENTORY_FAILED`)
|
||||
expect(scoped).toContain('oplog-failure:scoped:')
|
||||
expect(scoped).toContain('SPU-1:SKU-1')
|
||||
})
|
||||
|
||||
it('不同产品的同类失败各自推送，同一产品重复失败只推一次', async () => {
|
||||
const env = baseDeps()
|
||||
const service = serviceWith(env.deps)
|
||||
|
||||
expect((await service.notify(productWriteFailure({ productSpu: 'LRSF0095' }))).outcome).toBe('sent')
|
||||
expect((await service.notify(productWriteFailure({ productSpu: 'LRSF0096' }))).outcome).toBe('sent')
|
||||
expect((await service.notify(productWriteFailure({ productSpu: 'LRSF0095' }))).outcome).toBe('skipped-duplicate')
|
||||
|
||||
expect(env.sent).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('超出时间窗后同一产品再次失败会重新推送', async () => {
|
||||
const env = baseDeps()
|
||||
let now = FIXED_NOW
|
||||
const service = new OperationLogFailureNotificationService({
|
||||
...env.deps,
|
||||
now: () => now,
|
||||
})
|
||||
|
||||
expect((await service.notify(productWriteFailure())).outcome).toBe('sent')
|
||||
now = new Date(FIXED_NOW.getTime() + 11 * 60_000)
|
||||
expect((await service.notify(productWriteFailure())).outcome).toBe('sent')
|
||||
expect(env.sent).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('收件人未绑定钉钉时记录不可投递，不调用发送', async () => {
|
||||
const env = baseDeps({ recipient: activeRecipient({ dingtalkUserId: null }) })
|
||||
const result = await serviceWith(env.deps).notify(productWriteFailure())
|
||||
|
||||
expect(result.outcome).toBe('undeliverable')
|
||||
expect(env.sent).toHaveLength(0)
|
||||
expect(env.undeliverable).toEqual([{ code: 'RECIPIENT_DINGTALK_UNBOUND', recipientName: '徐霞客' }])
|
||||
})
|
||||
|
||||
it('收件人不存在时记录不可投递', async () => {
|
||||
const env = baseDeps({ recipient: null })
|
||||
const result = await serviceWith(env.deps).notify(productWriteFailure())
|
||||
|
||||
expect(result.outcome).toBe('undeliverable')
|
||||
expect(env.undeliverable[0].code).toBe('RECIPIENT_NOT_FOUND')
|
||||
})
|
||||
|
||||
it('通知中心调用失败时不抛出，返回 error 供调用方忽略', async () => {
|
||||
const env = baseDeps({ sendFails: true })
|
||||
const result = await serviceWith(env.deps).notify(productWriteFailure())
|
||||
|
||||
expect(result.outcome).toBe('error')
|
||||
expect(result.detail).toContain('通知中心连接失败')
|
||||
})
|
||||
|
||||
it('每小时超过推送上限后熔断', async () => {
|
||||
process.env.OPERATION_LOG_NOTIFIER_HOURLY_LIMIT = '2'
|
||||
const env = baseDeps()
|
||||
const service = serviceWith(env.deps)
|
||||
|
||||
expect((await service.notify(productWriteFailure({ productSpu: 'SPU-1' }))).outcome).toBe('sent')
|
||||
expect((await service.notify(productWriteFailure({ productSpu: 'SPU-2' }))).outcome).toBe('sent')
|
||||
const third = await service.notify(productWriteFailure({ productSpu: 'SPU-3' }))
|
||||
|
||||
expect(third.outcome).toBe('skipped-rate-limit')
|
||||
expect(env.sent).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('收件人名称可通过环境变量覆盖', async () => {
|
||||
process.env.OPERATION_LOG_NOTIFIER_RECIPIENT = '徐霞客'
|
||||
const env = baseDeps()
|
||||
const result = await serviceWith(env.deps).notify(productWriteFailure())
|
||||
expect(result.recipientName).toBe('徐霞客')
|
||||
})
|
||||
})
|
||||
|
||||
describe('buildOperationLogFailureMessage', () => {
|
||||
it('缺少产品、节点、错误码时不产生空行占位', () => {
|
||||
const { content } = buildOperationLogFailureMessage({
|
||||
action: OperationLogAction.INTAKE_FILE_UPLOAD,
|
||||
status: OperationLogStatus.RETRY,
|
||||
source: 'USER',
|
||||
actorName: '依依',
|
||||
message: '文件上传失败：网络中断',
|
||||
createdAt: FIXED_NOW,
|
||||
})
|
||||
|
||||
expect(content.split('\n\n')).toEqual([
|
||||
'动作：文件上传',
|
||||
'时间：2026-09-17 14:40:14',
|
||||
'人员：依依（人工操作）',
|
||||
'原因：网络中断',
|
||||
])
|
||||
})
|
||||
|
||||
it('超长原因按上限截断', () => {
|
||||
const { content } = buildOperationLogFailureMessage({
|
||||
action: OperationLogAction.INTAKE_PRODUCT_WRITE,
|
||||
status: OperationLogStatus.FAILED,
|
||||
source: 'USER',
|
||||
actorName: '槐米',
|
||||
errorCode: 'P2028',
|
||||
message: `产品资料保存失败：${'x'.repeat(600)}`,
|
||||
createdAt: FIXED_NOW,
|
||||
})
|
||||
|
||||
const reasonLine = content.split('\n\n').find((line) => line.startsWith('原因：')) ?? ''
|
||||
expect(reasonLine.endsWith('…')).toBe(true)
|
||||
expect(reasonLine.length).toBeLessThanOrEqual('原因：'.length + 401)
|
||||
})
|
||||
})
|
||||
@ -0,0 +1,107 @@
|
||||
import { OperationLogAction, OperationLogStatus } from '@prisma/client'
|
||||
|
||||
const mockNotify = jest.fn()
|
||||
const mockCreate = jest.fn()
|
||||
|
||||
// jest.mock 的工厂会被提升到 import 之前执行，这里用惰性包装避免访问尚未初始化的变量。
|
||||
jest.mock('@/server/services/operation-log-failure-notification.service', () => ({
|
||||
operationLogFailureNotificationService: {
|
||||
notify: (...args: unknown[]) => mockNotify(...args),
|
||||
},
|
||||
}))
|
||||
|
||||
jest.mock('@/server/repositories/operation-log.repository', () => ({
|
||||
operationLogRepository: {
|
||||
create: (...args: unknown[]) => mockCreate(...args),
|
||||
},
|
||||
}))
|
||||
|
||||
import { operationLogService } from '@/server/services/operation-log.service'
|
||||
|
||||
const ORIGINAL_NODE_ENV = process.env.NODE_ENV
|
||||
|
||||
function flushBackgroundTasks() {
|
||||
return new Promise((resolve) => setTimeout(resolve, 0))
|
||||
}
|
||||
|
||||
describe('OperationLogService 失败日志通知挂载', () => {
|
||||
beforeEach(() => {
|
||||
mockNotify.mockReset().mockResolvedValue({ outcome: 'sent' })
|
||||
mockCreate.mockReset().mockResolvedValue({ id: 'log-1' })
|
||||
// 绕过测试环境静默守卫，验证挂载逻辑本身。
|
||||
;(process.env as Record<string, string>).NODE_ENV = 'development'
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
;(process.env as Record<string, string>).NODE_ENV = ORIGINAL_NODE_ENV as string
|
||||
})
|
||||
|
||||
const failureInput = {
|
||||
action: OperationLogAction.INTAKE_PRODUCT_WRITE,
|
||||
status: OperationLogStatus.FAILED,
|
||||
actorName: '槐米',
|
||||
productSpu: 'LRSF0095',
|
||||
workspaceCategory: '样包测量',
|
||||
errorCode: 'P2028',
|
||||
message: '产品资料保存失败：Transaction API error',
|
||||
requestId: 'req-1',
|
||||
}
|
||||
|
||||
it('失败日志写入后自动触发通知，参数与日志一致', async () => {
|
||||
await operationLogService.record(failureInput)
|
||||
await flushBackgroundTasks()
|
||||
|
||||
expect(mockCreate).toHaveBeenCalledTimes(1)
|
||||
expect(mockNotify).toHaveBeenCalledTimes(1)
|
||||
expect(mockNotify).toHaveBeenCalledWith({
|
||||
action: OperationLogAction.INTAKE_PRODUCT_WRITE,
|
||||
status: OperationLogStatus.FAILED,
|
||||
source: undefined,
|
||||
actorName: '槐米',
|
||||
productSpu: 'LRSF0095',
|
||||
skuCode: undefined,
|
||||
workspaceCategory: '样包测量',
|
||||
errorCode: 'P2028',
|
||||
message: '产品资料保存失败：Transaction API error',
|
||||
requestId: 'req-1',
|
||||
})
|
||||
})
|
||||
|
||||
it('成功日志不触发通知', async () => {
|
||||
await operationLogService.record({
|
||||
...failureInput,
|
||||
status: OperationLogStatus.SUCCESS,
|
||||
})
|
||||
await flushBackgroundTasks()
|
||||
|
||||
expect(mockNotify).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('RETRY 状态同样触发通知', async () => {
|
||||
await operationLogService.record({
|
||||
...failureInput,
|
||||
status: OperationLogStatus.RETRY,
|
||||
})
|
||||
await flushBackgroundTasks()
|
||||
|
||||
expect(mockNotify).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('日志写入失败时不抛错，也不触发通知', async () => {
|
||||
mockCreate.mockRejectedValueOnce(new Error('database down'))
|
||||
|
||||
await expect(operationLogService.record(failureInput)).resolves.toBeUndefined()
|
||||
await flushBackgroundTasks()
|
||||
|
||||
expect(mockNotify).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('通知本身抛错时 record 依然静默完成', async () => {
|
||||
mockNotify.mockRejectedValueOnce(new Error('notification exploded'))
|
||||
|
||||
await expect(operationLogService.record(failureInput)).resolves.toBeUndefined()
|
||||
await flushBackgroundTasks()
|
||||
|
||||
expect(mockNotify).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
})
|
||||
@ -0,0 +1,358 @@
|
||||
import 'server-only'
|
||||
|
||||
import { OperationLogAction, OperationLogStatus } from '@prisma/client'
|
||||
import {
|
||||
notificationDeliveryRepository,
|
||||
type NotificationRecipient,
|
||||
} from '@/server/repositories/notification-delivery.repository'
|
||||
import {
|
||||
notificationDeliveryService,
|
||||
type RecordUndeliverableNotificationInput,
|
||||
type SendEmployeeNotificationInput,
|
||||
} from './notification-delivery.service'
|
||||
|
||||
export const OPERATION_LOG_FAILURE_BUSINESS_EVENT = 'operation_log_failure'
|
||||
|
||||
const FAILURE_STATUSES = new Set<OperationLogStatus>([
|
||||
OperationLogStatus.FAILED,
|
||||
OperationLogStatus.RETRY,
|
||||
])
|
||||
|
||||
/**
|
||||
* 定时任务批量写入的错误码：一次运行会产生上千条失败日志，
|
||||
* 按长窗口折叠成一条通知；其余错误码按业务对象在短窗口内合并。
|
||||
*/
|
||||
const FOLDED_ERROR_CODES = new Set<string>(['JUSHUITAN_FIXED_VIRTUAL_INVENTORY_FAILED'])
|
||||
const FOLDED_WINDOW_MINUTES = 60
|
||||
const DEFAULT_WINDOW_MINUTES = 10
|
||||
const DEFAULT_RECIPIENT_NAME = '徐霞客'
|
||||
const DEFAULT_HOURLY_LIMIT = 30
|
||||
const RECIPIENT_CACHE_TTL_MS = 10 * 60 * 1000
|
||||
const REASON_MAX_LENGTH = 400
|
||||
const SHANGHAI_OFFSET_MS = 8 * 60 * 60 * 1000
|
||||
|
||||
const ACTION_LABELS: Record<OperationLogAction, string> = {
|
||||
INTAKE_PRODUCT_CREATE: '新建产品开发明细',
|
||||
INTAKE_PRODUCT_WRITE: '产品资料保存',
|
||||
INTAKE_PRODUCT_DELETE: '删除产品开发明细',
|
||||
INTAKE_FILE_UPLOAD: '文件上传',
|
||||
INTAKE_WORKSPACE_CONFIRMATION: '节点确认',
|
||||
JUSHUITAN_PRODUCT_PUSH: '聚水潭商品推送',
|
||||
JUSHUITAN_VIRTUAL_INVENTORY_PUSH: '聚水潭虚拟库存推送',
|
||||
SUPPLY_CHAIN_DROPSHIP_ISSUE_WRITE: '代发问题单写入',
|
||||
}
|
||||
|
||||
export type OperationLogFailurePayload = {
|
||||
action: OperationLogAction
|
||||
status: OperationLogStatus
|
||||
source?: 'USER' | 'SYSTEM' | null
|
||||
actorName: string
|
||||
productSpu?: string | null
|
||||
skuCode?: string | null
|
||||
workspaceCategory?: string | null
|
||||
errorCode?: string | null
|
||||
message: string
|
||||
requestId?: string | null
|
||||
createdAt?: Date | null
|
||||
}
|
||||
|
||||
export type OperationLogFailureNotificationOutcome =
|
||||
| 'sent'
|
||||
| 'ignored'
|
||||
| 'skipped-duplicate'
|
||||
| 'skipped-rate-limit'
|
||||
| 'undeliverable'
|
||||
| 'error'
|
||||
|
||||
export type OperationLogFailureNotificationResult = {
|
||||
outcome: OperationLogFailureNotificationOutcome
|
||||
idempotencyKey: string | null
|
||||
recipientName: string | null
|
||||
detail: string | null
|
||||
}
|
||||
|
||||
export type OperationLogFailureNotificationDeps = {
|
||||
findRecipientsByNames(names: string[]): Promise<NotificationRecipient[]>
|
||||
findByIdempotencyKey(idempotencyKey: string): Promise<{ id: string } | null>
|
||||
sendToEmployee(input: SendEmployeeNotificationInput): Promise<unknown>
|
||||
recordUndeliverable(input: RecordUndeliverableNotificationInput): Promise<unknown>
|
||||
now(): Date
|
||||
}
|
||||
|
||||
function resolveRecipientName() {
|
||||
return (process.env.OPERATION_LOG_NOTIFIER_RECIPIENT || DEFAULT_RECIPIENT_NAME).trim()
|
||||
}
|
||||
|
||||
function pad(value: number) {
|
||||
return String(value).padStart(2, '0')
|
||||
}
|
||||
|
||||
/** 统一按北京时间展示，不依赖服务器时区。 */
|
||||
export function formatShanghaiTimestamp(date: Date) {
|
||||
const shifted = new Date(date.getTime() + SHANGHAI_OFFSET_MS)
|
||||
return [
|
||||
`${shifted.getUTCFullYear()}-${pad(shifted.getUTCMonth() + 1)}-${pad(shifted.getUTCDate())}`,
|
||||
`${pad(shifted.getUTCHours())}:${pad(shifted.getUTCMinutes())}:${pad(shifted.getUTCSeconds())}`,
|
||||
].join(' ')
|
||||
}
|
||||
|
||||
function collapseWhitespace(value: string) {
|
||||
return value.replace(/\s+/g, ' ').trim()
|
||||
}
|
||||
|
||||
function truncate(value: string, maxLength: number) {
|
||||
const normalized = collapseWhitespace(value)
|
||||
return normalized.length <= maxLength ? normalized : `${normalized.slice(0, maxLength)}…`
|
||||
}
|
||||
|
||||
function stripActionPrefix(message: string, actionLabel: string) {
|
||||
const normalized = message.trim()
|
||||
const prefix = `${actionLabel}失败：`
|
||||
return normalized.startsWith(prefix) ? normalized.slice(prefix.length).trim() : normalized
|
||||
}
|
||||
|
||||
export function isFoldedOperationLogError(errorCode: string | null | undefined) {
|
||||
return Boolean(errorCode && FOLDED_ERROR_CODES.has(errorCode))
|
||||
}
|
||||
|
||||
export function describeOperationLogAction(action: OperationLogAction) {
|
||||
return ACTION_LABELS[action] ?? action
|
||||
}
|
||||
|
||||
/**
|
||||
* 通知与幂等键对同一窗口必须保持稳定：折叠类不含业务对象，
|
||||
* 因此一次定时任务的上千条失败只会命中同一个幂等键。
|
||||
*/
|
||||
export function buildOperationLogFailureNotificationKey(input: {
|
||||
action: OperationLogAction
|
||||
errorCode?: string | null
|
||||
productSpu?: string | null
|
||||
skuCode?: string | null
|
||||
now: Date
|
||||
}) {
|
||||
const folded = isFoldedOperationLogError(input.errorCode)
|
||||
const windowMs = (folded ? FOLDED_WINDOW_MINUTES : DEFAULT_WINDOW_MINUTES) * 60 * 1000
|
||||
const bucket = Math.floor(input.now.getTime() / windowMs)
|
||||
const parts = [
|
||||
'oplog-failure',
|
||||
folded ? 'folded' : 'scoped',
|
||||
String(bucket),
|
||||
input.action,
|
||||
input.errorCode?.trim() || 'UNKNOWN',
|
||||
]
|
||||
if (!folded) {
|
||||
parts.push(input.productSpu?.trim() || '-', input.skuCode?.trim() || '-')
|
||||
}
|
||||
return parts.join(':')
|
||||
}
|
||||
|
||||
export function buildOperationLogFailureMessage(input: OperationLogFailurePayload) {
|
||||
const actionLabel = describeOperationLogAction(input.action)
|
||||
const title = `${actionLabel}失败`
|
||||
const folded = isFoldedOperationLogError(input.errorCode)
|
||||
const sourceLabel = (input.source ?? 'USER') === 'SYSTEM' ? '系统任务' : '人工操作'
|
||||
|
||||
const identity = [
|
||||
input.productSpu?.trim() ? `产品：${input.productSpu.trim()}` : null,
|
||||
input.skuCode?.trim() ? `SKU：${input.skuCode.trim()}` : null,
|
||||
input.workspaceCategory?.trim() ? `节点：${input.workspaceCategory.trim()}` : null,
|
||||
].filter((line): line is string => Boolean(line))
|
||||
|
||||
const lines = [
|
||||
`动作：${actionLabel}`,
|
||||
`时间：${formatShanghaiTimestamp(input.createdAt ?? new Date())}`,
|
||||
`人员：${input.actorName.trim() || '未知'}（${sourceLabel}）`,
|
||||
...(identity.length > 0 ? [identity.join(' ／ ')] : []),
|
||||
...(input.errorCode?.trim() ? [`错误码：${input.errorCode.trim()}`] : []),
|
||||
`原因：${truncate(stripActionPrefix(input.message, actionLabel), REASON_MAX_LENGTH)}`,
|
||||
]
|
||||
|
||||
if (folded) {
|
||||
lines.push(`说明：该错误码在 ${FOLDED_WINDOW_MINUTES} 分钟内多次出现时只通知一次`)
|
||||
}
|
||||
|
||||
// 钉钉按 Markdown 渲染：单个 \n 会被合并成空格，必须用空行分段才能每条独立成行。
|
||||
return { title, content: lines.join('\n\n') }
|
||||
}
|
||||
|
||||
export class OperationLogFailureNotificationService {
|
||||
private recipientCache: { name: string; recipient: NotificationRecipient; at: number } | null = null
|
||||
private readonly hourlyCounters = new Map<string, number>()
|
||||
|
||||
constructor(
|
||||
private readonly deps: OperationLogFailureNotificationDeps = {
|
||||
findRecipientsByNames: (names) => notificationDeliveryRepository.findRecipientsByNames(names),
|
||||
findByIdempotencyKey: (key) => notificationDeliveryRepository.findByIdempotencyKey(key),
|
||||
sendToEmployee: (input) => notificationDeliveryService.sendToEmployee(input),
|
||||
recordUndeliverable: (input) => notificationDeliveryService.recordUndeliverable(input),
|
||||
now: () => new Date(),
|
||||
},
|
||||
) {}
|
||||
|
||||
/**
|
||||
* 失败日志通知的唯一入口。调用方必须 fire-and-forget，本方法自身绝不抛错。
|
||||
*/
|
||||
async notify(input: OperationLogFailurePayload): Promise<OperationLogFailureNotificationResult> {
|
||||
try {
|
||||
return await this.dispatch(input)
|
||||
} catch (error) {
|
||||
console.error('[OperationLogFailureNotify] 推送失败日志通知时发生异常', error)
|
||||
return {
|
||||
outcome: 'error',
|
||||
idempotencyKey: null,
|
||||
recipientName: null,
|
||||
detail: error instanceof Error ? error.message : String(error),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async dispatch(
|
||||
input: OperationLogFailurePayload,
|
||||
): Promise<OperationLogFailureNotificationResult> {
|
||||
if (!FAILURE_STATUSES.has(input.status)) {
|
||||
return { outcome: 'ignored', idempotencyKey: null, recipientName: null, detail: '非失败状态' }
|
||||
}
|
||||
|
||||
const now = this.deps.now()
|
||||
const idempotencyKey = buildOperationLogFailureNotificationKey({
|
||||
action: input.action,
|
||||
errorCode: input.errorCode,
|
||||
productSpu: input.productSpu,
|
||||
skuCode: input.skuCode,
|
||||
now,
|
||||
})
|
||||
|
||||
const existing = await this.deps.findByIdempotencyKey(idempotencyKey)
|
||||
if (existing) {
|
||||
return {
|
||||
outcome: 'skipped-duplicate',
|
||||
idempotencyKey,
|
||||
recipientName: null,
|
||||
detail: '同一时间窗内已通知过',
|
||||
}
|
||||
}
|
||||
|
||||
if (!this.consumeRateLimit(now)) {
|
||||
console.warn(`[OperationLogFailureNotify] 超过每小时推送上限，已跳过：${idempotencyKey}`)
|
||||
return {
|
||||
outcome: 'skipped-rate-limit',
|
||||
idempotencyKey,
|
||||
recipientName: null,
|
||||
detail: '超过每小时推送上限',
|
||||
}
|
||||
}
|
||||
|
||||
const { title, content } = buildOperationLogFailureMessage(input)
|
||||
const recipient = await this.resolveRecipient(now)
|
||||
|
||||
if (!recipient || !recipient.dingtalkUserId?.trim()) {
|
||||
const name = resolveRecipientName()
|
||||
const errorCode = recipient ? 'RECIPIENT_DINGTALK_UNBOUND' : 'RECIPIENT_NOT_FOUND'
|
||||
const errorMessage = recipient
|
||||
? `收件人 ${recipient.name} 尚未绑定钉钉账号`
|
||||
: `未找到系统用户：${name}`
|
||||
await this.deps.recordUndeliverable({
|
||||
idempotencyKey,
|
||||
businessEvent: OPERATION_LOG_FAILURE_BUSINESS_EVENT,
|
||||
businessRecordId: input.requestId?.trim() || undefined,
|
||||
source: 'SYSTEM',
|
||||
recipient: recipient
|
||||
? {
|
||||
id: recipient.id,
|
||||
name: recipient.name,
|
||||
email: recipient.email,
|
||||
dingtalkUserId: recipient.dingtalkUserId ?? '',
|
||||
}
|
||||
: {
|
||||
id: name,
|
||||
name,
|
||||
email: '',
|
||||
dingtalkUserId: '',
|
||||
},
|
||||
title,
|
||||
content,
|
||||
errorCode,
|
||||
errorMessage,
|
||||
})
|
||||
return { outcome: 'undeliverable', idempotencyKey, recipientName: recipient?.name ?? null, detail: errorCode }
|
||||
}
|
||||
|
||||
await this.deps.sendToEmployee({
|
||||
idempotencyKey,
|
||||
businessEvent: OPERATION_LOG_FAILURE_BUSINESS_EVENT,
|
||||
businessRecordId: input.requestId?.trim() || undefined,
|
||||
source: 'SYSTEM',
|
||||
recipientUserId: recipient.id,
|
||||
title,
|
||||
content,
|
||||
retryExisting: false,
|
||||
})
|
||||
|
||||
return { outcome: 'sent', idempotencyKey, recipientName: recipient.name, detail: null }
|
||||
}
|
||||
|
||||
private async resolveRecipient(now: Date): Promise<NotificationRecipient | null> {
|
||||
const name = resolveRecipientName()
|
||||
const cached = this.recipientCache
|
||||
if (cached && cached.name === name && now.getTime() - cached.at < RECIPIENT_CACHE_TTL_MS) {
|
||||
return cached.recipient
|
||||
}
|
||||
|
||||
const users = await this.deps.findRecipientsByNames([name])
|
||||
const recipient = users.find((user) => user.name === name) ?? null
|
||||
if (recipient) this.recipientCache = { name, recipient, at: now.getTime() }
|
||||
return recipient
|
||||
}
|
||||
|
||||
/** 逐小时熔断：只在确认要真正发送时计数，被幂等键挡下的重复不计入。 */
|
||||
private consumeRateLimit(now: Date) {
|
||||
const configured = Number(process.env.OPERATION_LOG_NOTIFIER_HOURLY_LIMIT)
|
||||
const limit = Number.isFinite(configured) && configured > 0 ? configured : DEFAULT_HOURLY_LIMIT
|
||||
const bucket = String(Math.floor(now.getTime() / 3_600_000))
|
||||
|
||||
for (const key of [...this.hourlyCounters.keys()]) {
|
||||
if (key !== bucket) this.hourlyCounters.delete(key)
|
||||
}
|
||||
|
||||
const used = this.hourlyCounters.get(bucket) ?? 0
|
||||
if (used >= limit) return false
|
||||
this.hourlyCounters.set(bucket, used + 1)
|
||||
return true
|
||||
}
|
||||
|
||||
/** 手动测试入口：绕过幂等键，直接给收件人发一条测试通知。 */
|
||||
async sendTestNotification() {
|
||||
const now = this.deps.now()
|
||||
const recipient = await this.resolveRecipient(now)
|
||||
if (!recipient) {
|
||||
throw new Error(`未找到系统用户：${resolveRecipientName()}`)
|
||||
}
|
||||
|
||||
const { title, content } = buildOperationLogFailureMessage({
|
||||
action: OperationLogAction.INTAKE_PRODUCT_WRITE,
|
||||
status: OperationLogStatus.FAILED,
|
||||
source: 'USER',
|
||||
actorName: recipient.name,
|
||||
productSpu: 'TEST-SPU',
|
||||
workspaceCategory: '样包测量',
|
||||
errorCode: 'P2028',
|
||||
message: '产品资料保存失败：链路测试消息，收到即代表失败日志通知已接通',
|
||||
createdAt: now,
|
||||
})
|
||||
|
||||
const idempotencyKey = `oplog-failure-test:${now.getTime()}`
|
||||
await this.deps.sendToEmployee({
|
||||
idempotencyKey,
|
||||
businessEvent: `${OPERATION_LOG_FAILURE_BUSINESS_EVENT}_test`,
|
||||
source: 'SYSTEM',
|
||||
recipientUserId: recipient.id,
|
||||
title: `（测试）${title}`,
|
||||
content,
|
||||
retryExisting: false,
|
||||
})
|
||||
|
||||
return { idempotencyKey, recipientName: recipient.name, title, content }
|
||||
}
|
||||
}
|
||||
|
||||
export const operationLogFailureNotificationService = new OperationLogFailureNotificationService()
|
||||
@ -9,6 +9,7 @@ import {
|
||||
type OperationLogFilter,
|
||||
type OperationLogWithActor,
|
||||
} from '@/server/repositories/operation-log.repository'
|
||||
import { operationLogFailureNotificationService } from './operation-log-failure-notification.service'
|
||||
|
||||
export { OperationLogAction, OperationLogStatus }
|
||||
export type OperationLogInput = CreateOperationLogInput
|
||||
@ -40,12 +41,40 @@ class OperationLogService {
|
||||
async record(input: OperationLogInput): Promise<void> {
|
||||
try {
|
||||
await operationLogRepository.create(input)
|
||||
this.notifyFailureInBackground(input)
|
||||
} catch (error) {
|
||||
// Logging must never prevent the business operation from completing.
|
||||
console.error('[OperationLog] Failed to record operation', error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 失败日志通知是旁路能力：不 await、不抛错。
|
||||
* 既不拖慢保存等业务响应，也不影响日志写入本身。
|
||||
*/
|
||||
private notifyFailureInBackground(input: OperationLogInput) {
|
||||
if (input.status === OperationLogStatus.SUCCESS) return
|
||||
// 单测环境保持静默，避免自动化测试真的发出外部通知。
|
||||
if (process.env.NODE_ENV === 'test') return
|
||||
|
||||
void operationLogFailureNotificationService
|
||||
.notify({
|
||||
action: input.action,
|
||||
status: input.status,
|
||||
source: input.source,
|
||||
actorName: input.actorName,
|
||||
productSpu: input.productSpu,
|
||||
skuCode: input.skuCode,
|
||||
workspaceCategory: input.workspaceCategory,
|
||||
errorCode: input.errorCode,
|
||||
message: input.message,
|
||||
requestId: input.requestId,
|
||||
})
|
||||
.catch((error) => {
|
||||
console.error('[OperationLog] Failed to notify failure log', error)
|
||||
})
|
||||
}
|
||||
|
||||
async list(options: OperationLogListOptions = {}) {
|
||||
const page = Math.max(1, options.page ?? 1)
|
||||
const pageSize = Math.min(200, Math.max(1, options.pageSize ?? 50))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user