pushCoordinator.test.ts 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551
  1. import { generateKeyPairSync } from 'node:crypto'
  2. import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from 'node:fs'
  3. import { tmpdir } from 'node:os'
  4. import { dirname, join } from 'node:path'
  5. import { afterEach, describe, expect, it } from 'vitest'
  6. import type { APNsRequest, APNsSendResult } from '../../src/server/apns.js'
  7. import type { FCMRequest } from '../../src/server/fcm.js'
  8. import { loadServerConfig, type APNsProviderConfig } from '../../src/server/config.js'
  9. import {
  10. PushCoordinator,
  11. type FCMSender,
  12. type PushSender,
  13. } from '../../src/server/pushCoordinator.js'
  14. const roots: string[] = []
  15. afterEach(() => {
  16. for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true })
  17. })
  18. function root(): string {
  19. const value = mkdtempSync(join(tmpdir(), 'yaoyao-push-'))
  20. roots.push(value)
  21. return value
  22. }
  23. const apns: APNsProviderConfig = {
  24. keyFile: '/private/not-read-when-provider-is-injected.p8',
  25. keyId: 'KEY1234567',
  26. teamId: 'TEAM123456',
  27. topic: 'cn.samien.yaoyao.hermes',
  28. }
  29. class FakeSender implements PushSender {
  30. readonly requests: APNsRequest[] = []
  31. readonly results: APNsSendResult[] = []
  32. closed = false
  33. async send(request: APNsRequest): Promise<APNsSendResult> {
  34. this.requests.push(request)
  35. return this.results.shift() ?? { disposition: 'success', status: 200 }
  36. }
  37. close(): void { this.closed = true }
  38. }
  39. function register(coordinator: PushCoordinator, suffix = '1') {
  40. return coordinator.registerInstallation({
  41. userId: 'user-a',
  42. installationId: `phone-${suffix}`,
  43. clientAccountId: 'account-a',
  44. deviceToken: 'ab'.repeat(16 + Number(suffix)),
  45. environment: 'development',
  46. appVersion: '1.2.3',
  47. })
  48. }
  49. describe('push configuration', () => {
  50. it('keeps APNs optional and reports partial configuration without preventing startup', () => {
  51. const home = root()
  52. expect(loadServerConfig({ HERMES_YAOYAO_HOME: home }).apns).toBeUndefined()
  53. const partial = loadServerConfig({ HERMES_YAOYAO_HOME: home, HERMES_YAOYAO_APNS_KEY_ID: 'KEY1234567' })
  54. expect(partial.apns).toBeUndefined()
  55. expect(partial.apnsConfigurationError).toMatch(/absolute local path/)
  56. })
  57. it('loads a complete provider configuration and the default topic', () => {
  58. const directory = root()
  59. const keyFile = join(directory, 'AuthKey.p8')
  60. const { privateKey } = generateKeyPairSync('ec', { namedCurve: 'prime256v1' })
  61. writeFileSync(keyFile, privateKey.export({ format: 'pem', type: 'pkcs8' }))
  62. const config = loadServerConfig({
  63. HERMES_YAOYAO_HOME: directory,
  64. HERMES_YAOYAO_APNS_KEY_FILE: keyFile,
  65. HERMES_YAOYAO_APNS_KEY_ID: 'KEY1234567',
  66. HERMES_YAOYAO_APNS_TEAM_ID: 'TEAM123456',
  67. })
  68. expect(config.apns).toEqual({
  69. keyFile: realpathSync(keyFile),
  70. keyId: 'KEY1234567',
  71. teamId: 'TEAM123456',
  72. topic: 'cn.samien.yaoyao.hermes',
  73. environments: ['development', 'production'],
  74. })
  75. })
  76. it('disables an unreadable provider key as a capability error', () => {
  77. const keyFile = join(root(), 'AuthKey.p8')
  78. writeFileSync(keyFile, 'not a private key')
  79. const config = loadServerConfig({
  80. HERMES_YAOYAO_HOME: dirname(keyFile),
  81. HERMES_YAOYAO_APNS_KEY_FILE: keyFile,
  82. HERMES_YAOYAO_APNS_KEY_ID: 'KEY1234567',
  83. HERMES_YAOYAO_APNS_TEAM_ID: 'TEAM123456',
  84. })
  85. expect(config.apns).toBeUndefined()
  86. expect(config.apnsConfigurationError).toMatch(/ES256 private key/)
  87. })
  88. })
  89. describe('PushCoordinator durable state and delivery', () => {
  90. it('atomically migrates schema 1 APNs state to schema 2 without changing durable recovery data', () => {
  91. const home = root()
  92. const directory = join(home, 'push')
  93. mkdirSync(directory)
  94. const statePath = join(directory, 'state.json')
  95. writeFileSync(statePath, JSON.stringify({
  96. schemaVersion: 1,
  97. installations: [{
  98. userId: 'user-a', installationId: 'phone-1', clientAccountId: 'account-a',
  99. deviceToken: 'ab'.repeat(32), environment: 'development', badge: 2, updatedAt: '2026-01-01T00:00:00.000Z',
  100. }],
  101. groupSubscriptions: [{
  102. userId: 'user-a', roomId: 'room-1', enabled: true, updatedAt: '2026-01-01T00:00:00.000Z',
  103. }],
  104. outbox: [{
  105. id: 'outbox-1', eventId: 'pending-event', userId: 'user-a', installationId: 'phone-1',
  106. clientAccountId: 'account-a', kind: 'chat.completed', title: '完成', body: '正文',
  107. data: { sessionId: 'session-1' }, createdAt: 1, expiresAt: 4_000_000_000_000,
  108. attempts: 2, nextAttemptAt: 2_000_000_000_000,
  109. }],
  110. processedEvents: { existing: 123 },
  111. groupWatch: { epoch: 'epoch-1', cursor: 7 },
  112. chatJobs: [],
  113. ambiguousChatSessions: { ambiguous: 456 },
  114. status: { lastSuccessAt: '2026-01-01T00:00:00.000Z' },
  115. }))
  116. const coordinator = new PushCoordinator({ home, autoFlush: false })
  117. const migrated = JSON.parse(readFileSync(statePath, 'utf8')) as Record<string, any>
  118. expect(migrated).toMatchObject({
  119. schemaVersion: 2,
  120. installations: [{ platform: 'ios', environment: 'development' }],
  121. outbox: [{ platform: 'ios', id: 'outbox-1', attempts: 2 }],
  122. processedEvents: { existing: 123 },
  123. groupWatch: { epoch: 'epoch-1', cursor: 7 },
  124. ambiguousChatSessions: { ambiguous: 456 },
  125. providerStatus: { ios: { lastSuccessAt: '2026-01-01T00:00:00.000Z' }, android: {} },
  126. })
  127. expect(migrated).not.toHaveProperty('status')
  128. coordinator.close()
  129. })
  130. it('isolates an unexpected APNs sender exception from the FCM queue and redacts its target', async () => {
  131. const deviceToken = 'ab'.repeat(32)
  132. const fcmRequests: FCMRequest[] = []
  133. const coordinator = new PushCoordinator({
  134. home: root(),
  135. apns,
  136. provider: { send: async () => { throw new Error(`transport leaked ${deviceToken}`) } },
  137. fcm: {
  138. serviceAccountFile: '/private/not-read-when-provider-is-injected.json',
  139. projectId: 'yaoyao-test-project',
  140. packageName: 'cn.samien.yaoyao.hermes',
  141. },
  142. fcmProvider: {
  143. send: async request => {
  144. fcmRequests.push(request)
  145. return { disposition: 'success', status: 200 }
  146. },
  147. },
  148. autoFlush: false,
  149. })
  150. coordinator.registerInstallation({
  151. userId: 'user-a', installationId: 'phone-ios', clientAccountId: 'account-a',
  152. deviceToken, environment: 'development',
  153. })
  154. coordinator.registerInstallation({
  155. platform: 'android', userId: 'user-a', installationId: 'phone-android', clientAccountId: 'account-a',
  156. fid: 'fcm-registration-id-1234567890',
  157. })
  158. expect(coordinator.enqueue({
  159. eventId: 'dual-provider-event', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '正文',
  160. })).toBe(2)
  161. await expect(coordinator.flushDue()).resolves.toMatchObject({ delivered: 1, retried: 1 })
  162. expect(fcmRequests).toHaveLength(1)
  163. expect(coordinator.status()).toMatchObject({ pendingCount: 1, healthy: false })
  164. expect(coordinator.status().lastError).not.toContain(deviceToken)
  165. expect(coordinator.fcmStatus()).toMatchObject({ pendingCount: 0, healthy: true })
  166. coordinator.close()
  167. })
  168. it('keeps Android FIDs private and delivers high-priority FCM data independently of APNs', async () => {
  169. const requests: FCMRequest[] = []
  170. const sender: FCMSender = {
  171. send: async request => {
  172. requests.push(request)
  173. return { disposition: 'success', status: 200 }
  174. },
  175. }
  176. const coordinator = new PushCoordinator({
  177. home: root(),
  178. fcm: {
  179. serviceAccountFile: '/private/not-read-when-provider-is-injected.json',
  180. projectId: 'yaoyao-test-project',
  181. packageName: 'cn.samien.yaoyao.hermes',
  182. },
  183. fcmProvider: sender,
  184. autoFlush: false,
  185. })
  186. const fid = 'fcm-registration-id-1234567890'
  187. expect(coordinator.registerInstallation({
  188. platform: 'android', userId: 'user-a', installationId: 'phone-android', clientAccountId: 'account-a',
  189. fid, appVersion: '1.0.0',
  190. })).toEqual(expect.objectContaining({ platform: 'android', installationId: 'phone-android' }))
  191. expect(coordinator.registerInstallation({
  192. platform: 'android', userId: 'user-a', installationId: 'phone-android', clientAccountId: 'account-a', fid,
  193. })).not.toHaveProperty('fid')
  194. expect(coordinator.capabilities()).toMatchObject({
  195. enabled: false,
  196. platforms: { ios: { enabled: false }, android: { enabled: true, provider: 'fcm' } },
  197. })
  198. expect(coordinator.isAnyProviderEnabled()).toBe(true)
  199. expect(coordinator.enqueue({
  200. eventId: 'android-event', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '正文',
  201. collapseId: 'chat:1', data: { sessionId: 'session-1' },
  202. })).toBe(1)
  203. await expect(coordinator.flushDue()).resolves.toMatchObject({ delivered: 1 })
  204. expect(requests).toEqual([expect.objectContaining({
  205. fid,
  206. priority: 'high',
  207. data: expect.objectContaining({
  208. eventId: 'android-event', kind: 'chat.completed', title: '完成', body: '正文',
  209. collapseId: 'chat:1', sessionId: 'session-1',
  210. }),
  211. })])
  212. expect(coordinator.status()).toMatchObject({ configured: false, registrationCount: 0, pendingCount: 0 })
  213. expect(coordinator.fcmStatus()).toMatchObject({ configured: true, healthy: true, registrationCount: 1, pendingCount: 0 })
  214. coordinator.close()
  215. })
  216. it('hot-swaps during an active flush without closing the in-flight provider early', async () => {
  217. const home = root()
  218. let releaseSend!: (value: APNsSendResult) => void
  219. const oldSender: PushSender & { closed: boolean } = {
  220. closed: false,
  221. send: () => new Promise(resolve => { releaseSend = resolve }),
  222. close() { this.closed = true },
  223. }
  224. const replacement = new FakeSender()
  225. const coordinator = new PushCoordinator({
  226. home, apns, provider: oldSender, autoFlush: false,
  227. providerFactory: () => replacement,
  228. })
  229. register(coordinator)
  230. coordinator.enqueue({
  231. eventId: 'active-flush', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '已完成',
  232. })
  233. const flush = coordinator.flushDue()
  234. const replacementOperation = coordinator.configureAPNs({ ...apns, keyId: 'NEWKEY1234' })
  235. await replacementOperation
  236. expect(oldSender.closed).toBe(false)
  237. expect(coordinator.capabilities().enabled).toBe(true)
  238. releaseSend({ disposition: 'unregister', status: 400, reason: 'BadDeviceToken' })
  239. await flush
  240. await Promise.resolve()
  241. expect(oldSender.closed).toBe(true)
  242. expect(coordinator.status()).toMatchObject({ registrationCount: 1, pendingCount: 1 })
  243. await coordinator.flushDue()
  244. expect(replacement.requests).toHaveLength(1)
  245. expect(coordinator.status().pendingCount).toBe(0)
  246. })
  247. it('hot-enables providers and only queues installations from selected environments', async () => {
  248. const home = root()
  249. const senders: FakeSender[] = []
  250. const coordinator = new PushCoordinator({
  251. home,
  252. autoFlush: false,
  253. providerFactory: () => {
  254. const sender = new FakeSender()
  255. senders.push(sender)
  256. return sender
  257. },
  258. })
  259. const changes: boolean[] = []
  260. coordinator.onEnabledChange(enabled => { changes.push(enabled) })
  261. coordinator.registerInstallation({
  262. userId: 'user-a', installationId: 'phone-development', clientAccountId: 'account-a',
  263. deviceToken: 'ab'.repeat(32), environment: 'development',
  264. })
  265. coordinator.registerInstallation({
  266. userId: 'user-a', installationId: 'phone-production', clientAccountId: 'account-a',
  267. deviceToken: 'cd'.repeat(32), environment: 'production',
  268. })
  269. await coordinator.configureAPNs({ ...apns, environments: ['production'] })
  270. expect(coordinator.capabilities()).toMatchObject({ enabled: true, environments: ['production'] })
  271. expect(coordinator.enqueue({
  272. eventId: 'environment-event', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '已完成',
  273. })).toBe(1)
  274. await coordinator.flushDue()
  275. expect(senders[0]!.requests).toHaveLength(1)
  276. expect(senders[0]!.requests[0]!.environment).toBe('production')
  277. await coordinator.configureAPNs()
  278. expect(senders[0]!.closed).toBe(true)
  279. expect(coordinator.capabilities().enabled).toBe(false)
  280. expect(changes).toEqual([true, false])
  281. })
  282. it('persists private registrations, subscriptions, outbox delivery, and badge state', async () => {
  283. const home = root()
  284. const sender = new FakeSender()
  285. const coordinator = new PushCoordinator({ home, apns, provider: sender, autoFlush: false })
  286. expect(coordinator.capabilities()).toMatchObject({
  287. protocolVersion: 1, enabled: true, topic: apns.topic, maxSummaryCharacters: 180,
  288. })
  289. expect(register(coordinator)).not.toHaveProperty('deviceToken')
  290. expect(coordinator.setGroupSubscription('user-a', 'room-1', true)).toMatchObject({ roomId: 'room-1', enabled: true })
  291. expect(coordinator.isGroupSubscribed('user-a', 'room-1')).toBe(true)
  292. expect(coordinator.subscribedUserIds('room-1')).toEqual(['user-a'])
  293. expect(coordinator.enqueue({
  294. eventId: 'event-1',
  295. userId: 'user-a',
  296. kind: 'group.message.completed',
  297. title: '任务完成',
  298. body: '好'.repeat(250),
  299. roomId: 'room-1',
  300. threadId: 'topic-1',
  301. collapseId: 'message-1',
  302. data: { roomId: 'room-1', aps: 'untrusted', version: 99 },
  303. })).toBe(1)
  304. expect(coordinator.enqueue({
  305. eventId: 'event-1', userId: 'user-a', kind: 'group.message.completed', title: '重复', body: '重复',
  306. })).toBe(0)
  307. expect(coordinator.status()).toMatchObject({ registrationCount: 1, pendingCount: 1, topic: apns.topic })
  308. await expect(coordinator.flushDue()).resolves.toMatchObject({ delivered: 1 })
  309. expect(sender.requests).toHaveLength(1)
  310. expect(sender.requests[0]!.payload).toMatchObject({
  311. version: 1,
  312. eventId: 'event-1',
  313. kind: 'group.message.completed',
  314. clientAccountId: 'account-a',
  315. aps: { alert: { title: '任务完成' }, badge: 1, 'thread-id': 'topic-1' },
  316. })
  317. expect(Array.from(((sender.requests[0]!.payload.aps as { alert: { body: string } }).alert.body))).toHaveLength(180)
  318. expect(sender.requests[0]!.payload).not.toHaveProperty('userId')
  319. expect(coordinator.status()).toMatchObject({ healthy: true, pendingCount: 0 })
  320. const statePath = join(home, 'push', 'state.json')
  321. expect(statSync(statePath).mode & 0o777).toBe(0o600)
  322. const stored = readFileSync(statePath, 'utf8')
  323. expect(stored).toContain('ab'.repeat(17))
  324. expect(stored).not.toContain(apns.keyFile)
  325. coordinator.close()
  326. const restored = new PushCoordinator({ home, apns, provider: new FakeSender(), autoFlush: false })
  327. expect(restored.status()).toMatchObject({ registrationCount: 1, subscriptionCount: 1 })
  328. expect(restored.listGroupSubscriptions('user-a')).toEqual([
  329. expect.objectContaining({ roomId: 'room-1', enabled: true }),
  330. ])
  331. restored.close()
  332. })
  333. it('retries transient failures exponentially and removes invalid registrations', async () => {
  334. const home = root()
  335. let now = 1_000_000
  336. const sender = new FakeSender()
  337. sender.results.push(
  338. { disposition: 'retry', status: 503, reason: 'Shutdown' },
  339. { disposition: 'unregister', status: 410, reason: 'Unregistered' },
  340. )
  341. const coordinator = new PushCoordinator({
  342. home, apns, provider: sender, autoFlush: false, now: () => now, baseRetryMilliseconds: 100,
  343. })
  344. register(coordinator)
  345. coordinator.enqueue({ eventId: 'retry-1', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '正文' })
  346. await expect(coordinator.flushDue()).resolves.toMatchObject({ retried: 1 })
  347. expect(coordinator.status()).toMatchObject({ pendingCount: 1, healthy: false })
  348. await expect(coordinator.flushDue()).resolves.toMatchObject({ delivered: 0, retried: 0 })
  349. expect(sender.requests).toHaveLength(1)
  350. now += 100
  351. await expect(coordinator.flushDue()).resolves.toMatchObject({ removedRegistrations: 1 })
  352. expect(coordinator.status()).toMatchObject({ registrationCount: 0, pendingCount: 0 })
  353. coordinator.close()
  354. })
  355. it('does not delete a token that rotated while an APNs request was in flight', async () => {
  356. let release!: (result: APNsSendResult) => void
  357. let started!: () => void
  358. const began = new Promise<void>(resolve => { started = resolve })
  359. const first = new Promise<APNsSendResult>(resolve => { release = resolve })
  360. const requests: APNsRequest[] = []
  361. const sender: PushSender = {
  362. send: async request => {
  363. requests.push(request)
  364. if (requests.length === 1) {
  365. started()
  366. return first
  367. }
  368. return { disposition: 'success', status: 200 }
  369. },
  370. }
  371. const coordinator = new PushCoordinator({ home: root(), apns, provider: sender, autoFlush: false })
  372. register(coordinator)
  373. coordinator.enqueue({ eventId: 'rotated-1', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '正文' })
  374. const flushing = coordinator.flushDue()
  375. await began
  376. coordinator.registerInstallation({
  377. userId: 'user-a', installationId: 'phone-1', clientAccountId: 'account-a',
  378. deviceToken: 'cd'.repeat(32), environment: 'development', appVersion: '1.2.4',
  379. })
  380. release({ disposition: 'unregister', status: 410, reason: 'Unregistered' })
  381. await expect(flushing).resolves.toMatchObject({ retried: 1, removedRegistrations: 0 })
  382. expect(coordinator.status()).toMatchObject({ registrationCount: 1, pendingCount: 1 })
  383. await expect(coordinator.flushDue()).resolves.toMatchObject({ delivered: 1 })
  384. expect(requests[0]!.deviceToken).not.toBe(requests[1]!.deviceToken)
  385. expect(requests[1]!.deviceToken).toBe('cd'.repeat(32))
  386. coordinator.close()
  387. })
  388. it('fails closed when a user is disabled or their authentication version changes', async () => {
  389. let active = true
  390. let authVersion = 1
  391. const sender = new FakeSender()
  392. const coordinator = new PushCoordinator({
  393. home: root(), apns, provider: sender, autoFlush: false,
  394. isUserActive: () => active,
  395. userAuthorizationVersion: () => authVersion,
  396. })
  397. coordinator.registerInstallation({
  398. userId: 'user-a', installationId: 'phone-a', clientAccountId: 'account-a',
  399. deviceToken: 'ab'.repeat(32), environment: 'development', authorizationVersion: 1,
  400. })
  401. coordinator.enqueue({ eventId: 'version-1', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '正文' })
  402. authVersion = 2
  403. await expect(coordinator.flushDue()).resolves.toMatchObject({ failed: 1, delivered: 0 })
  404. expect(sender.requests).toHaveLength(0)
  405. expect(coordinator.status()).toMatchObject({ registrationCount: 0, pendingCount: 0 })
  406. coordinator.registerInstallation({
  407. userId: 'user-a', installationId: 'phone-a', clientAccountId: 'account-a',
  408. deviceToken: 'cd'.repeat(32), environment: 'development', authorizationVersion: 2,
  409. })
  410. active = false
  411. expect(coordinator.enqueue({
  412. eventId: 'disabled', userId: 'user-a', kind: 'chat.completed', title: '不应发送', body: '正文',
  413. })).toBe(0)
  414. coordinator.close()
  415. })
  416. it('cancels pending team notifications on unsubscribe and removes all user-owned state', () => {
  417. const home = root()
  418. const coordinator = new PushCoordinator({ home, apns, provider: new FakeSender(), autoFlush: false })
  419. register(coordinator)
  420. coordinator.setGroupSubscription('user-a', 'room-1', true)
  421. expect(coordinator.enqueueGroupEvent({
  422. eventID: 'group-1', localUserID: 'user-a', kind: 'group.message.completed',
  423. title: '团队消息', body: '正文', collapseID: 'group-1', data: {}, roomID: 'room-1',
  424. })).toBe('enqueued')
  425. coordinator.setGroupSubscription('user-a', 'room-1', false)
  426. expect(coordinator.status().pendingCount).toBe(0)
  427. coordinator.setGroupSubscription('user-a', 'room-2', true)
  428. const removed = coordinator.removeUser('user-a')
  429. expect(removed).toMatchObject({ registrations: 1, subscriptions: 1 })
  430. expect(coordinator.status()).toMatchObject({ registrationCount: 0, subscriptionCount: 0 })
  431. expect(coordinator.listGroupSubscriptions('user-a')).toEqual([])
  432. coordinator.close()
  433. })
  434. it('persists the global group cursor and resumable chat jobs in the same state file', () => {
  435. const home = root()
  436. const coordinator = new PushCoordinator({ home, apns, provider: new FakeSender(), autoFlush: false })
  437. expect(coordinator.resetGroupCursor({ epoch: 'epoch-1', cursor: 10, reason: 'initial' })).toMatchObject({
  438. epoch: 'epoch-1', cursor: 10,
  439. })
  440. coordinator.advanceGroupCursor({ epoch: 'epoch-1', cursor: 11 })
  441. expect(() => coordinator.advanceGroupCursor({ epoch: 'epoch-1', cursor: 9 })).toThrow(/backwards/)
  442. coordinator.saveChatJob({
  443. id: 'job-1', localUserID: 'user-a', profile: 'default', runtimeSessionID: 'runtime-1',
  444. storedSessionID: 'stored-1', requestID: 'request-1', queued: false, phase: 'watching',
  445. submittedAt: 1, expiresAt: Date.now() + 60_000,
  446. })
  447. coordinator.close()
  448. const restored = new PushCoordinator({ home, apns, provider: new FakeSender(), autoFlush: false })
  449. expect(restored.groupWatchAnchor()).toMatchObject({ epoch: 'epoch-1', cursor: 11 })
  450. expect(restored.pendingChatJobs()).toEqual([expect.objectContaining({ id: 'job-1', localUserID: 'user-a' })])
  451. restored.completeChatJob('job-1')
  452. expect(restored.pendingChatJobs()).toEqual([])
  453. restored.close()
  454. })
  455. it('fails closed when more than one local user has written the same stored session', () => {
  456. const coordinator = new PushCoordinator({ home: root(), apns, provider: new FakeSender(), autoFlush: false })
  457. const job = (id: string, localUserID: string) => ({
  458. id, localUserID, profile: 'default', runtimeSessionID: `runtime-${id}`, storedSessionID: 'shared',
  459. requestID: `request-${id}`, queued: false, phase: 'accepted' as const,
  460. submittedAt: 1, expiresAt: Date.now() + 60_000,
  461. })
  462. const first = job('a', 'user-a')
  463. coordinator.saveChatJob(first)
  464. expect(coordinator.chatJobRecoveryAllowed(first)).toBe(true)
  465. const second = job('b', 'user-b')
  466. coordinator.saveChatJob(second)
  467. expect(coordinator.chatJobRecoveryAllowed(first)).toBe(false)
  468. expect(coordinator.chatJobRecoveryAllowed(second)).toBe(false)
  469. coordinator.completeChatJob('b')
  470. expect(coordinator.chatJobRecoveryAllowed(first)).toBe(false)
  471. coordinator.close()
  472. })
  473. it('keeps prompt correlation private and stable across restart', () => {
  474. const home = root()
  475. const first = new PushCoordinator({ home, apns, provider: new FakeSender(), autoFlush: false })
  476. const digest = first.promptDigest('user-a', '继续执行')
  477. expect(digest).not.toContain('继续执行')
  478. expect(first.promptDigest('user-b', '继续执行')).not.toBe(digest)
  479. first.close()
  480. const restored = new PushCoordinator({ home, apns, provider: new FakeSender(), autoFlush: false })
  481. expect(restored.promptDigest('user-a', '继续执行')).toBe(digest)
  482. expect(statSync(join(home, 'push', 'correlation.key')).mode & 0o777).toBe(0o600)
  483. restored.close()
  484. })
  485. it('automatically flushes newly queued work and closes the injected provider', async () => {
  486. const sender = new FakeSender()
  487. const coordinator = new PushCoordinator({ home: root(), apns, provider: sender })
  488. register(coordinator)
  489. coordinator.enqueue({ eventId: 'auto-1', userId: 'user-a', kind: 'chat.completed', title: '完成', body: '正文' })
  490. await expect.poll(() => sender.requests.length).toBe(1)
  491. expect(coordinator.status().pendingCount).toBe(0)
  492. coordinator.close()
  493. expect(sender.closed).toBe(true)
  494. })
  495. it('protects and preserves a malformed existing state file instead of overwriting it', () => {
  496. const home = root()
  497. const directory = join(home, 'push')
  498. const statePath = join(directory, 'state.json')
  499. mkdirSync(directory)
  500. writeFileSync(statePath, '{malformed', { mode: 0o644, flag: 'w' })
  501. const coordinator = new PushCoordinator({ home, apns, provider: new FakeSender(), autoFlush: false })
  502. expect(coordinator.capabilities()).toMatchObject({ enabled: false })
  503. expect(coordinator.status().lastError).toMatch(/避免覆盖/)
  504. expect(() => register(coordinator)).toThrow(/推送状态文件无法读取/)
  505. expect(readFileSync(statePath, 'utf8')).toBe('{malformed')
  506. expect(statSync(statePath).mode & 0o777).toBe(0o600)
  507. coordinator.close()
  508. })
  509. })