update system metadata service to be observable-first

Signed-off-by: aramissennyeydd <aramis.sennyey@doordash.com>

packages/backend-defaults/package.json

@@ -198,6 +198,7 @@
     "winston-transport": "^4.5.0",
     "yauzl": "^3.0.0",
     "yn": "^4.0.0",
+    "zen-observable": "^0.10.0",
     "zod": "^3.22.4",
     "zod-to-json-schema": "^3.20.4"
   },

packages/backend-defaults/report-systemMetadata.api.md

@@ -19,6 +19,7 @@ export const actionsServiceFactory: ServiceFactory<
   ActionsService,
 import { BackstageInstance } from '@backstage/backend-plugin-api';
 import { LoggerService } from '@backstage/backend-plugin-api';
+import { Observable } from '@backstage/types';
 import { RootConfigService } from '@backstage/backend-plugin-api';
 import { ServiceFactory } from '@backstage/backend-plugin-api';
 import { SystemMetadataService } from '@backstage/backend-plugin-api';
@@ -32,9 +33,7 @@ export class DefaultSystemMetadataService implements SystemMetadataService {
     config: RootConfigService;
   }): DefaultSystemMetadataService;
   // (undocumented)
-  introspect(): Promise<{
-    instances: BackstageInstance[];
-  }>;
+  instances(): Observable<BackstageInstance[]>;
 }
 
 // @alpha

packages/backend-defaults/src/entrypoints/systemMetadata/SystemMetadataService.test.ts

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { createSpecializedBackend } from '@backstage/backend-app-api';
+import { Backend, createSpecializedBackend } from '@backstage/backend-app-api';
 import { systemMetadataServiceFactory } from './systemMetadataServiceFactory';
 import { mockServices } from '@backstage/backend-test-utils';
 import getPort from 'get-port';
