feat: Replace WebSocketSubscription2021 with WebSocketChannel2023

test/unit/server/notifications/WebSocketChannel2023/WebSocket2023Emitter.test.ts

@@ -0,0 +1,80 @@
+import { EventEmitter } from 'events';
+import type { WebSocket } from 'ws';
+import { BasicRepresentation } from '../../../../../src/http/representation/BasicRepresentation';
+import type { NotificationChannel } from '../../../../../src/server/notifications/NotificationChannel';
+import {
+  WebSocket2023Emitter,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocket2023Emitter';
+import type { SetMultiMap } from '../../../../../src/util/map/SetMultiMap';
+import { WrappedSetMultiMap } from '../../../../../src/util/map/WrappedSetMultiMap';
+
+describe('A WebSocket2023Emitter', (): void => {
+  const channel: NotificationChannel = {
+    id: 'id',
+    topic: 'http://example.com/foo',
+    type: 'type',
+  };
+
+  let webSocket: jest.Mocked<WebSocket>;
+  let socketMap: SetMultiMap<string, WebSocket>;
+  let emitter: WebSocket2023Emitter;
+
+  beforeEach(async(): Promise<void> => {
+    webSocket = new EventEmitter() as any;
+    webSocket.send = jest.fn();
+    webSocket.close = jest.fn();
+
+    socketMap = new WrappedSetMultiMap();
+
+    emitter = new WebSocket2023Emitter(socketMap);
+  });
+
+  it('emits notifications to the stored WebSockets.', async(): Promise<void> => {
+    socketMap.add(channel.id, webSocket);
+
+    const representation = new BasicRepresentation('notification', 'text/plain');
+    await expect(emitter.handle({ channel, representation })).resolves.toBeUndefined();
+    expect(webSocket.send).toHaveBeenCalledTimes(1);
+    expect(webSocket.send).toHaveBeenLastCalledWith('notification');
+  });
+
+  it('destroys the representation if there is no matching WebSocket.', async(): Promise<void> => {
+    const representation = new BasicRepresentation('notification', 'text/plain');
+    await expect(emitter.handle({ channel, representation })).resolves.toBeUndefined();
+    expect(webSocket.send).toHaveBeenCalledTimes(0);
+    expect(representation.data.destroyed).toBe(true);
+  });
+
+  it('can send to multiple matching WebSockets.', async(): Promise<void> => {
+    const webSocket2: jest.Mocked<WebSocket> = new EventEmitter() as any;
+    webSocket2.send = jest.fn();
+
+    socketMap.add(channel.id, webSocket);
+    socketMap.add(channel.id, webSocket2);
+
+    const representation = new BasicRepresentation('notification', 'text/plain');
+    await expect(emitter.handle({ channel, representation })).resolves.toBeUndefined();
+    expect(webSocket.send).toHaveBeenCalledTimes(1);
+    expect(webSocket.send).toHaveBeenLastCalledWith('notification');
+    expect(webSocket2.send).toHaveBeenCalledTimes(1);
+    expect(webSocket2.send).toHaveBeenLastCalledWith('notification');
+  });
+
+  it('only sends to the matching WebSockets.', async(): Promise<void> => {
+    const webSocket2: jest.Mocked<WebSocket> = new EventEmitter() as any;
+    webSocket2.send = jest.fn();
+    const channel2: NotificationChannel = {
+      ...channel,
+      id: 'other',
+    };
+
+    socketMap.add(channel.id, webSocket);
+    socketMap.add(channel2.id, webSocket2);
+
+    const representation = new BasicRepresentation('notification', 'text/plain');
+    await expect(emitter.handle({ channel, representation })).resolves.toBeUndefined();
+    expect(webSocket.send).toHaveBeenCalledTimes(1);
+    expect(webSocket.send).toHaveBeenLastCalledWith('notification');
+    expect(webSocket2.send).toHaveBeenCalledTimes(0);
+  });
+});

test/unit/server/notifications/WebSocketChannel2023/WebSocket2023Listener.test.ts

@@ -0,0 +1,122 @@
+import { EventEmitter } from 'events';
+import type { Server } from 'http';
+import type { WebSocket } from 'ws';
+import {
+  AbsolutePathInteractionRoute,
+} from '../../../../../src/identity/interaction/routing/AbsolutePathInteractionRoute';
+import type { HttpRequest } from '../../../../../src/server/HttpRequest';
+import type { NotificationChannel } from '../../../../../src/server/notifications/NotificationChannel';
+import type {
+  NotificationChannelStorage,
+} from '../../../../../src/server/notifications/NotificationChannelStorage';
+import type {
+  WebSocket2023Handler,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocket2023Handler';
+import {
+  WebSocket2023Listener,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocket2023Listener';
+import { flushPromises } from '../../../../util/Util';
+
+jest.mock('ws', (): any => ({
+  // eslint-disable-next-line @typescript-eslint/naming-convention
+  WebSocketServer: jest.fn().mockImplementation((): any => ({
+    handleUpgrade(upgradeRequest: any, socket: any, head: any, callback: any): void {
+      callback(socket, upgradeRequest);
+    },
+  })),
+}));
+
+describe('A WebSocket2023Listener', (): void => {
+  const channel: NotificationChannel = {
+    id: 'id',
+    topic: 'http://example.com/foo',
+    type: 'type',
+  };
+  const auth = '123456';
+  let server: Server;
+  let webSocket: WebSocket;
+  let upgradeRequest: HttpRequest;
+  let storage: jest.Mocked<NotificationChannelStorage>;
+  let handler: jest.Mocked<WebSocket2023Handler>;
+  const route = new AbsolutePathInteractionRoute('http://example.com/foo');
+  let listener: WebSocket2023Listener;
+
+  beforeEach(async(): Promise<void> => {
+    server = new EventEmitter() as any;
+    webSocket = new EventEmitter() as any;
+    webSocket.send = jest.fn();
+    webSocket.close = jest.fn();
+
+    upgradeRequest = { url: `/foo?auth=${auth}` } as any;
+
+    storage = {
+      get: jest.fn().mockResolvedValue(channel),
+    } as any;
+
+    handler = {
+      handleSafe: jest.fn(),
+    } as any;
+
+    listener = new WebSocket2023Listener(storage, handler, route);
+    await listener.handle(server);
+  });
+
+  it('rejects request targeting an unknown path.', async(): Promise<void> => {
+    upgradeRequest.url = '/wrong';
+    server.emit('upgrade', upgradeRequest, webSocket);
+
+    await flushPromises();
+
+    expect(webSocket.send).toHaveBeenCalledTimes(1);
+    expect(webSocket.send).toHaveBeenLastCalledWith('Unknown WebSocket target.');
+    expect(webSocket.close).toHaveBeenCalledTimes(1);
+    expect(handler.handleSafe).toHaveBeenCalledTimes(0);
+  });
+
+  it('rejects request with no url.', async(): Promise<void> => {
+    delete upgradeRequest.url;
+    server.emit('upgrade', upgradeRequest, webSocket);
+
+    await flushPromises();
+
+    expect(webSocket.send).toHaveBeenCalledTimes(1);
+    expect(webSocket.send).toHaveBeenLastCalledWith('Unknown WebSocket target.');
+    expect(webSocket.close).toHaveBeenCalledTimes(1);
+    expect(handler.handleSafe).toHaveBeenCalledTimes(0);
+  });
+
+  it('rejects requests without an auth parameter.', async(): Promise<void> => {
+    upgradeRequest.url = '/foo';
+    server.emit('upgrade', upgradeRequest, webSocket);
+
+    await flushPromises();
+
+    expect(webSocket.send).toHaveBeenCalledTimes(1);
+    expect(webSocket.send).toHaveBeenLastCalledWith('Missing auth parameter from WebSocket URL.');
+    expect(webSocket.close).toHaveBeenCalledTimes(1);
+    expect(handler.handleSafe).toHaveBeenCalledTimes(0);
+  });
+
+  it('rejects requests with an unknown auth parameter.', async(): Promise<void> => {
+    storage.get.mockResolvedValue(undefined);
+    server.emit('upgrade', upgradeRequest, webSocket);
+
+    await flushPromises();
+
+    expect(webSocket.send).toHaveBeenCalledTimes(1);
+    expect(webSocket.send).toHaveBeenLastCalledWith(`Notification channel has expired`);
+    expect(webSocket.close).toHaveBeenCalledTimes(1);
+    expect(handler.handleSafe).toHaveBeenCalledTimes(0);
+  });
+
+  it('calls the handler when receiving a valid request.', async(): Promise<void> => {
+    server.emit('upgrade', upgradeRequest, webSocket);
+
+    await flushPromises();
+
+    expect(webSocket.send).toHaveBeenCalledTimes(0);
+    expect(webSocket.close).toHaveBeenCalledTimes(0);
+    expect(handler.handleSafe).toHaveBeenCalledTimes(1);
+    expect(handler.handleSafe).toHaveBeenLastCalledWith({ webSocket, channel });
+  });
+});

test/unit/server/notifications/WebSocketChannel2023/WebSocket2023Storer.test.ts

@@ -0,0 +1,94 @@
+import { EventEmitter } from 'events';
+import type { WebSocket } from 'ws';
+import type { NotificationChannel } from '../../../../../src/server/notifications/NotificationChannel';
+import type {
+  NotificationChannelStorage,
+} from '../../../../../src/server/notifications/NotificationChannelStorage';
+
+import {
+  WebSocket2023Storer,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocket2023Storer';
+import type { SetMultiMap } from '../../../../../src/util/map/SetMultiMap';
+import { WrappedSetMultiMap } from '../../../../../src/util/map/WrappedSetMultiMap';
+import { flushPromises } from '../../../../util/Util';
+
+describe('A WebSocket2023Storer', (): void => {
+  const channel: NotificationChannel = {
+    id: 'id',
+    topic: 'http://example.com/foo',
+    type: 'type',
+  };
+  let webSocket: jest.Mocked<WebSocket>;
+  let storage: jest.Mocked<NotificationChannelStorage>;
+  let socketMap: SetMultiMap<string, WebSocket>;
+  let storer: WebSocket2023Storer;
+
+  beforeEach(async(): Promise<void> => {
+    webSocket = new EventEmitter() as any;
+    webSocket.close = jest.fn();
+
+    storage = {
+      get: jest.fn(),
+    } as any;
+
+    socketMap = new WrappedSetMultiMap();
+
+    storer = new WebSocket2023Storer(storage, socketMap);
+  });
+
+  it('stores WebSockets.', async(): Promise<void> => {
+    await expect(storer.handle({ channel, webSocket })).resolves.toBeUndefined();
+    expect([ ...socketMap.keys() ]).toHaveLength(1);
+    expect(socketMap.has(channel.id)).toBe(true);
+  });
+
+  it('removes closed WebSockets.', async(): Promise<void> => {
+    await expect(storer.handle({ channel, webSocket })).resolves.toBeUndefined();
+    expect(socketMap.has(channel.id)).toBe(true);
+    webSocket.emit('close');
+    expect(socketMap.has(channel.id)).toBe(false);
+  });
+
+  it('removes erroring WebSockets.', async(): Promise<void> => {
+    await expect(storer.handle({ channel, webSocket })).resolves.toBeUndefined();
+    expect(socketMap.has(channel.id)).toBe(true);
+    webSocket.emit('error');
+    expect(socketMap.has(channel.id)).toBe(false);
+  });
+
+  it('removes expired WebSockets.', async(): Promise<void> => {
+    jest.useFakeTimers();
+
+    // Need to create class after fake timers have been enabled
+    storer = new WebSocket2023Storer(storage, socketMap);
+
+    const webSocket2: jest.Mocked<WebSocket> = new EventEmitter() as any;
+    webSocket2.close = jest.fn();
+    const webSocketOther: jest.Mocked<WebSocket> = new EventEmitter() as any;
+    webSocketOther.close = jest.fn();
+    const channelOther: NotificationChannel = {
+      ...channel,
+      id: 'other',
+    };
+    await expect(storer.handle({ channel, webSocket })).resolves.toBeUndefined();
+    await expect(storer.handle({ channel, webSocket: webSocket2 })).resolves.toBeUndefined();
+    await expect(storer.handle({ channel: channelOther, webSocket: webSocketOther })).resolves.toBeUndefined();
+
+    // `channel` expired, `channelOther` did not
+    storage.get.mockImplementation((id): any => {
+      if (id === channelOther.id) {
+        return channelOther;
+      }
+    });
+
+    jest.advanceTimersToNextTimer();
+
+    await flushPromises();
+
+    expect(webSocket.close).toHaveBeenCalledTimes(1);
+    expect(webSocket2.close).toHaveBeenCalledTimes(1);
+    expect(webSocketOther.close).toHaveBeenCalledTimes(0);
+
+    jest.useRealTimers();
+  });
+});

test/unit/server/notifications/WebSocketChannel2023/WebSocket2023Util.test.ts

@@ -0,0 +1,27 @@
+import type { IncomingMessage } from 'http';
+import {
+  generateWebSocketUrl, parseWebSocketRequest,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocket2023Util';
+
+describe('WebSocket2023Util', (): void => {
+  describe('#generateWebSocketUrl', (): void => {
+    it('generates a WebSocket link with a query parameter.', async(): Promise<void> => {
+      expect(generateWebSocketUrl('http://example.com/', '123456')).toBe('ws://example.com/?auth=123456');
+
+      expect(generateWebSocketUrl('https://example.com/foo/bar', '123456'))
+        .toBe('wss://example.com/foo/bar?auth=123456');
+    });
+  });
+
+  describe('#parseWebSocketRequest', (): void => {
+    it('parses the request.', async(): Promise<void> => {
+      const request: IncomingMessage = { url: '/foo/bar?auth=123%24456' } as any;
+      expect(parseWebSocketRequest(request)).toEqual({ path: '/foo/bar', id: '123$456' });
+    });
+
+    it('returns an empty path and no id if the url parameter is undefined.', async(): Promise<void> => {
+      const request: IncomingMessage = {} as any;
+      expect(parseWebSocketRequest(request)).toEqual({ path: '/' });
+    });
+  });
+});

test/unit/server/notifications/WebSocketChannel2023/WebSocketChannel2023Type.test.ts

@@ -0,0 +1,58 @@
+import { DataFactory, Store } from 'n3';
+import {
+  AbsolutePathInteractionRoute,
+} from '../../../../../src/identity/interaction/routing/AbsolutePathInteractionRoute';
+import type { NotificationChannel } from '../../../../../src/server/notifications/NotificationChannel';
+
+import {
+  generateWebSocketUrl,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocket2023Util';
+import type {
+  WebSocketChannel2023,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocketChannel2023Type';
+import {
+  isWebSocket2023Channel,
+  WebSocketChannel2023Type,
+} from '../../../../../src/server/notifications/WebSocketChannel2023/WebSocketChannel2023Type';
+import { NOTIFY, RDF } from '../../../../../src/util/Vocabularies';
+import quad = DataFactory.quad;
+import blankNode = DataFactory.blankNode;
+import namedNode = DataFactory.namedNode;
+
+jest.mock('uuid', (): any => ({ v4: (): string => '4c9b88c1-7502-4107-bb79-2a3a590c7aa3' }));
+
+describe('A WebSocketChannel2023', (): void => {
+  let data: Store;
+  let channel: WebSocketChannel2023;
+  const subject = blankNode();
+  const topic = 'https://storage.example/resource';
+  const route = new AbsolutePathInteractionRoute('http://example.com/foo');
+  let channelType: WebSocketChannel2023Type;
+
+  beforeEach(async(): Promise<void> => {
+    data = new Store();
+    data.addQuad(quad(subject, RDF.terms.type, NOTIFY.terms.WebSocketChannel2023));
+    data.addQuad(quad(subject, NOTIFY.terms.topic, namedNode(topic)));
+
+    const id = 'http://example.com/foo/4c9b88c1-7502-4107-bb79-2a3a590c7aa3';
+    channel = {
+      id,
+      type: NOTIFY.WebSocketChannel2023,
+      topic,
+      receiveFrom: generateWebSocketUrl(route.getPath(), id),
+    };
+
+    channelType = new WebSocketChannel2023Type(route);
+  });
+
+  it('exposes a utility function to verify if a channel is a websocket channel.', async(): Promise<void> => {
+    expect(isWebSocket2023Channel(channel)).toBe(true);
+
+    (channel as NotificationChannel).type = 'something else';
+    expect(isWebSocket2023Channel(channel)).toBe(false);
+  });
+
+  it('correctly parses notification channel bodies.', async(): Promise<void> => {
+    await expect(channelType.initChannel(data, {})).resolves.toEqual(channel);
+  });
+});