Move pre-insertion to final_entity to stitcher

Signed-off-by: Johan Haals <johan.haals@gmail.com>

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

@@ -110,6 +110,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       return;
     }
 
+    // TODO: replace Promise.all with something more sophisticated for parallel processing.
     await Promise.all(
       items.map(async item => {
         const { id, state, unprocessedEntity } = item;

plugins/catalog-backend/src/next/Stitcher.test.ts

@@ -73,15 +73,6 @@ describe('Stitcher', () => {
       },
     ]);
 
-    await db<DbFinalEntitiesRow>('final_entities')
-      .insert({
-        entity_id: 'my-id',
-        hash: '',
-        stitch_ticket: '',
-      })
-      .onConflict('entity_id')
-      .ignore();
-
     await stitcher.stitch(new Set(['k:ns/n']));
 
     entities = await db<DbFinalEntitiesRow>('final_entities');

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

@@ -27,6 +27,7 @@ import { Knex } from 'knex';
 import { Logger } from 'winston';
 import { buildEntitySearch, DbSearchRow } from './search';
 import { v4 as uuid } from 'uuid';
+import { DbRefreshStateRow } from './database/DefaultProcessingDatabase';
 
 // The number of items that are sent per batch to the database layer, when
 // doing .batchInsert calls to knex. This needs to be low enough to not cause
@@ -61,21 +62,28 @@ export class Stitcher {
   async stitch(entityRefs: Set<string>) {
     for (const entityRef of entityRefs) {
       try {
-        const ticket = uuid();
-        const ticketRows = await this.database<DbFinalEntitiesRow>(
-          'final_entities',
+        const entityResult = await this.database<DbRefreshStateRow>(
+          'refresh_state',
         )
-          .update('stitch_ticket', ticket)
-          .whereIn('entity_id', qb =>
-            qb
-              .select('entity_id')
-              .from('refresh_state')
-              .where({ entity_ref: entityRef }),
-          );
-        if (ticketRows === 0) {
-          // No ticket written.
+          .where({ entity_ref: entityRef })
+          .limit(1)
+          .select('entity_id');
+        if (!entityResult.length) {
+          // Entity does no exist in refresh state table, no stitching required.
           continue;
         }
+
+        // Insert stitching ticket that will be compared before inserting the final entity.
+        const ticket = uuid();
+        await this.database<DbFinalEntitiesRow>('final_entities')
+          .insert({
+            entity_id: entityResult[0].entity_id,
+            hash: '',
+            stitch_ticket: ticket,
+          })
+          .onConflict('entity_id')
+          .merge(['stitch_ticket']);
+
         // Selecting from refresh_state and final_entities should yield exactly
         // one row (except in abnormal cases where the stitch was invoked for
         // something that didn't exist at all, in which case it's zero rows).

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts

@@ -27,7 +27,6 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
 import * as uuid from 'uuid';
 import { getVoidLogger } from '@backstage/backend-common';
 import { JsonObject } from '@backstage/config';
-import { DbFinalEntitiesRow } from '../Stitcher';
 
 describe('Default Processing Database', () => {
   let db: Knex;
@@ -116,10 +115,6 @@ describe('Default Processing Database', () => {
       );
       expect(entities[0].cache).toEqual(JSON.stringify(state));
       expect(entities[0].errors).toEqual("['something broke']");
-      const finalEntities = await db<DbFinalEntitiesRow>('final_entities')
-        .where('entity_id', id)
-        .select();
-      expect(finalEntities.length).toBe(1);
     });
 
     it('removes old relations and stores the new relationships', async () => {

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts

@@ -32,7 +32,6 @@ import type { Logger } from 'winston';
 
 import { v4 as uuid } from 'uuid';
 import { JsonObject } from '@backstage/config';
-import { DbFinalEntitiesRow } from '../Stitcher';
 
 export type DbRefreshStateRow = {
   entity_id: string;
@@ -122,15 +121,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       this.deduplicateRelations(relationRows),
       BATCH_SIZE,
     );
-
-    await tx<DbFinalEntitiesRow>('final_entities')
-      .insert({
-        entity_id: id,
-        hash: '',
-        stitch_ticket: '',
-      })
-      .onConflict('entity_id')
-      .ignore();
   }
 
   async updateProcessedEntityErrors(