Outbox prisma adapter

.github/workflows/linting.yml

@@ -30,5 +30,9 @@ jobs:
         run: |
           npm install --ignore-scripts
 
+      - name: Build
+        run: |
+          npm run build
+
       - name: Run lint
         run: npm run lint

.gitignore

@@ -130,3 +130,6 @@ dist
 .pnp.*
 /.idea
 /package-lock.json
+
+# prisma
+db-client

biome.json

@@ -1,12 +1,15 @@
 {
-	"$schema": "./node_modules/@biomejs/biome/configuration_schema.json",
-	"extends": ["./node_modules/@kibertoad/biome-config/configs/biome-package.json"],
-	"linter": {
-		"rules": {
-			"performance": {
-				"noBarrelFile": "off",
-				"noReExportAll": "off"
-			}
-		}
-	}
+  "$schema": "./node_modules/@biomejs/biome/configuration_schema.json",
+  "extends": ["./node_modules/@kibertoad/biome-config/configs/biome-package.json"],
+  "linter": {
+    "rules": {
+      "performance": {
+        "noBarrelFile": "off",
+        "noReExportAll": "off"
+      }
+    }
+  },
+  "files": {
+    "ignore": ["db-client"]
+  }
 }

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(),
     })
 

packages/outbox-prisma-adapter/README.md

@@ -0,0 +1,18 @@
+# outbox-prisma-adapter
+
+This package provides a Prisma adapter for the Outbox pattern.
+
+### Development
+
+#### Tests
+
+To run the tests, you need to have a PostgreSQL database running. You can use the following command to start a PostgreSQL database using Docker:
+
+```sh
+docker-compose up -d
+```
+
+Then update Prisma client:
+```sh
+npx prisma generate --schema=./test/schema.prisma
+```

packages/outbox-prisma-adapter/docker-compose.yml

@@ -0,0 +1,10 @@
+services:
+
+    postgres:
+        image: postgres:16.2
+        environment:
+            POSTGRES_USER: prisma
+            POSTGRES_PASSWORD: prisma
+            POSTGRES_DB: prisma
+        ports:
+            - 5432:5432

packages/outbox-prisma-adapter/index.ts

@@ -0,0 +1 @@
+export * from './lib/outbox-prisma-adapter'

packages/outbox-prisma-adapter/lib/outbox-prisma-adapter.ts

