From 81b609f01049ef8902bd47c405eabc33460731e1 Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Fri, 30 Apr 2021 09:55:44 +0200
Subject: [PATCH 01/12] chore: test for getProcessableEntities

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 .../DefaultProcessingDatabase.test.ts         | 44 +++++++++++++++++++
 1 file changed, 44 insertions(+)

diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
index d76bd74260..ae17746799 100644
--- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
@@ -514,4 +514,48 @@ describe('Default Processing Database', () => {
       });
     });
   });
+
+  describe('getProcessableEntities', () => {
+    it('should return entities to process', async () => {
+      const entity = JSON.stringify({
+        apiVersion: '1.0.0',
+        metadata: {
+          name: 'xyz',
+        },
+        kind: 'Location',
+      } as Entity);
+      await db<DbRefreshStateRow>('refresh_state').insert({
+        entity_id: '2',
+        entity_ref: 'location:default/new-root',
+        unprocessed_entity: entity,
+        errors: '',
+        next_update_at: '2019-01-01 23:00:00',
+        last_discovery_at: 'now()',
+      });
+      await db<DbRefreshStateRow>('refresh_state').insert({
+        entity_id: '1',
+        entity_ref: 'location:default/foobar',
+        unprocessed_entity: entity,
+        errors: '',
+        next_update_at: '2042-01-01 23:00:00',
+        last_discovery_at: 'now()',
+      });
+
+      await processingDatabase.transaction(async tx => {
+        // request two items but only one can be processed.
+        const result = await processingDatabase.getProcessableEntities(tx, {
+          processBatchSize: 2,
+        });
+        expect(result.items.length).toEqual(1);
+        expect(result.items[0].entityRef).toEqual('location:default/new-root');
+
+        // should not return the same item as there's nothing left to process.
+        await expect(
+          processingDatabase.getProcessableEntities(tx, {
+            processBatchSize: 2,
+          }),
+        ).resolves.toEqual({ items: [] });
+      });
+    });
+  });
 });

From f0f77473e0a75b45014e8a40ae5dd0d989f8eda6 Mon Sep 17 00:00:00 2001
From: blam <ben@blam.sh>
Date: Thu, 29 Apr 2021 17:51:42 +0200
Subject: [PATCH 02/12] chore: add some more tests for TransactionValue

Signed-off-by: blam <ben@blam.sh>
---
 .../src/next/Context/TransactionValue.test.ts | 37 +++++++++++++++++++
 1 file changed, 37 insertions(+)
 create mode 100644 plugins/catalog-backend/src/next/Context/TransactionValue.test.ts

diff --git a/plugins/catalog-backend/src/next/Context/TransactionValue.test.ts b/plugins/catalog-backend/src/next/Context/TransactionValue.test.ts
new file mode 100644
index 0000000000..97e554d0f6
--- /dev/null
+++ b/plugins/catalog-backend/src/next/Context/TransactionValue.test.ts
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2021 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { TransactionValue } from './TransactionValue';
+import { Knex } from 'knex';
+import { BackgroundContext } from './BackgroundContext';
+
+describe('TransactionValue Context', () => {
+  it('should be able to store tx values and retrieve them from a context', () => {
+    const tx = {} as Knex.Transaction;
+    const ctx = new BackgroundContext();
+
+    const nextCtx = TransactionValue.in(ctx, tx);
+
+    expect(TransactionValue.from(nextCtx)).toBe(tx);
+  });
+
+  it('should throw when there is no tx value in the context', () => {
+    const ctx = new BackgroundContext();
+
+    expect(() => TransactionValue.from(ctx)).toThrow(
+      /No transaction available in context/,
+    );
+  });
+});

From 8ecfa8e1ed494906068cbe52849d45b126309d20 Mon Sep 17 00:00:00 2001
From: blam <ben@blam.sh>
Date: Fri, 30 Apr 2021 09:53:52 +0200
Subject: [PATCH 03/12] chore: added some moar tests for ProcessingDatabase

Signed-off-by: blam <ben@blam.sh>
---
 .../DefaultProcessingDatabase.test.ts         | 266 +++++++++++-------
 .../database/DefaultProcessingDatabase.ts     |   1 +
 2 files changed, 170 insertions(+), 97 deletions(-)

diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
index ae17746799..46545771c0 100644
--- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
@@ -23,15 +23,26 @@ import {
   DefaultProcessingDatabase,
 } from './DefaultProcessingDatabase';
 
-import { Entity, EntityRelationSpec } from '@backstage/catalog-model';
+import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
 import * as uuid from 'uuid';
 import { getVoidLogger } from '@backstage/backend-common';
