Improve error handling in notification service

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",

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<typeof import('web-push')>('web-push')
+  return {
+    __esModule: true,
+    WebPushError: actual.WebPushError,
+    default: {
+      ...(actual ?? {}),
+      sendNotification: jest.fn()
+    }
+  }
+})
+
+const sendNotificationMock = webpush.sendNotification as jest.MockedFunction<typeof webpush.sendNotification>
+
+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<PushSubscription>,
+    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)
+  })
+})

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<unknown> }> {
+  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<void>): Promise<void> {
+  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')
+    })
+  })
+})

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)
 })