diff --git a/.changeset/rotten-colts-decide.md b/.changeset/rotten-colts-decide.md new file mode 100644 index 0000000000..9fddafe6f8 --- /dev/null +++ b/.changeset/rotten-colts-decide.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor +--- + +Allow incremental event handlers to be async; Force event handler to indicate if it made a change diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index c2fad4576c..4088b05bb5 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -48,18 +48,24 @@ export class IncrementalCatalogBuilder { ): Promise; } +// @public +export type IncrementalEntityEventResult = + | { + type: 'ignored'; + } + | { + type: 'delta'; + added: DeferredEntity[]; + removed: { + entityRef: string; + }[]; + }; + // @public export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; eventHandler?: { - onEvent: (params: EventParams) => - | undefined - | { - added: DeferredEntity[]; - removed: { - entityRef: string; - }[]; - }; + onEvent: (params: EventParams) => Promise; supportsEventTopics: () => string[]; }; getProviderName(): string; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index 13d5bfff71..0262dfd418 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -347,10 +347,10 @@ export class IncrementalIngestionEngine return; } - const delta = provider.eventHandler.onEvent(params); + const result = await provider.eventHandler.onEvent(params); - if (delta) { - if (delta.added.length > 0) { + if (result.type === 'delta') { + if (result.added.length > 0) { const ingestionRecord = await this.manager.getCurrentIngestionRecord( providerName, ); @@ -370,24 +370,21 @@ export class IncrementalIngestionEngine `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, ); } - await this.manager.createMarkEntities(mark.id, delta.added); + await this.manager.createMarkEntities(mark.id, result.added); } } - if (delta.removed.length > 0) { - await this.manager.deleteEntityRecordsByRef(delta.removed); + if (result.removed.length > 0) { + await this.manager.deleteEntityRecordsByRef(result.removed); } - await connection.applyMutation({ - type: 'delta', - ...delta, - }); + await connection.applyMutation(result); logger.debug( `incremental-engine: ${providerName} processed delta from '${topic}' event`, ); } else { - logger.warn( - `incremental-engine: Rejected delta from '${topic}' event - empty or invalid`, + logger.debug( + `incremental-engine: ${providerName} ignored event from topic '${topic}'`, ); } } diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/index.ts b/plugins/catalog-backend-module-incremental-ingestion/src/index.ts index ff1ee1faa1..f381572149 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/index.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/index.ts @@ -23,6 +23,7 @@ export * from './service'; export { type EntityIteratorResult, + type IncrementalEntityEventResult, type IncrementalEntityProvider, type IncrementalEntityProviderOptions, type PluginEnvironment, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 5f9d0489e6..68ad2b8b04 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -91,15 +91,12 @@ export interface IncrementalEntityProvider { * optionally maps the payload to an object containing a delta * mutation. * - * If a valid delta is returned by this method, it will be ingested - * automatically by the provider. + * If a delta result is returned by this method, it will be ingested + * automatically by the provider. Alternatively, if an "ignored" result is + * returned, then it is understood that this event should not cause anything + * to be ingested. */ - onEvent: (params: EventParams) => - | undefined - | { - added: DeferredEntity[]; - removed: { entityRef: string }[]; - }; + onEvent: (params: EventParams) => Promise; /** * This method returns an array of topics for the IncrementalEntityProvider @@ -109,6 +106,22 @@ export interface IncrementalEntityProvider { }; } +/** + * An object returned by event handler to indicate whether to ignore the event + * or to apply a delta in response to the event. + * + * @public + */ +export type IncrementalEntityEventResult = + | { + type: 'ignored'; + } + | { + type: 'delta'; + added: DeferredEntity[]; + removed: { entityRef: string }[]; + }; + /** * Value returned by an {@link IncrementalEntityProvider} to provide a * single page of entities to ingest.