diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts index 1d273a79..5ee44a92 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"; @@ -831,36 +829,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" }); @@ -914,36 +887,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/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 495e2539..f215460e 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/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 ac6a42b8..5acb5187 100644 --- a/backend/tests/integration/stream-actions.test.ts +++ b/backend/tests/integration/stream-actions.test.ts @@ -94,41 +94,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)); @@ -144,41 +138,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/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'; diff --git a/backend/tests/stream.controller.test.ts b/backend/tests/stream.controller.test.ts index 4471655d..849e0f4b 100644 --- a/backend/tests/stream.controller.test.ts +++ b/backend/tests/stream.controller.test.ts @@ -295,7 +295,6 @@ describe("Stream Controller", () => { it("should cap outgoing and incoming streams to MAX_USER_STREAMS (Issue #1246)", async () => { req.params = { address: "GUSER1" }; - // Simulate a wallet with more streams than the cap const manyStreams = Array.from({ length: MAX_USER_STREAMS + 100 }, (_, i) => ({ streamId: i, ratePerSecond: "10", @@ -310,11 +309,10 @@ describe("Stream Controller", () => { updatedAt: new Date(), })); - // findMany is called twice (outgoing + incoming), each returns at most MAX_USER_STREAMS const cappedStreams = manyStreams.slice(0, MAX_USER_STREAMS); (prisma.stream.findMany as any) - .mockResolvedValueOnce(cappedStreams) // outgoing - .mockResolvedValueOnce(cappedStreams); // incoming + .mockResolvedValueOnce(cappedStreams) + .mockResolvedValueOnce(cappedStreams); (claimableAmountService.getClaimableAmount as any).mockReturnValue({ claimableAmount: "0", @@ -324,9 +322,8 @@ describe("Stream Controller", () => { expect(res.status).toHaveBeenCalledWith(200); const body = (res.json as any).mock.calls[0][0]; - // The response should reflect only the capped result set expect(body.totalStreamsCreated).toBe(MAX_USER_STREAMS); - // Both findMany calls should have received take: MAX_USER_STREAMS + const findManyCalls = (prisma.stream.findMany as any).mock.calls; for (const call of findManyCalls) { expect(call[0].take).toBe(MAX_USER_STREAMS); @@ -382,7 +379,6 @@ describe("Stream Controller", () => { })); const empty: any[] = []; - // Outgoing hits cap, incoming is empty (prisma.stream.findMany as any) .mockResolvedValueOnce(atCap) .mockResolvedValueOnce(empty); @@ -399,35 +395,42 @@ describe("Stream Controller", () => { }); describe("pauseStream", () => { + it("should return 501 when pausing is not implemented", async () => { + await pauseStream(req as any, res as Response); + + expect(res.status).toHaveBeenCalledWith(501); + }); + it("should pause stream", 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", @@ -435,17 +438,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