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/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/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/, + ); + }); +}); 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..376af6e748 --- /dev/null +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -0,0 +1,161 @@ +/* + * 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'; +import waitForExpect from 'wait-for-expect'; + +describe('DefaultCatalogProcessingEngine', () => { + const db = ({ + transaction: jest.fn(), + getProcessableEntities: jest.fn(), + updateProcessedEntity: jest.fn(), + } as unknown) as jest.Mocked; + const orchestrator: jest.Mocked = { + process: jest.fn(), + }; + const stitcher = ({ + stitch: jest.fn(), + } as unknown) as jest.Mocked; + + 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 waitForExpect(() => { + expect(orchestrator.process).toBeCalledTimes(1); + expect(orchestrator.process).toBeCalledWith({ + entity: { + apiVersion: '1', + kind: 'Location', + metadata: { name: 'test' }, + }, + state: expect.anything(), + }); + }); + 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 e724fe79d8..55cb8e2029 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 { + const db = this.config.processingDatabase; if (mutation.type === 'full') { - await this.config.stateManager.replaceProcessingItems({ - sourceKey: this.config.id, - type: 'full', - items: mutation.entities, + await 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, + await 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,52 +72,77 @@ 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(); + 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(); + } + } + } - const result = await this.orchestrator.process({ - entity, - state: initialState, + private async process() { + 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) { + // No items to process, wait and try again. + await this.wait(); + return; + } - if (!result.ok) { - return; - } + const { id, state, unprocessedEntity } = items[0]; - result.completedEntity.metadata.uid = id; - await this.stateManager.setProcessingItemResult({ + 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, - entity: result.completedEntity, + processedEntity: result.completedEntity, state: result.state, - errors: result.errors, + errors: JSON.stringify(result.errors), relations: result.relations, deferredEntities: result.deferredEntities, }); + }); - const setOfThingsToStitch = new Set([ - stringifyEntityRef(result.completedEntity), - ...result.relations.map(relation => - stringifyEntityRef(relation.source), - ), - ]); - await this.stitcher.stitch(setOfThingsToStitch); - } + const setOfThingsToStitch = new Set([ + stringifyEntityRef(result.completedEntity), + ...result.relations.map(relation => stringifyEntityRef(relation.source)), + ]); + await this.stitcher.stitch(setOfThingsToStitch); + } + + private async wait() { + await new Promise(resolve => setTimeout(resolve, 1000)); } async stop() { 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 { - 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 { - 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(resolve => setTimeout(resolve, 1000)); - } - } -} diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts index 10500fdc15..0158cc9fc9 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,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, @@ -269,7 +267,7 @@ export class NextCatalogBuilder { const processingEngine = new DefaultCatalogProcessingEngine( logger, [locationStore, configLocationProvider], - stateManager, + processingDatabase, orchestrator, stitcher, ); diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index d76bd74260..7a0094dbf0 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('refresh_state_references').insert( + ref, + ); + }; + + const insertRefreshStateRow = async (ref: DbRefreshStateRow) => { + await db('refresh_state').insert(ref); + }; + beforeEach(async () => { db = await DatabaseManager.createTestDatabaseConnection(); await DatabaseManager.createDatabase(db); @@ -39,17 +50,162 @@ describe('Default Processing Database', () => { processingDatabase = new DefaultProcessingDatabase(db, logger); }); - describe('replaceUnprocessedEntities', () => { - const insertRefRow = async (ref: DbRefreshStateReferencesRow) => { - return db('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(), + 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: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + const state = new Map(); + 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('refresh_state').insert(ref); - }; + const entities = await db('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 () => { + await insertRefreshStateRow({ + entity_id: id, + entity_ref: 'location:default/fakelocation', + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + 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(), + relations: relations, + deferredEntities: [], + }), + ); + + const savedRelations = await db('relations') + .where({ originating_entity_id: id }) + .select(); + expect(savedRelations.length).toBe(1); + expect(savedRelations[0]).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: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + const deferredEntities = [ + { + apiVersion: '1', + kind: 'Location', + metadata: { + name: 'next', + }, + }, + ]; + + await processingDatabase.transaction(tx => + processingDatabase.updateProcessedEntity(tx, { + id, + processedEntity, + state: new Map(), + relations: [], + deferredEntities, + }), + ); + + const refreshStateEntries = await db('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({ @@ -57,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', }); } }; @@ -101,8 +257,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 +270,8 @@ describe('Default Processing Database', () => { kind: 'Location', } as Entity, ], - }); - }); + }), + ); const currentRefreshState = await db( 'refresh_state', @@ -431,86 +587,48 @@ 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('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 = { + describe('getProcessableEntities', () => { + it('should return entities to process', async () => { + const entity = JSON.stringify({ + kind: 'Location', apiVersion: '1.0.0', metadata: { - name: 'deferred', + name: 'xyz', }, - kind: 'Location', - } as Entity; + } 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], - }); + await db('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: '2021-04-01 13:37:00', }); - const deferredResult = await db('refresh_state') - .where({ entity_ref: 'location:default/deferred' }) - .select(); - expect(deferredResult.length).toBe(1); + await db('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: '2021-04-01 13:37:00', + }); - const [relations] = await db('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', + 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: [] }); }); }); }); 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; } - -export type ProcessingItemResult = { - id: string; - entity: Entity; - state: Map; - errors: Error[]; - relations: EntityRelationSpec[]; - deferredEntities: Entity[]; -}; - -export type ProcessingItem = { - id: string; - entity: Entity; - state: Map; -}; - -export type ReplaceProcessingItemsRequest = - | { - sourceKey: string; - items: Entity[]; - type: 'full'; - } - | { - sourceKey: string; - added: Entity[]; - removed: Entity[]; - type: 'delta'; - }; - -export interface ProcessingStateManager { - setProcessingItemResult(result: ProcessingItemResult): Promise; - getNextProcessingItem(): Promise; - replaceProcessingItems(request: ReplaceProcessingItemsRequest): Promise; -} diff --git a/yarn.lock b/yarn.lock index bffef20f24..f778b2d5cf 100644 --- a/yarn.lock +++ b/yarn.lock @@ -26444,6 +26444,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"