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-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,167 @@
+import type {
+  OutboxAccumulator,
+  OutboxEntry,
+  OutboxStorage,
+} from '@message-queue-toolkit/outbox-core'
+import { type CommonEventDefinition, getMessageType } from '@message-queue-toolkit/schemas'
+import type { PrismaClient } from '@prisma/client'
+
+type ModelDelegate = {
+  // biome-ignore lint/suspicious/noExplicitAny: <explanation>
+  create: (args: any) => Promise<any>
+  // biome-ignore lint/suspicious/noExplicitAny: <explanation>
+  findMany: (args: any) => Promise<any>
+  // biome-ignore lint/suspicious/noExplicitAny: <explanation>
+  createMany: (args: any) => Promise<any>
+  // biome-ignore lint/suspicious/noExplicitAny: <explanation>
+  updateMany: (args: any) => Promise<any>
+}
+
+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
+
+    // @ts-ignore
+    const messageType = getMessageType(outboxEntry.event)
+    return prismaModel.create({
+      data: {
+        id: outboxEntry.id,
+        type: messageType,
+        created: outboxEntry.created,
+        updated: outboxEntry.updated,
+        data: outboxEntry.data,
+        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
+
+    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 ModelDelegate
+      await this.handleSuccesses(prismaModel, entries, existingEntries)
+      await this.handleFailures(prismaModel, failedEntries, existingEntries)
+    })
+  }
+
+  private async handleSuccesses(
+    prismaModel: ModelDelegate,
+    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,
+          // @ts-ignore
+          type: getMessageType(entry.event),
+          created: entry.created,
+          updated: new Date(),
+          data: entry.data,
+          status: 'SUCCESS',
+        })),
+      })
+    }
+
+    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,
+    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,
+          // @ts-ignore
+          type: getMessageType(entry.event),
+          created: entry.created,
+          updated: new Date(),
+          data: entry.data,
+          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
+
+    return prismaModel.findMany({
+      where: {
+        retryCount: {
+          lte: maxRetryCount,
+        },
+      },
+    })
+  }
+}

packages/outbox-prisma-adapter/package.json

@@ -0,0 +1,66 @@
+{
+  "name": "@message-queue-toolkit/outbox-prisma-adapter",
+  "version": "0.1.0",
+  "private": false,
+  "license": "MIT",
+  "description": "OutboxStorage implementation for @message-queue-toolkit/outbox-core package.",
+  "maintainers": [
+    {
+      "name": "Igor Savin",
+      "email": "[email protected]"
+    }
+  ],
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
+  "scripts": {
+    "build": "del-cli dist && npm run db:update-client && tsc",
+    "build:release": "del-cli dist && del-cli coverage && npm run lint && tsc --project tsconfig.release.json",
+    "test": "vitest",
+    "test:coverage": "npm run docker:start:dev && npm run db:wait && npm test -- --coverage && npm run docker:stop:dev",
+    "test:ci": "npm run test:coverage",
+    "lint": "biome check . && tsc --project tsconfig.json --noEmit",
+    "lint:fix": "biome check --write .",
+    "docker:start:dev": "docker compose up -d",
+    "docker:stop:dev": "docker compose down",
+    "db:wait": "while ! echo \"SELECT 1;\" | prisma db execute --stdin; do sleep 1; done",
+    "db:update-client": "prisma generate",
+    "prepublishOnly": "npm run build:release"
+  },
+  "peerDependencies": {
+    "@message-queue-toolkit/core": ">=14.0.0",
+    "@message-queue-toolkit/outbox-core": ">=0.1.0",
+    "@message-queue-toolkit/schemas": ">=4.0.0",
+    "prisma": "^5.22.0",
+    "@prisma/client": "^5.19.1"
+  },
+  "devDependencies": {
+    "@biomejs/biome": "1.8.3",
+    "@kibertoad/biome-config": "^1.2.1",
+    "@prisma/client": "^5.22.0",
+    "@types/node": "^22.0.0",
+    "@vitest/coverage-v8": "^2.0.4",
+    "del-cli": "^5.1.0",
+    "prisma": "^5.22.0",
+    "typescript": "^5.5.3",
+    "uuidv7": "^1.0.2",
+    "vitest": "^2.0.4",
+    "zod": "^3.23.8"
+  },
+  "homepage": "https://github.com/kibertoad/message-queue-toolkit",
+  "repository": {
+    "type": "git",
+    "url": "git://github.com/kibertoad/message-queue-toolkit.git"
+  },
+  "keywords": [
+    "message",
+    "queue",
+    "queues",
+    "abstract",
+    "common",
+    "utils",
+    "notification",
+    "outbox",
+    "pattern"
+  ],
+  "files": ["README.md", "LICENSE", "dist/*"]
+}

packages/outbox-prisma-adapter/prisma/schema.prisma

@@ -0,0 +1,20 @@
+datasource db {
+  provider = "postgresql"
+  url      = "postgresql://prisma:prisma@localhost:5432/prisma?schema=prisma"
+}
+
+model OutboxEntry {
+  id         String   @id @default(uuid()) @db.Uuid
+  created    DateTime @default(now())
+  updated    DateTime @default(now()) @updatedAt
+  type       String
+  retryCount Int      @default(0) @map("retry_count")
+  data       Json
+  status     String
+
+  @@map("outbox_entry")
+}
+
+generator client {
+  provider = "prisma-client-js"
+}