Replace placeholder final_entities insert with upsert

Now that immediate mode is removed, the two-step pattern of inserting a
placeholder row and then updating it is unnecessary. Replace with a
single upsert that creates or updates the final_entities row in one
operation, removing a round-trip and simplifying the flow.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Fredrik Adelöw <freben@gmail.com>

plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts

@@ -317,70 +317,6 @@ describe.each(databases.eachSupportedId())(
   databaseId => {
     const logger = mockServices.logger.mock();
 
-    it('handles conflicts with past stitches', async () => {
-      if (databaseId === 'MYSQL_8') {
-        // MySQL doesn't handle conflicts in the same way as the other two, most
-        // likely due to the conflict probably being handled with a merged even
-        // if it's for the entity_ref. This probably means that MySQL will have
-        // inconsistencies in the final_entities table in some cases, but they
-        // should heal fairly quickly.
-        return;
-      }
-      const knex = await databases.init(databaseId);
-      await applyDatabaseMigrations(knex);
-
-      // Drop the foreign key constraint so we can insert a conflicting entity
-      // where the ID is inconsistent with the entity ref across refresh_state
-      // and final_entities
-      await knex.schema.alterTable('final_entities', table => {
-        table.dropForeign('entity_id');
-      });
-
-      await knex<DbRefreshStateRow>('refresh_state').insert([
-        {
-          entity_id: 'other-id',
-          entity_ref: 'k:ns/n',
-          unprocessed_entity: JSON.stringify({}),
-          processed_entity: JSON.stringify({
-            apiVersion: 'a',
-            kind: 'k',
-            metadata: {
-              name: 'n',
-              namespace: 'ns',
-            },
-            spec: {
-              k: 'v',
-            },
-          }),
-          errors: '[]',
-          next_update_at: knex.fn.now(),
-          last_discovery_at: knex.fn.now(),
-        },
-      ]);
-      await knex<DbFinalEntitiesRow>('final_entities').insert([
-        {
-          entity_id: 'my-id',
-          entity_ref: 'k:ns/n',
-          hash: '',
-          final_entity: JSON.stringify({}),
-        },
-      ]);
-
-      const stitchLogger = mockServices.logger.mock();
-      await expect(
-        performStitching({
-          knex,
-          logger: stitchLogger,
-          entityRef: 'k:ns/n',
-        }),
-      ).resolves.toBe('abandoned');
-
-      expect(stitchLogger.debug).toHaveBeenCalledWith(
-        'Skipping stitching of k:ns/n, conflict',
-        expect.anything(),
-      );
-    });
-
     it('stitches when final_entities row already exists', async () => {
       const knex = await databases.init(databaseId);
       await applyDatabaseMigrations(knex);

plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts

@@ -34,10 +34,7 @@ import {
 import { buildEntitySearch } from './buildEntitySearch';
 import { markDeferredStitchCompleted } from './markDeferredStitchCompleted';
 import { syncSearchRows } from './syncSearchRows';
-import {
-  LoggerService,
-  isDatabaseConflictError,
-} from '@backstage/backend-plugin-api';
+import { LoggerService } from '@backstage/backend-plugin-api';
 
 function generateStableHash(entity: Entity) {
   return createHash('sha1')
@@ -79,34 +76,10 @@ export async function performStitching(options: {
       return 'abandoned';
     }
 
-    // Ensure that a final_entities row exists for this entity.
-    try {
-      await knex<DbFinalEntitiesRow>('final_entities')
-        .insert({
-          entity_id: entityResult[0].entity_id,
-          hash: '',
-          entity_ref: entityRef,
-        })
-        .onConflict('entity_id')
-        .ignore();
-    } catch (error) {
-      // It's possible to hit a race where a refresh_state table delete + insert
-      // is done just after we read the entity_id from it. This conflict is safe
-      // to ignore because the current stitching operation will be triggered by
-      // the old entry, and the new entry will trigger it's own stitching that
-      // will update the entity.
-      if (isDatabaseConflictError(error)) {
-        logger.debug(`Skipping stitching of ${entityRef}, conflict`, error);
-        return 'abandoned';
-      }
-
-      throw error;
-    }
-
-    // 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).
-    // The join with the temporary incoming_references still gives one row.
+    // Selecting from refresh_state (with an optional left join to
+    // final_entities for the previous hash) should yield exactly one row,
+    // except in abnormal cases where the entity was deleted between the
+    // stitch request and now.
     const [processedResult, relationsResult] = await Promise.all([
       knex
         .with('incoming_references', function incomingReferences(builder) {
@@ -236,31 +209,31 @@ export async function performStitching(options: {
     // to write the search index.
     const searchEntries = buildEntitySearch(entityId, entity);
 
-    let updateQuery = knex<DbFinalEntitiesRow>('final_entities')
-      .update({
+    // Guard against concurrent stitchers by checking that the stitch_ticket
+    // in stitch_queue still matches what we were given.
+    if (stitchTicket) {
+      const ticketValid = await knex<DbStitchQueueRow>('stitch_queue')
+        .where('entity_ref', entityRef)
+        .where('stitch_ticket', stitchTicket)
+        .first();
+      if (!ticketValid) {
+        logger.debug(
+          `Entity ${entityRef} is already stitched, skipping write.`,
+        );
+        return 'abandoned';
+      }
+    }
+
+    await knex<DbFinalEntitiesRow>('final_entities')
+      .insert({
+        entity_id: entityId,
+        entity_ref: entityRef,
         final_entity: JSON.stringify(entity),
         hash,
         last_updated_at: knex.fn.now(),
       })
-      .where('entity_id', entityId);
-
-    // Guard against concurrent stitchers by checking that the stitch_ticket
-    // in stitch_queue still matches what we were given.
-    if (stitchTicket) {
-      updateQuery = updateQuery.whereExists(
-        knex<DbStitchQueueRow>('stitch_queue')
-          .where('stitch_queue.entity_ref', entityRef)
-          .where('stitch_queue.stitch_ticket', stitchTicket)
-          .select(knex.raw('1')),
-      );
-    }
-
-    const amountOfRowsChanged = await updateQuery;
-
-    if (amountOfRowsChanged === 0) {
-      logger.debug(`Entity ${entityRef} is already stitched, skipping write.`);
-      return 'abandoned';
-    }
+      .onConflict('entity_id')
+      .merge(['final_entity', 'hash', 'last_updated_at']);
 
     await syncSearchRows(knex, entityId, searchEntries);