From 43b2eb8f7050373bf7c6d0192cbfa164d9a4d35a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Sat, 25 Nov 2023 11:06:17 +0100 Subject: [PATCH] ensure that incremental ingestion cursors are json decoded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/rich-singers-join.md | 5 ++ ...ncrementalIngestionDatabaseManager.test.ts | 79 +++++++++++++++++++ .../IncrementalIngestionDatabaseManager.ts | 19 ++++- .../src/database/tables.ts | 2 +- 4 files changed, 101 insertions(+), 4 deletions(-) create mode 100644 .changeset/rich-singers-join.md create mode 100644 plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts diff --git a/.changeset/rich-singers-join.md b/.changeset/rich-singers-join.md new file mode 100644 index 0000000000..1a21fa9fc8 --- /dev/null +++ b/.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 diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts new file mode 100644 index 0000000000..a8862cfc2a --- /dev/null +++ b/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, + }, + ]); + }, + ); +}); diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts index 527d345706..e47954f47e 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/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('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(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 diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/tables.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/tables.ts index 00d40fb4a1..e1a47c040c 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/tables.ts +++ b/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; }