Simplified outbox entry.

Author: kamilwylegala

packages/outbox-core/lib/objects.ts

@@ -15,8 +15,7 @@ export type OutboxEntryStatus = 'CREATED' | 'ACKED' | 'SUCCESS' | 'FAILED'
 
 export type OutboxEntry<SupportedEvent extends CommonEventDefinition> = {
   id: string
-  event: SupportedEvent
-  data: Omit<CommonEventDefinitionPublisherSchemaType<SupportedEvent>, 'type'>
+  event: CommonEventDefinitionPublisherSchemaType<SupportedEvent>
   precedingMessageMetadata?: Partial<ConsumerMessageMetadataType>
   status: OutboxEntryStatus
   created: Date

packages/outbox-core/lib/outbox.ts

@@ -6,6 +6,7 @@ import type {
   ConsumerMessageMetadataType,
   DomainEventEmitter,
 } from '@message-queue-toolkit/core'
+import { enrichMessageSchemaWithBase } from '@message-queue-toolkit/schemas'
 import { PromisePool } from '@supercharge/promise-pool'
 import { uuidv7 } from 'uuidv7'
 import type { OutboxAccumulator } from './accumulators'
@@ -50,7 +51,11 @@ export class OutboxProcessor<SupportedEvents extends CommonEventDefinition[]> {
       .withConcurrency(this.outboxProcessorConfiguration.emitBatchSize)
       .process(async (entry) => {
         try {
-          await eventEmitter.emit(entry.event, entry.data, entry.precedingMessageMetadata)
+          const event = entry.event
+
+          const schema = { ...enrichMessageSchemaWithBase(event.type, event.payload) }
+
+          await eventEmitter.emit(schema, entry.event, entry.precedingMessageMetadata)
           await outboxAccumulator.add(entry)
         } catch (e) {
           context.logger.error({ error: e }, 'Failed to process outbox entry.')
@@ -133,14 +138,12 @@ export class OutboxEventEmitter<SupportedEvents extends CommonEventDefinition[]>
    * @param precedingMessageMetadata
    */
   public async emit<SupportedEvent extends SupportedEvents[number]>(
-    supportedEvent: SupportedEvent,
-    data: Omit<CommonEventDefinitionPublisherSchemaType<SupportedEvent>, 'type'>,
+    data: CommonEventDefinitionPublisherSchemaType<SupportedEvent>,
     precedingMessageMetadata?: Partial<ConsumerMessageMetadataType>,
   ) {
     await this.storage.createEntry({
       id: uuidv7(),
-      event: supportedEvent,
-      data,
+      event: data,
       precedingMessageMetadata,
       status: 'CREATED',
       created: new Date(),

packages/outbox-core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@message-queue-toolkit/outbox-core",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "private": false,
   "license": "MIT",
   "description": "Outbox pattern implementation for message queue toolkit",

packages/outbox-core/test/outbox.spec.ts

@@ -91,7 +91,7 @@ describe('outbox', () => {
   })
 
   it('saves outbox entry to storage', async () => {
-    await outboxEventEmitter.emit(TestEvents.created, createdEventPayload, {
+    await outboxEventEmitter.emit(createdEventPayload, {
       correlationId: randomUUID(),
     })
 
@@ -101,7 +101,7 @@ describe('outbox', () => {
   })
 
   it('saves outbox entry and process it', async () => {
-    await outboxEventEmitter.emit(TestEvents.created, createdEventPayload, {
+    await outboxEventEmitter.emit(createdEventPayload, {
       correlationId: randomUUID(),
     })
 
@@ -140,7 +140,7 @@ describe('outbox', () => {
       }),
     )
 
-    await outboxEventEmitter.emit(TestEvents.created, createdEventPayload, {
+    await outboxEventEmitter.emit(createdEventPayload, {
       correlationId: randomUUID(),
     })
 
@@ -184,7 +184,7 @@ describe('outbox', () => {
     })
 
     //Persist the event
-    await outboxEventEmitter.emit(TestEvents.created, createdEventPayload, {
+    await outboxEventEmitter.emit(createdEventPayload, {
       correlationId: randomUUID(),
     })
 
@@ -229,7 +229,7 @@ describe('outbox', () => {
   it("doesn't emit event again if it's already present in accumulator", async () => {
     const mockedEventEmitter = vi.spyOn(eventEmitter, 'emit')
 
-    await outboxEventEmitter.emit(TestEvents.created, createdEventPayload, {
+    await outboxEventEmitter.emit(createdEventPayload, {
       correlationId: randomUUID(),
     })