From b1cc10696f2f23033430997bba9e808b35035dfe Mon Sep 17 00:00:00 2001 From: Charles Lowell Date: Mon, 12 Jun 2023 20:47:10 +0300 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8Make=20incremental=20event=20handler?= =?UTF-8?q?=20to=20be=20both=20explicity=20and=20async?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The point of the event hook is to map incoming changes into catalog deltas. However, this is not always a synchronous process, and might require accessing other resources to get the actual delta. This just converts the hook into an async function and awaits it on the other end. This also cleans up the API a little bit by not relying on `undefined` to indicate that the event was ignored, but by explicitly forcing the event handler to indicate that the event is to be ignore by returning { type: "ignored" }. Since ignoring events is a perfectly acceptable outcome, the warning if the event was dropped is just ignored. If there is an error, it should be propagated as an exception. This is purely an aesthetic change designed to communicate via usage how to use the API. Signed-off-by: Charles Lowell --- .changeset/rotten-colts-decide.md | 5 ++++ .../api-report.md | 22 +++++++++----- .../src/engine/IncrementalIngestionEngine.ts | 21 ++++++-------- .../src/index.ts | 1 + .../src/types.ts | 29 ++++++++++++++----- 5 files changed, 50 insertions(+), 28 deletions(-) create mode 100644 .changeset/rotten-colts-decide.md 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.