From 7add62ae2b01e061c8c8335471db0ef11cef15d8 Mon Sep 17 00:00:00 2001 From: Artyom Savchenko Date: Sun, 22 Mar 2026 03:23:41 +0700 Subject: [PATCH] Improve error handling in notification service (#10662) Signed-off-by: Artem Savchenko --- common/config/rush/pnpm-lock.yaml | 11 +- .../pod-notification/package.json | 3 + .../src/__tests__/push.test.ts | 158 ++++++++++++++++++ .../src/__tests__/server.test.ts | 129 ++++++++++++++ .../pod-notification/src/index.ts | 39 ++++- .../notification/pod-notification/src/main.ts | 58 +++---- .../notification/pod-notification/src/push.ts | 64 +++++++ .../pod-notification/src/server.ts | 10 +- 8 files changed, 426 insertions(+), 46 deletions(-) create mode 100644 services/notification/pod-notification/src/__tests__/push.test.ts create mode 100644 services/notification/pod-notification/src/__tests__/server.test.ts create mode 100644 services/notification/pod-notification/src/push.ts diff --git a/common/config/rush/pnpm-lock.yaml b/common/config/rush/pnpm-lock.yaml index b5ae38fa90..f2566db58f 100644 --- a/common/config/rush/pnpm-lock.yaml +++ b/common/config/rush/pnpm-lock.yaml @@ -40178,6 +40178,12 @@ importers: ../../services/notification/pod-notification: dependencies: + '@hcengineering/analytics': + specifier: workspace:^0.7.17 + version: link:../../../foundations/core/packages/analytics + '@hcengineering/analytics-service': + specifier: workspace:^0.7.17 + version: link:../../../foundations/core/packages/analytics-service '@hcengineering/client': specifier: workspace:^0.7.18 version: link:../../../foundations/core/packages/client @@ -40193,6 +40199,9 @@ importers: '@hcengineering/platform': specifier: workspace:^0.7.19 version: link:../../../foundations/core/packages/platform + '@hcengineering/server-core': + specifier: workspace:^0.7.18 + version: link:../../../foundations/server/packages/core '@hcengineering/server-token': specifier: workspace:^0.7.17 version: link:../../../foundations/core/packages/token @@ -61548,7 +61557,7 @@ snapshots: node-loader@2.0.0(webpack@5.102.1): dependencies: loader-utils: 2.0.4 - webpack: 5.102.1(esbuild@0.25.12)(webpack-cli@5.1.4) + webpack: 5.102.1 node-localstorage@2.2.1: dependencies: diff --git a/services/notification/pod-notification/package.json b/services/notification/pod-notification/package.json index b7c1f001ed..55827d9637 100644 --- a/services/notification/pod-notification/package.json +++ b/services/notification/pod-notification/package.json @@ -54,11 +54,14 @@ "@types/web-push": "^3.6.4" }, "dependencies": { + "@hcengineering/analytics": "workspace:^0.7.17", + "@hcengineering/analytics-service": "workspace:^0.7.17", "@hcengineering/client": "workspace:^0.7.18", "@hcengineering/client-resources": "workspace:^0.7.18", "@hcengineering/core": "workspace:^0.7.24", "@hcengineering/notification": "workspace:^0.7.0", "@hcengineering/platform": "workspace:^0.7.19", + "@hcengineering/server-core": "workspace:^0.7.18", "@hcengineering/server-token": "workspace:^0.7.17", "cors": "^2.8.5", "dotenv": "^16.4.5", diff --git a/services/notification/pod-notification/src/__tests__/push.test.ts b/services/notification/pod-notification/src/__tests__/push.test.ts new file mode 100644 index 0000000000..364176cc77 --- /dev/null +++ b/services/notification/pod-notification/src/__tests__/push.test.ts @@ -0,0 +1,158 @@ +// +// Copyright © 2026 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import type { MeasureContext, Ref } from '@hcengineering/core' +import type { PushData, PushSubscription } from '@hcengineering/notification' +import webpush, { WebPushError } from 'web-push' +import { isDisposableSubscriptionError, sendPushToSubscription, webPushErrorBodyString } from '../push' + +jest.mock('web-push', () => { + const actual = jest.requireActual('web-push') + return { + __esModule: true, + WebPushError: actual.WebPushError, + default: { + ...(actual ?? {}), + sendNotification: jest.fn() + } + } +}) + +const sendNotificationMock = webpush.sendNotification as jest.MockedFunction + +function mkWebPushError (body: string, statusCode: number = 410): WebPushError { + return new WebPushError('push failed', statusCode, {}, body, 'https://push.example/ep') +} + +function createMockMeasureContext (): MeasureContext { + return { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + newChild: jest.fn(), + with: jest.fn(), + withSync: jest.fn(), + extractMeta: jest.fn(() => ({})), + contextData: {}, + getParams: jest.fn(() => ({})), + measure: jest.fn(), + end: jest.fn() + } as unknown as MeasureContext +} + +function mkSubscription (id: string): PushSubscription { + const sub: PushSubscription = { + _id: id as Ref, + endpoint: 'https://push.example/ep', + keys: { p256dh: 'p256', auth: 'auth' }, + user: 'user-1' as never, + space: 'space-1' as never, + modifiedOn: 0, + modifiedBy: 'user-1' as never + } as any + return sub +} + +const sampleData: PushData = { title: 't', body: 'b' } + +describe('webPushErrorBodyString', () => { + it('returns string body as-is', () => { + const err = mkWebPushError('subscription expired') + expect(webPushErrorBodyString(err)).toBe('subscription expired') + }) +}) + +describe('isDisposableSubscriptionError', () => { + it.each([ + ['expired', 'subscription expired'], + ['Unregistered', 'push subscription Unregistered'], + ['No such subscription', 'No such subscription'] + ])('matches disposable pattern %s', (_name, body) => { + expect(isDisposableSubscriptionError(mkWebPushError(body))).toBe(true) + }) + + it('returns false for other push errors', () => { + expect(isDisposableSubscriptionError(mkWebPushError('Rate limit exceeded', 429))).toBe(false) + }) +}) + +describe('sendPushToSubscription', () => { + beforeEach(() => { + sendNotificationMock.mockReset() + }) + + it('returns empty when all sends succeed', async () => { + sendNotificationMock.mockResolvedValue({ statusCode: 201, body: '', headers: {} }) + const ctx = createMockMeasureContext() + const subs = [mkSubscription('s1'), mkSubscription('s2')] + const result = await sendPushToSubscription(ctx, subs, sampleData) + expect(result).toEqual([]) + expect(sendNotificationMock).toHaveBeenCalledTimes(2) + expect(ctx.warn).not.toHaveBeenCalled() + expect(ctx.error).not.toHaveBeenCalled() + }) + + it('collects subscription id for disposable WebPushError', async () => { + sendNotificationMock.mockRejectedValueOnce(mkWebPushError('Unregistered')) + const ctx = createMockMeasureContext() + const subs = [mkSubscription('drop-me')] + const result = await sendPushToSubscription(ctx, subs, sampleData) + expect(result).toEqual(['drop-me']) + expect(ctx.warn).not.toHaveBeenCalled() + expect(ctx.error).not.toHaveBeenCalled() + }) + + it('warns but does not collect id for non-disposable WebPushError', async () => { + const wpe = mkWebPushError('Internal error', 500) + sendNotificationMock.mockRejectedValueOnce(wpe) + const ctx = createMockMeasureContext() + const subs = [mkSubscription('keep-me')] + const result = await sendPushToSubscription(ctx, subs, sampleData) + expect(result).toEqual([]) + expect(ctx.warn).toHaveBeenCalledWith('Web push failed for subscription', { + statusCode: 500, + body: 'Internal error', + subscriptionId: 'keep-me' + }) + expect(ctx.error).not.toHaveBeenCalled() + }) + + it('logs unexpected errors', async () => { + const boom = new TypeError('network') + sendNotificationMock.mockRejectedValueOnce(boom) + const ctx = createMockMeasureContext() + const subs = [mkSubscription('sub-x')] + const result = await sendPushToSubscription(ctx, subs, sampleData) + expect(result).toEqual([]) + expect(ctx.error).toHaveBeenCalledWith('Unexpected error sending web push', { + error: boom, + subscriptionId: 'sub-x' + }) + expect(ctx.warn).not.toHaveBeenCalled() + }) + + it('processes subscriptions independently', async () => { + sendNotificationMock + .mockResolvedValueOnce({ statusCode: 201, body: '', headers: {} }) + .mockRejectedValueOnce(mkWebPushError('expired')) + .mockRejectedValueOnce(new Error('weird')) + const ctx = createMockMeasureContext() + const subs = [mkSubscription('a'), mkSubscription('b'), mkSubscription('c')] + const result = await sendPushToSubscription(ctx, subs, sampleData) + expect(result).toEqual(['b']) + expect(ctx.warn).not.toHaveBeenCalled() + expect(ctx.error).toHaveBeenCalledTimes(1) + }) +}) diff --git a/services/notification/pod-notification/src/__tests__/server.test.ts b/services/notification/pod-notification/src/__tests__/server.test.ts new file mode 100644 index 0000000000..4e28b04081 --- /dev/null +++ b/services/notification/pod-notification/src/__tests__/server.test.ts @@ -0,0 +1,129 @@ +// +// Copyright © 2026 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import http from 'http' +import { createServer } from '../server' +import { ApiError } from '../error' +import type { Endpoint } from '../types' + +function httpRequest ( + url: string, + options: { method?: string, body?: string } = {} +): Promise<{ status: number, json: () => Promise }> { + return new Promise((resolve, reject) => { + const u = new URL(url) + const req = http.request( + { + hostname: u.hostname, + port: u.port, + path: u.pathname + u.search, + method: options.method ?? 'GET', + headers: + options.body !== undefined + ? { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(options.body) } + : undefined + }, + (res) => { + const chunks: Buffer[] = [] + res.on('data', (c) => { + chunks.push(c) + }) + res.on('end', () => { + const text = Buffer.concat(chunks).toString('utf8') + resolve({ + status: res.statusCode ?? 0, + json: async () => JSON.parse(text) as unknown + }) + }) + } + ) + req.on('error', reject) + if (options.body !== undefined) { + req.write(options.body) + } + req.end() + }) +} + +function withServer (endpoints: Endpoint[], fn: (baseUrl: string) => Promise): Promise { + const app = createServer(endpoints) + return new Promise((resolve, reject) => { + const srv = app.listen(0, '127.0.0.1', () => { + const addr = srv.address() + const port = typeof addr === 'object' && addr !== null ? addr.port : 0 + const baseUrl = `http://127.0.0.1:${port}` + void fn(baseUrl) + .then(() => { + srv.close((err) => { + err != null ? reject(err) : resolve() + }) + }) + .catch((e) => { + srv.close(() => { + reject(e) + }) + }) + }) + srv.on('error', reject) + }) +} + +describe('createServer', () => { + it('returns 404 for unknown routes', async () => { + await withServer([], async (baseUrl) => { + const res = await httpRequest(`${baseUrl}/missing`) + expect(res.status).toBe(404) + const body = (await res.json()) as { message: string } + expect(body.message).toBe('Not found') + }) + }) + + it('maps ApiError to 400 with code', async () => { + const endpoints: Endpoint[] = [ + { + endpoint: '/err', + type: 'post', + handler: async (_req, _res) => { + throw new ApiError('INVALID', 'bad input') + } + } + ] + await withServer(endpoints, async (baseUrl) => { + const res = await httpRequest(`${baseUrl}/err`, { method: 'POST', body: '{}' }) + expect(res.status).toBe(400) + const body = (await res.json()) as { code: string, message: string } + expect(body.code).toBe('INVALID') + expect(body.message).toBe('bad input') + }) + }) + + it('maps unexpected errors to 500', async () => { + const endpoints: Endpoint[] = [ + { + endpoint: '/boom', + type: 'post', + handler: async (_req, _res) => { + throw new Error('boom') + } + } + ] + await withServer(endpoints, async (baseUrl) => { + const res = await httpRequest(`${baseUrl}/boom`, { method: 'POST', body: '{}' }) + expect(res.status).toBe(500) + const body = (await res.json()) as { message: string } + expect(body.message).toBe('boom') + }) + }) +}) diff --git a/services/notification/pod-notification/src/index.ts b/services/notification/pod-notification/src/index.ts index c225f89a0f..380371a129 100644 --- a/services/notification/pod-notification/src/index.ts +++ b/services/notification/pod-notification/src/index.ts @@ -1,5 +1,5 @@ // -// Copyright © 2023 Hardcore Engineering Inc. +// Copyright © 2026 Hardcore Engineering Inc. // // Licensed under the Eclipse Public License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. You may @@ -13,10 +13,39 @@ // limitations under the License. // +import { Analytics } from '@hcengineering/analytics' +import { SplitLogger, configureAnalytics, createOpenTelemetryMetricsContext } from '@hcengineering/analytics-service' +import { newMetrics } from '@hcengineering/core' +import { initStatisticsContext } from '@hcengineering/server-core' +import { join } from 'path' import { main } from './main' -void main().catch((err) => { - if (err != null) { - console.error(err) - } +configureAnalytics('notification', process.env.VERSION ?? '0.7.0') +const metricsContext = initStatisticsContext('notification', { + factory: () => + createOpenTelemetryMetricsContext( + 'notification', + {}, + {}, + newMetrics(), + new SplitLogger('notification-service', { + root: join(process.cwd(), 'logs'), + enableConsole: (process.env.ENABLE_CONSOLE ?? 'true') === 'true' + }) + ) +}) + +Analytics.setTag('application', 'notification-service') + +process.on('uncaughtException', (e) => { + metricsContext.error('UncaughtException', { error: e }) +}) + +process.on('unhandledRejection', (reason, promise) => { + metricsContext.error('Unhandled Rejection at:', { promise, reason }) +}) + +void main(metricsContext).catch((err) => { + metricsContext.error('Failed to start', { error: err }) + process.exit(1) }) diff --git a/services/notification/pod-notification/src/main.ts b/services/notification/pod-notification/src/main.ts index 1d1324909e..7db735c486 100644 --- a/services/notification/pod-notification/src/main.ts +++ b/services/notification/pod-notification/src/main.ts @@ -1,5 +1,5 @@ // -// Copyright © 2023 Hardcore Engineering Inc. +// Copyright © 2026 Hardcore Engineering Inc. // // Licensed under the Eclipse Public License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. You may @@ -13,47 +13,34 @@ // limitations under the License. // -import type { Ref } from '@hcengineering/core' +import type { MeasureContext } from '@hcengineering/core' import { PushSubscription, type PushData } from '@hcengineering/notification' import type { Request, Response } from 'express' -import webpush, { WebPushError } from 'web-push' +import webpush from 'web-push' import config from './config' +import { sendPushToSubscription } from './push' import { createServer, listen } from './server' import { Endpoint } from './types' -const errorMessages = ['expired', 'Unregistered', 'No such subscription'] -async function sendPushToSubscription ( - subscriptions: PushSubscription[], - data: PushData -): Promise[]> { - const result: Ref[] = [] - for (const subscription of subscriptions) { - try { - await webpush.sendNotification(subscription, JSON.stringify(data)) - } catch (err: any) { - if (err instanceof WebPushError) { - if (errorMessages.some((p) => JSON.stringify(err.body).includes(p))) { - result.push(subscription._id) - } - } - } - } - return result -} - -export const main = async (): Promise => { - console.log('Notification service has been started') +export const main = async (ctx: MeasureContext): Promise => { + ctx.info('Notification service starting') let webpushInitDone = false if (config.PushPublicKey !== undefined && config.PushPrivateKey !== undefined) { try { const subj = config.PushSubject ?? 'mailto:hey@huly.io' - console.log('Setting VAPID details', subj, config.PushPublicKey.length, config.PushPrivateKey.length) - webpush.setVapidDetails(config.PushSubject ?? 'mailto:hey@huly.io', config.PushPublicKey, config.PushPrivateKey) + ctx.info('Setting VAPID details', { + subject: subj, + publicKeyLen: config.PushPublicKey.length, + privateKeyLen: config.PushPrivateKey.length + }) + webpush.setVapidDetails(subj, config.PushPublicKey, config.PushPrivateKey) webpushInitDone = true - } catch (err: any) { - console.error(err) + } catch (err: unknown) { + ctx.error('Failed to set VAPID details', { error: err }) } + } else { + ctx.warn('VAPID keys not configured; /web-push will return empty results until keys are set') } const checkAuth = (req: Request, res: Response): boolean => { @@ -92,15 +79,18 @@ export const main = async (): Promise => { return } - const result = await sendPushToSubscription(subscriptions, data) + const result = await sendPushToSubscription(ctx, subscriptions, data) res.json({ result }).end() } } ] - const server = listen(createServer(endpoints), config.Port) + const server = listen(createServer(endpoints), config.Port, undefined, () => { + ctx.info('Notification service listening', { port: config.Port }) + }) const shutdown = (): void => { + ctx.info('Closed') server.close(() => { process.exit() }) @@ -108,10 +98,4 @@ export const main = async (): Promise => { process.on('SIGINT', shutdown) process.on('SIGTERM', shutdown) - process.on('uncaughtException', (e) => { - console.error(e) - }) - process.on('unhandledRejection', (e) => { - console.error(e) - }) } diff --git a/services/notification/pod-notification/src/push.ts b/services/notification/pod-notification/src/push.ts new file mode 100644 index 0000000000..610cfcbb5f --- /dev/null +++ b/services/notification/pod-notification/src/push.ts @@ -0,0 +1,64 @@ +// +// Copyright © 2026 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import type { MeasureContext, Ref } from '@hcengineering/core' +import { type PushData, type PushSubscription } from '@hcengineering/notification' +import webpush, { WebPushError } from 'web-push' + +/** Push endpoints return these when the subscription should be removed — not actionable server errors. */ +const disposableSubscriptionPatterns = ['expired', 'Unregistered', 'No such subscription'] + +export function webPushErrorBodyString (err: WebPushError): string { + const b = err.body + if (typeof b === 'string') return b + try { + return JSON.stringify(b) + } catch { + return String(b) + } +} + +export function isDisposableSubscriptionError (err: WebPushError): boolean { + const body = webPushErrorBodyString(err) + return disposableSubscriptionPatterns.some((p) => body.includes(p)) +} + +export async function sendPushToSubscription ( + ctx: MeasureContext, + subscriptions: PushSubscription[], + data: PushData +): Promise[]> { + const result: Ref[] = [] + for (const subscription of subscriptions) { + try { + await webpush.sendNotification(subscription, JSON.stringify(data)) + } catch (err: unknown) { + if (err instanceof WebPushError) { + if (isDisposableSubscriptionError(err)) { + result.push(subscription._id) + } else { + ctx.warn('Web push failed for subscription', { + statusCode: err.statusCode, + body: err.body, + subscriptionId: subscription._id + }) + } + } else { + ctx.error('Unexpected error sending web push', { error: err, subscriptionId: subscription._id }) + } + } + } + return result +} diff --git a/services/notification/pod-notification/src/server.ts b/services/notification/pod-notification/src/server.ts index e07e7ed029..194772a576 100644 --- a/services/notification/pod-notification/src/server.ts +++ b/services/notification/pod-notification/src/server.ts @@ -1,5 +1,5 @@ // -// Copyright © 2023 Hardcore Engineering Inc. +// Copyright © 2026 Hardcore Engineering Inc. // // Licensed under the Eclipse Public License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. You may @@ -59,9 +59,13 @@ export function createServer (endpoints: Endpoint[]): Express { return app } -export function listen (e: Express, port: number, host?: string): Server { +export function listen (e: Express, port: number, host?: string, onListening?: () => void): Server { const cb = (): void => { - console.log(`Notification service has been started at ${host ?? '*'}:${port}`) + if (onListening !== undefined) { + onListening() + } else { + console.log(`Notification service has been started at ${host ?? '*'}:${port}`) + } } return host !== undefined ? e.listen(port, host, cb) : e.listen(port, cb)