From f06211e44757939d5fc6076e768bd8b332aa8c98 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Thu, 12 Sep 2024 22:18:57 +0200 Subject: [PATCH] events-backend: couple of simple review fixes Signed-off-by: Patrik Oldsberg --- .../migrations/20240523100528_init.js | 3 +- .../src/service/EventsPlugin.test.ts | 16 ++++---- .../src/service/hub/DatabaseEventBusStore.ts | 23 ++++++----- .../src/service/hub/createEventBusRouter.ts | 40 +++++++++---------- 4 files changed, 41 insertions(+), 41 deletions(-) diff --git a/plugins/events-backend/migrations/20240523100528_init.js b/plugins/events-backend/migrations/20240523100528_init.js index 90cdb770a5..7ec85d7917 100644 --- a/plugins/events-backend/migrations/20240523100528_init.js +++ b/plugins/events-backend/migrations/20240523100528_init.js @@ -82,6 +82,7 @@ exports.up = async function up(knex) { */ exports.down = async function down(knex) { if (knex.client.config.client === 'pg') { - await knex.schema.dropTable('events'); + await knex.schema.dropTable('event_bus_subscriptions'); + await knex.schema.dropTable('event_bus_events'); } }; diff --git a/plugins/events-backend/src/service/EventsPlugin.test.ts b/plugins/events-backend/src/service/EventsPlugin.test.ts index 0cfb68eb05..ba9a236800 100644 --- a/plugins/events-backend/src/service/EventsPlugin.test.ts +++ b/plugins/events-backend/src/service/EventsPlugin.test.ts @@ -150,7 +150,7 @@ describe('eventsPlugin', () => { 'should be possible to publish events as a service, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); @@ -177,7 +177,7 @@ describe('eventsPlugin', () => { 'should be possible to subscribe as a service and receive an event, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); @@ -217,7 +217,7 @@ describe('eventsPlugin', () => { 'should only send an event for each subscriber once, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); @@ -264,7 +264,7 @@ describe('eventsPlugin', () => { 'should not notify subscribers that have already consumed the event, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); @@ -309,7 +309,7 @@ describe('eventsPlugin', () => { 'should return multiple events in order, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); @@ -355,7 +355,7 @@ describe('eventsPlugin', () => { 'should skip publishing if all subscribers have already consumed the event, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); @@ -373,7 +373,7 @@ describe('eventsPlugin', () => { 'should time out when no events are available, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); await helper.subscribe('tester', ['test']).expect(201); @@ -423,7 +423,7 @@ describe('eventsPlugin', () => { 'should refuse listen without a subscription, %p', async databaseId => { const backend = await startTestBackend({ - features: [eventsPlugin(), await mockKnexFactory(databaseId)], + features: [eventsPlugin, await mockKnexFactory(databaseId)], }); const helper = new ReqHelper(backend); diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts index c9e97e22b1..e8f0da444d 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts @@ -26,7 +26,7 @@ import { import { ForwardedError, NotFoundError } from '@backstage/errors'; import { HumanDuration, durationToMilliseconds } from '@backstage/types'; -const WINDOW_SIZE_DEFAULT = 10_000; +const WINDOW_MAX_COUNT_DEFAULT = 10_000; const WINDOW_MIN_AGE_DEFAULT = { minutes: 10 }; const WINDOW_MAX_AGE_DEFAULT = { days: 1 }; @@ -254,7 +254,7 @@ export class DatabaseEventBusStore implements EventBusStore { /** Events outside of this age will always be deleted */ maxAge?: HumanDuration; /** Events outside of this count will be deleted if they are outside the minAge window */ - size?: number; + maxCount?: number; }; }): Promise { const db = await options.database.getClient(); @@ -278,12 +278,12 @@ export class DatabaseEventBusStore implements EventBusStore { db, options.logger, listener, - options.window?.size ?? WINDOW_SIZE_DEFAULT, + options.window?.maxCount ?? WINDOW_MAX_COUNT_DEFAULT, durationToMilliseconds(options.window?.minAge ?? WINDOW_MIN_AGE_DEFAULT), durationToMilliseconds(options.window?.maxAge ?? WINDOW_MAX_AGE_DEFAULT), ); - options.scheduler.scheduleTask({ + await options.scheduler.scheduleTask({ id: 'event-bus-cleanup', frequency: { seconds: 10 }, timeout: { minutes: 1 }, @@ -301,7 +301,7 @@ export class DatabaseEventBusStore implements EventBusStore { readonly #db: Knex; readonly #logger: LoggerService; readonly #listener: DatabaseEventBusListener; - readonly #windowSize: number; + readonly #windowMaxCount: number; readonly #windowMinAge: number; readonly #windowMaxAge: number; @@ -309,14 +309,14 @@ export class DatabaseEventBusStore implements EventBusStore { db: Knex, logger: LoggerService, listener: DatabaseEventBusListener, - windowSize: number, + windowMaxCount: number, windowMinAge: number, windowMaxAge: number, ) { this.#db = db; this.#logger = logger; this.#listener = listener; - this.#windowSize = windowSize; + this.#windowMaxCount = windowMaxCount; this.#windowMinAge = windowMinAge; this.#windowMaxAge = windowMaxAge; } @@ -393,9 +393,10 @@ export class DatabaseEventBusStore implements EventBusStore { id, updated_at: this.#db.fn.now(), topics, - read_until: this.#db.raw( - `( SELECT COALESCE(MAX("id"), 0) FROM "${TABLE_EVENTS}" )`, - ), + // TODO(Rugvip): Might be that there's a more performant way to do this + read_until: this.#db.raw(`( SELECT COALESCE(MAX("id"), 0) FROM ?? )`, [ + TABLE_EVENTS, + ]), }) .onConflict('id') .merge(['topics', 'updated_at']) @@ -523,7 +524,7 @@ export class DatabaseEventBusStore implements EventBusStore { .select('id') .from(TABLE_EVENTS) .orderBy('id', 'desc') - .offset(this.#windowSize), + .offset(this.#windowMaxCount), ) .wrap('ANY(ARRAY(', '))'), ) diff --git a/plugins/events-backend/src/service/hub/createEventBusRouter.ts b/plugins/events-backend/src/service/hub/createEventBusRouter.ts index a3ba7779b1..532ceaf2b8 100644 --- a/plugins/events-backend/src/service/hub/createEventBusRouter.ts +++ b/plugins/events-backend/src/service/hub/createEventBusRouter.ts @@ -114,6 +114,23 @@ cause unnecessary storage of events. */ +async function createEventBusStore(deps: { + logger: LoggerService; + database: DatabaseService; + scheduler: SchedulerService; + lifecycle: LifecycleService; + httpAuth: HttpAuthService; +}): Promise { + const db = await deps.database.getClient(); + if (db.client.config.client === 'pg') { + deps.logger.info('Database is PostgreSQL, using database store'); + return await DatabaseEventBusStore.create(deps); + } + + deps.logger.info('Database is not PostgreSQL, using memory store'); + return new MemoryEventBusStore(); +} + /** * Creates a new event bus router * @internal @@ -126,30 +143,11 @@ export async function createEventBusRouter(options: { httpAuth: HttpAuthService; notifyTimeoutMs?: number; // for testing }): Promise { - const { - database, - httpAuth, - scheduler, - lifecycle, - notifyTimeoutMs = DEFAULT_NOTIFY_TIMEOUT_MS, - } = options; + const { httpAuth, notifyTimeoutMs = DEFAULT_NOTIFY_TIMEOUT_MS } = options; const logger = options.logger.child({ type: 'EventBus' }); const router = Router(); - let store: EventBusStore; - const db = await database.getClient(); - if (db.client.config.client === 'pg') { - logger.info('Database is PostgreSQL, using database store'); - store = await DatabaseEventBusStore.create({ - database, - logger, - scheduler, - lifecycle, - }); - } else { - logger.info('Database is not PostgreSQL, using memory store'); - store = new MemoryEventBusStore(); - } + const store = await createEventBusStore(options); const apiRouter = await createOpenApiRouter();