fix(indexer): wrap event handlers in Prisma transaction to prevent poison-pill events (#168)

server/src/indexer/parsers/event-parser.service.spec.ts

@@ -3,15 +3,20 @@ import { EventParserService } from './event-parser.service';
 import type { RawSorobanEvent } from '../types/soroban-events.types';
 
 type MockPrisma = {
+  $transaction: jest.Mock;
   transaction: { findUnique: jest.Mock; create: jest.Mock };
-  campaign: { upsert: jest.Mock; update: jest.Mock };
+  campaign: {
+    upsert: jest.Mock;
+    update: jest.Mock;
+    updateMany: jest.Mock;
+  };
   user: { upsert: jest.Mock };
   investment: { create: jest.Mock };
   tranche: { create: jest.Mock };
   dispute: { create: jest.Mock; update: jest.Mock; findFirst: jest.Mock };
 };
 
-function makeMockPrisma(): MockPrisma {
+function makeModelMocks() {
   return {
     transaction: {
       findUnique: jest.fn().mockResolvedValue(null),
@@ -20,6 +25,7 @@ function makeMockPrisma(): MockPrisma {
     campaign: {
       upsert: jest.fn().mockResolvedValue({}),
       update: jest.fn().mockResolvedValue({}),
+      updateMany: jest.fn().mockResolvedValue({ count: 1 }),
     },
     user: {
       upsert: jest.fn().mockResolvedValue({}),
@@ -38,6 +44,18 @@ function makeMockPrisma(): MockPrisma {
   };
 }
 
+function makeMockPrisma(): MockPrisma {
+  const models = makeModelMocks();
+  return {
+    $transaction: jest.fn(async (callback: (tx: any) => Promise<void>) => {
+      // Pass the same model mocks as the transaction client so all calls
+      // within the transaction are recorded on the same mocks.
+      await callback(models);
+    }),
+    ...models,
+  };
+}
+
 function rawEvent(
   id: string,
   topic: unknown[],
@@ -689,9 +707,12 @@ describe('EventParserService', () => {
   describe('broadcast-after-persist wiring', () => {
     it('emits a realtime event only after the DB write succeeds', async () => {
       const emitCampaignEvent = jest.fn();
-      const withRealtime = new EventParserService(prisma as any, {
-        emitCampaignEvent,
-      } as any);
+      const withRealtime = new EventParserService(
+        prisma as any,
+        {
+          emitCampaignEvent,
+        } as any,
+      );
 
       await withRealtime.processEvent(
         rawEvent(
@@ -713,9 +734,12 @@ describe('EventParserService', () => {
     it('does not emit for already-persisted (replayed) events', async () => {
       const emitCampaignEvent = jest.fn();
       prisma.transaction.findUnique.mockResolvedValueOnce({ id: 'e-rt2' });
-      const withRealtime = new EventParserService(prisma as any, {
-        emitCampaignEvent,
-      } as any);
+      const withRealtime = new EventParserService(
+        prisma as any,
+        {
+          emitCampaignEvent,
+        } as any,
+      );
 
       await withRealtime.processEvent(
         rawEvent(
@@ -730,9 +754,12 @@ describe('EventParserService', () => {
 
     it('does not emit when the parsed event has no campaignId', async () => {
       const emitCampaignEvent = jest.fn();
-      const withRealtime = new EventParserService(prisma as any, {
-        emitCampaignEvent,
-      } as any);
+      const withRealtime = new EventParserService(
+        prisma as any,
+        {
+          emitCampaignEvent,
+        } as any,
+      );
 
       await withRealtime.processEvent(
         rawEvent(
@@ -745,4 +772,90 @@ describe('EventParserService', () => {
       expect(emitCampaignEvent).not.toHaveBeenCalled();
     });
   });
+
+  describe('transaction atomicity (poison-pill prevention)', () => {
+    it('rolls back all writes when a handler fails mid-way and allows retry', async () => {
+      // First call: campaign.update fails after investment.create would have succeeded.
+      // The transaction should roll back, leaving zero rows.
+      prisma.campaign.update.mockRejectedValueOnce(
+        new Error('transient DB error'),
+      );
+
+      // processEvent catches and logs the error (does not rethrow).
+      await service.processEvent(
+        rawEvent(
+          'e-tx1',
+          ['ContribReceived', CAMPAIGN_ID],
+          [INVESTOR, 1700000000n, 250n],
+        ),
+      );
+
+      // Error should be logged.
+      expect(errorSpy).toHaveBeenCalled();
+
+      // KEY ASSERTION: Transaction row (idempotency marker) NOT created on failure,
+      // so the event can be retried on next poll.
+      expect(prisma.transaction.create).not.toHaveBeenCalled();
+
+      // Second call: same event, transient error resolved -> succeeds.
+      prisma.campaign.update.mockResolvedValueOnce({});
+      errorSpy.mockClear();
+
+      await service.processEvent(
+        rawEvent(
+          'e-tx1', // same event ID - idempotency key
+          ['ContribReceived', CAMPAIGN_ID],
+          [INVESTOR, 1700000000n, 250n],
+        ),
+      );
+
+      // Now the full transaction commits: Transaction row created (event marked processed).
+      expect(prisma.transaction.create).toHaveBeenCalledTimes(1);
+      expect(prisma.transaction.create).toHaveBeenCalledWith(
+        expect.objectContaining({
+          data: expect.objectContaining({
+            id: 'e-tx1',
+            type: 'campaign.invested',
+          }),
+        }),
+      );
+      // No error on retry.
+      expect(errorSpy).not.toHaveBeenCalled();
+    });
+
+    it('marks event as processed only after full transaction commits', async () => {
+      // Simulate a handler that fails on the second write.
+      prisma.campaign.update.mockRejectedValueOnce(
+        new Error('constraint violation'),
+      );
+
+      await service.processEvent(
+        rawEvent(
+          'e-tx2',
+          ['CampaignFunded', CAMPAIGN_ID],
+          [1700000000n, 5000n],
+        ),
+      );
+
+      expect(errorSpy).toHaveBeenCalled();
+      errorSpy.mockClear();
+
+      // Event NOT marked processed (no Transaction row) -> can be retried.
+      expect(prisma.transaction.create).not.toHaveBeenCalled();
+
+      // Retry succeeds.
+      prisma.campaign.update.mockResolvedValueOnce({});
+      await service.processEvent(
+        rawEvent(
+          'e-tx2',
+          ['CampaignFunded', CAMPAIGN_ID],
+          [1700000000n, 5000n],
+        ),
+      );
+
+      // Now marked processed exactly once.
+      expect(prisma.transaction.create).toHaveBeenCalledTimes(1);
+      expect(errorSpy).not.toHaveBeenCalled();
+    });
+  });
 });