From 97e6abd83f09710ee7333f8e305e558bc57ef1b5 Mon Sep 17 00:00:00 2001 From: d1a-m3n Date: Sun, 30 Aug 2026 02:08:39 +0100 Subject: [PATCH 1/2] fix: make soroban event processing idempotent --- backend/src/workers/soroban-event-worker.ts | 54 +++++++------ backend/tests/soroban-event-worker.test.ts | 88 ++++++++++++++++----- 2 files changed, 98 insertions(+), 44 deletions(-) diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 86b76396..a99e79e0 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -1013,39 +1013,41 @@ export class SorobanEventWorker { const token = decodeAddress(body["token"]); const timestamp = Math.floor(Date.now() / 1000); - const existingEvent = await prisma.streamEvent.findUnique({ - where: { - transactionHash_eventType: { - transactionHash: event.txHash, - eventType: "FEE_COLLECTED", - }, - }, - select: { id: true }, - }); - if (existingEvent) { - logger.warn( - `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=FEE_COLLECTED`, - ); - } else { - await prisma.streamEvent.upsert({ + await prisma.$transaction(async (tx: Prisma.TransactionClient) => { + const existingEvent = await tx.streamEvent.findUnique({ where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: "FEE_COLLECTED", }, }, - create: { - streamId, - eventType: "FEE_COLLECTED", - amount: feeAmount, - transactionHash: event.txHash, - ledgerSequence: event.ledger, - timestamp, - metadata: JSON.stringify({ treasury, token }), - }, - update: {}, + select: { id: true }, }); - } + if (existingEvent) { + logger.warn( + `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=FEE_COLLECTED`, + ); + } else { + await tx.streamEvent.upsert({ + where: { + transactionHash_eventType: { + transactionHash: event.txHash, + eventType: "FEE_COLLECTED", + }, + }, + create: { + streamId, + eventType: "FEE_COLLECTED", + amount: feeAmount, + transactionHash: event.txHash, + ledgerSequence: event.ledger, + timestamp, + metadata: JSON.stringify({ treasury, token }), + }, + update: {}, + }); + } + }); // Broadcast to admin channel for treasury reporting sseService.broadcastToAdmin("stream.fee_collected", { diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 8d7cf386..5e598a1b 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -228,47 +228,99 @@ describe('SorobanEventWorker', () => { inSuccessfulContractCall: true, topic: [ { switch: () => ({ value: 0 }), sym: () => 'fee_collected' } as any, - { switch: () => ({ value: 1 }), u64: () => ({ toString: () => streamId.toString() }) } as any, + { + switch: () => ({ value: 1 }), + u64: () => ({ toString: () => streamId.toString() }), + } as any, ], value: { switch: () => ({ value: 4 }), map: () => [ - { key: () => ({ sym: () => 'treasury' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, - { key: () => ({ sym: () => 'fee_amount' }), val: () => ({ i128: () => ({ hi: () => ({ toString: () => '0' }), lo: () => ({ toString: () => '1000' }) }) }) }, - { key: () => ({ sym: () => 'token' }), val: () => ({ address: () => ({ switch: () => ({ value: 1 }), contractId: () => Buffer.alloc(32) }) }) }, + { + key: () => ({ sym: () => 'treasury' }), + val: () => ({ + address: () => ({ + switch: () => ({ value: 0 }), + accountId: () => ({ + ed25519: () => Buffer.alloc(32), + }), + }), + }), + }, + { + key: () => ({ sym: () => 'fee_amount' }), + val: () => ({ + i128: () => ({ + hi: () => ({ toString: () => '0' }), + lo: () => ({ toString: () => '1000' }), + }), + }), + }, + { + key: () => ({ sym: () => 'token' }), + val: () => ({ + address: () => ({ + switch: () => ({ value: 1 }), + contractId: () => Buffer.alloc(32), + }), + }), + }, ] as any, } as any, }; - // First call: event doesn't exist - (prisma.streamEvent.findUnique as ReturnType).mockResolvedValueOnce(null); - (prisma.streamEvent.upsert as ReturnType).mockResolvedValueOnce({ + const mockTx = { + streamEvent: { + findUnique: vi.fn(), + upsert: vi.fn(), + }, + }; + + (prisma.$transaction as ReturnType).mockImplementation( + (callback) => callback(mockTx), + ); + + // First call: event doesn't exist. + mockTx.streamEvent.findUnique.mockResolvedValueOnce(null); + mockTx.streamEvent.upsert.mockResolvedValueOnce({ id: 'fee-event-1', transactionHash: txHash, eventType: 'FEE_COLLECTED', }); - await (worker as any).handleFeeCollected(mockEvent, mockEvent.topic![1]); - expect(prisma.streamEvent.findUnique).toHaveBeenCalledTimes(1); - expect(prisma.streamEvent.upsert).toHaveBeenCalledTimes(1); + await (worker as any).handleFeeCollected( + mockEvent, + mockEvent.topic![1], + ); + + expect(prisma.$transaction).toHaveBeenCalledTimes(1); + expect(mockTx.streamEvent.findUnique).toHaveBeenCalledTimes(1); + expect(mockTx.streamEvent.upsert).toHaveBeenCalledTimes(1); expect(logger.warn).not.toHaveBeenCalled(); - // Reset mocks vi.clearAllMocks(); - // Second call: event exists (duplicate) - (prisma.streamEvent.findUnique as ReturnType).mockResolvedValueOnce({ + // Second call: event already exists, so no second row is created. + mockTx.streamEvent.findUnique.mockResolvedValueOnce({ id: 'fee-event-1', }); - await (worker as any).handleFeeCollected(mockEvent, mockEvent.topic![1]); - expect(prisma.streamEvent.findUnique).toHaveBeenCalledTimes(1); - expect(prisma.streamEvent.upsert).not.toHaveBeenCalled(); + (prisma.$transaction as ReturnType).mockImplementation( + (callback) => callback(mockTx), + ); + + await (worker as any).handleFeeCollected( + mockEvent, + mockEvent.topic![1], + ); + + expect(prisma.$transaction).toHaveBeenCalledTimes(1); + expect(mockTx.streamEvent.findUnique).toHaveBeenCalledTimes(1); + expect(mockTx.streamEvent.upsert).not.toHaveBeenCalled(); expect(logger.warn).toHaveBeenCalledWith( - expect.stringContaining('Duplicate StreamEvent skipped') + expect.stringContaining('Duplicate StreamEvent skipped'), ); }); - it('should process fee_config_updated events successfully', async () => { const txHash = 'fee-config-tx-hash'; From ae79336f412e05ab3cf523c1e380a3a4e1625a91 Mon Sep 17 00:00:00 2001 From: d1a-m3n Date: Mon, 31 Aug 2026 00:45:22 +0100 Subject: [PATCH 2/2] fix: restore unimplemented pause and resume behavior --- backend/src/controllers/stream.controller.ts | 72 ++--------- .../pause-resume.regression.test.ts | 2 +- .../tests/integration/stream-actions.test.ts | 112 ++++++++---------- backend/tests/stream.controller.test.ts | 46 +++---- 4 files changed, 85 insertions(+), 147 deletions(-) diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts index 394c11ad..2b0f672e 100644 --- a/backend/src/controllers/stream.controller.ts +++ b/backend/src/controllers/stream.controller.ts @@ -9,8 +9,6 @@ import { getClaimableFromChain, isStale, topUpStream, - pauseStream as sorobanPauseStream, - resumeStream as sorobanResumeStream, } from "../services/sorobanService.js"; import type { AuthenticatedRequest } from "../types/auth.types.js"; import { parseStreamId } from "../lib/stream-id.js"; @@ -804,36 +802,11 @@ export const pauseStream = async (req: Request, res: Response) => { }); } - try { - // Call Soroban service to verify the pause operation would succeed - const result = await sorobanPauseStream( - authReq.user.publicKey, - parsedStreamId, - ); - - logger.info( - `Stream ${parsedStreamId} pause simulated by ${authReq.user.publicKey}`, - ); - - return res.status(200).json({ - success: true, - streamId: parsedStreamId, - txHash: result.txHash, - stream, - }); - } catch (sorobanError) { - logger.error( - `Soroban pause failed for stream ${parsedStreamId}:`, - sorobanError, - ); - return res.status(400).json({ - error: "Failed to pause stream on chain", - message: - sorobanError instanceof Error - ? sorobanError.message - : "Unknown error", - }); - } + return res.status(501).json({ + error: "Not Implemented", + message: + "Pausing streams is not currently supported because the on-chain transaction is not yet submitted.", + }); } catch (error) { logger.error("Error pausing stream:", error); return res.status(500).json({ error: "Internal server error" }); @@ -887,36 +860,11 @@ export const resumeStream = async (req: Request, res: Response) => { }); } - try { - // Call Soroban service to verify the resume operation would succeed - const result = await sorobanResumeStream( - authReq.user.publicKey, - parsedStreamId, - ); - - logger.info( - `Stream ${parsedStreamId} resume simulated by ${authReq.user.publicKey}`, - ); - - return res.status(200).json({ - success: true, - streamId: parsedStreamId, - txHash: result.txHash, - stream, - }); - } catch (sorobanError) { - logger.error( - `Soroban resume failed for stream ${parsedStreamId}:`, - sorobanError, - ); - return res.status(400).json({ - error: "Failed to resume stream on chain", - message: - sorobanError instanceof Error - ? sorobanError.message - : "Unknown error", - }); - } + return res.status(501).json({ + error: "Not Implemented", + message: + "Resuming streams is not currently supported because the on-chain transaction is not yet submitted.", + }); } catch (error) { logger.error("Error resuming stream:", error); return res.status(500).json({ error: "Internal server error" }); diff --git a/backend/tests/integration/pause-resume.regression.test.ts b/backend/tests/integration/pause-resume.regression.test.ts index c5076f24..04af2d12 100644 --- a/backend/tests/integration/pause-resume.regression.test.ts +++ b/backend/tests/integration/pause-resume.regression.test.ts @@ -97,7 +97,7 @@ describe('Regression #804: Pause/resume controller duplicate StreamEvent', () => .post(`/v1/streams/${streamId}/pause`) .set('Authorization', `Bearer ${token}`); - expect(pauseRes.status).toBe(200); + expect(pauseRes.status).toBe(501); // Controller should NOT write to DB for PAUSED event expect(mockPrisma.streamEvent.create).not.toHaveBeenCalled(); diff --git a/backend/tests/integration/stream-actions.test.ts b/backend/tests/integration/stream-actions.test.ts index 07211c3a..3be7215d 100644 --- a/backend/tests/integration/stream-actions.test.ts +++ b/backend/tests/integration/stream-actions.test.ts @@ -92,41 +92,35 @@ describe('stream action routes', () => { mockPrisma.streamEvent.count.mockResolvedValue(0); }); - it('POST /v1/streams/:streamId/pause pauses an active sender-owned stream', async () => { - const sender = makeKeypair(); - const token = await getValidJwt(sender); - - mockPrisma.stream.findUnique.mockResolvedValue({ - streamId: 7, - sender: sender.publicKey(), - recipient: makeKeypair().publicKey(), - isActive: true, - isPaused: false, - pausedAt: null, - totalPausedDuration: 0, - }); - mockPauseStream.mockResolvedValue({ txHash: 'pause-tx-hash' }); - mockPrisma.stream.update.mockResolvedValue({ - streamId: 7, - isActive: true, - isPaused: true, - pausedAt: 1700000000, - totalPausedDuration: 0, - }); + it('POST /v1/streams/:streamId/pause returns 501 when pausing is not implemented', async () => { + const sender = makeKeypair(); + const token = await getValidJwt(sender); + + mockPrisma.stream.findUnique.mockResolvedValue({ + streamId: 7, + sender: sender.publicKey(), + recipient: makeKeypair().publicKey(), + isActive: true, + isPaused: false, + pausedAt: null, + totalPausedDuration: 0, + }); - const response = await request(app) - .post('/v1/streams/7/pause') - .set('Authorization', `Bearer ${token}`); + const response = await request(app) + .post('/v1/streams/7/pause') + .set('Authorization', `Bearer ${token}`); - expect(response.status).toBe(200); - expect(response.body).toMatchObject({ - success: true, - streamId: 7, - txHash: 'pause-tx-hash', - }); - expect(mockPauseStream).toHaveBeenCalledWith(sender.publicKey(), 7n); + expect(response.status).toBe(501); + expect(response.body).toMatchObject({ + error: 'Not Implemented', + message: + 'Pausing streams is not currently supported because the on-chain transaction is not yet submitted.', }); + expect(mockPauseStream).not.toHaveBeenCalled(); + expect(mockPrisma.stream.update).not.toHaveBeenCalled(); +}); + it('rejects a raw signed transaction bearer token without a JWT', async () => { const sender = makeKeypair(); const rawToken = buildSignedTransaction(sender, '00'.repeat(32)); @@ -142,41 +136,35 @@ describe('stream action routes', () => { }); }); - it('POST /v1/streams/:streamId/resume resumes a paused sender-owned stream', async () => { - const sender = makeKeypair(); - const token = await getValidJwt(sender); - - mockPrisma.stream.findUnique.mockResolvedValue({ - streamId: 9, - sender: sender.publicKey(), - recipient: makeKeypair().publicKey(), - isActive: true, - isPaused: true, - pausedAt: Math.floor(Date.now() / 1000) - 30, - totalPausedDuration: 10, - }); - mockResumeStream.mockResolvedValue({ txHash: 'resume-tx-hash' }); - mockPrisma.stream.update.mockResolvedValue({ - streamId: 9, - isActive: true, - isPaused: false, - pausedAt: null, - totalPausedDuration: 40, - }); + it('POST /v1/streams/:streamId/resume returns 501 when resuming is not implemented', async () => { + const sender = makeKeypair(); + const token = await getValidJwt(sender); + + mockPrisma.stream.findUnique.mockResolvedValue({ + streamId: 9, + sender: sender.publicKey(), + recipient: makeKeypair().publicKey(), + isActive: true, + isPaused: true, + pausedAt: Math.floor(Date.now() / 1000) - 30, + totalPausedDuration: 10, + }); - const response = await request(app) - .post('/v1/streams/9/resume') - .set('Authorization', `Bearer ${token}`); + const response = await request(app) + .post('/v1/streams/9/resume') + .set('Authorization', `Bearer ${token}`); - expect(response.status).toBe(200); - expect(response.body).toMatchObject({ - success: true, - streamId: 9, - txHash: 'resume-tx-hash', - }); - expect(mockResumeStream).toHaveBeenCalledWith(sender.publicKey(), 9n); + expect(response.status).toBe(501); + expect(response.body).toMatchObject({ + error: 'Not Implemented', + message: + 'Resuming streams is not currently supported because the on-chain transaction is not yet submitted.', }); + expect(mockResumeStream).not.toHaveBeenCalled(); + expect(mockPrisma.stream.update).not.toHaveBeenCalled(); +}); + it('POST /v1/streams/:streamId/withdraw withdraws the claimable amount for the recipient', async () => { const recipient = makeKeypair(); const token = await getValidJwt(recipient); diff --git a/backend/tests/stream.controller.test.ts b/backend/tests/stream.controller.test.ts index e53c9999..0407525c 100644 --- a/backend/tests/stream.controller.test.ts +++ b/backend/tests/stream.controller.test.ts @@ -283,36 +283,37 @@ describe("Stream Controller", () => { }); }); - describe("pauseStream", () => { - it("should pause stream", async () => { + describe("pauseStream", () => { + it("should return 501 when pausing is not implemented", async () => { req.params = { streamId: "123" }; - req.body = { secret: "S123" }; (req as any).user = { publicKey: "GUSER1" }; + (prisma.stream.findUnique as any).mockResolvedValue({ streamId: 123, sender: "GUSER1", isPaused: false, isActive: true, }); - (sorobanService.pauseStream as any).mockResolvedValue({ - txHash: "tx123", - }); - (prisma.stream.update as any).mockResolvedValue({ - streamId: 123, - isPaused: true, - }); await pauseStream(req as Request, res as Response); - expect(res.status).toHaveBeenCalledWith(200); + expect(res.status).toHaveBeenCalledWith(501); + expect(res.json).toHaveBeenCalledWith({ + error: "Not Implemented", + message: + "Pausing streams is not currently supported because the on-chain transaction is not yet submitted.", + }); + + expect(sorobanService.pauseStream).not.toHaveBeenCalled(); + expect(prisma.stream.update).not.toHaveBeenCalled(); }); }); describe("resumeStream", () => { - it("should resume stream", async () => { + it("should return 501 when resuming is not implemented", async () => { req.params = { streamId: "123" }; - req.body = { secret: "S123" }; (req as any).user = { publicKey: "GUSER1" }; + (prisma.stream.findUnique as any).mockResolvedValue({ streamId: 123, sender: "GUSER1", @@ -320,17 +321,18 @@ describe("Stream Controller", () => { isActive: true, pausedAt: Math.floor(Date.now() / 1000), }); - (sorobanService.resumeStream as any).mockResolvedValue({ - txHash: "tx123", - }); - (prisma.stream.update as any).mockResolvedValue({ - streamId: 123, - isPaused: false, - }); await resumeStream(req as Request, res as Response); - expect(res.status).toHaveBeenCalledWith(200); + expect(res.status).toHaveBeenCalledWith(501); + expect(res.json).toHaveBeenCalledWith({ + error: "Not Implemented", + message: + "Resuming streams is not currently supported because the on-chain transaction is not yet submitted.", + }); + + expect(sorobanService.resumeStream).not.toHaveBeenCalled(); + expect(prisma.stream.update).not.toHaveBeenCalled(); }); }); -}); +}); \ No newline at end of file