AP-5046 config + dependencies object.

Author: kamilwylegala

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

@@ -14,7 +14,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { z } from 'zod'
 import { InMemoryOutboxAccumulator, type OutboxAccumulator } from './accumulators'
 import type { OutboxEntry } from './objects'
-import { OutboxEventEmitter, OutboxProcessor } from './outbox'
+import { type OutboxDependencies, OutboxEventEmitter, OutboxProcessor } from './outbox'
 import type { OutboxStorage } from './storage'
 
 const TestEvents = {
@@ -145,10 +145,12 @@ describe('outbox', () => {
     outboxStorage = new InMemoryOutboxStorage<TestEventsType>()
     outboxEventEmitter = new OutboxEventEmitter<TestEventsType>(outboxStorage)
     outboxProcessor = new OutboxProcessor<TestEventsType>(
-      outboxStorage,
-      //@ts-ignore
-      new InMemoryOutboxAccumulator(),
-      eventEmitter,
+      {
+        outboxStorage,
+        //@ts-ignore
+        outboxAccumulator: new InMemoryOutboxAccumulator(),
+        eventEmitter,
+      } satisfies OutboxDependencies<TestEventsType>,
       MAX_RETRY_COUNT,
       1,
     )

packages/outbox-core/lib/outbox.ts

@@ -11,38 +11,50 @@ import { uuidv7 } from 'uuidv7'
 import type { OutboxAccumulator } from './accumulators'
 import type { OutboxStorage } from './storage'
 
+export type OutboxDependencies<SupportedEvents extends CommonEventDefinition[]> = {
+  outboxStorage: OutboxStorage<SupportedEvents>
+  outboxAccumulator: OutboxAccumulator<SupportedEvents>
+  eventEmitter: DomainEventEmitter<SupportedEvents>
+}
+
+export type OutboxConfiguration = {
+  maxRetryCount: number
+  emitBatchSize: number
+  jobIntervalInMs: number
+}
+
 /**
  * Main logic for handling outbox entries.
  *
  * If entry is rejected, it is NOT going to be handled during the same execution. Next execution will pick it up.
  */
 export class OutboxProcessor<SupportedEvents extends CommonEventDefinition[]> {
   constructor(
-    private readonly outboxStorage: OutboxStorage<SupportedEvents>,
-    private readonly outboxAccumulator: OutboxAccumulator<SupportedEvents>,
-    private readonly eventEmitter: DomainEventEmitter<SupportedEvents>,
+    private readonly outboxDependencies: OutboxDependencies<SupportedEvents>,
     private readonly maxRetryCount: number,
     private readonly emitBatchSize: number,
   ) {}
 
   public async processOutboxEntries(context: JobExecutionContext) {
-    const entries = await this.outboxStorage.getEntries(this.maxRetryCount)
+    const { outboxStorage, eventEmitter, outboxAccumulator } = this.outboxDependencies
+
+    const entries = await outboxStorage.getEntries(this.maxRetryCount)
 
     await PromisePool.for(entries)
       .withConcurrency(this.emitBatchSize)
       .process(async (entry) => {
         try {
-          await this.eventEmitter.emit(entry.event, entry.data, entry.precedingMessageMetadata)
-          await this.outboxAccumulator.add(entry)
+          await eventEmitter.emit(entry.event, entry.data, entry.precedingMessageMetadata)
+          await outboxAccumulator.add(entry)
         } catch (e) {
           context.logger.error({ error: e }, 'Failed to process outbox entry.')
 
-          await this.outboxAccumulator.addFailure(entry)
+          await outboxAccumulator.addFailure(entry)
         }
       })
 
-    await this.outboxStorage.flush(this.outboxAccumulator)
-    await this.outboxAccumulator.clear()
+    await outboxStorage.flush(outboxAccumulator)
+    await outboxAccumulator.clear()
   }
 }
 
@@ -59,19 +71,15 @@ export class OutboxPeriodicJob<
   private readonly outboxProcessor: OutboxProcessor<SupportedEvents>
 
   constructor(
-    outboxStorage: OutboxStorage<SupportedEvents>,
-    outboxAccumulator: OutboxAccumulator<SupportedEvents>,
-    eventEmitter: DomainEventEmitter<SupportedEvents>,
+    outboxDependencies: OutboxDependencies<SupportedEvents>,
+    outboxConfiguration: OutboxConfiguration,
     dependencies: PeriodicJobDependencies,
-    maxRetryCount: number,
-    emitBatchSize: number,
-    intervalInMs: number,
   ) {
     super(
       {
         jobId: 'OutboxJob',
         schedule: {
-          intervalInMs: intervalInMs,
+          intervalInMs: outboxConfiguration.jobIntervalInMs,
         },
         singleConsumerMode: {
           enabled: true,
@@ -87,11 +95,9 @@ export class OutboxPeriodicJob<
     )
 
     this.outboxProcessor = new OutboxProcessor<SupportedEvents>(
-      outboxStorage,
-      outboxAccumulator,
-      eventEmitter,
-      maxRetryCount,
-      emitBatchSize,
+      outboxDependencies,
+      outboxConfiguration.maxRetryCount,
+      outboxConfiguration.emitBatchSize,
     )
   }