YouYiDianNotificationCenter/src/__tests__/worker.test.ts

import assert from 'node:assert/strict'
import test from 'node:test'
import { DingTalkError, sendRobotMessage } from '../dingtalk'
import prisma from '../prisma'
import { classifyDeliveryFailure, quarantineExpiredSendingLeases } from '../worker'

test('发送请求超时属于结果不确定并停止自动重试', 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'
  let calls = 0
  global.fetch = (async () => {
    calls += 1
    if (calls === 1) return new Response(JSON.stringify({ errcode: 0, access_token: 'test-token', expires_in: 7200 }), { status: 200 })
    throw new DOMException('aborted', 'AbortError')
  }) 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.outcomeUnknown, true)
    assert.deepEqual(classifyDeliveryFailure({ error: caught, attemptCount: 1, maxAttempts: 4 }), {
      failed: true,
      errorCode: 'DELIVERY_OUTCOME_UNKNOWN',
      errorMessage: '钉钉发送结果不确定，请先在钉钉侧人工核验，系统不会自动重试',
    })
  } finally {
    global.fetch = originalFetch
    process.env.DINGTALK_APP_KEY = oldAppKey
    process.env.DINGTALK_APP_SECRET = oldAppSecret
    process.env.DINGTALK_ROBOT_CODE = oldRobotCode
  }
})

test('机器人 ActionCard 原样发送统一标题、分段正文和按钮跳转', 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'
  let batchRequest: Record<string, any> | undefined
  global.fetch = (async (input, init) => {
    const url = String(input)
    if (url.includes('/gettoken?')) {
      return new Response(JSON.stringify({ errcode: 0, access_token: 'test-token', expires_in: 7200 }), { status: 200 })
    }
    assert.equal(url, 'https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend')
    batchRequest = JSON.parse(String(init?.body || '{}'))
    return new Response(JSON.stringify({ processQueryKey: 'process-query-key-1' }), { status: 200 })
  }) as typeof fetch

  const input = {
    recipientUserId: 'ding-user-1',
    title: '【固定资产】资产报废待审批',
    content: '### 【固定资产】资产报废待审批\n\n**申请时间：** 2026年7月16日 14:30\n\n**资产：** 主机（JS-PC-0010）',
    actionTitle: '立即审批',
    actionUrl: 'https://asset.example.com/workflows/disposals/53',
  }

  try {
    assert.equal(await sendRobotMessage(input), 'process-query-key-1')
    assert.ok(batchRequest)
    assert.equal(batchRequest.robotCode, 'test-robot-code')
    assert.deepEqual(batchRequest.userIds, ['ding-user-1'])
    assert.equal(batchRequest.msgKey, 'sampleActionCard')
    assert.deepEqual(JSON.parse(batchRequest.msgParam), {
      title: input.title,
      text: input.content,
      singleTitle: input.actionTitle,
      singleURL: input.actionUrl,
    })
  } 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(classifyDeliveryFailure({
    error: new DingTalkError('限流', 'FLOW_CONTROLLED'), attemptCount: 1, maxAttempts: 4,
  }).failed, false)
  assert.equal(classifyDeliveryFailure({
    error: new DingTalkError('用户无效', 'INVALID_USER', true), attemptCount: 1, maxAttempts: 4,
  }).failed, true)
  assert.equal(classifyDeliveryFailure({
    error: new DingTalkError('限流', 'FLOW_CONTROLLED'), attemptCount: 4, maxAttempts: 5,
  }).failed, false)
  assert.equal(classifyDeliveryFailure({
    error: new DingTalkError('限流', 'FLOW_CONTROLLED'), attemptCount: 5, maxAttempts: 5,
  }).failed, true)
})

test('进程崩溃遗留的过期 sending 租约转人工核验而不是再次发送', async () => {
  await prisma.notificationDelivery.deleteMany({ where: { id: 'expired-sending-test' } })
  await prisma.notificationDelivery.create({
    data: {
      id: 'expired-sending-test',
      projectCode: 'fixed_asset',
      idempotencyKey: 'expired-sending-test',
      requestHash: 'hash',
      recipientUserId: 'ding-user-1',
      businessTitle: '测试',
      renderedTitle: '测试',
      content: '测试',
      actionTitle: '查看',
      actionUrl: 'https://asset.example.com/dashboard',
      status: 'sending',
      attemptCount: 1,
      leaseOwner: 'crashed-worker',
      leaseExpiresAt: new Date(Date.now() - 1000),
    },
  })
  const result = await quarantineExpiredSendingLeases()
  assert.equal(result.count, 1)
  const row = await prisma.notificationDelivery.findUniqueOrThrow({ where: { id: 'expired-sending-test' } })
  assert.equal(row.status, 'failed')
  assert.equal(row.lastErrorCode, 'DELIVERY_OUTCOME_UNKNOWN')
  assert.equal(row.attemptCount, 1)
  assert.equal(row.leaseOwner, null)
  assert.equal(row.leaseExpiresAt, null)
})