From b8a73e963c96b2a3ea6012ed979d8a2c9155aab4 Mon Sep 17 00:00:00 2001 From: chenpei Date: Wed, 26 Aug 2026 11:35:24 +0800 Subject: [PATCH] =?UTF-8?q?=E9=80=9A=E7=9F=A5=E4=B8=AD=E5=BF=83=E5=8F=AF?= =?UTF-8?q?=E9=9D=A0=E6=80=A7=E5=8A=A0=E5=9B=BA=EF=BC=9Atoken=E5=A4=B1?= =?UTF-8?q?=E6=95=88=E9=87=8D=E8=AF=95=E3=80=81=E5=A4=B1=E8=B4=A5=E5=8F=AF?= =?UTF-8?q?=E8=A7=81=E6=80=A7=E3=80=81SQLite=20WAL=E3=80=81=E4=BF=9D?= =?UTF-8?q?=E7=95=99=E6=9C=9F=E6=B8=85=E7=90=86=E3=80=81=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E6=B6=88=E8=B4=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docker-compose.yml | 3 +- src/__tests__/deployment.test.ts | 3 +- src/__tests__/server.test.ts | 20 ++++++ src/__tests__/worker.test.ts | 106 ++++++++++++++++++++++++++++++- src/dingtalk.ts | 100 +++++++++++++++++------------ src/index.ts | 21 +++++- src/prisma.ts | 20 ++++++ src/server.ts | 34 +++++++++- src/worker.ts | 46 +++++++++++++- 9 files changed, 303 insertions(+), 50 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 634fc50..19daccb 100644 --- a/docker-compose.yml +++ b/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} diff --git a/src/__tests__/deployment.test.ts b/src/__tests__/deployment.test.ts index 6ec4a7a..30c9961 100644 --- a/src/__tests__/deployment.test.ts +++ b/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"/) diff --git a/src/__tests__/server.test.ts b/src/__tests__/server.test.ts index 9760035..421cd5b 100644 --- a/src/__tests__/server.test.ts +++ b/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, diff --git a/src/__tests__/worker.test.ts b/src/__tests__/worker.test.ts index b65119d..4d71d21 100644 --- a/src/__tests__/worker.test.ts +++ b/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)['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 diff --git a/src/dingtalk.ts b/src/dingtalk.ts index 6a89d67..b7c05f9 100644 --- a/src/dingtalk.ts +++ b/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 (fn: () => Promise): Promise => { + 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,51 +126,55 @@ export const sendRobotMessage = async (input: { actionUrl: string }) => { const { robotCode } = getDingTalkConfig() - const token = await getToken() - const data = await requestJson('https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend', { - method: 'POST', - headers: { 'x-acs-dingtalk-access-token': token }, - body: JSON.stringify({ - robotCode, - userIds: [input.recipientUserId], - msgKey: input.actionUrl ? 'sampleActionCard' : 'sampleMarkdown', - msgParam: JSON.stringify(input.actionUrl ? { - title: input.title.slice(0, 120), - text: input.content.slice(0, 4800), - singleTitle: input.actionTitle.slice(0, 20), - singleURL: input.actionUrl, - } : { - title: input.title.slice(0, 120), - text: input.content.slice(0, 4800), + return withFreshTokenRetry(async () => { + const token = await getToken() + const data = await requestJson('https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend', { + method: 'POST', + headers: { 'x-acs-dingtalk-access-token': token }, + body: JSON.stringify({ + robotCode, + userIds: [input.recipientUserId], + msgKey: input.actionUrl ? 'sampleActionCard' : 'sampleMarkdown', + msgParam: JSON.stringify(input.actionUrl ? { + title: input.title.slice(0, 120), + text: input.content.slice(0, 4800), + singleTitle: input.actionTitle.slice(0, 20), + singleURL: input.actionUrl, + } : { + title: input.title.slice(0, 120), + text: input.content.slice(0, 4800), + }), }), - }), - }, { deliveryOutcomeCanBeUnknown: true }) + }, { deliveryOutcomeCanBeUnknown: true }) - const invalid = stringList(data.invalidStaffIdList) - const filtered = stringList(data.filteredStaffIdList) - const flowControlled = stringList(data.flowControlledStaffIdList) - if (invalid.includes(input.recipientUserId)) throw new DingTalkError('收件人钉钉 userId 无效', 'INVALID_USER', true) - if (filtered.includes(input.recipientUserId)) throw new DingTalkError('收件人不在机器人可见范围内', 'FILTERED_USER', true) - if (flowControlled.includes(input.recipientUserId)) throw new DingTalkError('钉钉机器人发送被限流', 'FLOW_CONTROLLED') - const processQueryKey = clean(data.processQueryKey) - if (!processQueryKey) { - throw new DingTalkError('钉钉未返回 processQueryKey，发送结果需要人工核验', 'PROCESS_QUERY_KEY_MISSING', false, true) - } - return processQueryKey + const invalid = stringList(data.invalidStaffIdList) + const filtered = stringList(data.filteredStaffIdList) + const flowControlled = stringList(data.flowControlledStaffIdList) + if (invalid.includes(input.recipientUserId)) throw new DingTalkError('收件人钉钉 userId 无效', 'INVALID_USER', true) + if (filtered.includes(input.recipientUserId)) throw new DingTalkError('收件人不在机器人可见范围内', 'FILTERED_USER', true) + if (flowControlled.includes(input.recipientUserId)) throw new DingTalkError('钉钉机器人发送被限流', 'FLOW_CONTROLLED') + const processQueryKey = clean(data.processQueryKey) + if (!processQueryKey) { + throw new DingTalkError('钉钉未返回 processQueryKey，发送结果需要人工核验', 'PROCESS_QUERY_KEY_MISSING', false, true) + } + return processQueryKey + }) } export const queryRobotReadStatus = async (processQueryKey: string) => { const { robotCode } = getDingTalkConfig() - const token = await getToken() - const url = new URL('https://api.dingtalk.com/v1.0/robot/oToMessages/readStatus') - url.searchParams.set('robotCode', robotCode) - url.searchParams.set('processQueryKey', processQueryKey) - const data = await requestJson(url.toString(), { - headers: { 'x-acs-dingtalk-access-token': token }, + 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) + url.searchParams.set('processQueryKey', processQueryKey) + const data = await requestJson(url.toString(), { + headers: { 'x-acs-dingtalk-access-token': token }, + }) + const rows = Array.isArray(data.messageReadInfoList) ? data.messageReadInfoList : [] + const read = rows.some((row) => clean(row?.readStatus).toLowerCase() === 'read') + const unread = rows.some((row) => clean(row?.readStatus).toLowerCase() === 'unread') + const failed = clean(data.sendStatus).toLowerCase().includes('fail') + return { read, unread, failed } }) - const rows = Array.isArray(data.messageReadInfoList) ? data.messageReadInfoList : [] - const read = rows.some((row) => clean(row?.readStatus).toLowerCase() === 'read') - const unread = rows.some((row) => clean(row?.readStatus).toLowerCase() === 'unread') - const failed = clean(data.sendStatus).toLowerCase().includes('fail') - return { read, unread, failed } } diff --git a/src/index.ts b/src/index.ts index 3766a42..0f3040a 100644 --- a/src/index.ts +++ b/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, () => { - console.log(`[NotificationCenter] listening on ${port}, projects=${projects.size}`) + +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() diff --git a/src/prisma.ts b/src/prisma.ts index 4590932..2f27701 100644 --- a/src/prisma.ts +++ b/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 diff --git a/src/server.ts b/src/server.ts index cb8ef2f..774e74f 100644 --- a/src/server.ts +++ b/src/server.ts @@ -75,7 +75,19 @@ export const createServer = (projects: Map, 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 = {} + 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, 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! diff --git a/src/worker.ts b/src/worker.ts index c2d1c19..d06615b 100644 --- a/src/worker.ts +++ b/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 => { 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() }