Tests WIP.

Author: kamilwylegala

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

@@ -0,0 +1,32 @@
+import {
+  type SnsAwareEventDefinition,
+  enrichEventSchemaWithBase,
+} from '@message-queue-toolkit/schemas'
+import { describe } from 'vitest'
+import { z } from 'zod'
+import type { OutboxStorage } from './outbox'
+
+const TEST_EVENT_SCHEMA = z.object({
+  name: z.string(),
+  age: z.number(),
+})
+
+const testEvents = {
+  'test-event.something-happened': {
+    ...enrichEventSchemaWithBase('test-event.something-happened', TEST_EVENT_SCHEMA),
+    snsTopic: 'TEST_TOPIC',
+    producedBy: ['unit tes'],
+  },
+} satisfies Record<string, SnsAwareEventDefinition>
+
+console.log(testEvents)
+
+const testEventsArray = [...Object.values(testEvents)] satisfies SnsAwareEventDefinition[]
+
+type TestEvents = typeof testEventsArray
+
+class InMemoryOutbox implements OutboxStorage<TestEvents> {}
+
+describe('outbox', () => {
+  it('saves outbox entry to process it later', async () => {})
+})

packages/outbox-core/lib/outbox.ts

@@ -53,6 +53,41 @@ export interface OutboxStorage<SupportedEvents extends CommonEventDefinition[]>
   getEntries(maxRetryCount: number): Promise<OutboxEntry<SupportedEvents[number]>[]>
 }
 
+export class OutboxProcessor<SupportedEvents extends CommonEventDefinition[]> {
+  constructor(
+    private readonly outboxStorage: OutboxStorage<SupportedEvents>,
+    private readonly eventEmitter: DomainEventEmitter<SupportedEvents>,
+    private readonly maxRetryCount: number,
+  ) {}
+
+  public async processOutboxEntries(context: JobExecutionContext) {
+    const entries = await this.outboxStorage.getEntries(this.maxRetryCount)
+
+    for (const entry of entries) {
+      try {
+        const updatedEntry = await this.outboxStorage.update({
+          ...entry,
+          updated: new Date(),
+          status: 'ACKED',
+        })
+
+        await this.eventEmitter.emit(entry.event, entry.data, entry.precedingMessageMetadata)
+
+        await this.outboxStorage.update({ ...updatedEntry, updated: new Date(), status: 'SUCCESS' })
+      } catch (e) {
+        context.logger.error({ error: e }, 'Failed to process outbox entry.')
+
+        await this.outboxStorage.update({
+          ...entry,
+          updated: new Date(),
+          status: 'FAILED',
+          retryCount: entry.retryCount + 1,
+        })
+      }
+    }
+  }
+}
+
 /**
  * Periodic job that processes outbox entries every second. If processing takes longer than 1 second, another subsequent job WILL NOT be started.
  *
@@ -63,10 +98,12 @@ export interface OutboxStorage<SupportedEvents extends CommonEventDefinition[]>
 export class OutboxPeriodicJob<
   SupportedEvents extends CommonEventDefinition[],
 > extends AbstractPeriodicJob {
+  private readonly outboxProcessor: OutboxProcessor<SupportedEvents>
+
   constructor(
-    private readonly outboxStorage: OutboxStorage<SupportedEvents>,
-    private readonly eventEmitter: DomainEventEmitter<SupportedEvents>,
-    private readonly maxRetryCount: number,
+    outboxStorage: OutboxStorage<SupportedEvents>,
+    eventEmitter: DomainEventEmitter<SupportedEvents>,
+    maxRetryCount: number,
     dependencies: PeriodicJobDependencies,
   ) {
     super(
@@ -87,33 +124,16 @@ export class OutboxPeriodicJob<
         scheduler: dependencies.scheduler,
       },
     )
+
+    this.outboxProcessor = new OutboxProcessor<SupportedEvents>(
+      outboxStorage,
+      eventEmitter,
+      maxRetryCount,
+    )
   }
 
   protected async processInternal(context: JobExecutionContext): Promise<void> {
-    const entries = await this.outboxStorage.getEntries(this.maxRetryCount)
-
-    for (const entry of entries) {
-      try {
-        const updatedEntry = await this.outboxStorage.update({
-          ...entry,
-          updated: new Date(),
-          status: 'ACKED',
-        })
-
-        await this.eventEmitter.emit(entry.event, entry.data, entry.precedingMessageMetadata)
-
-        await this.outboxStorage.update({ ...updatedEntry, updated: new Date(), status: 'SUCCESS' })
-      } catch (e) {
-        context.logger.error({ error: e }, 'Failed to process outbox entry.')
-
-        await this.outboxStorage.update({
-          ...entry,
-          updated: new Date(),
-          status: 'FAILED',
-          retryCount: entry.retryCount + 1,
-        })
-      }
-    }
+    await this.outboxProcessor.processOutboxEntries(context)
   }
 }
 

packages/outbox-core/package.json

@@ -29,17 +29,18 @@
     "@lokalise/id-utils": "^2.2.0"
   },
   "peerDependencies": {
-    "@message-queue-toolkit/core": ">=14.0.0"
+    "@message-queue-toolkit/core": ">=14.0.0",
+    "@message-queue-toolkit/schemas": ">=4.0.0"
   },
   "devDependencies": {
     "@biomejs/biome": "1.8.3",
     "@kibertoad/biome-config": "^1.2.1",
-    "@message-queue-toolkit/core": "*",
     "@types/node": "^22.0.0",
     "@vitest/coverage-v8": "^2.0.4",
     "del-cli": "^5.1.0",
     "typescript": "^5.5.3",
-    "vitest": "^2.0.4"
+    "vitest": "^2.0.4",
+    "zod": "^3.23.8"
   },
   "homepage": "https://github.com/kibertoad/message-queue-toolkit",
   "repository": {