diff --git a/plugins/events-backend/migrations/20240523100528_init.js b/plugins/events-backend/migrations/20240523100528_init.js new file mode 100644 index 0000000000..90cdb770a5 --- /dev/null +++ b/plugins/events-backend/migrations/20240523100528_init.js @@ -0,0 +1,87 @@ +/* + * Copyright 2024 The Backstage Authors + * + * 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. + */ + +// @ts-check + +/** + * @param { import("knex").Knex } knex + * @returns { Promise } + */ +exports.up = async function up(knex) { + // The event bus only supports PostgresSQL + if (knex.client.config.client === 'pg') { + await knex.schema.createTable('event_bus_events', table => { + table.comment('Events published to the events bus'); + table + .bigIncrements('id') + .primary() + .comment('The unique ID of this event'); + table + .dateTime('created_at') + .defaultTo(knex.fn.now()) + .notNullable() + .comment('The time that the event was created'); + table.text('topic').notNullable().comment('The topic of the event'); + table + .text('data_json') + .notNullable() + .comment('The payload data of this event'); + table + .specificType('consumed_by', 'text ARRAY') + .comment( + 'The IDs of the subscribers that have already consumed this event', + ); + }); + + await knex.schema.createTable('event_bus_subscriptions', table => { + table.comment('Subscriptions to the event bus'); + table + .string('id') + .primary() + .notNullable() + .comment('The unique ID of this particular subscription'); + table + .dateTime('created_at') + .defaultTo(knex.fn.now()) + .notNullable() + .comment('The time that the subscription was created'); + table + .dateTime('updated_at') + .defaultTo(knex.fn.now()) + .notNullable() + .comment('The time that the subscription was last updated'); + table + .bigInteger('read_until') + .notNullable() + .comment( + 'The sequence counter until which the subscription has read events', + ); + table + .specificType('topics', 'text ARRAY') + .comment('The topics that this subscriber is interested in'); + }); + } +}; + +/** + * @param { import("knex").Knex } knex + * @returns { Promise } + */ +exports.down = async function down(knex) { + if (knex.client.config.client === 'pg') { + await knex.schema.dropTable('events'); + } +}; diff --git a/plugins/events-backend/package.json b/plugins/events-backend/package.json index 2f9db1b081..f7a5ecc0c2 100644 --- a/plugins/events-backend/package.json +++ b/plugins/events-backend/package.json @@ -39,7 +39,8 @@ }, "files": [ "config.d.ts", - "dist" + "dist", + "migrations" ], "scripts": { "build": "backstage-cli package build", @@ -62,6 +63,7 @@ "@types/express": "^4.17.6", "express": "^4.17.1", "express-promise-router": "^4.1.0", + "knex": "^3.0.0", "winston": "^3.2.1" }, "devDependencies": { diff --git a/plugins/events-backend/src/service/EventsPlugin.ts b/plugins/events-backend/src/service/EventsPlugin.ts index b7892a5079..089ea928bc 100644 --- a/plugins/events-backend/src/service/EventsPlugin.ts +++ b/plugins/events-backend/src/service/EventsPlugin.ts @@ -75,11 +75,12 @@ export const eventsPlugin = createBackendPlugin({ deps: { config: coreServices.rootConfig, events: eventsServiceRef, + database: coreServices.database, logger: coreServices.logger, httpAuth: coreServices.httpAuth, router: coreServices.httpRouter, }, - async init({ config, events, logger, httpAuth, router }) { + async init({ config, events, database, logger, httpAuth, router }) { const ingresses = Object.fromEntries( extensionPoint.httpPostIngresses.map(ingress => [ ingress.topic, @@ -96,7 +97,7 @@ export const eventsPlugin = createBackendPlugin({ const eventsRouter = Router(); http.bind(eventsRouter); - const hub = await EventHub.create({ logger, httpAuth }); + const hub = await EventHub.create({ database, logger, httpAuth }); router.use(hub.handler()); router.use(eventsRouter); diff --git a/plugins/events-backend/src/service/hub/DatabaseEventHubStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventHubStore.ts new file mode 100644 index 0000000000..646812482d --- /dev/null +++ b/plugins/events-backend/src/service/hub/DatabaseEventHubStore.ts @@ -0,0 +1,202 @@ +/* + * Copyright 2024 The Backstage Authors + * + * 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 { EventParams } from '@backstage/plugin-events-node'; +import { EventHubStore } from './types'; +import { Knex } from 'knex'; +import { + DatabaseService, + LoggerService, + resolvePackagePath, +} from '@backstage/backend-plugin-api'; + +const MAX_BATCH_SIZE = 10; + +const TABLE_EVENTS = 'event_bus_events'; +const TABLE_SUBSCRIPTIONS = 'event_bus_subscriptions'; + +type EventsRow = { + id: string; + created_at: Date; + topic: string; + data_json: string; + consumed_by: string[]; +}; + +type SubscriptionsRow = { + id: string; + created_at: Date; + updated_at: Date; + read_until: string; + topics: string[]; +}; + +const migrationsDir = resolvePackagePath( + '@backstage/plugin-events-backend', + 'migrations', +); + +export class DatabaseEventHubStore implements EventHubStore { + static async create(options: { + database: DatabaseService; + logger: LoggerService; + }): Promise { + const db = await options.database.getClient(); + + if (db.client.config.client !== 'pg') { + throw new Error( + `DatabaseEventHubStore only supports PostgreSQL, got '${db.client.config.client}'`, + ); + } + + if (!options.database.migrations?.skip) { + await db.migrate.latest({ + directory: migrationsDir, + }); + options.logger.info('DatabaseEventHubStore migrations ran successfully'); + } + + return new DatabaseEventHubStore(db); + } + + readonly #db: Knex; + + private constructor(db: Knex) { + this.#db = db; + } + + async publish(options: { + params: EventParams; + subscriberIds: string[]; + }): Promise<{ id: string }> { + const result = await this.#db(TABLE_EVENTS) + .insert({ + topic: options.params.topic, + data_json: JSON.stringify({ + payload: options.params.eventPayload, + metadata: options.params.metadata, + }), + consumed_by: options.subscriberIds, + }) + .returning('id'); + + if (result.length !== 1) { + throw new Error(`Failed to insert event, updated ${result.length} rows`); + } + + const { id } = result[0]; + + // TODO: notify + + return { id }; + } + + async upsertSubscription(id: string, topics: string[]): Promise { + const result = await this.#db(TABLE_SUBSCRIPTIONS) + .insert({ + id, + updated_at: this.#db.fn.now(), + topics, + read_until: this.#db(TABLE_EVENTS).max('id') as any, // TODO: figure out TS, + }) + .onConflict('id') + .merge(['topics', 'updated_at']) + .returning('*'); + + if (result.length !== 1) { + throw new Error( + `Failed to upsert subscription, updated ${result.length} rows`, + ); + } + } + + async readSubscription(id: string): Promise<{ events: EventParams[] }> { + const result = await this.#db(TABLE_SUBSCRIPTIONS) + // Read the target subscription so that we can use the read marker and topics + .with('sub', q => q.select().from(TABLE_SUBSCRIPTIONS).where({ id })) + // Read the next batch of events for the subscription from its read marker + .with('events', q => + q + .select('event.*') + .from({ event: 'event_bus_events', sub: 'sub' }) + // For each event, check if it matches any of the topics that we're subscribed to + .where( + 'event.topic', + '=', + this.#db.ref('topics').withSchema('sub').wrap('ANY(', ')'), + ) + // Skip events that have already been consumed by this subscription + .where( + this.#db.raw('?', id), + '<>', + this.#db.ref('consumed_by').withSchema('event').wrap('ANY(', ')'), + ) + .where('event.id', '>', this.#db.ref('read_until').withSchema('sub')) + .orderBy('event.id', 'asc') + .limit(MAX_BATCH_SIZE), + ) + // Find the ID of the last event in the batch, for use as the new read_until marker + .with('last_event_id', q => q.max({ last_event_id: 'id' }).from('events')) + // Aggregate the events into a JSON array so that we can return all of them with the UPDATE + .with('events_array', q => + q + .select({ events: this.#db.raw('json_agg(row_to_json(events))') }) + .from('events'), + ) + // Update the read_until marker to the ID of the last event, or if no + // events where read, the last ID out of all events + .update({ + read_until: this.#db.raw( + 'COALESCE(last_event_id, (SELECT MAX(id) FROM event_bus_events))', + ), + }) + .updateFrom({ + events_array: 'events_array', + last_event_id: 'last_event_id', + }) + .where(`${TABLE_SUBSCRIPTIONS}.id`, id) + .returning<[{ events: EventsRow[] }]>('events_array.events'); + + if (result.length !== 1) { + throw new Error( + `Failed to upsert subscription, updated ${result.length} rows`, + ); + } + + const rows = result[0].events; + if (!rows || rows.length === 0) { + return { events: [] }; + } + + return { + events: result[0].events.map(row => { + const { payload, metadata } = JSON.parse(row.data_json); + return { + topic: row.topic, + eventPayload: payload, + metadata, + }; + }), + }; + } + + async listen( + _subscriptionId: string, + _onNotify: (topicId: string) => void, + ): Promise<() => void> { + // TODO + return () => {}; + } +} diff --git a/plugins/events-backend/src/service/hub/EventHub.ts b/plugins/events-backend/src/service/hub/EventHub.ts index 594713da17..c78a8359ef 100644 --- a/plugins/events-backend/src/service/hub/EventHub.ts +++ b/plugins/events-backend/src/service/hub/EventHub.ts @@ -14,24 +14,33 @@ * limitations under the License. */ -import { HttpAuthService, LoggerService } from '@backstage/backend-plugin-api'; +import { + DatabaseService, + HttpAuthService, + LoggerService, +} from '@backstage/backend-plugin-api'; import { Handler } from 'express'; import Router from 'express-promise-router'; import { spec, createOpenApiRouter } from '../../schema/openapi.generated'; import { internal } from '@backstage/backend-openapi-utils'; import { MemoryEventHubStore } from './MemoryEventHubStore'; +import { DatabaseEventHubStore } from './DatabaseEventHubStore'; import { EventHubStore } from './types'; import { EventParams } from '@backstage/plugin-events-node'; export class EventHub { static async create(options: { logger: LoggerService; + database: DatabaseService; httpAuth: HttpAuthService; }) { - const { httpAuth } = options; + const { database, httpAuth } = options; const logger = options.logger.child({ type: 'EventHub' }); const router = Router(); - const store = new MemoryEventHubStore(); + const store = await DatabaseEventHubStore.create({ + database, + logger, + }); const hub = new EventHub(router, logger, httpAuth, store); @@ -83,16 +92,19 @@ export class EventHub { const credentials = await this.#httpAuth.credentials(req, { allow: ['service'], }); - await this.#store.publish({ + const { id } = await this.#store.publish({ params: { topic: req.body.event.topic, eventPayload: req.body.event.payload, } as EventParams, subscriberIds: req.body.subscriptionIds ?? [], }); - this.#logger.info(`Published event to '${req.body.event.topic}'`, { - subject: credentials.principal.subject, - }); + this.#logger.info( + `Published event to '${req.body.event.topic}' with ID '${id}'`, + { + subject: credentials.principal.subject, + }, + ); res.status(201).end(); }; diff --git a/plugins/events-backend/src/service/hub/types.ts b/plugins/events-backend/src/service/hub/types.ts index 8e1124252a..8adedc69c7 100644 --- a/plugins/events-backend/src/service/hub/types.ts +++ b/plugins/events-backend/src/service/hub/types.ts @@ -20,7 +20,7 @@ export type EventHubStore = { publish(options: { params: EventParams; subscriberIds: string[]; - }): Promise; + }): Promise<{ id: string }>; upsertSubscription(id: string, topics: string[]): Promise; diff --git a/yarn.lock b/yarn.lock index 2aeb54be7f..306e96a2c8 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6152,6 +6152,7 @@ __metadata: "@types/express": ^4.17.6 express: ^4.17.1 express-promise-router: ^4.1.0 + knex: ^3.0.0 supertest: ^7.0.0 winston: ^3.2.1 languageName: unknown