Skip to content

Commit 10216a4

Browse files
ogazboizogazboiz
authored
Merge pull request #397 from thewealthyplace/feature/pause-resume-stream-enhancements
feature: pause/resume stream enhancements with notification fixes
2 parents aa8f561 + 28c9157 commit 10216a4

9 files changed

Lines changed: 378 additions & 74 deletions

File tree

‎backend/prisma/migrations/20260428000000_add_pause_state_fields/migration.sql‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
-- AlterTable
2+
ALTER TABLE "Stream" ADD COLUMN "isPaused" BOOLEAN NOT NULL DEFAULT false,
3+
ADD COLUMN "pausedAt" INTEGER,
4+
ADD COLUMN "totalPausedDuration" INTEGER NOT NULL DEFAULT 0;

‎backend/src/index.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@ const startServer = async () => {
2020
// Connect Redis (graceful fallback to single-instance mode when absent)
2121
await connectRedis();
2222
await sseService.initRedisSubscription();
23+
24+
// Start SSE heartbeat for connection management
25+
sseService.startHeartbeat();
2326

2427
const port = process.env.PORT || 3001;
2528
const server = app.listen(port, () => {

‎backend/src/services/sse.service.ts‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@ interface SSEClient {
1212

1313
const MAX_CONNECTIONS_PER_IP = 5;
1414
const RETRY_AFTER_SECONDS = 60;
15+
const HEARTBEAT_INTERVAL_MS = 30000; // 30 seconds
16+
const IDLE_TIMEOUT_MS = 300000; // 5 minutes
1517

1618
interface SSECapacityCheckResult {
1719
allowed: boolean;
@@ -88,6 +90,71 @@ export class SSEService {
8890
logger.info('[SSEService] Redis pub/sub subscription active.');
8991
}
9092

93+
startHeartbeat(): void {
94+
if (this.heartbeatTimer) return;
95+
96+
this.heartbeatTimer = setInterval(() => {
97+
this.sendHeartbeat();
98+
this.removeIdleConnections();
99+
}, HEARTBEAT_INTERVAL_MS);
100+
101+
logger.info('[SSEService] Heartbeat started');
102+
}
103+
104+
stopHeartbeat(): void {
105+
if (this.heartbeatTimer) {
106+
clearInterval(this.heartbeatTimer);
107+
this.heartbeatTimer = null;
108+
logger.info('[SSEService] Heartbeat stopped');
109+
}
110+
}
111+
112+
private sendHeartbeat(): void {
113+
const heartbeatMessage = ': keep-alive\n\n';
114+
let sentCount = 0;
115+
116+
for (const client of this.clients.values()) {
117+
try {
118+
client.res.write(heartbeatMessage);
119+
sentCount++;
120+
} catch (err) {
121+
logger.warn(`[SSEService] Failed to send heartbeat to client ${client.id}:`, err);
122+
// Remove client on write failure
123+
this.removeClient(client.id);
124+
}
125+
}
126+
127+
if (sentCount > 0) {
128+
logger.debug(`[SSEService] Sent heartbeat to ${sentCount} clients`);
129+
}
130+
}
131+
132+
private removeIdleConnections(): void {
133+
const now = Date.now();
134+
const idleClients: string[] = [];
135+
136+
for (const [clientId, client] of this.clients.entries()) {
137+
if (now - client.lastActivityAt > IDLE_TIMEOUT_MS) {
138+
idleClients.push(clientId);
139+
}
140+
}
141+
142+
if (idleClients.length > 0) {
143+
logger.info(`[SSEService] Removing ${idleClients.length} idle connections`);
144+
for (const clientId of idleClients) {
145+
try {
146+
const client = this.clients.get(clientId);
147+
if (client) {
148+
client.res.end();
149+
}
150+
} catch (err) {
151+
logger.warn(`[SSEService] Error closing idle client ${clientId}:`, err);
152+
}
153+
this.removeClient(clientId);
154+
}
155+
}
156+
}
157+
91158
checkCapacity(ip: string): SSECapacityCheckResult {
92159
if (this.clients.size >= this.maxConnections) {
93160
return {

‎backend/src/workers/soroban-event-worker.ts‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -821,6 +821,126 @@ export class SorobanEventWorker {
821821
timestamp,
822822
});
823823
}
824+
825+
private async handleStreamPaused(
826+
event: rpc.Api.EventResponse,
827+
streamIdTopic: xdr.ScVal,
828+
): Promise<void> {
829+
const streamId = Number(decodeU64(streamIdTopic));
830+
const body = decodeMap(event.value);
831+
832+
if (!body['sender'] || !body['paused_at']) {
833+
throw new Error(`StreamPaused #${streamId}: missing body fields`);
834+
}
835+
836+
const sender = decodeAddress(body['sender']);
837+
const pausedAt = Number(decodeU64(body['paused_at']));
838+
const timestamp = Math.floor(Date.now() / 1000);
839+
840+
await prisma.$transaction(async (tx: any) => {
841+
// Get current stream to preserve totalPausedDuration
842+
const currentStream = await tx.stream.findUniqueOrThrow({
843+
where: { streamId },
844+
select: { totalPausedDuration: true },
845+
});
846+
847+
await tx.stream.update({
848+
where: { streamId },
849+
data: {
850+
isPaused: true,
851+
pausedAt,
852+
lastUpdateTime: timestamp,
853+
},
854+
});
855+
856+
await tx.streamEvent.create({
857+
data: {
858+
streamId,
859+
eventType: 'PAUSED',
860+
transactionHash: event.txHash,
861+
ledgerSequence: event.ledger,
862+
timestamp,
863+
metadata: JSON.stringify({ sender, pausedAt }),
864+
},
865+
});
866+
});
867+
868+
sseService.broadcastToStream(String(streamId), 'stream.paused', {
869+
streamId,
870+
sender,
871+
pausedAt,
872+
transactionHash: event.txHash,
873+
ledger: event.ledger,
874+
timestamp,
875+
});
876+
}
877+
878+
private async handleStreamResumed(
879+
event: rpc.Api.EventResponse,
880+
streamIdTopic: xdr.ScVal,
881+
): Promise<void> {
882+
const streamId = Number(decodeU64(streamIdTopic));
883+
const body = decodeMap(event.value);
884+
885+
if (!body['sender'] || !body['new_end_time']) {
886+
throw new Error(`StreamResumed #${streamId}: missing body fields`);
887+
}
888+
889+
const sender = decodeAddress(body['sender']);
890+
const newEndTime = Number(decodeU64(body['new_end_time']));
891+
const timestamp = Math.floor(Date.now() / 1000);
892+
893+
await prisma.$transaction(async (tx: any) => {
894+
// Get current stream to calculate paused duration
895+
const currentStream = await tx.stream.findUniqueOrThrow({
896+
where: { streamId },
897+
select: { pausedAt: true, totalPausedDuration: true },
898+
});
899+
900+
// Calculate the duration of this pause interval
901+
let additionalPausedDuration = 0;
902+
if (currentStream.pausedAt) {
903+
additionalPausedDuration = timestamp - currentStream.pausedAt;
904+
}
905+
906+
const newTotalPausedDuration = currentStream.totalPausedDuration + additionalPausedDuration;
907+
908+
await tx.stream.update({
909+
where: { streamId },
910+
data: {
911+
isPaused: false,
912+
pausedAt: null,
913+
totalPausedDuration: newTotalPausedDuration,
914+
lastUpdateTime: timestamp,
915+
},
916+
});
917+
918+
await tx.streamEvent.create({
919+
data: {
920+
streamId,
921+
eventType: 'RESUMED',
922+
transactionHash: event.txHash,
923+
ledgerSequence: event.ledger,
924+
timestamp,
925+
metadata: JSON.stringify({
926+
sender,
927+
newEndTime,
928+
pausedDuration: additionalPausedDuration,
929+
totalPausedDuration: newTotalPausedDuration
930+
}),
931+
},
932+
});
933+
});
934+
935+
sseService.broadcastToStream(String(streamId), 'stream.resumed', {
936+
streamId,
937+
sender,
938+
newEndTime,
939+
transactionHash: event.txHash,
940+
ledger: event.ledger,
941+
timestamp,
942+
});
943+
}
824944
}
825945

826946
export const sorobanEventWorker = new SorobanEventWorker();

‎backend/tests/claimable.service.test.ts‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,9 @@ describe('ClaimableAmountService', () => {
1919
pausedAt: null,
2020
totalPausedDuration: 0,
2121
isActive: true,
22+
isPaused: false,
23+
pausedAt: null,
24+
totalPausedDuration: 0,
2225
});
2326

2427
// elapsed = 10 - 7 = 3
@@ -47,6 +50,9 @@ describe('ClaimableAmountService', () => {
4750
pausedAt: null,
4851
totalPausedDuration: 0,
4952
isActive: true,
53+
isPaused: false,
54+
pausedAt: null,
55+
totalPausedDuration: 0,
5056
});
5157

5258
expect(result.claimableAmount).toBe('100');
@@ -62,14 +68,17 @@ describe('ClaimableAmountService', () => {
6268
const result = service.getClaimableAmount({
6369
streamId: 3,
6470
ratePerSecond: '10',
65-
depositedAmount: '1000',
71+
depositedAmount: '100',
6672
withdrawnAmount: '100',
6773
lastUpdateTime: 0,
6874
startTime: 0,
6975
isPaused: false,
7076
pausedAt: null,
7177
totalPausedDuration: 0,
7278
isActive: false,
79+
isPaused: false,
80+
pausedAt: null,
81+
totalPausedDuration: 0,
7382
});
7483

7584
expect(result.claimableAmount).toBe('0');
@@ -93,6 +102,9 @@ describe('ClaimableAmountService', () => {
93102
pausedAt: null,
94103
totalPausedDuration: 0,
95104
isActive: true,
105+
isPaused: false,
106+
pausedAt: null,
107+
totalPausedDuration: 0,
96108
});
97109

98110
expect(result.claimableAmount).toBe('0');
@@ -117,6 +129,9 @@ describe('ClaimableAmountService', () => {
117129
pausedAt: null,
118130
totalPausedDuration: 0,
119131
isActive: true,
132+
isPaused: false,
133+
pausedAt: null,
134+
totalPausedDuration: 0,
120135
};
121136

122137
const first = service.getClaimableAmount(input, 5);
@@ -150,6 +165,9 @@ describe('ClaimableAmountService', () => {
150165
pausedAt: null,
151166
totalPausedDuration: 0,
152167
isActive: true,
168+
isPaused: false,
169+
pausedAt: null,
170+
totalPausedDuration: 0,
153171
});
154172

155173
// calculatedAt = floor(1_000_000 / 1000) = 1000

‎backend/tsconfig.json‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@
3535
"jsx": "react-jsx",
3636
"verbatimModuleSyntax": true,
3737
"isolatedModules": true,
38-
"noUncheckedSideEffectImports": true,
3938
"moduleDetection": "force",
4039
"skipLibCheck": true,
4140
},

0 commit comments

Comments
 (0)