diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 1052afda74..3fc932ad2d 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -10,6 +10,7 @@ import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; import type { Config } from '@backstage/config'; import type { DeferredEntity } from '@backstage/plugin-catalog-backend'; import type { DurationObjectUnits } from 'luxon'; +import { EventParams } from '@backstage/plugin-events-node'; import type { Logger } from 'winston'; import type { PermissionEvaluator } from '@backstage/plugin-permission-common'; import type { PluginDatabaseManager } from '@backstage/backend-common'; @@ -55,16 +56,18 @@ export interface IncrementalEntityProvider { context: TContext, cursor?: TCursor, ): Promise>; - onEvent?: (payload: unknown) => { - delta: - | { - added: DeferredEntity[]; - removed: { - entityRef: string; - }[]; - } - | undefined; - }; + onEvent?: (params: EventParams) => + | { + delta: + | { + added: DeferredEntity[]; + removed: { + entityRef: string; + }[]; + } + | undefined; + } + | undefined; } // @public (undocumented) 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 a21092a838..c1a62b7a73 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -336,14 +336,14 @@ export class IncrementalIngestionEngine } async onEvent(params: EventParams): Promise { - const { topic, eventPayload } = params; + const { topic } = params; if (topic !== this.providerEventTopic) { return; } const { logger, provider, connection } = this.options; const providerName = provider.getProviderName(); - logger.info( + logger.debug( `incremental-engine: Received ${this.providerEventTopic} event`, ); @@ -351,48 +351,50 @@ export class IncrementalIngestionEngine return; } - const update = provider.onEvent(eventPayload); + const update = provider.onEvent(params); - if (update.delta) { - if (update.delta.added.length > 0) { - const ingestionRecord = await this.manager.getCurrentIngestionRecord( - providerName, - ); - - if (!ingestionRecord) { - logger.debug( - `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`, + if (update) { + if (update.delta) { + if (update.delta.added.length > 0) { + const ingestionRecord = await this.manager.getCurrentIngestionRecord( + providerName, ); - } else { - const mark = - ingestionRecord.status === 'resting' - ? await this.manager.getLastMark(ingestionRecord.id) - : await this.manager.getFirstMark(ingestionRecord.id); - if (!mark) { - throw new Error( - `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, + if (!ingestionRecord) { + logger.debug( + `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`, ); + } else { + const mark = + ingestionRecord.status === 'resting' + ? await this.manager.getLastMark(ingestionRecord.id) + : await this.manager.getFirstMark(ingestionRecord.id); + + if (!mark) { + throw new Error( + `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, + ); + } + await this.manager.createMarkEntities(mark.id, update.delta.added); } - await this.manager.createMarkEntities(mark.id, update.delta.added); } - } - if (update.delta.removed.length > 0) { - await this.manager.deleteEntityRecordsByRef(update.delta.removed); - } + if (update.delta.removed.length > 0) { + await this.manager.deleteEntityRecordsByRef(update.delta.removed); + } - await connection.applyMutation({ - type: 'delta', - ...update.delta, - }); - logger.info( - `incremental-engine: Processed ${this.providerEventTopic} event`, - ); - } else { - logger.warn( - `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`, - ); + await connection.applyMutation({ + type: 'delta', + ...update.delta, + }); + logger.debug( + `incremental-engine: Processed ${this.providerEventTopic} event`, + ); + } else { + logger.warn( + `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`, + ); + } } } diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts index d03619d5a4..b3e3ae5792 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts @@ -239,7 +239,7 @@ export class IncrementalProviderRouter implements EventPublisher { }); }); - router.post(`${PROVIDER_BASE_PATH}/delta`, async (req, res) => { + router.post(`${PROVIDER_BASE_PATH}/event`, async (req, res) => { const { provider } = req.params; const topic = `${provider}-push`; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 315757d88a..efadee3a76 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -27,6 +27,7 @@ import type { DeferredEntity, EntityProviderConnection, } from '@backstage/plugin-catalog-backend'; +import { EventParams } from '@backstage/plugin-events-node'; import type { PermissionEvaluator } from '@backstage/plugin-permission-common'; import type { DurationObjectUnits } from 'luxon'; import type { Logger } from 'winston'; @@ -77,20 +78,23 @@ export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; /** - * If present, this method accepts an incoming event, and maps the - * payload to an object containing a delta mutation. + * This method accepts an incoming event for the provider, and + * (optionally) maps the payload to an object containing a delta + * mutation. * - * If a delta is present, the incremental entity provider will apply - * it automatically. + * If a valid delta is returned by this method, it will be ingested + * automatically by the provider. */ - onEvent?: (payload: unknown) => { - delta: - | { - added: DeferredEntity[]; - removed: { entityRef: string }[]; - } - | undefined; - }; + onEvent?: (params: EventParams) => + | { + delta: + | { + added: DeferredEntity[]; + removed: { entityRef: string }[]; + } + | undefined; + } + | undefined; } /**