feat: move all signal handling to backend plugin

Signed-off-by: Heikki Hellgren <heikki.hellgren@op.fi>

plugins/signals-node/README.md

@@ -27,10 +27,8 @@ function makeCreateEnv(config: Config) {
   // ...
 
   const eventBroker = new DefaultEventBroker(root.child({ type: 'plugin' }));
-  const signalService = SignalService.create({
-    logger: root,
-    eventBroker, // EventBroker is optional
-    identity,
+  const signalService = DefaultSignalService.create({
+    eventBroker,
   });
 
   return (plugin: string): PluginEnvironment => {
@@ -51,16 +49,38 @@ To allow connections from the frontend, you should also install the `@backstage/
 
 Once you have both of the backend plugins installed, you can utilize the signal service by calling the
 `publish` method. This will publish the message to all subscribers in the frontend. To send message to
-all subscribers, you can use `*` as `to` parameter.
+all subscribers, you can use `null` as `recipients` parameter.
 
 ```ts
 // Periodic sending example
 setInterval(async () => {
-  await signalService.publish('*', 'plugin:topic', {
-    message: 'hello world',
+  await signalService.publish({
+    recipients: null,
+    channel: 'my_plugin',
+    message: {
+      message: 'hello world',
+    },
   });
 }, 5000);
 ```
 
 To receive this message in the frontend, check the documentation of `@backstage/plugin-signals` and
 `@backstage/plugin-signals-react`.
+
+## Using event broker directly
+
+Other way to send signals is to utilize the `EventBroker` directly. This requires that the payload is correct for it
+to work:
+
+```ts
+eventBroker.publish({
+  topic: 'signals',
+  eventPayload: {
+    recipients: ['user:default/user1'],
+    message: {
+      message: 'hello world',
+    },
+    channel: 'my_plugin',
+  },
+});
+```

plugins/signals-node/api-report.md

@@ -3,63 +3,35 @@
 > Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
 
 ```ts
-/// <reference types="node" />
-
-import { Duplex } from 'stream';
 import { EventBroker } from '@backstage/plugin-events-node';
-import http from 'http';
-import https from 'https';
-import { IdentityApi } from '@backstage/plugin-auth-node';
-import { IncomingMessage } from 'http';
 import { JsonObject } from '@backstage/types';
-import { LoggerService } from '@backstage/backend-plugin-api';
 import { ServiceRef } from '@backstage/backend-plugin-api';
 
 // @public (undocumented)
 export class DefaultSignalService implements SignalService {
   // (undocumented)
-  static create(options: ServiceOptions): DefaultSignalService;
-  handleUpgrade(options: SignalServiceUpgradeOptions): Promise<void>;
-  publish(
-    to: string | string[],
-    topic: string,
-    message: JsonObject,
-  ): Promise<void>;
+  static create(options: SignalServiceOptions): DefaultSignalService;
+  publish(signal: SignalPayload): Promise<void>;
 }
 
 // @public (undocumented)
-export type ServiceOptions = {
-  eventBroker?: EventBroker;
-  logger: LoggerService;
-  identity: IdentityApi;
-};
-
-// @public (undocumented)
-export type SignalEventBrokerPayload = {
-  recipients?: string[];
-  topic?: string;
-  message?: JsonObject;
+export type SignalPayload = {
+  recipients: string[] | null;
+  channel: string;
+  message: JsonObject;
 };
 
 // @public (undocumented)
 export type SignalService = {
-  publish(
-    to: string | string[],
-    topic: string,
-    message: JsonObject,
-  ): Promise<void>;
-  handleUpgrade(options: SignalServiceUpgradeOptions): Promise<void>;
+  publish(signal: SignalPayload): Promise<void>;
 };
 
 // @public (undocumented)
 export const signalService: ServiceRef<SignalService, 'plugin'>;
 
 // @public (undocumented)
-export type SignalServiceUpgradeOptions = {
-  server: https.Server | http.Server;
-  request: IncomingMessage;
-  socket: Duplex;
-  head: Buffer;
+export type SignalServiceOptions = {
+  eventBroker?: EventBroker;
 };
 
 // (No @packageDocumentation comment for this package)

plugins/signals-node/src/DefaultSignalService.ts

@@ -13,226 +13,38 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { EventBroker, EventParams } from '@backstage/plugin-events-node';
-import {
-  ServiceOptions,
-  SignalConnection,
-  SignalEventBrokerPayload,
-} from './types';
-import { RawData, WebSocket, WebSocketServer } from 'ws';
-import { IncomingMessage } from 'http';
-import { v4 as uuid } from 'uuid';
-import { JsonObject } from '@backstage/types';
-import {
-  BackstageIdentityResponse,
-  IdentityApi,
-  IdentityApiGetIdentityRequest,
-} from '@backstage/plugin-auth-node';
-import { LoggerService } from '@backstage/backend-plugin-api';
-import { SignalService, SignalServiceUpgradeOptions } from './SignalService';
+import { EventBroker } from '@backstage/plugin-events-node';
+import { SignalPayload, SignalServiceOptions } from './types';
+import { SignalService } from './SignalService';
 
 /** @public */
 export class DefaultSignalService implements SignalService {
-  private connections: Map<string, SignalConnection> = new Map<
-    string,
-    SignalConnection
-  >();
+  // TODO: Remove this to be optional when events-backend has eventBroker as service
   private eventBroker?: EventBroker;
-  private logger: LoggerService;
-  private identity: IdentityApi;
-  private server: WebSocketServer;
 
-  static create(options: ServiceOptions) {
+  static create(options: SignalServiceOptions) {
     return new DefaultSignalService(options);
   }
 
-  private constructor(options: ServiceOptions) {
-    ({
-      eventBroker: this.eventBroker,
-      logger: this.logger,
-      identity: this.identity,
-    } = options);
-
-    this.server = new WebSocketServer({
-      noServer: true,
-      clientTracking: false,
-    });
-
-    this.eventBroker?.subscribe({
-      supportsEventTopics: () => ['signals'],
-      onEvent: (params: EventParams<SignalEventBrokerPayload>) =>
-        this.onEventBrokerEvent(params),
-    });
-  }
-
-  /**
-   * Handles request upgrade to websocket and adds the connection to internal
-   * list for publish/subscribe functionality
-   * @param req - Request
-   */
-  async handleUpgrade(options: SignalServiceUpgradeOptions) {
-    const { request, socket, head } = options;
-    let identity: BackstageIdentityResponse | undefined = undefined;
-
-    // Authentication token is passed in Sec-WebSocket-Protocol header as there
-    // is no other way to pass the token with plain websockets
-    const token = request.headers['sec-websocket-protocol'];
-    if (token) {
-      identity = await this.identity.getIdentity({
-        request: {
-          headers: { authorization: token },
-        },
-      } as IdentityApiGetIdentityRequest);
-    }
-
-    this.server.handleUpgrade(
-      request,
-      socket,
-      head,
-      (ws: WebSocket, __: IncomingMessage) => {
-        this.addConnection(ws, identity);
-      },
-    );
-  }
-
-  private addConnection(ws: WebSocket, identity?: BackstageIdentityResponse) {
-    const id = uuid();
-
-    const conn = {
-      id,
-      user: identity?.identity.userEntityRef ?? 'user:default/guest',
-      ws,
-      ownershipEntityRefs: identity?.identity.ownershipEntityRefs ?? [],
-      subscriptions: new Set<string>(),
-    };
-
-    this.connections.set(id, conn);
-
-    ws.on('error', (err: Error) => {
-      this.logger.info(
-        `Error occurred with connection ${id}: ${err}, closing connection`,
-      );
-      ws.close();
-      this.connections.delete(id);
-    });
-
-    ws.on('close', (code: number, reason: Buffer) => {
-      this.logger.info(
-        `Connection ${id} closed with code ${code}, reason: ${reason}`,
-      );
-      this.connections.delete(id);
-    });
-
-    ws.on('message', (data: RawData, isBinary: boolean) => {
-      this.logger.debug(`Received message from connection ${id}: ${data}`);
-      if (isBinary) {
-        return;
-      }
-      try {
-        const json = JSON.parse(data.toString()) as JsonObject;
-        this.handleMessage(conn, json);
-      } catch (err: any) {
-        this.logger.error(
-          `Invalid message received from connection ${id}: ${err}`,
-        );
-      }
-    });
-  }
-
-  private handleMessage(connection: SignalConnection, message: JsonObject) {
-    if (message.action === 'subscribe' && message.topic) {
-      this.logger.info(
-        `Connection ${connection.id} subscribed to ${message.topic}`,
-      );
-      connection.subscriptions.add(message.topic as string);
-    }
-
-    if (message.action === 'unsubscribe' && message.topic) {
-      this.logger.info(
-        `Connection ${connection.id} unsubscribed from ${message.topic}`,
-      );
-      connection.subscriptions.delete(message.topic as string);
-    }
+  private constructor(options: SignalServiceOptions) {
+    ({ eventBroker: this.eventBroker } = options);
   }
 
   /**
    * Publishes a message to user refs to specific topic
-   * @param to - string or array of user ref strings to publish message to
+   * @param recipients - string or array of user ref strings to publish message to
    * @param topic - message topic
    * @param message - message to publish
    */
-  async publish(to: string | string[], topic: string, message: JsonObject) {
-    await this.publishInternal(
-      Array.isArray(to) ? to : [to],
-      topic,
-      message,
-      false,
-    );
-  }
-
-  private async publishInternal(
-    recipients: string[],
-    topic: string,
-    message: JsonObject,
-    brokedEvent: boolean,
-  ) {
-    const jsonMessage = JSON.stringify({ topic, message });
-    if (jsonMessage.length === 0) {
-      return;
-    }
-
-    // If there is event broker, use that to publish the message to
-    // all signal services, including this one.
-    if (this.eventBroker && !brokedEvent) {
-      await this.eventBroker.publish({
-        topic: 'signals',
-        eventPayload: {
-          recipients,
-          message,
-          topic,
-        },
-      });
-      return;
-    }
-
-    // Actual websocket message sending
-    this.connections.forEach(conn => {
-      if (!conn.subscriptions.has(topic)) {
-        return;
-      }
-      // Sending to all users can be done with '*'
-      if (
-        !recipients.includes('*') &&
-        !conn.ownershipEntityRefs.some(ref => recipients.includes(ref))
-      ) {
-        return;
-      }
-
-      if (conn.ws.readyState !== WebSocket.OPEN) {
-        return;
-      }
-
-      conn.ws.send(jsonMessage);
+  async publish(signal: SignalPayload) {
+    const { recipients, channel, message } = signal;
+    await this.eventBroker?.publish({
+      topic: 'signals',
+      eventPayload: {
+        recipients,
+        message,
+        channel,
+      },
     });
   }
-
-  private async onEventBrokerEvent(
-    params: EventParams<SignalEventBrokerPayload>,
-  ): Promise<void> {
-    const { eventPayload } = params;
-    if (
-      !eventPayload?.recipients ||
-      !eventPayload.topic ||
-      !eventPayload.message
-    ) {
-      return;
-    }
-
-    await this.publishInternal(
-      eventPayload.recipients,
-      eventPayload.topic,
-      eventPayload.message,
-      true,
-    );
-  }
 }

plugins/signals-node/src/SignalService.ts

@@ -13,32 +13,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { JsonObject } from '@backstage/types';
-import http, { IncomingMessage } from 'http';
-import { Duplex } from 'stream';
-import https from 'https';
-
-/** @public */
-export type SignalServiceUpgradeOptions = {
-  server: https.Server | http.Server;
-  request: IncomingMessage;
-  socket: Duplex;
-  head: Buffer;
-};
+import { SignalPayload } from './types';
 
 /** @public */
 export type SignalService = {
   /**
    * Publishes a message to user refs to specific topic
    */
-  publish(
-    to: string | string[],
-    topic: string,
-    message: JsonObject,
-  ): Promise<void>;
-
-  /**
-   * Handles request upgrade
-   */
-  handleUpgrade(options: SignalServiceUpgradeOptions): Promise<void>;
+  publish(signal: SignalPayload): Promise<void>;
 };

plugins/signals-node/src/lib.ts

@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 import {
-  coreServices,
   createServiceFactory,
   createServiceRef,
 } from '@backstage/backend-plugin-api';
@@ -29,12 +28,11 @@ export const signalService = createServiceRef<SignalService>({
     createServiceFactory({
       service,
       deps: {
-        logger: coreServices.logger,
-        identity: coreServices.identity,
-        // TODO: EventBroker
+        // TODO: EventBroker. It is optional for now but it's actually required so waiting for the new backend system
+        //       for the events-backend for this to work.
       },
-      factory({ logger, identity }) {
-        return DefaultSignalService.create({ identity, logger });
+      factory({}) {
+        return DefaultSignalService.create({});
       },
     }),
 });

plugins/signals-node/src/types.ts

@@ -13,35 +13,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { IdentityApi } from '@backstage/plugin-auth-node';
 import { EventBroker } from '@backstage/plugin-events-node';
-import { WebSocket } from 'ws';
 import { JsonObject } from '@backstage/types';
-import { LoggerService } from '@backstage/backend-plugin-api';
 
 /**
  * @public
  */
-export type ServiceOptions = {
+export type SignalServiceOptions = {
   eventBroker?: EventBroker;
-  logger: LoggerService;
-  identity: IdentityApi;
 };
 
 /** @public */
-export type SignalEventBrokerPayload = {
-  recipients?: string[];
-  topic?: string;
-  message?: JsonObject;
-};
-
-/**
- * @internal
- */
-export type SignalConnection = {
-  id: string;
-  user: string;
-  ws: WebSocket;
-  ownershipEntityRefs: string[];
-  subscriptions: Set<string>;
+export type SignalPayload = {
+  recipients: string[] | null;
+  channel: string;
+  message: JsonObject;
 };