feat: notifications in PlatformClient

.changeset/notifications-client.md

@@ -0,0 +1,23 @@
+---
+'@thatopen/services': minor
+---
+
+Add notifications to `PlatformClient`.
+
+Read and manage the signed-in user's notifications with `getNotifications`,
+`getUnreadNotificationCount`, `markNotificationsRead` and
+`markAllNotificationsRead`, and manage automation subscriptions with
+`getNotificationSubscriptions`, `subscribeToAutomation`,
+`updateAutomationSubscription` and `unsubscribeFromAutomation`. All scoped to
+the signed-in user via the bearer token an app already has.
+
+`onNotification` subscribes to new notifications live. Unlike
+`onExecutionProgress` it stays connected for the session rather than closing on
+a terminal event, and it returns a function that disconnects.
+
+The notification types mirror the backend's wire DTOs in `src/types`, the same
+as every other type here. Read `outcome` to tell how an automation run ended
+rather than matching on the copy: `title` is built from the user's own
+automation name, so an automation called "Failover sync" makes every successful
+run look failed to anything parsing the text. `groupKey` and `groupLabel` are
+what a client needs to collapse a busy automation's runs into one row.

src/core/client.ts

@@ -299,6 +299,15 @@ export class EngineServicesClient {
    * the new token is picked up on every request — expired tokens no
    * longer stick around.
    */
+  /**
+   * Socket origin without namespace or query, for gateways other than the
+   * execution one. `wsUrl` already carries a token that may be stale when a
+   * provider is in play, so callers append their own.
+   */
+  protected get socketOrigin(): string {
+    return this.wsUrl.split('?')[0];
+  }
+
   protected async resolveAccessToken(): Promise<string> {
     return this.accessToken;
   }

src/core/platform-client.live.test.ts

@@ -0,0 +1,74 @@
+import { describe, it, expect, beforeEach, vi } from 'vitest';
+
+const handlers = new Map<string, (payload: unknown) => void>();
+const disconnect = vi.fn();
+const ioMock = vi.fn(() => ({
+  on: (event: string, handler: (payload: unknown) => void) => {
+    handlers.set(event, handler);
+  },
+  disconnect,
+}));
+
+vi.mock('socket.io-client', () => ({ io: (...args: unknown[]) => ioMock(...(args as [])) }));
+
+const { PlatformClient } = await import('./platform-client');
+
+const API = 'https://api.example.com';
+
+describe('PlatformClient — live notifications', () => {
+  let client: InstanceType<typeof PlatformClient>;
+
+  beforeEach(() => {
+    handlers.clear();
+    ioMock.mockClear();
+    disconnect.mockClear();
+    client = new PlatformClient('jwt-1', API);
+  });
+
+  it('connects to the notifications namespace with the token', async () => {
+    await client.onNotification(() => {});
+
+    const url = (ioMock.mock.calls[0] as unknown as string[])[0];
+    expect(url).toContain('/notifications');
+    expect(url).toContain('accessToken=jwt-1');
+    // No /api on a socket URL; that prefix is for REST only.
+    expect(url).not.toContain('/api/');
+  });
+
+  // A provider-backed client must open the socket with a current token, not
+  // the one it happened to be constructed with.
+  it('resolves the token per connection when a provider is used', async () => {
+    const provider = vi.fn().mockResolvedValue('fresh-token');
+    const providerClient = new PlatformClient(provider, API);
+
+    await providerClient.onNotification(() => {});
+
+    expect(provider).toHaveBeenCalled();
+    expect((ioMock.mock.calls[0] as unknown as string[])[0]).toContain(
+      'accessToken=fresh-token',
+    );
+  });
+
+  it('maps each server event onto one callback shape', async () => {
+    const seen: unknown[] = [];
+    await client.onNotification((event) => seen.push(event));
+
+    handlers.get('notification.created')?.({ id: 'n1' });
+    handlers.get('notification.read')?.({ id: 'n2' });
+    handlers.get('notifications.allRead')?.({ batch: 42 });
+
+    expect(seen).toEqual([
+      { type: 'created', id: 'n1' },
+      { type: 'read', id: 'n2' },
+      { type: 'allRead', batch: 42 },
+    ]);
+  });
+
+  it('returns a disconnect function', async () => {
+    const stop = await client.onNotification(() => {});
+
+    expect(disconnect).not.toHaveBeenCalled();
+    stop();
+    expect(disconnect).toHaveBeenCalledTimes(1);
+  });
+});

src/core/platform-client.notifications.test.ts

@@ -0,0 +1,214 @@
+import {
+  describe,
+  it,
+  expect,
+  beforeEach,
+  afterEach,
+  vi,
+  type Mock,
+} from 'vitest';
+import { PlatformClient } from './platform-client';
+
+const API = 'https://api.example.com';
+const JWT = 'test-jwt';
+
+function okResponse(data: unknown): Response {
+  return {
+    ok: true,
+    status: 200,
+    statusText: 'OK',
+    text: async () => JSON.stringify(data),
+    json: async () => data,
+  } as unknown as Response;
+}
+
+function callUrl(fetchMock: Mock, index = 0): URL {
+  return new URL(fetchMock.mock.calls[index][0] as string);
+}
+
+function callInit(fetchMock: Mock, index = 0): RequestInit {
+  return fetchMock.mock.calls[index][1] as RequestInit;
+}
+
+const emptyPage = { items: [], nextCursor: null };
+
+describe('PlatformClient — notifications', () => {
+  let fetchMock: Mock;
+  let client: PlatformClient;
+
+  beforeEach(() => {
+    fetchMock = vi.fn();
+    globalThis.fetch = fetchMock as unknown as typeof fetch;
+    client = new PlatformClient(JWT, API);
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  describe('getNotifications', () => {
+    it('reads the account notifications with no query when unpaged', async () => {
+      fetchMock.mockResolvedValue(okResponse(emptyPage));
+
+      await client.getNotifications();
+
+      const url = callUrl(fetchMock);
+      expect(url.pathname).toContain('/notifications');
+      expect(url.searchParams.get('cursor')).toBeNull();
+      expect(url.searchParams.get('limit')).toBeNull();
+    });
+
+    it('passes cursor and limit through', async () => {
+      fetchMock.mockResolvedValue(okResponse(emptyPage));
+
+      await client.getNotifications({ cursor: 'abc_123', limit: 50 });
+
+      const url = callUrl(fetchMock);
+      expect(url.searchParams.get('cursor')).toBe('abc_123');
+      expect(url.searchParams.get('limit')).toBe('50');
+    });
+
+    // The cursor is opaque and round-trips verbatim; anything that mangles it
+    // silently breaks pagination rather than erroring.
+    it('does not mangle a cursor containing an ISO timestamp', async () => {
+      fetchMock.mockResolvedValue(okResponse(emptyPage));
+      const cursor = '2026-08-12T09:30:00.000Z_6a7c95b780c5fd7e84758c32';
+
+      await client.getNotifications({ cursor });
+
+      expect(callUrl(fetchMock).searchParams.get('cursor')).toBe(cursor);
+    });
+
+    it('returns the page as sent, ids and timestamps as strings', async () => {
+      const page = {
+        items: [
+          {
+            _id: '6a7c95b780c5fd7e84758c32',
+            accountId: '6a3bb8b0f32c03c0f86897f2',
+            type: 'automation.run.finished',
+            category: 'automation',
+            title: 'Nightly report failed',
+            body: 'IFC Converter ended with ERROR.',
+            link: '/dashboard/projects/p1/automation-runs',
+            muted: false,
+            readAt: null,
+            createdAt: '2026-08-12T09:30:00.000Z',
+          },
+        ],
+        nextCursor: null,
+      };
+      fetchMock.mockResolvedValue(okResponse(page));
+
+      const result = await client.getNotifications();
+
+      expect(result).toEqual(page);
+      expect(typeof result.items[0]._id).toBe('string');
+      expect(typeof result.items[0].createdAt).toBe('string');
+    });
+  });
+
+  it('unwraps the unread count to a number', async () => {
+    fetchMock.mockResolvedValue(okResponse({ count: 7 }));
+
+    await expect(client.getUnreadNotificationCount()).resolves.toBe(7);
+    expect(callUrl(fetchMock).pathname).toContain('/notifications/unread-count');
+  });
+
+  it('marks specific notifications read as a JSON body', async () => {
+    fetchMock.mockResolvedValue(okResponse({ updated: 2 }));
+
+    const result = await client.markNotificationsRead(['id-1', 'id-2']);
+
+    const init = callInit(fetchMock);
+    expect(init.method).toBe('POST');
+    expect(JSON.parse(init.body as string)).toEqual({ ids: ['id-1', 'id-2'] });
+    expect(result.updated).toBe(2);
+  });
+
+  it('marks all read without a body', async () => {
+    fetchMock.mockResolvedValue(okResponse({ updated: 9 }));
+
+    await client.markAllNotificationsRead();
+
+    expect(callUrl(fetchMock).pathname).toContain(
+      '/notifications/mark-all-read',
+    );
+    expect(callInit(fetchMock).method).toBe('POST');
+  });
+
+  it('lists subscriptions', async () => {
+    fetchMock.mockResolvedValue(okResponse([]));
+
+    await expect(client.getNotificationSubscriptions()).resolves.toEqual([]);
+    expect(callUrl(fetchMock).pathname).toContain('/notifications/subscriptions');
+  });
+
+  it('unsubscribes by hook id', async () => {
+    fetchMock.mockResolvedValue(okResponse({ unsubscribed: true }));
+
+    const result = await client.unsubscribeFromAutomation('hook-1');
+
+    expect(callInit(fetchMock).method).toBe('DELETE');
+    expect(callUrl(fetchMock).pathname).toContain(
+      '/notifications/subscriptions/hook-1',
+    );
+    expect(result.unsubscribed).toBe(true);
+  });
+
+  it('sends the bearer token on notification routes', async () => {
+    fetchMock.mockResolvedValue(okResponse(emptyPage));
+
+    await client.getNotifications();
+
+    const headers = callInit(fetchMock).headers as Record<string, string>;
+    expect(headers.Authorization).toBe(`Bearer ${JWT}`);
+  });
+
+  describe('subscribing to an automation', () => {
+    const PROJECT = 'proj-1';
+    const HOOK = 'hook-1';
+
+    it('subscribes through the project route', async () => {
+      fetchMock.mockResolvedValue(okResponse({ subscribed: true }));
+
+      await client.subscribeToAutomation(PROJECT, HOOK, {
+        filter: 'failures',
+        channels: { email: true },
+      });
+
+      const init = callInit(fetchMock);
+      expect(init.method).toBe('POST');
+      expect(callUrl(fetchMock).pathname).toContain(
+        `/project/${PROJECT}/events/hooks/${HOOK}/subscription`,
+      );
+      expect(JSON.parse(init.body as string)).toEqual({
+        filter: 'failures',
+        channels: { email: true },
+      });
+    });
+
+    it('sends an empty body when no options are given', async () => {
+      fetchMock.mockResolvedValue(okResponse({ subscribed: true }));
+
+      await client.subscribeToAutomation(PROJECT, HOOK);
+
+      expect(JSON.parse(callInit(fetchMock).body as string)).toEqual({});
+    });
+
+    // PATCH is a merge server-side, so sending channels alone must not carry
+    // a filter along with it and reset one that was already set.
+    it('patches only what it is given', async () => {
+      fetchMock.mockResolvedValue(okResponse({ updated: true }));
+
+      await client.updateAutomationSubscription(PROJECT, HOOK, {
+        channels: { email: false },
+      });
+
+      const init = callInit(fetchMock);
+      expect(init.method).toBe('PATCH');
+      expect(JSON.parse(init.body as string)).toEqual({
+        channels: { email: false },
+      });
+    });
+  });
+});