Skip to content

Commit f58d55e

Browse files
authored
Merge pull request #981 from Banx17/fix/804-duplicate-pause-resume-events
Fix #804: Stop pause/resume controllers from duplicating PAUSED/RESUMED events
2 parents abf657a + a9bc408 commit f58d55e

3 files changed

Lines changed: 156 additions & 90 deletions

File tree

backend/src/controllers/stream.controller.ts

Lines changed: 4 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -672,36 +672,13 @@ export const pauseStream = async (req: Request, res: Response) => {
672672
// Call Soroban service to verify the pause operation would succeed
673673
const result = await sorobanPauseStream(authReq.user.publicKey, parsedStreamId);
674674

675-
// Update the database to mark stream as paused
676-
const now = Math.floor(Date.now() / 1000);
677-
const updatedStream = await prisma.stream.update({
678-
where: { streamId: parsedStreamId },
679-
data: {
680-
isPaused: true,
681-
pausedAt: now,
682-
lastUpdateTime: now,
683-
},
684-
});
685-
686-
// Create a PAUSED event
687-
await prisma.streamEvent.create({
688-
data: {
689-
streamId: parsedStreamId,
690-
eventType: 'PAUSED',
691-
transactionHash: result.txHash,
692-
ledgerSequence: 0, // Will be updated by event indexer
693-
timestamp: now,
694-
metadata: JSON.stringify({ pausedBy: authReq.user.publicKey }),
695-
},
696-
});
697-
698-
logger.info(`Stream ${parsedStreamId} paused by ${authReq.user.publicKey}`);
675+
logger.info(`Stream ${parsedStreamId} pause simulated by ${authReq.user.publicKey}`);
699676

700677
return res.status(200).json({
701678
success: true,
702679
streamId: parsedStreamId,
703680
txHash: result.txHash,
704-
stream: updatedStream,
681+
stream,
705682
});
706683
} catch (sorobanError) {
707684
logger.error(`Soroban pause failed for stream ${parsedStreamId}:`, sorobanError);
@@ -766,44 +743,13 @@ export const resumeStream = async (req: Request, res: Response) => {
766743
// Call Soroban service to verify the resume operation would succeed
767744
const result = await sorobanResumeStream(authReq.user.publicKey, parsedStreamId);
768745

769-
// Calculate pause duration and update the database
770-
const now = Math.floor(Date.now() / 1000);
771-
const pausedAt = stream.pausedAt ?? now;
772-
const pauseDuration = Math.max(0, now - pausedAt);
773-
const totalPausedDuration = (stream.totalPausedDuration ?? 0) + pauseDuration;
774-
775-
const updatedStream = await prisma.stream.update({
776-
where: { streamId: parsedStreamId },
777-
data: {
778-
isPaused: false,
779-
pausedAt: null,
780-
totalPausedDuration,
781-
lastUpdateTime: now,
782-
},
783-
});
784-
785-
// Create a RESUMED event
786-
await prisma.streamEvent.create({
787-
data: {
788-
streamId: parsedStreamId,
789-
eventType: 'RESUMED',
790-
transactionHash: result.txHash,
791-
ledgerSequence: 0, // Will be updated by event indexer
792-
timestamp: now,
793-
metadata: JSON.stringify({
794-
resumedBy: authReq.user.publicKey,
795-
pauseDuration,
796-
}),
797-
},
798-
});
799-
800-
logger.info(`Stream ${parsedStreamId} resumed by ${authReq.user.publicKey}`);
746+
logger.info(`Stream ${parsedStreamId} resume simulated by ${authReq.user.publicKey}`);
801747

802748
return res.status(200).json({
803749
success: true,
804750
streamId: parsedStreamId,
805751
txHash: result.txHash,
806-
stream: updatedStream,
752+
stream,
807753
});
808754
} catch (sorobanError) {
809755
logger.error(`Soroban resume failed for stream ${parsedStreamId}:`, sorobanError);
Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
import { describe, it, expect, vi, beforeEach } from 'vitest';
2+
import request from 'supertest';
3+
import * as StellarSdk from '@stellar/stellar-sdk';
4+
import { SorobanEventWorker } from '../../src/workers/soroban-event-worker.js';
5+
6+
const { mockPauseStream, mockResumeStream, mockPrisma } = vi.hoisted(() => ({
7+
mockPauseStream: vi.fn(),
8+
mockResumeStream: vi.fn(),
9+
mockPrisma: {
10+
stream: {
11+
findUnique: vi.fn(),
12+
update: vi.fn(),
13+
findUniqueOrThrow: vi.fn(),
14+
},
15+
streamEvent: {
16+
create: vi.fn(),
17+
upsert: vi.fn(),
18+
findUnique: vi.fn(),
19+
},
20+
$transaction: vi.fn(async (cb) => {
21+
// Mock the transaction client as mockPrisma itself
22+
return cb(mockPrisma);
23+
}),
24+
},
25+
}));
26+
27+
vi.mock('../../src/lib/prisma.js', () => ({
28+
default: mockPrisma,
29+
prisma: mockPrisma,
30+
}));
31+
32+
vi.mock('../../src/services/sorobanService.js', () => ({
33+
pauseStream: mockPauseStream,
34+
resumeStream: mockResumeStream,
35+
}));
36+
37+
import app from '../../src/app.js';
38+
39+
function makeKeypair() {
40+
return StellarSdk.Keypair.random();
41+
}
42+
43+
function buildSignedTransaction(keypair: StellarSdk.Keypair, nonce: string): string {
44+
const account = new StellarSdk.Account(keypair.publicKey(), '0');
45+
const tx = new StellarSdk.TransactionBuilder(account, {
46+
fee: '100',
47+
networkPassphrase: StellarSdk.Networks.TESTNET,
48+
})
49+
.addOperation(
50+
StellarSdk.Operation.manageData({
51+
name: 'auth',
52+
value: Buffer.from(nonce, 'hex'),
53+
}),
54+
)
55+
.setTimeout(60)
56+
.build();
57+
58+
tx.sign(keypair);
59+
return tx.toXDR();
60+
}
61+
62+
async function getValidJwt(keypair: StellarSdk.Keypair): Promise<string> {
63+
const challengeRes = await request(app)
64+
.post('/v1/auth/challenge')
65+
.send({ publicKey: keypair.publicKey() });
66+
67+
const nonce = challengeRes.body.nonce as string;
68+
const signedTransaction = buildSignedTransaction(keypair, nonce);
69+
70+
const verifyRes = await request(app)
71+
.post('/v1/auth/verify')
72+
.send({ publicKey: keypair.publicKey(), signedTransaction });
73+
74+
return verifyRes.body.token as string;
75+
}
76+
77+
describe('Regression #804: Pause/resume controller duplicate StreamEvent', () => {
78+
beforeEach(() => {
79+
vi.clearAllMocks();
80+
});
81+
82+
it('pauses a stream and only writes one PAUSED event via indexer', async () => {
83+
const sender = makeKeypair();
84+
const token = await getValidJwt(sender);
85+
const streamId = 77;
86+
87+
// 1. Controller flow
88+
mockPrisma.stream.findUnique.mockResolvedValue({
89+
streamId,
90+
sender: sender.publicKey(),
91+
isActive: true,
92+
isPaused: false,
93+
});
94+
mockPauseStream.mockResolvedValue({ txHash: 'simulated-pause-77' });
95+
96+
const pauseRes = await request(app)
97+
.post(`/v1/streams/${streamId}/pause`)
98+
.set('Authorization', `Bearer ${token}`);
99+
100+
expect(pauseRes.status).toBe(200);
101+
102+
// Controller should NOT write to DB for PAUSED event
103+
expect(mockPrisma.streamEvent.create).not.toHaveBeenCalled();
104+
expect(mockPrisma.stream.update).not.toHaveBeenCalled();
105+
106+
// 2. Indexer flow
107+
const worker = new SorobanEventWorker();
108+
109+
const mockEvent = {
110+
id: 'event1',
111+
ledger: 100,
112+
txHash: 'real-tx-hash',
113+
topic: [
114+
StellarSdk.xdr.ScVal.scvSymbol('stream_paused'),
115+
StellarSdk.nativeToScVal(streamId, { type: 'u64' }),
116+
],
117+
value: StellarSdk.xdr.ScVal.scvMap([
118+
new StellarSdk.xdr.ScMapEntry({
119+
key: StellarSdk.xdr.ScVal.scvSymbol('sender'),
120+
val: new StellarSdk.Address(sender.publicKey()).toScVal(),
121+
}),
122+
new StellarSdk.xdr.ScMapEntry({
123+
key: StellarSdk.xdr.ScVal.scvSymbol('paused_at'),
124+
val: StellarSdk.nativeToScVal(Math.floor(Date.now() / 1000), { type: 'u64' }),
125+
}),
126+
]),
127+
inSuccessfulContractCall: true,
128+
} as any;
129+
130+
mockPrisma.streamEvent.findUnique.mockResolvedValue(null);
131+
132+
await worker.processEvent(mockEvent);
133+
134+
// Indexer should write exactly one PAUSED event
135+
expect(mockPrisma.streamEvent.upsert).toHaveBeenCalledTimes(1);
136+
expect(mockPrisma.streamEvent.upsert).toHaveBeenCalledWith(
137+
expect.objectContaining({
138+
create: expect.objectContaining({
139+
eventType: 'PAUSED',
140+
transactionHash: 'real-tx-hash',
141+
}),
142+
}),
143+
);
144+
expect(mockPrisma.stream.update).toHaveBeenCalledTimes(1);
145+
expect(mockPrisma.stream.update).toHaveBeenCalledWith(
146+
expect.objectContaining({
147+
where: { streamId },
148+
data: expect.objectContaining({ isPaused: true }),
149+
}),
150+
);
151+
});
152+
});

backend/tests/integration/stream-actions.test.ts

Lines changed: 0 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -124,22 +124,6 @@ describe('stream action routes', () => {
124124
txHash: 'pause-tx-hash',
125125
});
126126
expect(mockPauseStream).toHaveBeenCalledWith(sender.publicKey(), 7);
127-
expect(mockPrisma.stream.update).toHaveBeenCalledWith(
128-
expect.objectContaining({
129-
where: { streamId: 7 },
130-
data: expect.objectContaining({
131-
isPaused: true,
132-
}),
133-
}),
134-
);
135-
expect(mockPrisma.streamEvent.create).toHaveBeenCalledWith(
136-
expect.objectContaining({
137-
data: expect.objectContaining({
138-
eventType: 'PAUSED',
139-
transactionHash: 'pause-tx-hash',
140-
}),
141-
}),
142-
);
143127
});
144128

145129
it('rejects a raw signed transaction bearer token without a JWT', async () => {
@@ -190,22 +174,6 @@ describe('stream action routes', () => {
190174
txHash: 'resume-tx-hash',
191175
});
192176
expect(mockResumeStream).toHaveBeenCalledWith(sender.publicKey(), 9);
193-
expect(mockPrisma.stream.update).toHaveBeenCalledWith(
194-
expect.objectContaining({
195-
where: { streamId: 9 },
196-
data: expect.objectContaining({
197-
isPaused: false,
198-
}),
199-
}),
200-
);
201-
expect(mockPrisma.streamEvent.create).toHaveBeenCalledWith(
202-
expect.objectContaining({
203-
data: expect.objectContaining({
204-
eventType: 'RESUMED',
205-
transactionHash: 'resume-tx-hash',
206-
}),
207-
}),
208-
);
209177
});
210178

211179
it('POST /v1/streams/:streamId/withdraw withdraws the claimable amount for the recipient', async () => {

0 commit comments

Comments
 (0)