通知中心可靠性加固：token失效重试、失败可见性、SQLite WAL、保留期清理、批量消费

docker-compose.yml

@@ -8,7 +8,8 @@ services:
       - "127.0.0.1:${PORT:-3088}:3088"
     environment:
       PORT: 3088
-      DATABASE_URL: file:/app/data/notifications.db
+      # 单连接 + 启动期 PRAGMA(WAL/busy_timeout)：SQLite 写锁不再放大为整库请求超时。
+      DATABASE_URL: file:/app/data/notifications.db?connection_limit=1
       DINGTALK_APP_KEY: ${DINGTALK_APP_KEY}
       DINGTALK_APP_SECRET: ${DINGTALK_APP_SECRET}
       DINGTALK_ROBOT_CODE: ${DINGTALK_ROBOT_CODE}

src/__tests__/deployment.test.ts

@@ -19,7 +19,8 @@ test('生产发布面透传免登配置并保留已有项目 login', () => {
     assert.match(deploy, new RegExp(key))
   }
   assert.match(deploy, /old_login = by_code\.get\(code, \{\}\)\.get\('login'\)/)
-  assert.match(deploy, /project\['login'\] = old_login/)
+  assert.match(deploy, /def project_with_preserved_login/)
+  assert.match(deploy, /project\['login'\] = login/)
   assert.match(deploy, /docker pause youyidian-notification-center/)
   assert.match(deploy, /docker cp youyidian-notification-center:\/app\/data\/\. "\$data_backup\/"/)
   assert.match(deploy, /test -s "\$data_backup\/notifications\.db"/)

src/__tests__/server.test.ts

@@ -61,6 +61,26 @@ test('通知中心验证签名、项目域名和幂等键', async () => {
     const rejected = await fetch(`http://127.0.0.1:${port}${path}`, { method: 'POST', headers: headers('POST', path, badBody), body: badBody })
     assert.equal(rejected.status, 422)
 
+    // 按状态列出本项目投递记录，接入方对账失败用（签名按纯路径，与现有协议一致）
+    await prisma.notificationDelivery.update({
+      where: { id: firstPayload.id },
+      data: { status: 'failed', lastErrorCode: 'DELIVERY_OUTCOME_UNKNOWN', lastErrorMessage: '人工核验' },
+    })
+    const listResponse = await fetch(`http://127.0.0.1:${port}/v1/notifications?status=failed&limit=5`, {
+      method: 'GET',
+      headers: headers('GET', '/v1/notifications', ''),
+    })
+    assert.equal(listResponse.status, 200)
+    const listPayload = await listResponse.json() as any
+    assert.equal(listPayload.items.length, 1)
+    assert.equal(listPayload.items[0].id, firstPayload.id)
+
+    const healthResponse = await fetch(`http://127.0.0.1:${port}/health`)
+    const healthPayload = await healthResponse.json() as any
+    assert.equal(healthPayload.ok, true)
+    assert.equal(healthPayload.deliveries.counts.failed, 1)
+    assert.equal(healthPayload.deliveries.outcomeUnknown, 1)
+
     const reservedTitleBody = JSON.stringify({ ...JSON.parse(body), idempotencyKey: 'reserved-title', title: '【固定资产】待审批报废' })
     const reservedTitle = await fetch(`http://127.0.0.1:${port}${path}`, {
       method: 'POST', headers: headers('POST', path, reservedTitleBody), body: reservedTitleBody,

src/__tests__/worker.test.ts

@@ -2,7 +2,111 @@ 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'
+import { classifyDeliveryFailure, cleanupDeliveries, quarantineExpiredSendingLeases } from '../worker'
+
+test('token 失效先刷新重试一次，避免缓存过期导致批量误判终态', 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'
+  const batchSendTokens: string[] = []
+  global.fetch = (async (input, init) => {
+    const url = String(input)
+    if (url.includes('/gettoken?')) {
+      const issued = batchSendTokens.length === 0 ? 'stale-token' : 'fresh-token'
+      return new Response(JSON.stringify({ errcode: 0, access_token: issued, expires_in: 7200 }), { status: 200 })
+    }
+    batchSendTokens.push(String((init?.headers as Record<string, string>)['x-acs-dingtalk-access-token'] || ''))
+    if (batchSendTokens.length === 1) {
+      // 旧 token 已在钉钉侧失效
+      return new Response(JSON.stringify({ errcode: 40014, errmsg: 'access token expired' }), { status: 200 })
+    }
+    return new Response(JSON.stringify({ processQueryKey: 'pqk-refreshed' }), { status: 200 })
+  }) as typeof fetch
+
+  try {
+    const key = await sendRobotMessage({ recipientUserId: 'ding-user-1', title: '测试', content: '测试', actionTitle: '查看', actionUrl: 'https://example.com' })
+    assert.equal(key, 'pqk-refreshed')
+    assert.deepEqual(batchSendTokens, ['stale-token', 'fresh-token'])
+  } finally {
+    global.fetch = originalFetch
+    process.env.DINGTALK_APP_KEY = oldAppKey
+    process.env.DINGTALK_APP_SECRET = oldAppSecret
+    process.env.DINGTALK_ROBOT_CODE = oldRobotCode
+  }
+})
+
+test('token 刷新后仍失效则维持确定性失败语义', 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 batchSends = 0
+  let tokenIssues = 0
+  global.fetch = (async (input) => {
+    const url = String(input)
+    if (url.includes('/gettoken?')) {
+      tokenIssues += 1
+      return new Response(JSON.stringify({ errcode: 0, access_token: `token-${tokenIssues}`, expires_in: 7200 }), { status: 200 })
+    }
+    batchSends += 1
+    return new Response(JSON.stringify({ errcode: 40014, errmsg: 'access token expired' }), { status: 200 })
+  }) 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, '40014')
+    // 刷新重试恰好一次，之后按确定性错误上抛
+    assert.equal(batchSends, 2)
+    assert.equal(classifyDeliveryFailure({ error: caught, attemptCount: 1, maxAttempts: 4 }).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('长期未读转 expired 停止回查，终态记录按保留期清理且 failed 保留', async () => {
+  // 测试文件按字母序串行共享同一个临时库，清全表保证清理计数不被前序用例残留影响。
+  await prisma.notificationDelivery.deleteMany({})
+  const daysAgo = (days: number) => new Date(Date.now() - days * 24 * 60 * 60 * 1000)
+  const base = {
+    projectCode: 'fixed_asset',
+    requestHash: 'hash',
+    recipientUserId: 'ding-user-1',
+    businessTitle: '测试',
+    renderedTitle: '测试',
+    content: '测试',
+    actionTitle: '查看',
+    actionUrl: 'https://asset.example.com/dashboard',
+  }
+  await prisma.notificationDelivery.create({ data: { ...base, id: 'cleanup-unread-old', idempotencyKey: 'cleanup-unread-old', status: 'unread', processQueryKey: 'pqk', updatedAt: daysAgo(15), createdAt: daysAgo(20) } })
+  await prisma.notificationDelivery.create({ data: { ...base, id: 'cleanup-sent-purge', idempotencyKey: 'cleanup-sent-purge', status: 'sent', processQueryKey: 'pqk', updatedAt: daysAgo(91), createdAt: daysAgo(95) } })
+  await prisma.notificationDelivery.create({ data: { ...base, id: 'cleanup-read-keep', idempotencyKey: 'cleanup-read-keep', status: 'read', readAt: daysAgo(10), updatedAt: daysAgo(10), createdAt: daysAgo(20) } })
+  await prisma.notificationDelivery.create({ data: { ...base, id: 'cleanup-failed-keep', idempotencyKey: 'cleanup-failed-keep', status: 'failed', lastErrorCode: 'DELIVERY_OUTCOME_UNKNOWN', updatedAt: daysAgo(120), createdAt: daysAgo(130) } })
+
+  const result = await cleanupDeliveries()
+  // 先删后转：91 天的 sent-purge 按原始时间戳直接删除；15 天的 unread-old 只转 expired 止血。
+  assert.equal(result.expiredUnread, 1)
+  assert.equal(result.purged, 1)
+  assert.equal((await prisma.notificationDelivery.findUnique({ where: { id: 'cleanup-unread-old' } }))?.status, 'expired')
+  assert.equal(await prisma.notificationDelivery.findUnique({ where: { id: 'cleanup-sent-purge' } }), null)
+  assert.equal((await prisma.notificationDelivery.findUnique({ where: { id: 'cleanup-read-keep' } }))?.status, 'read')
+  assert.equal((await prisma.notificationDelivery.findUnique({ where: { id: 'cleanup-failed-keep' } }))?.status, 'failed')
+})
 
 test('发送请求超时属于结果不确定并停止自动重试', async () => {
   const originalFetch = global.fetch

src/dingtalk.ts

@@ -68,6 +68,22 @@ 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))
 
+// token 过期属于可恢复错误：先作废本地缓存、用新 token 重试一次；仍失败再按原语义上抛。
+const AUTH_EXPIRED_CODES = new Set(['40014', 'HTTP_401', 'InvalidAuthentication'])
+const isAuthExpiredError = (error: unknown) =>
+  error instanceof DingTalkError &&
+  (AUTH_EXPIRED_CODES.has(error.code) || /invalid\s*(token|authentication)|token\s*(expired|invalid)/i.test(`${error.code} ${error.message}`))
+
+const withFreshTokenRetry = async <T>(fn: () => Promise<T>): Promise<T> => {
+  try {
+    return await fn()
+  } catch (error) {
+    if (!isAuthExpiredError(error)) throw error
+    tokenCache = null
+    return fn()
+  }
+}
+
 const getToken = async () => {
   if (tokenCache && tokenCache.expiresAt > Date.now() + 60_000) return tokenCache.token
   if (tokenPromise) return tokenPromise
@@ -110,6 +126,7 @@ export const sendRobotMessage = async (input: {
   actionUrl: string
 }) => {
   const { robotCode } = getDingTalkConfig()
+  return withFreshTokenRetry(async () => {
     const token = await getToken()
     const data = await requestJson('https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend', {
       method: 'POST',
@@ -141,10 +158,12 @@ export const sendRobotMessage = async (input: {
       throw new DingTalkError('钉钉未返回 processQueryKey，发送结果需要人工核验', 'PROCESS_QUERY_KEY_MISSING', false, true)
     }
     return processQueryKey
+  })
 }
 
 export const queryRobotReadStatus = async (processQueryKey: string) => {
   const { robotCode } = getDingTalkConfig()
+  return withFreshTokenRetry(async () => {
     const token = await getToken()
     const url = new URL('https://api.dingtalk.com/v1.0/robot/oToMessages/readStatus')
     url.searchParams.set('robotCode', robotCode)
@@ -157,4 +176,5 @@ export const queryRobotReadStatus = async (processQueryKey: string) => {
     const unread = rows.some((row) => clean(row?.readStatus).toLowerCase() === 'unread')
     const failed = clean(data.sendStatus).toLowerCase().includes('fail')
     return { read, unread, failed }
+  })
 }

src/index.ts

@@ -1,5 +1,6 @@
 import 'dotenv/config'
 import { assertRuntimeConfig, loadProjectConfigs } from './config'
+import prisma, { ensureDatabaseReady } from './prisma'
 import { createServer } from './server'
 import { startWorker } from './worker'
 
@@ -7,7 +8,21 @@ const projects = loadProjectConfigs()
 assertRuntimeConfig(projects)
 
 const port = Math.max(1, Number(process.env.PORT) || 3088)
-createServer(projects).listen(port, () => {
+
+const main = async () => {
+  await ensureDatabaseReady()
+  createServer(projects).listen(port, () => {
     console.log(`[NotificationCenter] listening on ${port}, projects=${projects.size}`)
+  })
+  startWorker()
+}
+
+main().catch((error) => {
+  console.error('[NotificationCenter] startup failed:', error)
+  process.exitCode = 1
+})
+
+process.on('SIGTERM', async () => {
+  await prisma.$disconnect().catch(() => undefined)
+  process.exit(0)
 })
-startWorker()

src/prisma.ts

@@ -2,4 +2,24 @@ import { PrismaClient } from '@prisma/client'
 
 const prisma = new PrismaClient()
 
+// SQLite 默认 DELETE journal 全库写锁；WAL 持久化到库文件，busy_timeout 需每个连接设置。
+// 启动期数据库尚未就绪时按递增间隔重试，避免 worker 首查就抛 PrismaClientKnownRequestError。
+export const ensureDatabaseReady = async (maxAttempts = 10) => {
+  let lastError: unknown
+  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+    try {
+      await prisma.$queryRawUnsafe('PRAGMA journal_mode=WAL;')
+      await prisma.$queryRawUnsafe('PRAGMA busy_timeout=5000;')
+      await prisma.$queryRaw`SELECT 1`
+      return
+    } catch (error) {
+      lastError = error
+      if (attempt < maxAttempts) {
+        await new Promise((resolve) => setTimeout(resolve, 300 * attempt))
+      }
+    }
+  }
+  throw lastError
+}
+
 export default prisma

src/server.ts

@@ -75,7 +75,19 @@ export const createServer = (projects: Map<string, ProjectConfig>, dependencies:
   app.get('/health', async (_req, res) => {
     try {
       await prisma.$queryRaw`SELECT 1`
-      res.json({ ok: true, service: 'youyidian-notification-center', database: 'ready' })
+      // 接入方只能看到「已投递给通知中心」，这里的失败计数是对账的唯一切口。
+      const groups = await prisma.notificationDelivery.groupBy({ by: ['status'], _count: { _all: true } })
+      const counts: Record<string, number> = {}
+      for (const group of groups) counts[group.status] = group._count._all
+      const outcomeUnknown = await prisma.notificationDelivery.count({
+        where: { status: 'failed', lastErrorCode: 'DELIVERY_OUTCOME_UNKNOWN' },
+      })
+      res.json({
+        ok: true,
+        service: 'youyidian-notification-center',
+        database: 'ready',
+        deliveries: { counts, failed: counts.failed || 0, outcomeUnknown },
+      })
     } catch {
       res.status(503).json({ ok: false, service: 'youyidian-notification-center', database: 'unavailable' })
     }
@@ -279,6 +291,26 @@ export const createServer = (projects: Map<string, ProjectConfig>, dependencies:
     }
   })
 
+  // 按状态分页列出本项目投递记录，供接入方对账失败/未读等异常。
+  app.get('/v1/notifications', authenticate, async (req: SignedRequest, res, next) => {
+    try {
+      const status = clean(req.query.status)
+      const limit = Math.min(50, Math.max(1, Number(req.query.limit) || 20))
+      const rows = await prisma.notificationDelivery.findMany({
+        where: {
+          projectCode: req.project!.code,
+          ...(status ? { status } : {}),
+        },
+        orderBy: { createdAt: 'desc' },
+        take: limit,
+      })
+      res.set('Cache-Control', 'no-store')
+      res.json({ items: rows.map(deliveryView) })
+    } catch (error) {
+      next(error)
+    }
+  })
+
   app.post('/v1/auth/tickets/redeem', authenticate, async (req: SignedRequest, res, next) => {
     try {
       const project = req.project!

src/worker.ts

@@ -4,6 +4,12 @@ import { DingTalkError, queryRobotReadStatus, sendRobotMessage } from './dingtal
 
 const workerId = `${os.hostname()}:${process.pid}:${Math.random().toString(36).slice(2, 10)}`
 const backoffMinutes = [1, 5, 15, 30]
+// 单 tick 批量消费，避免积压时每秒只推进一条。
+const MAX_DELIVERIES_PER_TICK = 10
+// sent/unread 超过该天数仍无终态则停止回查；终态记录保留该天数后清理。
+const READ_STATUS_EXPIRY_DAYS = 14
+const DELIVERY_RETENTION_DAYS = 90
+const CLEANUP_INTERVAL_MS = 60 * 60 * 1000
 let busy = false
 let readBusy = false
 const outcomeUnknownMessage = '钉钉发送结果不确定，请先在钉钉侧人工核验，系统不会自动重试'
@@ -66,9 +72,9 @@ const claimNext = async () => {
   return prisma.notificationDelivery.findUnique({ where: { id: candidate.id } })
 }
 
-const processOne = async () => {
+const processOne = async (): Promise<boolean> => {
   const delivery = await claimNext()
-  if (!delivery) return
+  if (!delivery) return false
   try {
     const processQueryKey = await sendRobotMessage({
       recipientUserId: delivery.recipientUserId,
@@ -109,6 +115,31 @@ const processOne = async () => {
       },
     })
   }
+  return true
+}
+
+/**
+ * 周期清理：终态记录超过保留期后删除；长期未读停止回查并转 expired。
+ * 必须先删后转——updateMany 会刷新 updatedAt，先转换会让老记录重新计时 90 天。
+ * failed（人工核验线索）不自动删除。
+ */
+export const cleanupDeliveries = async (now = new Date()) => {
+  const readExpiry = new Date(now.getTime() - READ_STATUS_EXPIRY_DAYS * 24 * 60 * 60 * 1000)
+  const retentionExpiry = new Date(now.getTime() - DELIVERY_RETENTION_DAYS * 24 * 60 * 60 * 1000)
+  const purged = await prisma.notificationDelivery.deleteMany({
+    where: {
+      status: { in: ['sent', 'read', 'expired'] },
+      updatedAt: { lt: retentionExpiry },
+    },
+  })
+  const expired = await prisma.notificationDelivery.updateMany({
+    where: {
+      status: { in: ['sent', 'unread'] },
+      updatedAt: { lt: readExpiry },
+    },
+    data: { status: 'expired' },
+  })
+  return { expiredUnread: expired.count, purged: purged.count }
 }
 
 const refreshOne = async () => {
@@ -146,7 +177,10 @@ export const startWorker = () => {
     if (busy) return
     busy = true
     try {
-      await processOne()
+      for (let processed = 0; processed < MAX_DELIVERIES_PER_TICK; processed++) {
+        const claimed = await processOne()
+        if (!claimed) break
+      }
     } finally {
       busy = false
     }
@@ -157,5 +191,11 @@ export const startWorker = () => {
     readBusy = true
     void refreshOne().finally(() => { readBusy = false })
   }, readInterval).unref()
+  setInterval(() => {
+    void cleanupDeliveries().catch((error) => {
+      console.error('[worker] cleanup failed:', error instanceof Error ? error.message : String(error))
+    })
+  }, CLEANUP_INTERVAL_MS).unref()
   setTimeout(() => void tick(), 500).unref()
+  setTimeout(() => void cleanupDeliveries().catch(() => undefined), 30_000).unref()
 }