AP-5046 outbox-core package for transactional outbox pattern

packages/outbox-core/README.md

@@ -0,0 +1,42 @@
+# outbox-core
+
+Main package that contains the core functionality of the Outbox pattern to provide "at least once" delivery semantics for messages.
+
+## Installation
+
+```bash
+npm i -S @message-queue-toolkit/outbox-core
+```
+
+## Usage
+
+To process outbox entries and emit them to the message queue, you need to create an instance of the `OutboxPeriodicJob` class:
+
+```typescript
+import { OutboxPeriodicJob } from '@message-queue-toolkit/outbox-core';
+
+const job = new OutboxPeriodicJob(
+    //Implementation of OutboxStorage interface, TODO: Point to other packages in message-queue-toolkit
+    outboxStorage, 
+    //Default available accumulator for gathering outbox entries as the process job is progressing.
+    new InMemoryOutboxAccumulator(),
+    //DomainEventEmitter, it will be used to publish events, see @message-queue-toolkit/core
+    eventEmitter,
+    //See PeriodicJobDependencies from @lokalise/background-jobs-common
+    dependencies,
+    //Retry count, how many times outbox entries should be retried to be processed
+    3,
+    //emitBatchSize - how many outbox entries should be emitted at once
+    10,
+    //internalInMs - how often the job should be executed, e.g. below it runs every 1sec
+    1000
+)
+```
+
+Job will take care of processing outbox entries emitted by:
+```typescript
+const emitter = new OutboxEventEmitter(
+    //Same instance of outbox storage that is used by OutboxPeriodicJob
+    outboxStorage
+)
+```

packages/outbox-core/index.ts

@@ -0,0 +1,4 @@
+export * from './lib/outbox'
+export * from './lib/objects'
+export * from './lib/accumulators'
+export * from './lib/storage'

packages/outbox-core/lib/accumulators.ts

@@ -0,0 +1,73 @@
+import type { CommonEventDefinition } from '@message-queue-toolkit/schemas'
+import type { OutboxEntry } from './objects.ts'
+
+/**
+ * Accumulator is responsible for storing outbox entries in two cases:
+ * - successfully dispatched event
+ * - failed events
+ *
+ * Thanks to this, we can use aggregated result and persist in the storage in batches.
+ */
+export interface OutboxAccumulator<SupportedEvents extends CommonEventDefinition[]> {
+  /**
+   * Accumulates successfully dispatched event.
+   * @param outboxEntry
+   */
+  add(outboxEntry: OutboxEntry<SupportedEvents[number]>): Promise<void>
+
+  /**
+   * Accumulates failed event.
+   * @param outboxEntry
+   */
+  addFailure(outboxEntry: OutboxEntry<SupportedEvents[number]>): Promise<void>
+
+  /**
+   * It's meant to be used by OutboxStorage::flush() to get all entries that should be persisted as successful ones.
+   */
+  getEntries(): Promise<OutboxEntry<SupportedEvents[number]>[]>
+
+  /**
+   * Also used by OutboxStorage::flush() to get all entries that should be persisted as failed ones. Such entries will be retried + their retryCount will be incremented.
+   */
+  getFailedEntries(): Promise<OutboxEntry<SupportedEvents[number]>[]>
+
+  /**
+   * After running clear(), no entries should be returned by getEntries() and getFailedEntries().
+   *
+   * clear() is always called after flush() in OutboxStorage.
+   */
+  clear(): Promise<void>
+}
+
+export class InMemoryOutboxAccumulator<SupportedEvents extends CommonEventDefinition[]>
+  implements OutboxAccumulator<SupportedEvents>
+{
+  private entries: OutboxEntry<SupportedEvents[number]>[] = []
+  private failedEntries: OutboxEntry<SupportedEvents[number]>[] = []
+
+  public add(outboxEntry: OutboxEntry<SupportedEvents[number]>) {
+    this.entries = [...this.entries, outboxEntry]
+
+    return Promise.resolve()
+  }
+
+  public addFailure(outboxEntry: OutboxEntry<SupportedEvents[number]>) {
+    this.failedEntries = [...this.failedEntries, outboxEntry]
+
+    return Promise.resolve()
+  }
+
+  getEntries(): Promise<OutboxEntry<SupportedEvents[number]>[]> {
+    return Promise.resolve(this.entries)
+  }
+
+  getFailedEntries(): Promise<OutboxEntry<SupportedEvents[number]>[]> {
+    return Promise.resolve(this.failedEntries)
+  }
+
+  public clear(): Promise<void> {
+    this.entries = []
+    this.failedEntries = []
+    return Promise.resolve()
+  }
+}

packages/outbox-core/lib/objects.ts

@@ -0,0 +1,25 @@
+import type {
+  CommonEventDefinition,
+  CommonEventDefinitionPublisherSchemaType,
+  ConsumerMessageMetadataType,
+} from '@message-queue-toolkit/schemas'
+
+/**
+ * Status of the outbox entry.
+ * - CREATED - entry was created and is waiting to be processed to publish actual event
+ * - ACKED - entry was picked up by outbox job and is being processed
+ * - SUCCESS - entry was successfully processed, event was published
+ * - FAILED - entry processing failed, it will be retried
+ */
+export type OutboxEntryStatus = 'CREATED' | 'ACKED' | 'SUCCESS' | 'FAILED'
+
+export type OutboxEntry<SupportedEvent extends CommonEventDefinition> = {
+  id: string
+  event: SupportedEvent
+  data: Omit<CommonEventDefinitionPublisherSchemaType<SupportedEvent>, 'type'>
+  precedingMessageMetadata?: Partial<ConsumerMessageMetadataType>
+  status: OutboxEntryStatus
+  created: Date
+  updated?: Date
+  retryCount: number
+}