@@ -29,111 +29,171 @@ const baseFactories = [
 ];
 
 describe('SystemMetadataService', () => {
-  it('should list plugins across instances', async () => {
-    const instance1HttpPort = await getPort();
-    const instance2HttpPort = await getPort();
-    const instance1 = createSpecializedBackend({
-      defaultServiceFactories: [
-        ...baseFactories,
-        systemMetadataServiceFactory,
-        mockServices.rootConfig.factory({
-          data: {
-            backend: {
-              listen: {
-                port: instance1HttpPort,
-              },
-            },
-            discovery: {
-              instances: [
-                {
-                  baseUrl: `http://localhost:${instance1HttpPort}`,
-                },
-                {
-                  baseUrl: `http://localhost:${instance2HttpPort}`,
-                },
-              ],
+  describe('multiple backends testing', () => {
+    let instance1: Backend;
+    let instance2: Backend;
+    let instance1HttpPort: number;
+    let instance2HttpPort: number;
+
+    const configFactory = (port: number) =>
+      mockServices.rootConfig.factory({
+        data: {
+          backend: {
+            listen: {
+              port,
             },
           },
-        }),
-      ],
+          discovery: {
+            instances: [
+              {
+                baseUrl: `http://localhost:${instance1HttpPort}`,
+              },
+              {
+                baseUrl: `http://localhost:${instance2HttpPort}`,
+              },
+            ],
+          },
+        },
+      });
+    beforeEach(async () => {
+      instance1HttpPort = await getPort();
+      instance2HttpPort = await getPort();
+      // Setup code for multiple backend instances
+      instance1 = createSpecializedBackend({
+        defaultServiceFactories: [
+          ...baseFactories,
+          systemMetadataServiceFactory,
+          configFactory(instance1HttpPort),
+        ],
+      });
+
+      instance2 = createSpecializedBackend({
+        defaultServiceFactories: [
+          ...baseFactories,
+          systemMetadataServiceFactory,
+          configFactory(instance2HttpPort),
+        ],
+      });
     });
 
-    const instance2 = createSpecializedBackend({
-      defaultServiceFactories: [
-        ...baseFactories,
-        systemMetadataServiceFactory,
-        mockServices.rootConfig.factory({
-          data: {
-            backend: {
-              listen: {
-                port: instance2HttpPort,
+    it('should list plugins across instances', async () => {
+      instance1.add(
+        createBackendPlugin({
+          pluginId: 'test',
+          register(reg) {
+            reg.registerInit({
+              deps: {},
+              async init() {
+                // do nothing
               },
-            },
-            discovery: {
-              instances: [
-                {
-                  baseUrl: `http://localhost:${instance1HttpPort}`,
-                },
-                {
-                  baseUrl: `http://localhost:${instance2HttpPort}`,
-                },
-              ],
-            },
+            });
           },
         }),
-      ],
+      );
+
+      instance2.add(
+        createBackendPlugin({
+          pluginId: 'test-other',
+          register(reg) {
+            reg.registerInit({
+              deps: {},
+              async init() {
+                // do nothing
+              },
+            });
+          },
+        }),
+      );
+
+      await instance1.start();
+      await instance2.start();
+
+      const instance1Response = await fetch(
+        `http://localhost:${instance1HttpPort}/.backstage/systemMetadata/v1/features/installed`,
+      );
+
+      expect(instance1Response.status).toBe(200);
+      await expect(instance1Response.json()).resolves.toMatchObject({
+        test: [
+          {
+            externalUrl: `http://localhost:${instance1HttpPort}`,
+            internalUrl: `http://localhost:${instance1HttpPort}`,
+          },
+        ],
+        'test-other': [
+          {
+            externalUrl: `http://localhost:${instance2HttpPort}`,
+            internalUrl: `http://localhost:${instance2HttpPort}`,
+          },
+        ],
+      });
+
+      const instance2Response = await fetch(
+        `http://localhost:${instance2HttpPort}/.backstage/systemMetadata/v1/features/installed`,
+      );
+
+      expect(instance2Response.status).toBe(200);
+
+      await expect(instance2Response.json()).resolves.toMatchObject({
+        test: [
+          {
+            externalUrl: `http://localhost:${instance1HttpPort}`,
+            internalUrl: `http://localhost:${instance1HttpPort}`,
+          },
+        ],
+        'test-other': [
+          {
+            externalUrl: `http://localhost:${instance2HttpPort}`,
+            internalUrl: `http://localhost:${instance2HttpPort}`,
+          },
+        ],
+      });
     });
 
-    instance1.add(
-      createBackendPlugin({
-        pluginId: 'test',
-        register(reg) {
-          reg.registerInit({
-            deps: {},
-            async init() {
-              // do nothing
-            },
-          });
-        },
-      }),
-    );
+    it('should list all known instances', async () => {
+      await instance1.start();
+      await instance2.start();
 
-    instance2.add(
-      createBackendPlugin({
-        pluginId: 'test-other',
-        register(reg) {
-          reg.registerInit({
-            deps: {},
-            async init() {
-              // do nothing
-            },
-          });
-        },
-      }),
-    );
+      const instance1Response = await fetch(
+        `http://localhost:${instance1HttpPort}/.backstage/systemMetadata/v1/instances`,
+      );
 
-    await instance1.start();
-    await instance2.start();
+      expect(instance1Response.status).toBe(200);
+      await expect(instance1Response.json()).resolves.toMatchObject({
+        items: [
+          {
+            externalUrl: `http://localhost:${instance1HttpPort}`,
+            internalUrl: `http://localhost:${instance1HttpPort}`,
+          },
+          {
+            externalUrl: `http://localhost:${instance2HttpPort}`,
+            internalUrl: `http://localhost:${instance2HttpPort}`,
+          },
+        ],
+      });
 
-    const installedFeatures = await fetch(
-      `http://localhost:${instance1HttpPort}/.backstage/systemMetadata/v1/features/installed`,
-    );
+      const instance2Response = await fetch(
+        `http://localhost:${instance2HttpPort}/.backstage/systemMetadata/v1/instances`,
+      );
 
-    expect(installedFeatures.status).toBe(200);
+      expect(instance2Response.status).toBe(200);
+      await expect(instance2Response.json()).resolves.toMatchObject({
+        items: [
+          {
+            externalUrl: `http://localhost:${instance1HttpPort}`,
+            internalUrl: `http://localhost:${instance1HttpPort}`,
+          },
+          {
+            externalUrl: `http://localhost:${instance2HttpPort}`,
+            internalUrl: `http://localhost:${instance2HttpPort}`,
+          },
+        ],
+      });
+    });
 
-    await expect(installedFeatures.json()).resolves.toMatchObject({
-      test: [
-        {
-          externalUrl: `http://localhost:${instance1HttpPort}`,
-          internalUrl: `http://localhost:${instance1HttpPort}`,
-        },
-      ],
-      'test-other': [
-        {
-          externalUrl: `http://localhost:${instance2HttpPort}`,
-          internalUrl: `http://localhost:${instance2HttpPort}`,
-        },
-      ],
+    afterEach(async () => {
+      await instance1.stop();
+      await instance2.stop();
     });
   });
 });

packages/backend-defaults/src/entrypoints/systemMetadata/lib/DefaultSystemMetadataService.ts

@@ -22,18 +22,126 @@ import {
   BackstageInstance,
   SystemMetadataService,
 } from '@backstage/backend-plugin-api';
+import { Observable } from '@backstage/types';
 import z from 'zod';
+import ObservableImpl from 'zen-observable';
 
 const targetObjectSchema = z.object({
   internal: z.string(),
   external: z.string(),
 });
 
+/**
+ * A basic implementation of ReactiveX behavior subjects.
+ *
+ * A subject is a convenient way to create an observable when you want
+ * to fan out a single value to all subscribers.
+ *
+ * The BehaviorSubject will emit the most recently emitted value or error
+ * whenever a new observer subscribes to the subject.
+ *
+ * See http://reactivex.io/documentation/subject.html
+ *
+ * FORKED FROM core-app-api - where should this live?
+ */
+
+export class BehaviorSubject<T>
+  implements Observable<T>, ZenObservable.SubscriptionObserver<T>
+{
+  private isClosed: boolean;
+  private currentValue: T;
+  private terminatingError: Error | undefined;
+  private readonly observable: Observable<T>;
+
+  constructor(value: T) {
+    this.isClosed = false;
+    this.currentValue = value;
+    this.terminatingError = undefined;
+    this.observable = new ObservableImpl<T>(subscriber => {
+      if (this.isClosed) {
+        if (this.terminatingError) {
+          subscriber.error(this.terminatingError);
+        } else {
+          subscriber.complete();
+        }
+        return () => {};
+      }
+
+      subscriber.next(this.currentValue);
+
+      this.subscribers.add(subscriber);
+      return () => {
+        this.subscribers.delete(subscriber);
+      };
+    });
+  }
+
+  private readonly subscribers = new Set<
+    ZenObservable.SubscriptionObserver<T>
+  >();
+
+  [Symbol.observable]() {
+    return this;
+  }
+
+  get closed() {
+    return this.isClosed;
+  }
+
+  next(value: T) {
+    if (this.isClosed) {
+      throw new Error('BehaviorSubject is closed');
+    }
+    this.currentValue = value;
+    this.subscribers.forEach(subscriber => subscriber.next(value));
+  }
+
+  error(error: Error) {
+    if (this.isClosed) {
+      throw new Error('BehaviorSubject is closed');
+    }
+    this.isClosed = true;
+    this.terminatingError = error;
+    this.subscribers.forEach(subscriber => subscriber.error(error));
+  }
+
+  complete() {
+    if (this.isClosed) {
+      throw new Error('BehaviorSubject is closed');
+    }
+    this.isClosed = true;
+    this.subscribers.forEach(subscriber => subscriber.complete());
+  }
+
+  subscribe(observer: ZenObservable.Observer<T>): ZenObservable.Subscription;
+  subscribe(
+    onNext: (value: T) => void,
+    onError?: (error: any) => void,
+    onComplete?: () => void,
+  ): ZenObservable.Subscription;
+  subscribe(
+    onNext: ZenObservable.Observer<T> | ((value: T) => void),
+    onError?: (error: any) => void,
+    onComplete?: () => void,
+  ): ZenObservable.Subscription {
+    const observer =
+      typeof onNext === 'function'
+        ? {
+            next: onNext,
+            error: onError,
+            complete: onComplete,
+          }
+        : onNext;
+
+    return this.observable.subscribe(observer);
+  }
+}
+
 /**
  * @alpha
  */
 export class DefaultSystemMetadataService implements SystemMetadataService {
-  private instances: BackstageInstance[];
+  private instance$: BehaviorSubject<BackstageInstance[]>;
   constructor(
     private options: { logger: LoggerService; config: RootConfigService },
   ) {
@@ -60,9 +168,9 @@ export class DefaultSystemMetadataService implements SystemMetadataService {
       }
       return instances;
     };
-    this.instances = getInstances();
+    this.instance$ = new BehaviorSubject(getInstances());
     this.options.config.subscribe?.(() => {
-      this.instances = getInstances();
+      this.instance$.next(getInstances());
     });
   }
 
@@ -73,9 +181,7 @@ export class DefaultSystemMetadataService implements SystemMetadataService {
     return new DefaultSystemMetadataService(pluginEnv);
   }
 
-  async introspect() {
-    return {
-      instances: this.instances,
-    };
+  instances(): Observable<BackstageInstance[]> {
+    return this.instance$;
   }
 }

packages/backend-defaults/src/entrypoints/systemMetadata/lib/createSystemMetadataRouter.ts

@@ -16,7 +16,10 @@
 
 import { LoggerService } from '@backstage/backend-plugin-api';
 import { BackendFeatureMeta } from '@backstage/backend-plugin-api/alpha';
-import type { SystemMetadataService } from '@backstage/backend-plugin-api';
+import type {
+  BackstageInstance,
+  SystemMetadataService,
+} from '@backstage/backend-plugin-api';
 import Router from 'express-promise-router';
 
 export async function createSystemMetadataRouter(options: {
@@ -25,23 +28,22 @@ export async function createSystemMetadataRouter(options: {
 }) {
   const { logger, systemMetadata } = options;
 
-  async function getInstances() {
-    const instances = await systemMetadata.introspect();
-    return instances.instances;
-  }
+  let instances: BackstageInstance[] = [];
+  systemMetadata.instances().subscribe({
+    next: value => {
+      instances = value;
+    },
+  });
 
-  logger.info(
-    `Instances in this system: ${JSON.stringify(await getInstances())}`,
-  );
+  logger.info(`Instances in this system: ${JSON.stringify(instances)}`);
 
   const router = Router();
 
   router.get('/instances', async (_, res) => {
-    res.json(await getInstances());
+    res.json({ items: instances });
   });
 
   router.get('/features/installed', async (_, res) => {
-    const instances = await getInstances();
     const featurePromises = await Promise.allSettled(
       instances.map(async instance => {
         const response = await fetch(