Merge pull request #21567 from backstage/freben/incremental-cursors

ensure that incremental ingestion cursors are json decoded

.changeset/rich-singers-join.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch
+---
+
+Ensure that cursors always come back as JSON on sqlite too

plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2023 The Backstage Authors
+ *
+ * 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 { TestDatabases } from '@backstage/backend-test-utils';
+import { IncrementalIngestionDatabaseManager } from './IncrementalIngestionDatabaseManager';
+import { v4 as uuid } from 'uuid';
+
+const migrationsDir = `${__dirname}/../../migrations`;
+
+jest.setTimeout(60_000);
+
+describe('IncrementalIngestionDatabaseManager', () => {
+  const databases = TestDatabases.create({
+    ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
+  });
+
+  it.each(databases.eachSupportedId())(
+    'stores and retrieves marks, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+      await knex.migrate.latest({ directory: migrationsDir });
+
+      const manager = new IncrementalIngestionDatabaseManager({ client: knex });
+      const { ingestionId } = (await manager.createProviderIngestionRecord(
+        'myProvider',
+      ))!;
+
+      const cursorId = uuid();
+
+      await manager.createMark({
+        record: {
+          id: cursorId,
+          ingestion_id: ingestionId,
+          sequence: 1,
+          cursor: { data: 1 },
+        },
+      });
+
+      await expect(manager.getFirstMark(ingestionId)).resolves.toEqual({
+        created_at: expect.anything(),
+        cursor: { data: 1 },
+        id: cursorId,
+        ingestion_id: ingestionId,
+        sequence: 1,
+      });
+
+      await expect(manager.getLastMark(ingestionId)).resolves.toEqual({
+        created_at: expect.anything(),
+        cursor: { data: 1 },
+        id: cursorId,
+        ingestion_id: ingestionId,
+        sequence: 1,
+      });
+
+      await expect(manager.getAllMarks(ingestionId)).resolves.toEqual([
+        {
+          created_at: expect.anything(),
+          cursor: { data: 1 },
+          id: cursorId,
+          ingestion_id: ingestionId,
+          sequence: 1,
+        },
+      ]);
+    },
+  );
+});

plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts

@@ -542,7 +542,7 @@ export class IncrementalIngestionDatabaseManager {
         .where('ingestion_id', ingestionId)
         .orderBy('sequence', 'desc')
         .first();
-      return mark;
+      return this.#decodeMark(this.client, mark);
     });
   }
 
@@ -557,7 +557,7 @@ export class IncrementalIngestionDatabaseManager {
         .where('ingestion_id', ingestionId)
         .orderBy('sequence', 'asc')
         .first();
-      return mark;
+      return this.#decodeMark(this.client, mark);
     });
   }
 
@@ -566,7 +566,7 @@ export class IncrementalIngestionDatabaseManager {
       const marks = await tx<MarkRecord>('ingestion_marks')
         .where('ingestion_id', ingestionId)
         .orderBy('sequence', 'desc');
-      return marks;
+      return marks.map(m => this.#decodeMark(this.client, m));
     });
   }
 
@@ -580,6 +580,19 @@ export class IncrementalIngestionDatabaseManager {
       await tx('ingestion_marks').insert(record);
     });
   }
+
+  // Handles the fact that sqlite does not support json columns; they just
+  // persist the stringified data instead
+  #decodeMark<T extends MarkRecord | undefined>(knex: Knex, record: T): T {
+    if (record && knex.client.config.client.includes('sqlite3')) {
+      return {
+        ...record,
+        cursor: JSON.parse(record.cursor as string),
+      };
+    }
+    return record;
+  }
+
   /**
    * Performs an upsert to the `ingestion_mark_entities` table for all deferred entities.
    * @param markId - string

plugins/catalog-backend-module-incremental-ingestion/src/database/tables.ts

@@ -89,7 +89,7 @@ export interface MarkRecord {
   id: string;
   sequence: number;
   ingestion_id: string;
-  cursor: string;
+  cursor: unknown;
   created_at: string;
 }