Skip to content

Commit a9bc408

Browse files
committed
Fix #804: Stop pause/resume controllers from duplicating PAUSED/RESUMED events
1 parent f50d60b commit a9bc408

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
@@ -630,36 +630,13 @@ export const pauseStream = async (req: Request, res: Response) => {
630630
// Call Soroban service to verify the pause operation would succeed
631631
const result = await sorobanPauseStream(authReq.user.publicKey, parsedStreamId);
632632

633-
// Update the database to mark stream as paused
634-
const now = Math.floor(Date.now() / 1000);
635-
const updatedStream = await prisma.stream.update({
636-
where: { streamId: parsedStreamId },
637-
data: {
638-
isPaused: true,
639-
pausedAt: now,
640-
lastUpdateTime: now,
641-
},
642-
});
643-
644-
// Create a PAUSED event
645-
await prisma.streamEvent.create({
646-
data: {
647-
streamId: parsedStreamId,
648-
eventType: 'PAUSED',
649-
transactionHash: result.txHash,
650-
ledgerSequence: 0, // Will be updated by event indexer
651-
timestamp: now,
652-
metadata: JSON.stringify({ pausedBy: authReq.user.publicKey }),
653-
},
654-
});
655-
656-
logger.info(`Stream ${parsedStreamId} paused by ${authReq.user.publicKey}`);
633+
logger.info(`Stream ${parsedStreamId} pause simulated by ${authReq.user.publicKey}`);
657634

658635
return res.status(200).json({
659636
success: true,
660637
streamId: parsedStreamId,
661638
txHash: result.txHash,
662-
stream: updatedStream,
639+
stream,
663640
});
664641
} catch (sorobanError) {
665642
logger.error(`Soroban pause failed for stream ${parsedStreamId}:`, sorobanError);
@@ -724,44 +701,13 @@ export const resumeStream = async (req: Request, res: Response) => {
724701
// Call Soroban service to verify the resume operation would succeed
725702
const result = await sorobanResumeStream(authReq.user.publicKey, parsedStreamId);
726703

727-
// Calculate pause duration and update the database
728-
const now = Math.floor(Date.now() / 1000);
729-
const pausedAt = stream.pausedAt ?? now;
730-
const pauseDuration = Math.max(0, now - pausedAt);
731-
const totalPausedDuration = (stream.totalPausedDuration ?? 0) + pauseDuration;
732-
733-
const updatedStream = await prisma.stream.update({
734-
where: { streamId: parsedStreamId },
735-
data: {
736-
isPaused: false,
737-
pausedAt: null,
738-
totalPausedDuration,
739-
lastUpdateTime: now,
740-
},
741-
});
742-
743-
// Create a RESUMED event
744-
await prisma.streamEvent.create({
745-
data: {
746-
streamId: parsedStreamId,
747-
eventType: 'RESUMED',
748-
transactionHash: result.txHash,
749-
ledgerSequence: 0, // Will be updated by event indexer
750-
timestamp: now,
751-
metadata: JSON.stringify({
752-
resumedBy: authReq.user.publicKey,
753-
pauseDuration,
754-
}),
755-
},
756-
});
757-
758-
logger.info(`Stream ${parsedStreamId} resumed by ${authReq.user.publicKey}`);
704+
logger.info(`Stream ${parsedStreamId} resume simulated by ${authReq.user.publicKey}`);
759705

760706
return res.status(200).json({
761707
success: true,
762708
streamId: parsedStreamId,
763709
txHash: result.txHash,
764-
stream: updatedStream,
710+
stream,
765711
});
766712
} catch (sorobanError) {
767713
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)