diff --git a/plugins/events-backend/src/service/hub/DatabaseEventHubStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventHubStore.ts index e5e1334191..f310ae86a4 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventHubStore.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventHubStore.ts @@ -348,8 +348,11 @@ export class DatabaseEventHubStore implements EventHubStore { async listen( subscriptionId: string, - onNotify: (topicId: string) => void, - ): Promise<() => void> { + listeners: { + onNotify: (topicId: string) => void; + onError: () => void; + }, + ): Promise<{ cancel(): void }> { const result = await this.#db(TABLE_SUBSCRIPTIONS) .select('topics') .where({ id: subscriptionId }) @@ -361,6 +364,11 @@ export class DatabaseEventHubStore implements EventHubStore { ); } const topics = new Set(result.topics ?? []); - return this.#listener.listen(topics, onNotify, () => {}); + const cancel = await this.#listener.listen( + topics, + listeners.onNotify, + listeners.onError, + ); + return { cancel }; } } diff --git a/plugins/events-backend/src/service/hub/EventHub.ts b/plugins/events-backend/src/service/hub/EventHub.ts index c78a8359ef..97d641e107 100644 --- a/plugins/events-backend/src/service/hub/EventHub.ts +++ b/plugins/events-backend/src/service/hub/EventHub.ts @@ -118,21 +118,46 @@ export class EventHub { }); const id = req.params.subscriptionId; - const { events } = await this.#store.readSubscription(id); - - this.#logger.info( - `Reading subscription '${id}' resulted in ${events.length} events`, - { subject: credentials.principal.subject }, - ); - - if (events.length > 0) { - res.json({ events }); - return; - } - - this.#store.listen(id, () => { - res.status(204).end(); + let resolveShouldNotify: (shouldNotify: boolean) => void; + const shouldNotifyPromise = new Promise(resolve => { + resolveShouldNotify = resolve; }); + + const { cancel } = await this.#store.listen(id, { + onNotify() { + shouldNotifyPromise.then(shouldNotify => { + if (shouldNotify) { + res.status(204).end(); + } + }); + }, + onError() { + shouldNotifyPromise.then(shouldNotify => { + if (shouldNotify) { + res.status(500).end(); + } + }); + }, + }); + req.on('end', cancel); + + try { + const { events } = await this.#store.readSubscription(id); + + this.#logger.info( + `Reading subscription '${id}' resulted in ${events.length} events`, + { subject: credentials.principal.subject }, + ); + + if (events.length > 0) { + res.json({ events }); + resolveShouldNotify!(false); + } else { + resolveShouldNotify!(true); + } + } finally { + resolveShouldNotify!(false); + } }; #handlePutSubscription: internal.DocRequestHandler< diff --git a/plugins/events-backend/src/service/hub/types.ts b/plugins/events-backend/src/service/hub/types.ts index 8adedc69c7..0a0a27b1c1 100644 --- a/plugins/events-backend/src/service/hub/types.ts +++ b/plugins/events-backend/src/service/hub/types.ts @@ -28,6 +28,9 @@ export type EventHubStore = { listen( subscriptionId: string, - onNotify: (topicId: string) => void, - ): Promise<() => void>; + listeners: { + onNotify(topicId: string): void; + onError(): void; + }, + ): Promise<{ cancel(): void }>; };