Improve error handling in notification service (#10662)

Signed-off-by: Artem Savchenko <armisav@gmail.com>

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

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<Ref<PushSubscription>[]> {
-  const result: Ref<PushSubscription>[] = []
-  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<void> => {
-  console.log('Notification service has been started')
+export const main = async (ctx: MeasureContext): Promise<void> => {
+  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<any>, res: Response<any>): boolean => {
@@ -92,15 +79,18 @@ export const main = async (): Promise<void> => {
           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<void> => {
 
   process.on('SIGINT', shutdown)
   process.on('SIGTERM', shutdown)
-  process.on('uncaughtException', (e) => {
-    console.error(e)
-  })
-  process.on('unhandledRejection', (e) => {
-    console.error(e)
-  })
 }

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<Ref<PushSubscription>[]> {
+  const result: Ref<PushSubscription>[] = []
+  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
+}

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)