+import { JsonObject } from '@backstage/config';
 
 describe('Default Processing Database', () => {
   let db: Knex;
   let processingDatabase: DefaultProcessingDatabase;
   const logger = getVoidLogger();
 
+  const insertRefRow = async (ref: DbRefreshStateReferencesRow) => {
+    return db<DbRefreshStateReferencesRow>('refresh_state_references').insert(
+      ref,
+    );
+  };
+
+  const insertRefreshStateRow = async (ref: DbRefreshStateRow) => {
+    await db<DbRefreshStateRow>('refresh_state').insert(ref);
+  };
+
   beforeEach(async () => {
     db = await DatabaseManager.createTestDatabaseConnection();
     await DatabaseManager.createDatabase(db);
@@ -39,17 +50,160 @@ describe('Default Processing Database', () => {
     processingDatabase = new DefaultProcessingDatabase(db, logger);
   });
 
-  describe('replaceUnprocessedEntities', () => {
-    const insertRefRow = async (ref: DbRefreshStateReferencesRow) => {
-      return db<DbRefreshStateReferencesRow>('refresh_state_references').insert(
-        ref,
+  describe('updateProcessedEntity', () => {
+    let id: string;
+    let processedEntity: Entity;
+
+    beforeEach(() => {
+      id = uuid.v4();
+      processedEntity = {
+        apiVersion: '1',
+        kind: 'Location',
+        metadata: {
+          name: 'fakelocation',
+        },
+        spec: {
+          type: 'url',
+          target: 'somethingelse',
+        },
+      };
+    });
+
+    it('fails when there is no processing state for the entity', async () => {
+      await processingDatabase.transaction(async tx => {
+        await expect(() =>
+          processingDatabase.updateProcessedEntity(tx, {
+            id,
+            processedEntity,
+            state: new Map<string, JsonObject>(),
+            relations: [],
+            deferredEntities: [],
+          }),
+        ).rejects.toThrow(`Processing state not found for ${id}`);
+      });
+    });
+
+    it('updates the refresh state entry with the cache, processed entity and errors', async () => {
+      await insertRefreshStateRow({
+        entity_id: id,
+        entity_ref: 'location:default/fakelocation',
+        unprocessed_entity: '{}',
+        processed_entity: '{}',
+        errors: '',
+        next_update_at: 'now()',
+        last_discovery_at: 'now()',
+      });
+
+      const state = new Map<string, JsonObject>();
+      state.set('hello', { t: 'something' });
+
+      await processingDatabase.transaction(tx =>
+        processingDatabase.updateProcessedEntity(tx, {
+          id,
+          processedEntity,
+          state,
+          relations: [],
+          deferredEntities: [],
+          errors: 'something broke',
+        }),
       );
-    };
 
-    const insertRefreshStateRow = async (ref: DbRefreshStateRow) => {
-      await db<DbRefreshStateRow>('refresh_state').insert(ref);
-    };
+      const [entity] = await db<DbRefreshStateRow>('refresh_state').select();
 
+      expect(entity.processed_entity).toEqual(JSON.stringify(processedEntity));
+      expect(entity.cache).toEqual(JSON.stringify(state));
+      expect(entity.errors).toEqual('something broke');
+    });
+
+    it('removes old relations and stores the new relationships', async () => {
+      await insertRefreshStateRow({
+        entity_id: id,
+        entity_ref: 'location:default/fakelocation',
+        unprocessed_entity: '{}',
+        processed_entity: '{}',
+        errors: '',
+        next_update_at: 'now()',
+        last_discovery_at: 'now()',
+      });
+
+      const relations = [
+        {
+          source: {
+            kind: 'Component',
+            namespace: 'Default',
+            name: 'foo',
+          },
+          target: {
+            kind: 'Component',
+            namespace: 'Default',
+            name: 'foo',
+          },
+          type: 'memberOf',
+        },
+      ];
+
+      await processingDatabase.transaction(tx =>
+        processingDatabase.updateProcessedEntity(tx, {
+          id,
+          processedEntity,
+          state: new Map<string, JsonObject>(),
+          relations: relations,
+          deferredEntities: [],
+        }),
+      );
+
+      const [savedRelations] = await db<DbRelationsRow>('relations')
+        .where({ originating_entity_id: id })
+        .select();
+
+      expect(savedRelations).toEqual({
+        originating_entity_id: id,
+        source_entity_ref: 'component:default/foo',
+        type: 'memberOf',
+        target_entity_ref: 'component:default/foo',
+      });
+    });
+
+    it('adds deferred entities to the the refresh_state table to be picked up later', async () => {
+      await insertRefreshStateRow({
+        entity_id: id,
+        entity_ref: 'location:default/fakelocation',
+        unprocessed_entity: '{}',
+        processed_entity: '{}',
+        errors: '',
+        next_update_at: 'now()',
+        last_discovery_at: 'now()',
+      });
+
+      const deferredEntities = [
+        {
+          apiVersion: '1',
+          kind: 'Location',
+          metadata: {
+            name: 'next',
+          },
+        },
+      ];
+
+      await processingDatabase.transaction(tx =>
+        processingDatabase.updateProcessedEntity(tx, {
+          id,
+          processedEntity,
+          state: new Map<string, JsonObject>(),
+          relations: [],
+          deferredEntities,
+        }),
+      );
+
+      const refreshStateEntries = await db<DbRefreshStateRow>('refresh_state')
+        .where({ entity_ref: stringifyEntityRef(deferredEntities[0]) })
+        .select();
+
+      expect(refreshStateEntries).toHaveLength(1);
+    });
+  });
+
+  describe('replaceUnprocessedEntities', () => {
     const createLocations = async (entityRefs: string[]) => {
       for (const ref of entityRefs) {
         await insertRefreshStateRow({
@@ -101,8 +255,8 @@ describe('Default Processing Database', () => {
         target_entity_ref: 'location:default/root-2',
       });
 
-      await processingDatabase.transaction(async tx => {
-        await processingDatabase.replaceUnprocessedEntities(tx, {
+      await processingDatabase.transaction(tx =>
+        processingDatabase.replaceUnprocessedEntities(tx, {
           type: 'full',
           sourceKey: 'config',
           items: [
@@ -114,8 +268,8 @@ describe('Default Processing Database', () => {
               kind: 'Location',
             } as Entity,
           ],
-        });
-      });
+        }),
+      );
 
       const currentRefreshState = await db<DbRefreshStateRow>(
         'refresh_state',
@@ -431,90 +585,6 @@ describe('Default Processing Database', () => {
     });
   });
 
-  describe('updateProcessedEntity', () => {
-    it('should throw if the entity does not exist', async () => {
-      await processingDatabase.transaction(async tx => {
-        await expect(
-          processingDatabase.updateProcessedEntity(tx, {
-            id: '9',
-            processedEntity: {
-              apiVersion: '1.0.0',
-              metadata: {
-                name: 'new-root',
-              },
-              kind: 'Location',
-            } as Entity,
-            deferredEntities: [],
-            relations: [],
-          }),
-        ).rejects.toThrow('Processing state not found for 9');
-      });
-    });
-
-    it('should update a processed entity', async () => {
-      await db<DbRefreshStateRow>('refresh_state').insert({
-        entity_id: '321',
-        entity_ref: 'location:default/new-root',
-        unprocessed_entity: '',
-        errors: '',
-        next_update_at: 'now()',
-        last_discovery_at: 'now()',
-      });
-
-      const deferredEntity = {
-        apiVersion: '1.0.0',
-        metadata: {
-          name: 'deferred',
-        },
-        kind: 'Location',
-      } as Entity;
-
-      const relation: EntityRelationSpec = {
-        source: {
-          kind: 'Component',
-          namespace: 'Default',
-          name: 'foo',
-        },
-        target: {
-          kind: 'Component',
-          namespace: 'Default',
-          name: 'foo',
-        },
-        type: 'url',
-      };
-
-      await processingDatabase.transaction(async tx => {
-        await processingDatabase.updateProcessedEntity(tx, {
-          id: '321',
-          processedEntity: {
-            apiVersion: '1.0.0',
-            metadata: {
-              name: 'new-root',
-            },
-            kind: 'Location',
-          } as Entity,
-          deferredEntities: [deferredEntity],
-          relations: [relation],
-        });
-      });
-
-      const deferredResult = await db<DbRefreshStateRow>('refresh_state')
-        .where({ entity_ref: 'location:default/deferred' })
-        .select();
-      expect(deferredResult.length).toBe(1);
-
-      const [relations] = await db<DbRelationsRow>('relations')
-        .where({ originating_entity_id: '321' })
-        .select();
-      expect(relations).toEqual({
-        originating_entity_id: '321',
-        source_entity_ref: 'component:default/foo',
-        type: 'url',
-        target_entity_ref: 'component:default/foo',
-      });
-    });
-  });
-
   describe('getProcessableEntities', () => {
     it('should return entities to process', async () => {
       const entity = JSON.stringify({
@@ -524,6 +594,7 @@ describe('Default Processing Database', () => {
         },
         kind: 'Location',
       } as Entity);
+
       await db<DbRefreshStateRow>('refresh_state').insert({
         entity_id: '2',
         entity_ref: 'location:default/new-root',
@@ -532,6 +603,7 @@ describe('Default Processing Database', () => {
         next_update_at: '2019-01-01 23:00:00',
         last_discovery_at: 'now()',
       });
+
       await db<DbRefreshStateRow>('refresh_state').insert({
         entity_id: '1',
         entity_ref: 'location:default/foobar',
diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
index dc5e6036f5..984e7776ea 100644
--- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
@@ -87,6 +87,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       .update({
         processed_entity: JSON.stringify(processedEntity),
         cache: JSON.stringify(state),
+        // TODO: should this also be JSON stringfy?
         errors,
       })
       .where('entity_id', id);

From 45ca6c76992fa8f86e9977222f54cdbb46a89247 Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 10:53:36 +0200
Subject: [PATCH 04/12] Refactor StateManager into the ProcessingEngine

And add some tests

Co-authored-by: Ben Lambert <ben@blam.sh>
Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 packages/backend/src/plugins/catalog.ts       |   2 +-
 .../DefaultCatalogProcessingEngine.test.ts    |  95 +++++++++++++++
 .../next/DefaultCatalogProcessingEngine.ts    | 112 ++++++++++--------
 .../src/next/DefaultProcessingStateManager.ts |  69 -----------
 .../src/next/NextCatalogBuilder.ts            |   5 +-
 plugins/catalog-backend/src/next/types.ts     |  34 ------
 6 files changed, 160 insertions(+), 157 deletions(-)
 create mode 100644 plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
 delete mode 100644 plugins/catalog-backend/src/next/DefaultProcessingStateManager.ts

diff --git a/packages/backend/src/plugins/catalog.ts b/packages/backend/src/plugins/catalog.ts
index e58d8cd3eb..7651836e2a 100644
--- a/packages/backend/src/plugins/catalog.ts
+++ b/packages/backend/src/plugins/catalog.ts
@@ -42,7 +42,7 @@ export default async function createPlugin(
     } = await builder.build();
 
     // TODO(jhaals): run and manage in background.
-    processingEngine.start();
+    await processingEngine.start();
 
     return await createRouter({
       entitiesCatalog,
diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
new file mode 100644
index 0000000000..68ae8150d1
--- /dev/null
+++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2021 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { getVoidLogger } from '@backstage/backend-common';
+import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
+
+import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
+import { Stitcher } from './Stitcher';
+import { CatalogProcessingOrchestrator } from './types';
+
+describe('DefaultCatalogProcessingEngine', () => {
+  const db = ({
+    transaction: jest.fn(),
+    getProcessableEntities: jest.fn(),
+    updateProcessedEntity: jest.fn(),
+  } as unknown) as jest.Mocked<DefaultProcessingDatabase>;
+  const orchestrator: jest.Mocked<CatalogProcessingOrchestrator> = {
+    process: jest.fn(),
+  };
+  const stitcher = ({
+    stitch: jest.fn(),
+  } as unknown) as jest.Mocked<Stitcher>;
+
+  beforeEach(() => {
+    jest.resetAllMocks();
+  });
+  it('should process stuff', async () => {
+    orchestrator.process.mockResolvedValue({
+      ok: true,
+      completedEntity: {
+        apiVersion: '1',
+        kind: 'Location',
+        metadata: { name: 'test' },
+      },
+      relations: [],
+      errors: [],
+      deferredEntities: [],
+      state: new Map(),
+    });
+    const engine = new DefaultCatalogProcessingEngine(
+      getVoidLogger(),
+      [],
+      db,
+      orchestrator,
+      stitcher,
+    );
+
+    db.transaction.mockImplementation(cb => cb((() => {}) as any));
+
+    db.getProcessableEntities
+      .mockImplementation(async () => {
+        await engine.stop();
+        return { items: [] };
+      })
+      .mockResolvedValueOnce({
+        items: [
+          {
+            entityRef: 'foo',
+            id: '1',
+            unprocessedEntity: {
+              apiVersion: '1',
+              kind: 'Location',
+              metadata: { name: 'test' },
+            },
+            state: new Map(),
+            nextUpdateAt: '',
+            lastDiscoveryAt: '',
+          },
+        ],
+      });
+
+    await engine.start();
+    await new Promise<void>(resolve => setTimeout(resolve, 1));
+    expect(orchestrator.process).toBeCalledTimes(1);
+    expect(orchestrator.process).toBeCalledWith({
+      entity: { apiVersion: '1', kind: 'Location', metadata: { name: 'test' } },
+      state: expect.anything(),
+    });
+
+    await engine.stop();
+  });
+});
diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
index e724fe79d8..c8c49e4bd9 100644
--- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
@@ -16,6 +16,7 @@
 
 import { stringifyEntityRef } from '@backstage/catalog-model';
 import { Logger } from 'winston';
+import { ProcessingDatabase } from './database/types';
 import { Stitcher } from './Stitcher';
 import {
   CatalogProcessingEngine,
@@ -23,44 +24,46 @@ import {
   EntityProvider,
   EntityProviderConnection,
   EntityProviderMutation,
-  ProcessingStateManager,
 } from './types';
 
 class Connection implements EntityProviderConnection {
   constructor(
     private readonly config: {
-      stateManager: ProcessingStateManager;
+      processingDatabase: ProcessingDatabase;
       id: string;
     },
   ) {}
 
   async applyMutation(mutation: EntityProviderMutation): Promise<void> {
+    const db = this.config.processingDatabase;
     if (mutation.type === 'full') {
-      await this.config.stateManager.replaceProcessingItems({
-        sourceKey: this.config.id,
-        type: 'full',
-        items: mutation.entities,
+      db.transaction(async tx => {
+        await db.replaceUnprocessedEntities(tx, {
+          sourceKey: this.config.id,
+          type: 'full',
+          items: mutation.entities,
+        });
       });
-
       return;
     }
-
-    await this.config.stateManager.replaceProcessingItems({
-      sourceKey: this.config.id,
-      type: 'delta',
-      added: mutation.added,
-      removed: mutation.removed,
+    db.transaction(async tx => {
+      await db.replaceUnprocessedEntities(tx, {
+        sourceKey: this.config.id,
+        type: 'delta',
+        added: mutation.added,
+        removed: mutation.removed,
+      });
     });
   }
 }
 
 export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
-  private running: boolean = false;
+  private running = false;
 
   constructor(
     private readonly logger: Logger,
     private readonly entityProviders: EntityProvider[],
-    private readonly stateManager: ProcessingStateManager,
+    private readonly processingDatabase: ProcessingDatabase,
     private readonly orchestrator: CatalogProcessingOrchestrator,
     private readonly stitcher: Stitcher,
   ) {}
@@ -69,51 +72,60 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
     for (const provider of this.entityProviders) {
       await provider.connect(
         new Connection({
-          stateManager: this.stateManager,
+          processingDatabase: this.processingDatabase,
           id: provider.getProviderName(),
         }),
       );
     }
-
     this.running = true;
+    this.run();
+  }
 
+  private async run() {
     while (this.running) {
-      const {
-        id,
-        entity,
-        state: initialState,
-      } = await this.stateManager.getNextProcessingItem();
-
-      const result = await this.orchestrator.process({
-        entity,
-        state: initialState,
+      const { items } = await this.processingDatabase.transaction(async tx => {
+        return this.processingDatabase.getProcessableEntities(tx, {
+          processBatchSize: 1,
+        });
       });
 
-      for (const error of result.errors) {
-        this.logger.warn(error.message);
+      if (items.length) {
+        const { id, state, unprocessedEntity } = items[0];
+
+        const result = await this.orchestrator.process({
+          entity: unprocessedEntity,
+          state,
+        });
+        for (const error of result.errors) {
+          this.logger.warn(error.message);
+        }
+        if (!result.ok) {
+          return;
+        }
+
+        result.completedEntity.metadata.uid = id;
+        await this.processingDatabase.transaction(async tx => {
+          await this.processingDatabase.updateProcessedEntity(tx, {
+            id,
+            processedEntity: result.completedEntity,
+            state: result.state,
+            errors: JSON.stringify(result.errors),
+            relations: result.relations,
+            deferredEntities: result.deferredEntities,
+          });
+        });
+
+        const setOfThingsToStitch = new Set<string>([
+          stringifyEntityRef(result.completedEntity),
+          ...result.relations.map(relation =>
+            stringifyEntityRef(relation.source),
+          ),
+        ]);
+        await this.stitcher.stitch(setOfThingsToStitch);
+      } else {
+        // Wait one second before fetching next item.
+        await new Promise<void>(resolve => setTimeout(resolve, 1000));
       }
-
-      if (!result.ok) {
-        return;
-      }
-
-      result.completedEntity.metadata.uid = id;
-      await this.stateManager.setProcessingItemResult({
-        id,
-        entity: result.completedEntity,
-        state: result.state,
-        errors: result.errors,
-        relations: result.relations,
-        deferredEntities: result.deferredEntities,
-      });
-
-      const setOfThingsToStitch = new Set<string>([
-        stringifyEntityRef(result.completedEntity),
-        ...result.relations.map(relation =>
-          stringifyEntityRef(relation.source),
-        ),
-      ]);
-      await this.stitcher.stitch(setOfThingsToStitch);
     }
   }
 
diff --git a/plugins/catalog-backend/src/next/DefaultProcessingStateManager.ts b/plugins/catalog-backend/src/next/DefaultProcessingStateManager.ts
deleted file mode 100644
index 9c1d20ae9c..0000000000
--- a/plugins/catalog-backend/src/next/DefaultProcessingStateManager.ts
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Copyright 2021 Spotify AB
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import { ProcessingDatabase } from './database/types';
-import {
-  ProcessingItem,
-  ProcessingItemResult,
-  ProcessingStateManager,
-  ReplaceProcessingItemsRequest,
-} from './types';
-
-export class DefaultProcessingStateManager implements ProcessingStateManager {
-  constructor(private readonly db: ProcessingDatabase) {}
-
-  replaceProcessingItems(
-    request: ReplaceProcessingItemsRequest,
-  ): Promise<void> {
-    return this.db.transaction(async tx => {
-      await this.db.replaceUnprocessedEntities(tx, request);
-    });
-  }
-
-  async setProcessingItemResult(result: ProcessingItemResult) {
-    return this.db.transaction(async tx => {
-      await this.db.updateProcessedEntity(tx, {
-        id: result.id,
-        processedEntity: result.entity,
-        errors: JSON.stringify(result.errors),
-        state: result.state,
-        relations: result.relations,
-        deferredEntities: result.deferredEntities,
-      });
-    });
-  }
-
-  async getNextProcessingItem(): Promise<ProcessingItem> {
-    for (;;) {
-      const { items } = await this.db.transaction(async tx => {
-        return this.db.getProcessableEntities(tx, {
-          processBatchSize: 1,
-        });
-      });
-
-      if (items.length) {
-        const { id, state, unprocessedEntity } = items[0];
-        return {
-          id,
-          entity: unprocessedEntity,
-          state,
-        };
-      }
-
-      await new Promise<void>(resolve => setTimeout(resolve, 1000));
-    }
-  }
-}
diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts
index 10500fdc15..dc09013964 100644
--- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts
+++ b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts
@@ -69,7 +69,6 @@ import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'
 import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
 import { DefaultCatalogProcessingOrchestrator } from './DefaultCatalogProcessingOrchestrator';
 import { DefaultLocationStore } from './DefaultLocationStore';
-import { DefaultProcessingStateManager } from './DefaultProcessingStateManager';
 import { NextEntitiesCatalog } from './NextEntitiesCatalog';
 import { Stitcher } from './Stitcher';
 
@@ -252,7 +251,7 @@ export class NextCatalogBuilder {
     const db = new CommonDatabase(dbClient, logger);
 
     const processingDatabase = new DefaultProcessingDatabase(dbClient, logger);
-    const stateManager = new DefaultProcessingStateManager(processingDatabase);
+    // const stateManager = new DefaultProcessingStateManager(processingDatabase);
     const integrations = ScmIntegrations.fromConfig(config);
     const orchestrator = new DefaultCatalogProcessingOrchestrator({
       processors,
@@ -269,7 +268,7 @@ export class NextCatalogBuilder {
     const processingEngine = new DefaultCatalogProcessingEngine(
       logger,
       [locationStore, configLocationProvider],
-      stateManager,
+      processingDatabase,
       orchestrator,
       stitcher,
     );
diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts
index 202576fddd..410066eafd 100644
--- a/plugins/catalog-backend/src/next/types.ts
+++ b/plugins/catalog-backend/src/next/types.ts
@@ -80,37 +80,3 @@ export type EntityProcessingResult =
 export interface CatalogProcessingOrchestrator {
   process(request: EntityProcessingRequest): Promise<EntityProcessingResult>;
 }
-
-export type ProcessingItemResult = {
-  id: string;
-  entity: Entity;
-  state: Map<string, JsonObject>;
-  errors: Error[];
-  relations: EntityRelationSpec[];
-  deferredEntities: Entity[];
-};
-
-export type ProcessingItem = {
-  id: string;
-  entity: Entity;
-  state: Map<string, JsonObject>;
-};
-
-export type ReplaceProcessingItemsRequest =
-  | {
-      sourceKey: string;
-      items: Entity[];
-      type: 'full';
-    }
-  | {
-      sourceKey: string;
-      added: Entity[];
-      removed: Entity[];
-      type: 'delta';
-    };
-
-export interface ProcessingStateManager {
-  setProcessingItemResult(result: ProcessingItemResult): Promise<void>;
-  getNextProcessingItem(): Promise<ProcessingItem>;
-  replaceProcessingItems(request: ReplaceProcessingItemsRequest): Promise<void>;
-}

From 0bd4dfa9d2c1de30ab54a038393466b7b3b59d8a Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 13:52:05 +0200
Subject: [PATCH 05/12] Fix review comments

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 .../DefaultProcessingDatabase.test.ts         | 54 ++++++++++---------
 1 file changed, 28 insertions(+), 26 deletions(-)

diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
index 46545771c0..7a0094dbf0 100644
--- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
@@ -89,9 +89,9 @@ describe('Default Processing Database', () => {
         entity_ref: 'location:default/fakelocation',
         unprocessed_entity: '{}',
         processed_entity: '{}',
-        errors: '',
-        next_update_at: 'now()',
-        last_discovery_at: 'now()',
+        errors: '[]',
+        next_update_at: '2021-04-01 13:37:00',
+        last_discovery_at: '2021-04-01 13:37:00',
       });
 
       const state = new Map<string, JsonObject>();
@@ -104,15 +104,17 @@ describe('Default Processing Database', () => {
           state,
           relations: [],
           deferredEntities: [],
-          errors: 'something broke',
+          errors: "['something broke']",
         }),
       );
 
-      const [entity] = await db<DbRefreshStateRow>('refresh_state').select();
-
-      expect(entity.processed_entity).toEqual(JSON.stringify(processedEntity));
-      expect(entity.cache).toEqual(JSON.stringify(state));
-      expect(entity.errors).toEqual('something broke');
+      const entities = await db<DbRefreshStateRow>('refresh_state').select();
+      expect(entities.length).toBe(1);
+      expect(entities[0].processed_entity).toEqual(
+        JSON.stringify(processedEntity),
+      );
+      expect(entities[0].cache).toEqual(JSON.stringify(state));
+      expect(entities[0].errors).toEqual("['something broke']");
     });
 
     it('removes old relations and stores the new relationships', async () => {
@@ -121,9 +123,9 @@ describe('Default Processing Database', () => {
         entity_ref: 'location:default/fakelocation',
         unprocessed_entity: '{}',
         processed_entity: '{}',
-        errors: '',
-        next_update_at: 'now()',
-        last_discovery_at: 'now()',
+        errors: '[]',
+        next_update_at: '2021-04-01 13:37:00',
+        last_discovery_at: '2021-04-01 13:37:00',
       });
 
       const relations = [
@@ -152,11 +154,11 @@ describe('Default Processing Database', () => {
         }),
       );
 
-      const [savedRelations] = await db<DbRelationsRow>('relations')
+      const savedRelations = await db<DbRelationsRow>('relations')
         .where({ originating_entity_id: id })
         .select();
-
-      expect(savedRelations).toEqual({
+      expect(savedRelations.length).toBe(1);
+      expect(savedRelations[0]).toEqual({
         originating_entity_id: id,
         source_entity_ref: 'component:default/foo',
         type: 'memberOf',
@@ -170,9 +172,9 @@ describe('Default Processing Database', () => {
         entity_ref: 'location:default/fakelocation',
         unprocessed_entity: '{}',
         processed_entity: '{}',
-        errors: '',
-        next_update_at: 'now()',
-        last_discovery_at: 'now()',
+        errors: '[]',
+        next_update_at: '2021-04-01 13:37:00',
+        last_discovery_at: '2021-04-01 13:37:00',
       });
 
       const deferredEntities = [
@@ -211,9 +213,9 @@ describe('Default Processing Database', () => {
           entity_ref: ref,
           unprocessed_entity: '{}',
           processed_entity: '{}',
-          errors: '',
-          next_update_at: 'now()',
-          last_discovery_at: 'now()',
+          errors: '[]',
+          next_update_at: '2021-04-01 13:37:00',
+          last_discovery_at: '2021-04-01 13:37:00',
         });
       }
     };
@@ -588,29 +590,29 @@ describe('Default Processing Database', () => {
   describe('getProcessableEntities', () => {
     it('should return entities to process', async () => {
       const entity = JSON.stringify({
+        kind: 'Location',
         apiVersion: '1.0.0',
         metadata: {
           name: 'xyz',
         },
-        kind: 'Location',
       } as Entity);
 
       await db<DbRefreshStateRow>('refresh_state').insert({
         entity_id: '2',
         entity_ref: 'location:default/new-root',
         unprocessed_entity: entity,
-        errors: '',
+        errors: '[]',
         next_update_at: '2019-01-01 23:00:00',
-        last_discovery_at: 'now()',
+        last_discovery_at: '2021-04-01 13:37:00',
       });
 
       await db<DbRefreshStateRow>('refresh_state').insert({
         entity_id: '1',
         entity_ref: 'location:default/foobar',
         unprocessed_entity: entity,
-        errors: '',
+        errors: '[]',
         next_update_at: '2042-01-01 23:00:00',
-        last_discovery_at: 'now()',
+        last_discovery_at: '2021-04-01 13:37:00',
       });
 
       await processingDatabase.transaction(async tx => {

From 852b6bec95871b735cf2aacec3b8228a329dc7c0 Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 14:13:25 +0200
Subject: [PATCH 06/12] Wrap test in waitForExpect

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 plugins/catalog-backend/package.json            |  3 ++-
 .../next/DefaultCatalogProcessingEngine.test.ts | 17 +++++++++++------
 yarn.lock                                       | 16 +++++++++++++---
 3 files changed, 26 insertions(+), 10 deletions(-)

diff --git a/plugins/catalog-backend/package.json b/plugins/catalog-backend/package.json
index 7649067a44..99df0acd09 100644
--- a/plugins/catalog-backend/package.json
+++ b/plugins/catalog-backend/package.json
@@ -73,7 +73,8 @@
     "@types/yup": "^0.29.8",
     "msw": "^0.21.2",
     "sqlite3": "^5.0.0",
-    "supertest": "^6.1.3"
+    "supertest": "^6.1.3",
+    "wait-for-expect": "^3.0.2"
   },
   "files": [
     "dist",
diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
index 68ae8150d1..b6363a697d 100644
--- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
+++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
@@ -20,6 +20,7 @@ import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'
 import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
 import { Stitcher } from './Stitcher';
 import { CatalogProcessingOrchestrator } from './types';
+import waitForExpect from 'wait-for-expect';
 
 describe('DefaultCatalogProcessingEngine', () => {
   const db = ({
@@ -83,13 +84,17 @@ describe('DefaultCatalogProcessingEngine', () => {
       });
 
     await engine.start();
-    await new Promise<void>(resolve => setTimeout(resolve, 1));
-    expect(orchestrator.process).toBeCalledTimes(1);
-    expect(orchestrator.process).toBeCalledWith({
-      entity: { apiVersion: '1', kind: 'Location', metadata: { name: 'test' } },
-      state: expect.anything(),
+    await waitForExpect(() => {
+      expect(orchestrator.process).toBeCalledTimes(1);
+      expect(orchestrator.process).toBeCalledWith({
+        entity: {
+          apiVersion: '1',
+          kind: 'Location',
+          metadata: { name: 'test' },
+        },
+        state: expect.anything(),
+      });
     });
-
     await engine.stop();
   });
 });
diff --git a/yarn.lock b/yarn.lock
index bffef20f24..4d3c45ab5e 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -14482,7 +14482,7 @@ graphql-extensions@^0.12.8:
     apollo-server-env "^3.0.0"
     apollo-server-types "^0.6.3"
 
-graphql-language-service-interface@2.8.2, graphql-language-service-interface@^2.8.2:
+graphql-language-service-interface@^2.8.2:
   version "2.8.2"
   resolved "https://registry.npmjs.org/graphql-language-service-interface/-/graphql-language-service-interface-2.8.2.tgz#b3bb2aef7eaf0dff0b4ea419fa412c5f66fa268b"
   integrity sha512-otbOQmhgkAJU1QJgQkMztNku6SbJLu/uodoFOYOOtJsizTjrMs93vkYaHCcYnLA3oi1Goj27XcHjMnRCYQOZXQ==
@@ -14492,7 +14492,7 @@ graphql-language-service-interface@2.8.2, graphql-language-service-interface@^2.
     graphql-language-service-utils "^2.5.1"
     vscode-languageserver-types "^3.15.1"
 
-graphql-language-service-parser@1.9.0, graphql-language-service-parser@^1.9.0:
+graphql-language-service-parser@^1.9.0:
   version "1.9.0"
   resolved "https://registry.npmjs.org/graphql-language-service-parser/-/graphql-language-service-parser-1.9.0.tgz#79af21294119a0a7e81b6b994a1af36833bab724"
   integrity sha512-B5xPZLbBmIp0kHvpY1Z35I5DtPoDK9wGxQVRDIzcBaiIvAmlTrDvjo3bu7vKREdjFbYKvWNgrEWENuprMbF17Q==
@@ -25796,11 +25796,16 @@ typescript-json-schema@^0.49.0:
     typescript "^4.1.3"
     yargs "^16.2.0"
 
-typescript@^4.0.3, typescript@^4.1.3, typescript@~4.1.3:
+typescript@^4.0.3, typescript@^4.1.3:
   version "4.2.3"
   resolved "https://registry.npmjs.org/typescript/-/typescript-4.2.3.tgz#39062d8019912d43726298f09493d598048c1ce3"
   integrity sha512-qOcYwxaByStAWrBf4x0fibwZvMRG+r4cQoTjbPtUlrWjBHbmCAww1i448U0GJ+3cNNEtebDteo/cHOR3xJ4wEw==
 
+typescript@~4.1.3:
+  version "4.1.5"
+  resolved "https://registry.npmjs.org/typescript/-/typescript-4.1.5.tgz#123a3b214aaff3be32926f0d8f1f6e704eb89a72"
+  integrity sha512-6OSu9PTIzmn9TCDiovULTnET6BgXtDYL4Gg4szY+cGsc3JP1dQL8qvE8kShTRx1NIw4Q9IBHlwODjkjWEtMUyA==
+
 ua-parser-js@^0.7.18:
   version "0.7.21"
   resolved "https://registry.npmjs.org/ua-parser-js/-/ua-parser-js-0.7.21.tgz#853cf9ce93f642f67174273cc34565ae6f308777"
@@ -26444,6 +26449,11 @@ w3c-xmlserializer@^2.0.0:
   dependencies:
     xml-name-validator "^3.0.0"
 
+wait-for-expect@^3.0.2:
+  version "3.0.2"
+  resolved "https://registry.npmjs.org/wait-for-expect/-/wait-for-expect-3.0.2.tgz#d2f14b2f7b778c9b82144109c8fa89ceaadaa463"
+  integrity sha512-cfS1+DZxuav1aBYbaO/kE06EOS8yRw7qOFoD3XtjTkYvCvh3zUvNST8DXK/nPaeqIzIv3P3kL3lRJn8iwOiSag==
+
 wait-on@5.3.0:
   version "5.3.0"
   resolved "https://registry.npmjs.org/wait-on/-/wait-on-5.3.0.tgz#584e17d4b3fe7b46ac2b9f8e5e102c005c2776c7"

From fe84fb2fdcb355ad2e63c83d09add0b8ef9f1bc2 Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 14:14:57 +0200
Subject: [PATCH 07/12] Await transactions

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 .../src/next/DefaultCatalogProcessingEngine.ts                | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
index c8c49e4bd9..6bee54f153 100644
--- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
@@ -37,7 +37,7 @@ class Connection implements EntityProviderConnection {
   async applyMutation(mutation: EntityProviderMutation): Promise<void> {
     const db = this.config.processingDatabase;
     if (mutation.type === 'full') {
-      db.transaction(async tx => {
+      await db.transaction(async tx => {
         await db.replaceUnprocessedEntities(tx, {
           sourceKey: this.config.id,
           type: 'full',
@@ -46,7 +46,7 @@ class Connection implements EntityProviderConnection {
       });
       return;
     }
-    db.transaction(async tx => {
+    await db.transaction(async tx => {
       await db.replaceUnprocessedEntities(tx, {
         sourceKey: this.config.id,
         type: 'delta',

From e9d61bcbeba2d202be39b456e3ed46640ad1a91c Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 14:30:40 +0200
Subject: [PATCH 08/12] Ensure that processing continues despite errors
 Co-authored-by: Ben Lambert <ben@blam.sh>

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 .../DefaultCatalogProcessingEngine.test.ts    | 61 ++++++++++++
 .../next/DefaultCatalogProcessingEngine.ts    | 97 +++++++++++--------
 2 files changed, 118 insertions(+), 40 deletions(-)

diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
index b6363a697d..376af6e748 100644
--- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
+++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
@@ -97,4 +97,65 @@ describe('DefaultCatalogProcessingEngine', () => {
     });
     await engine.stop();
   });
+
+  it('should process stuff even if the first attempt fail', async () => {
+    orchestrator.process.mockResolvedValue({
+      ok: true,
+      completedEntity: {
+        apiVersion: '1',
+        kind: 'Location',
+        metadata: { name: 'test' },
+      },
+      relations: [],
+      errors: [],
+      deferredEntities: [],
+      state: new Map(),
+    });
+    const engine = new DefaultCatalogProcessingEngine(
+      getVoidLogger(),
+      [],
+      db,
+      orchestrator,
+      stitcher,
+    );
+
+    db.transaction.mockImplementation(cb => cb((() => {}) as any));
+
+    db.getProcessableEntities
+      .mockImplementation(async () => {
+        await engine.stop();
+        return { items: [] };
+      })
+      .mockRejectedValueOnce(new Error('I FAILED'))
+      .mockResolvedValueOnce({
+        items: [
+          {
+            entityRef: 'foo',
+            id: '1',
+            unprocessedEntity: {
+              apiVersion: '1',
+              kind: 'Location',
+              metadata: { name: 'test' },
+            },
+            state: new Map(),
+            nextUpdateAt: '',
+            lastDiscoveryAt: '',
+          },
+        ],
+      });
+
+    await engine.start();
+    await waitForExpect(() => {
+      expect(orchestrator.process).toBeCalledTimes(1);
+      expect(orchestrator.process).toBeCalledWith({
+        entity: {
+          apiVersion: '1',
+          kind: 'Location',
+          metadata: { name: 'test' },
+        },
+        state: expect.anything(),
+      });
+    });
+    await engine.stop();
+  });
 });
diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
index 6bee54f153..877c3075aa 100644
--- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
@@ -83,52 +83,69 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
   private async run() {
     while (this.running) {
-      const { items } = await this.processingDatabase.transaction(async tx => {
-        return this.processingDatabase.getProcessableEntities(tx, {
-          processBatchSize: 1,
+      try {
+        // TODO: We want to disconnect the queue popping and message processing
+        // so that if the queue popping fails we exponentially back off in order to give the DB room to sort itself out.
+        await this.process();
+      } catch (e) {
+        this.logger.warn('Processing failed with:', e);
+        // TODO: this can be a little smarter as mentioned in the above comment.
+        // But for now, if something fails, wait a brief time to pick up the next message.
+        await this.wait();
+      }
+    }
+  }
+
+  private async process() {
+    const { items } = await this.processingDatabase.transaction(async tx => {
+      return this.processingDatabase.getProcessableEntities(tx, {
+        processBatchSize: 1,
+      });
+    });
+
+    if (items.length) {
+      const { id, state, unprocessedEntity } = items[0];
+
+      const result = await this.orchestrator.process({
+        entity: unprocessedEntity,
+        state,
+      });
+      for (const error of result.errors) {
+        this.logger.warn(error.message);
+      }
+      if (!result.ok) {
+        return;
+      }
+
+      result.completedEntity.metadata.uid = id;
+      await this.processingDatabase.transaction(async tx => {
+        await this.processingDatabase.updateProcessedEntity(tx, {
+          id,
+          processedEntity: result.completedEntity,
+          state: result.state,
+          errors: JSON.stringify(result.errors),
+          relations: result.relations,
+          deferredEntities: result.deferredEntities,
         });
       });
 
-      if (items.length) {
-        const { id, state, unprocessedEntity } = items[0];
-
-        const result = await this.orchestrator.process({
-          entity: unprocessedEntity,
-          state,
-        });
-        for (const error of result.errors) {
-          this.logger.warn(error.message);
-        }
-        if (!result.ok) {
-          return;
-        }
-
-        result.completedEntity.metadata.uid = id;
-        await this.processingDatabase.transaction(async tx => {
-          await this.processingDatabase.updateProcessedEntity(tx, {
-            id,
-            processedEntity: result.completedEntity,
-            state: result.state,
-            errors: JSON.stringify(result.errors),
-            relations: result.relations,
-            deferredEntities: result.deferredEntities,
-          });
-        });
-
-        const setOfThingsToStitch = new Set<string>([
-          stringifyEntityRef(result.completedEntity),
-          ...result.relations.map(relation =>
-            stringifyEntityRef(relation.source),
-          ),
-        ]);
-        await this.stitcher.stitch(setOfThingsToStitch);
-      } else {
-        // Wait one second before fetching next item.
-        await new Promise<void>(resolve => setTimeout(resolve, 1000));
-      }
+      const setOfThingsToStitch = new Set<string>([
+        stringifyEntityRef(result.completedEntity),
+        ...result.relations.map(relation =>
+          stringifyEntityRef(relation.source),
+        ),
+      ]);
+      await this.stitcher.stitch(setOfThingsToStitch);
+    } else {
+      // No items to process, wait and try again.
+      await this.wait();
     }
   }
 
+  private async wait() {
+    await new Promise<void>(resolve => setTimeout(resolve, 1000));
+  }
+
   async stop() {
     this.running = false;
   }

From cd33a8cd26b95182eedc8e274a61686f00fbc9bb Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 14:42:06 +0200
Subject: [PATCH 09/12] Reduce if block

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 .../next/DefaultCatalogProcessingEngine.ts    | 67 +++++++++----------
 1 file changed, 33 insertions(+), 34 deletions(-)

diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
index 877c3075aa..55cb8e2029 100644
--- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
@@ -103,43 +103,42 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       });
     });
 
-    if (items.length) {
-      const { id, state, unprocessedEntity } = items[0];
-
-      const result = await this.orchestrator.process({
-        entity: unprocessedEntity,
-        state,
-      });
-      for (const error of result.errors) {
-        this.logger.warn(error.message);
-      }
-      if (!result.ok) {
-        return;
-      }
-
-      result.completedEntity.metadata.uid = id;
-      await this.processingDatabase.transaction(async tx => {
-        await this.processingDatabase.updateProcessedEntity(tx, {
-          id,
-          processedEntity: result.completedEntity,
-          state: result.state,
-          errors: JSON.stringify(result.errors),
-          relations: result.relations,
-          deferredEntities: result.deferredEntities,
-        });
-      });
-
-      const setOfThingsToStitch = new Set<string>([
-        stringifyEntityRef(result.completedEntity),
-        ...result.relations.map(relation =>
-          stringifyEntityRef(relation.source),
-        ),
-      ]);
-      await this.stitcher.stitch(setOfThingsToStitch);
-    } else {
+    if (!items.length) {
       // No items to process, wait and try again.
       await this.wait();
+      return;
     }
+
+    const { id, state, unprocessedEntity } = items[0];
+
+    const result = await this.orchestrator.process({
+      entity: unprocessedEntity,
+      state,
+    });
+    for (const error of result.errors) {
+      this.logger.warn(error.message);
+    }
+    if (!result.ok) {
+      return;
+    }
+
+    result.completedEntity.metadata.uid = id;
+    await this.processingDatabase.transaction(async tx => {
+      await this.processingDatabase.updateProcessedEntity(tx, {
+        id,
+        processedEntity: result.completedEntity,
+        state: result.state,
+        errors: JSON.stringify(result.errors),
+        relations: result.relations,
+        deferredEntities: result.deferredEntities,
+      });
+    });
+
+    const setOfThingsToStitch = new Set<string>([
+      stringifyEntityRef(result.completedEntity),
+      ...result.relations.map(relation => stringifyEntityRef(relation.source)),
+    ]);
+    await this.stitcher.stitch(setOfThingsToStitch);
   }
 
   private async wait() {

From cb10731dfa13c558e68ed3a85dc8e107c32a770b Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 14:54:57 +0200
Subject: [PATCH 10/12] Drop references to deleted code

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 plugins/catalog-backend/src/next/NextCatalogBuilder.ts | 1 -
 1 file changed, 1 deletion(-)

diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts
index dc09013964..0158cc9fc9 100644
--- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts
+++ b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts
@@ -251,7 +251,6 @@ export class NextCatalogBuilder {
     const db = new CommonDatabase(dbClient, logger);
 
     const processingDatabase = new DefaultProcessingDatabase(dbClient, logger);
-    // const stateManager = new DefaultProcessingStateManager(processingDatabase);
     const integrations = ScmIntegrations.fromConfig(config);
     const orchestrator = new DefaultCatalogProcessingOrchestrator({
       processors,

From 641d36b148b1dab696ce144611f9d97466aef4d9 Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 3 May 2021 15:47:30 +0200
Subject: [PATCH 11/12] chore: delete todo comment

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 .../src/next/database/DefaultProcessingDatabase.ts               | 1 -
 1 file changed, 1 deletion(-)

diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
index 984e7776ea..dc5e6036f5 100644
--- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
@@ -87,7 +87,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       .update({
         processed_entity: JSON.stringify(processedEntity),
         cache: JSON.stringify(state),
-        // TODO: should this also be JSON stringfy?
         errors,
       })
       .where('entity_id', id);

From bfe77a9dbdea852780b6b1ead6aa1bb0d3e7615b Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Tue, 4 May 2021 09:41:06 +0200
Subject: [PATCH 12/12] fix yarn lock

Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 yarn.lock | 11 +++--------
 1 file changed, 3 insertions(+), 8 deletions(-)

diff --git a/yarn.lock b/yarn.lock
index 4d3c45ab5e..f778b2d5cf 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -14482,7 +14482,7 @@ graphql-extensions@^0.12.8:
     apollo-server-env "^3.0.0"
     apollo-server-types "^0.6.3"
 
-graphql-language-service-interface@^2.8.2:
+graphql-language-service-interface@2.8.2, graphql-language-service-interface@^2.8.2:
   version "2.8.2"
   resolved "https://registry.npmjs.org/graphql-language-service-interface/-/graphql-language-service-interface-2.8.2.tgz#b3bb2aef7eaf0dff0b4ea419fa412c5f66fa268b"
   integrity sha512-otbOQmhgkAJU1QJgQkMztNku6SbJLu/uodoFOYOOtJsizTjrMs93vkYaHCcYnLA3oi1Goj27XcHjMnRCYQOZXQ==
@@ -14492,7 +14492,7 @@ graphql-language-service-interface@^2.8.2:
     graphql-language-service-utils "^2.5.1"
     vscode-languageserver-types "^3.15.1"
 
-graphql-language-service-parser@^1.9.0:
+graphql-language-service-parser@1.9.0, graphql-language-service-parser@^1.9.0:
   version "1.9.0"
   resolved "https://registry.npmjs.org/graphql-language-service-parser/-/graphql-language-service-parser-1.9.0.tgz#79af21294119a0a7e81b6b994a1af36833bab724"
   integrity sha512-B5xPZLbBmIp0kHvpY1Z35I5DtPoDK9wGxQVRDIzcBaiIvAmlTrDvjo3bu7vKREdjFbYKvWNgrEWENuprMbF17Q==
@@ -25796,16 +25796,11 @@ typescript-json-schema@^0.49.0:
     typescript "^4.1.3"
     yargs "^16.2.0"
 
-typescript@^4.0.3, typescript@^4.1.3:
+typescript@^4.0.3, typescript@^4.1.3, typescript@~4.1.3:
   version "4.2.3"
   resolved "https://registry.npmjs.org/typescript/-/typescript-4.2.3.tgz#39062d8019912d43726298f09493d598048c1ce3"
   integrity sha512-qOcYwxaByStAWrBf4x0fibwZvMRG+r4cQoTjbPtUlrWjBHbmCAww1i448U0GJ+3cNNEtebDteo/cHOR3xJ4wEw==
 
-typescript@~4.1.3:
-  version "4.1.5"
-  resolved "https://registry.npmjs.org/typescript/-/typescript-4.1.5.tgz#123a3b214aaff3be32926f0d8f1f6e704eb89a72"
-  integrity sha512-6OSu9PTIzmn9TCDiovULTnET6BgXtDYL4Gg4szY+cGsc3JP1dQL8qvE8kShTRx1NIw4Q9IBHlwODjkjWEtMUyA==
-
 ua-parser-js@^0.7.18:
   version "0.7.21"
   resolved "https://registry.npmjs.org/ua-parser-js/-/ua-parser-js-0.7.21.tgz#853cf9ce93f642f67174273cc34565ae6f308777"