Failed entries handling.

Author: kamilwylegala

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

@@ -25,52 +25,110 @@ export class OutboxPrismaAdapter<SupportedEvents extends CommonEventDefinition[]
         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: PrismaClient[typeof this.modelName] = this.prisma[this.modelName]
 
     const existingEntries = await prismaModel.findMany({
       where: {
         id: {
-          in: entries.map((entry) => entry.id),
+          in: [...entries.map((entry) => entry.id), ...failedEntries.map((entry) => entry.id)],
         },
       },
     })
 
+    await this.handleSuccesses(prismaModel, entries, existingEntries)
+    await this.handleFailures(prismaModel, failedEntries, existingEntries)
+  }
+
+  private async handleSuccesses(
+    prismaModel: PrismaClient[typeof this.modelName],
+    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),
     )
 
-    await prismaModel.createMany({
-      data: toCreate.map((entry) => ({
-        id: entry.id,
-        type: getMessageType(entry.event),
-        created: entry.created,
-        updated: new Date(),
-        data: entry.data,
-        status: 'SUCCESS',
-      })),
-    })
+    if (toCreate.length > 0) {
+      await prismaModel.createMany({
+        data: toCreate.map((entry) => ({
+          id: entry.id,
+          type: getMessageType(entry.event),
+          created: entry.created,
+          updated: new Date(),
+          data: entry.data,
+          status: 'SUCCESS',
+        })),
+      })
+    }
 
-    await prismaModel.updateMany({
-      where: {
-        id: {
-          in: toUpdate.map((entry) => entry.id),
+    if (toUpdate.length > 0) {
+      await prismaModel.updateMany({
+        where: {
+          id: {
+            in: toUpdate.map((entry) => entry.id),
+          },
         },
-      },
-      data: {
-        status: 'SUCCESS',
-        updated: new Date(),
-      },
-    })
+        data: {
+          status: 'SUCCESS',
+          updated: new Date(),
+        },
+      })
+    }
+  }
+
+  private async handleFailures(
+    prismaModel: PrismaClient[typeof this.modelName],
+    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: 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]>[]> {

packages/outbox-prisma-adapter/test/outbox-prisma-adapter.spec.ts

@@ -206,4 +206,40 @@ describe('outbox-prisma-adapter', () => {
       },
     ])
   })
+
+  it("should change failed entries' status to 'FAILED' and increment retry count", async () => {
+    const accumulator = new InMemoryOutboxAccumulator<SupportedEvents>()
+    accumulator.addFailure(ENTRY_1)
+
+    await outboxPrismaAdapter.flush(accumulator)
+
+    const afterFirstFlush = await outboxPrismaAdapter.getEntries(10)
+    expect(afterFirstFlush).toMatchObject([
+      {
+        id: ENTRY_1.id,
+        status: 'FAILED',
+        retryCount: 1,
+      },
+    ])
+  })
+
+  it('should change failed EXISTING entries status to FAILED and increment retry count', async () => {
+    const accumulator = new InMemoryOutboxAccumulator<SupportedEvents>()
+    const failedEntry = { ...ENTRY_1, retryCount: 3, status: 'FAILED' } satisfies OutboxEntry<
+      SupportedEvents[number]
+    >
+    accumulator.addFailure(failedEntry)
+
+    await outboxPrismaAdapter.createEntry(failedEntry)
+    await outboxPrismaAdapter.flush(accumulator)
+
+    const afterFirstFlush = await outboxPrismaAdapter.getEntries(10)
+    expect(afterFirstFlush).toMatchObject([
+      {
+        id: failedEntry.id,
+        status: 'FAILED',
+        retryCount: 4,
+      },
+    ])
+  })
 })