catalog-backend: WIP EntityProvider mutation interface

Co-authored-by: Fredrik Adelöw <freben@gmail.com>
Co-authored-by: blam <ben@blam.sh>
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts

@@ -19,6 +19,8 @@ import {
   CatalogProcessingEngine,
   EntityProvider,
   EntityMessage,
+  EntityProviderConnection,
+  EntityProviderMutation,
   ProcessingStateManager,
   CatalogProcessingOrchestrator,
 } from './types';
@@ -27,8 +29,13 @@ import { Logger } from 'winston';
 import { stringifyEntityRef } from '@backstage/catalog-model';
 import { Stitcher } from './Stitcher';
 
+class Connection implements EntityProviderConnection {
+  constructor(private readonly stateManager: ProcessingStateManager) {}
+
+  async applyMutation(mutation: EntityProviderMutation): Promise<void> {}
+}
+
 export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
-  private subscriptions: Subscription[] = [];
   private running: boolean = false;
 
   constructor(
@@ -41,11 +48,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
   async start() {
     for (const provider of this.entityProviders) {
-      const id = 'databaseProvider';
-      const subscription = provider
-        .entityChange$()
-        .subscribe({ next: m => this.onNext(id, m) });
-      this.subscriptions.push(subscription);
+      provider.connect(new Connection(this.stateManager));
     }
 
     this.running = true;

plugins/catalog-backend/src/next/DefaultLocationStore.ts

@@ -16,24 +16,29 @@
 
 import { LocationSpec, Location } from '@backstage/catalog-model';
 import { Database } from '../database';
-import { LocationStore } from './types';
+import {
+  LocationStore,
+  EntityProvider,
+  EntityProviderConnection,
+} from './types';
 import { v4 as uuidv4 } from 'uuid';
+import { locationSpecToLocationEntity } from './util';
 import { ConflictError } from '@backstage/errors';
-import { Observable } from '@backstage/core';
-import ObservableImpl from 'zen-observable';
 
 export type LocationMessage =
   | { all: Location[] }
   | { added: Location[]; removed: Location[] };
 
-export class DefaultLocationStore implements LocationStore {
-  private subscribers = new Set<
-    ZenObservable.SubscriptionObserver<LocationMessage>
-  >();
+export class DefaultLocationStore implements LocationStore, EntityProvider {
+  private _connection: EntityProviderConnection | undefined;
 
   constructor(private readonly db: Database) {}
 
   createLocation(spec: LocationSpec): Promise<Location> {
+    if (!this.connection) {
+      throw new Error('location store is not initialized');
+    }
+
     return this.db.transaction(async tx => {
       // TODO: id should really be type and target combined and not a uuid.
 
@@ -55,7 +60,11 @@ export class DefaultLocationStore implements LocationStore {
         target: spec.target,
       });
 
-      this.notifyAddition(location);
+      await this.connection.applyMutation({
+        type: 'delta',
+        added: [locationSpecToLocationEntity(location)],
+        removed: [],
+      });
 
       return location;
     });
@@ -75,40 +84,33 @@ export class DefaultLocationStore implements LocationStore {
   }
 
   deleteLocation(id: string): Promise<void> {
+    if (!this.connection) {
+      throw new Error('location store is not initialized');
+    }
+
     return this.db.transaction(async tx => {
       const location = await this.db.location(id);
       if (!location) {
         throw new ConflictError(`No location found with id: ${id}`);
       }
       await this.db.removeLocation(tx, id);
-      this.notifyDeletion(location);
-    });
-  }
-
-  private notifyAddition(location: Location) {
-    for (const subscriber of this.subscribers) {
-      subscriber.next({
-        added: [location],
-        removed: [],
-      });
-    }
-  }
-
-  private notifyDeletion(location: Location) {
-    for (const subscriber of this.subscribers) {
-      subscriber.next({
+      await this.connection.applyMutation({
+        type: 'delta',
         added: [],
-        removed: [location],
+        removed: [locationSpecToLocationEntity(location)],
       });
-    }
+    });
   }
 
-  location$(): Observable<LocationMessage> {
-    return new ObservableImpl<LocationMessage>(subscriber => {
-      this.subscribers.add(subscriber);
-      return () => {
-        this.subscribers.delete(subscriber);
-      };
-    });
+  private get connection(): EntityProviderConnection {
+    if (!this._connection) {
+      throw new Error('location store is not initialized');
+    }
+
+    return this._connection;
+  }
+
+  async connect(connection: EntityProviderConnection): Promise<void> {
+    this._connection = connection;
   }
 }

plugins/catalog-backend/src/next/NextCatalogBuilder.ts

@@ -272,11 +272,10 @@ export class NextCatalogBuilder {
     const entitiesCatalog = new NextEntitiesCatalog(dbClient);
 
     const locationStore = new DefaultLocationStore(db);
-    const dbLocationProvider = new DatabaseLocationProvider(locationStore);
     const stitcher = new Stitcher(dbClient, logger);
     const processingEngine = new DefaultCatalogProcessingEngine(
       logger,
-      [dbLocationProvider], // entityproviders
+      [locationStore], // entityproviders
       stateManager,
       orchestrator,
       stitcher,

plugins/catalog-backend/src/next/types.ts

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 import {
   Entity,
   EntityName,
@@ -21,7 +22,6 @@ import {
   EntityRelationSpec,
 } from '@backstage/catalog-model';
 import { JsonObject } from '@backstage/config';
-import { Observable } from '@backstage/core'; // << nooo
 
 export interface LocationEntity {
   apiVersion: 'backstage.io/v1alpha1';
@@ -45,20 +45,11 @@ export interface LocationService {
   deleteLocation(id: string): Promise<void>;
 }
 
-export type EntityMessage =
-  | { all: Entity[] }
-  | { added: Entity[]; removed: EntityName[] };
-
 export interface LocationStore {
-  // extends EntityProvider
   createLocation(spec: LocationSpec): Promise<Location>;
   listLocations(): Promise<Location[]>;
   getLocation(id: string): Promise<Location>;
   deleteLocation(id: string): Promise<void>;
-
-  location$(): Observable<
-    { all: Location[] } | { added: Location[]; removed: Location[] }
-  >;
 }
 
 export interface CatalogProcessingEngine {
@@ -66,8 +57,16 @@ export interface CatalogProcessingEngine {
   stop(): Promise<void>;
 }
 
+export type EntityProviderMutation =
+  | { type: 'full'; entities: Iterable<Entity> }
+  | { type: 'delta'; added: Iterable<Entity>; removed: Iterable<Entity> };
+
+export interface EntityProviderConnection {
+  applyMutation(mutation: EntityProviderMutation): Promise<void>;
+}
+
 export interface EntityProvider {
-  entityChange$(): Observable<EntityMessage>;
+  connect(connection: EntityProviderConnection): Promise<void>;
 }
 
 export type EntityProcessingRequest = {