@@ -0,0 +1,190 @@
+import type {
+  OutboxAccumulator,
+  OutboxEntry,
+  OutboxStorage,
+} from '@message-queue-toolkit/outbox-core'
+import type { CommonEventDefinition } from '@message-queue-toolkit/schemas'
+import type { Prisma, PrismaClient } from '@prisma/client'
+
+export type EnrichedOutboxEntry<Event extends CommonEventDefinition> = OutboxEntry<Event> & {
+  type: string
+}
+
+type ModelDelegate<Event extends CommonEventDefinition> = {
+  create: (args: { data: EnrichedOutboxEntry<Event> }) => Promise<EnrichedOutboxEntry<Event>>
+  findMany: (args: {
+    where:
+      | Partial<EnrichedOutboxEntry<Event>>
+      | {
+          id?: Prisma.StringFilter
+          retryCount?: Prisma.IntFilter
+        }
+  }) => Promise<EnrichedOutboxEntry<Event>[]>
+  createMany: (args: { data: EnrichedOutboxEntry<Event>[] }) => Promise<void>
+  updateMany: (args: {
+    where: {
+      id: {
+        in: string[]
+      }
+    }
+    data:
+      | Partial<OutboxEntry<Event>>
+      | {
+          retryCount?: number | Prisma.IntFieldUpdateOperationsInput
+        }
+  }) => Promise<void>
+}
+
+export class OutboxPrismaAdapter<
+  SupportedEvents extends CommonEventDefinition[],
+  ModelName extends keyof PrismaClient & string,
+> implements OutboxStorage<SupportedEvents>
+{
+  constructor(
+    private readonly prisma: PrismaClient,
+    private readonly modelName: ModelName,
+  ) {}
+
+  createEntry(
+    outboxEntry: OutboxEntry<SupportedEvents[number]>,
+  ): Promise<OutboxEntry<SupportedEvents[number]>> {
+    const prismaModel = this.prisma[this.modelName] as unknown as ModelDelegate<
+      SupportedEvents[number]
+    >
+
+    return prismaModel.create({
+      data: {
+        id: outboxEntry.id,
+        type: outboxEntry.event.type,
+        created: outboxEntry.created,
+        updated: outboxEntry.updated,
+        event: outboxEntry.event,
+        status: outboxEntry.status,
+        retryCount: outboxEntry.retryCount,
+      },
+    })
+  }
+
+  async flush(outboxAccumulator: OutboxAccumulator<SupportedEvents>): Promise<void> {
+    const entries = await outboxAccumulator.getEntries()
+    const failedEntries = await outboxAccumulator.getFailedEntries()
+    const prismaModel = this.prisma[this.modelName] as unknown as ModelDelegate<
+      SupportedEvents[number]
+    >
+
+    const existingEntries = await prismaModel.findMany({
+      where: {
+        id: {
+          in: [...entries.map((entry) => entry.id), ...failedEntries.map((entry) => entry.id)],
+        },
+      },
+    })
+
+    await this.prisma.$transaction(async (prisma) => {
+      const prismaModel = prisma[this.modelName] as unknown as ModelDelegate<
+        SupportedEvents[number]
+      >
+      await this.handleSuccesses(prismaModel, entries, existingEntries)
+      await this.handleFailures(prismaModel, failedEntries, existingEntries)
+    })
+  }
+
+  private async handleSuccesses(
+    prismaModel: ModelDelegate<SupportedEvents[number]>,
+    entries: OutboxEntry<SupportedEvents[number]>[],
+    existingEntries: OutboxEntry<SupportedEvents[number]>[],
+  ) {
+    const toCreate = entries.filter(
+      (entry) => !existingEntries.some((existingEntry) => existingEntry.id === entry.id),
+    )
+    const toUpdate = entries.filter((entry) =>
+      existingEntries.some((existingEntry) => existingEntry.id === entry.id),
+    )
+
+    if (toCreate.length > 0) {
+      await prismaModel.createMany({
+        data: toCreate.map((entry) => ({
+          id: entry.id,
+          type: entry.event.type,
+          created: entry.created,
+          updated: new Date(),
+          event: entry.event,
+          status: 'SUCCESS',
+          retryCount: entry.retryCount,
+        })),
+      })
+    }
+
+    if (toUpdate.length > 0) {
+      await prismaModel.updateMany({
+        where: {
+          id: {
+            in: toUpdate.map((entry) => entry.id),
+          },
+        },
+        data: {
+          status: 'SUCCESS',
+          updated: new Date(),
+        },
+      })
+    }
+  }
+
+  private async handleFailures(
+    prismaModel: ModelDelegate<SupportedEvents[number]>,
+    entries: OutboxEntry<SupportedEvents[number]>[],
+    existingEntries: OutboxEntry<SupportedEvents[number]>[],
+  ) {
+    const toCreate = entries.filter(
+      (entry) => !existingEntries.some((existingEntry) => existingEntry.id === entry.id),
+    )
+    const toUpdate = entries.filter((entry) =>
+      existingEntries.some((existingEntry) => existingEntry.id === entry.id),
+    )
+
+    if (toCreate.length > 0) {
+      await prismaModel.createMany({
+        data: toCreate.map((entry) => ({
+          id: entry.id,
+          type: entry.event.type,
+          created: entry.created,
+          updated: new Date(),
+          event: entry.event,
+          status: 'FAILED',
+          retryCount: 1,
+        })),
+      })
+    }
+
+    if (toUpdate.length > 0) {
+      await prismaModel.updateMany({
+        where: {
+          id: {
+            in: toUpdate.map((entry) => entry.id),
+          },
+        },
+        data: {
+          status: 'FAILED',
+          updated: new Date(),
+          retryCount: {
+            increment: 1,
+          },
+        },
+      })
+    }
+  }
+
+  getEntries(maxRetryCount: number): Promise<OutboxEntry<SupportedEvents[number]>[]> {
+    const prismaModel = this.prisma[this.modelName] as unknown as ModelDelegate<
+      SupportedEvents[number]
+    >
+
+    return prismaModel.findMany({
+      where: {
+        retryCount: {
+          lte: maxRetryCount,
+        },
+      },
+    })
+  }
+}