From 9edd559b611c7ee6c4bf89e26835dd6c5cd3e540 Mon Sep 17 00:00:00 2001 From: TCROWN10 Date: Mon, 27 Jul 2026 10:18:01 +0100 Subject: [PATCH] fix(backend): store on-chain streamId as BigInt for full u64 range Widen Stream.streamId / StreamEvent.streamId to Postgres bigint and keep decoded u64 values as JS bigint through the worker and API parsers so IDs above int4 max no longer fail inserts. Adds a migration and regression coverage for ids > 2^31-1. Closes #829 Co-authored-by: Cursor --- .../migration.sql | 12 ++ backend/prisma/schema.prisma | 12 +- backend/src/app.ts | 1 + backend/src/controllers/sse.controller.ts | 2 +- backend/src/controllers/stream.controller.ts | 39 +++-- backend/src/controllers/stream/cancel.ts | 6 +- backend/src/lib/stream-id.ts | 53 +++++++ backend/src/repositories/stream.repository.ts | 2 +- backend/src/routes/v1/streams/withdraw.ts | 5 +- backend/src/services/claimable.service.ts | 4 +- .../src/services/soroban-indexer.service.ts | 16 +- backend/src/services/sorobanService.ts | 16 +- backend/src/services/sse.service.ts | 138 +++++++++++++++++- backend/src/validators/stream.validator.ts | 15 +- backend/src/workers/soroban-event-worker.ts | 25 ++-- backend/tests/cancel.controller.test.ts | 4 +- .../tests/integration/indexer-worker.test.ts | 12 +- .../tests/integration/stream-actions.test.ts | 6 +- .../tests/integration/streams/cancel.test.ts | 4 +- .../integration/streams/withdraw.test.ts | 6 +- backend/tests/integration/top-up.test.ts | 4 +- backend/tests/soroban-event-worker.test.ts | 74 +++++++++- backend/tests/stream-id.test.ts | 30 ++++ backend/tests/stream.repository.test.ts | 16 +- backend/tests/stream.validator.test.ts | 19 +++ 25 files changed, 422 insertions(+), 99 deletions(-) create mode 100644 backend/prisma/migrations/20260727090000_stream_id_bigint/migration.sql create mode 100644 backend/src/lib/stream-id.ts create mode 100644 backend/tests/stream-id.test.ts diff --git a/backend/prisma/migrations/20260727090000_stream_id_bigint/migration.sql b/backend/prisma/migrations/20260727090000_stream_id_bigint/migration.sql new file mode 100644 index 00000000..438ce5a2 --- /dev/null +++ b/backend/prisma/migrations/20260727090000_stream_id_bigint/migration.sql @@ -0,0 +1,12 @@ +-- Convert on-chain stream identifiers from int4 to bigint (Soroban u64). +-- Drop the FK first so both columns can be widened, then recreate it. + +ALTER TABLE "StreamEvent" DROP CONSTRAINT IF EXISTS "StreamEvent_streamId_fkey"; + +ALTER TABLE "Stream" ALTER COLUMN "streamId" TYPE BIGINT USING ("streamId"::bigint); +ALTER TABLE "StreamEvent" ALTER COLUMN "streamId" TYPE BIGINT USING ("streamId"::bigint); + +ALTER TABLE "StreamEvent" + ADD CONSTRAINT "StreamEvent_streamId_fkey" + FOREIGN KEY ("streamId") REFERENCES "Stream"("streamId") + ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/backend/prisma/schema.prisma b/backend/prisma/schema.prisma index 49356254..8de870f9 100644 --- a/backend/prisma/schema.prisma +++ b/backend/prisma/schema.prisma @@ -27,7 +27,7 @@ model User { // Stream model - mirrors on-chain stream state for fast querying model Stream { id String @id @default(uuid()) - streamId Int @unique // On-chain stream ID + streamId BigInt @unique // On-chain stream ID (Soroban u64) sender String // Sender's Stellar public key recipient String // Recipient's Stellar public key tokenAddress String // Token contract address @@ -68,7 +68,7 @@ model IndexerState { // StreamEvent model - indexer events for tracking all on-chain stream activities model StreamEvent { id String @id @default(uuid()) - streamId Int // Reference to on-chain stream ID + streamId BigInt // Reference to on-chain stream ID (Soroban u64) eventType String // EventType: "CREATED", "TOPPED_UP", "WITHDRAWN", "CANCELLED", "COMPLETED", "PAUSED", "RESUMED" amount String? // Amount involved in the event (for top-ups, withdrawals) transactionHash String // Stellar transaction hash @@ -87,12 +87,4 @@ model StreamEvent { @@index([transactionHash]) @@index([createdAt]) @@index([streamId, createdAt]) - @@unique([transactionHash, eventType]) -} - -// IndexerState model - tracks indexer cursor for resumable event processing -model IndexerState { - id String @id @default("singleton") - lastLedger Int @default(0) - updatedAt DateTime @updatedAt } diff --git a/backend/src/app.ts b/backend/src/app.ts index 60f4610b..56c3035e 100644 --- a/backend/src/app.ts +++ b/backend/src/app.ts @@ -16,6 +16,7 @@ import { requestIdMiddleware } from "./middleware/requestId.js"; import v1Routes from "./routes/v1/index.js"; import healthRoutes from "./routes/health.routes.js"; +import "./lib/stream-id.js"; const app = express(); const isProduction = process.env.NODE_ENV === "production"; diff --git a/backend/src/controllers/sse.controller.ts b/backend/src/controllers/sse.controller.ts index 9b702257..dc91abd1 100644 --- a/backend/src/controllers/sse.controller.ts +++ b/backend/src/controllers/sse.controller.ts @@ -51,7 +51,7 @@ export const subscribe = async (req: Request, res: Response) => { where: { OR: [{ sender: publicKey }, { recipient: publicKey }] }, select: { streamId: true, sender: true, recipient: true }, }); - const ownedIds = new Set(ownedStreams.map((s: { streamId: number }) => String(s.streamId))); + const ownedIds = new Set(ownedStreams.map((s: { streamId: bigint }) => String(s.streamId))); const allowedUserKeys = new Set([publicKey]); for (const stream of ownedStreams) { allowedUserKeys.add(stream.sender); diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts index 968def20..3f80101e 100644 --- a/backend/src/controllers/stream.controller.ts +++ b/backend/src/controllers/stream.controller.ts @@ -13,6 +13,7 @@ import { resumeStream as sorobanResumeStream, } from "../services/sorobanService.js"; import type { AuthenticatedRequest } from "../types/auth.types.js"; +import { parseStreamId } from "../lib/stream-id.js"; import { DEFAULT_EVENTS_PAGE_SIZE, MAX_EVENTS_PAGE_SIZE, @@ -128,10 +129,10 @@ export const createStream = async (req: Request, res: Response) => { }); } - const parsedStreamId = Number.parseInt(streamId, 10); + const parsedStreamId = parseStreamId(streamId); const parsedStartTime = Number.parseInt(startTime, 10); - if (!Number.isFinite(parsedStreamId)) { + if (parsedStreamId === null) { return res .status(400) .json({ error: "Invalid streamId: must be a valid integer" }); @@ -347,9 +348,8 @@ export const getStream = async (req: Request, res: Response) => { const streamIdParam = Array.isArray(req.params.streamId) ? req.params.streamId[0] : req.params.streamId; - const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10); - - if (!Number.isFinite(parsedStreamId)) { + const parsedStreamId = parseStreamId(streamIdParam); + if (parsedStreamId === null) { return res.status(400).json({ error: "Invalid streamId parameter" }); } @@ -398,9 +398,8 @@ export const getStreamEvents = async (req: Request, res: Response) => { const streamIdParam = Array.isArray(req.params.streamId) ? req.params.streamId[0] : req.params.streamId; - const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10); - - if (!Number.isFinite(parsedStreamId)) { + const parsedStreamId = parseStreamId(streamIdParam); + if (parsedStreamId === null) { return res.status(400).json({ error: "Invalid streamId parameter" }); } @@ -489,9 +488,8 @@ export const getStreamClaimableAmount = async (req: Request, res: Response) => { const streamIdParam = Array.isArray(req.params.streamId) ? req.params.streamId[0] : req.params.streamId; - const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10); - - if (!Number.isFinite(parsedStreamId)) { + const parsedStreamId = parseStreamId(streamIdParam); + if (parsedStreamId === null) { return res.status(400).json({ error: "Invalid streamId parameter" }); } @@ -686,13 +684,12 @@ const topUpBodySchema = z.object({ * Adds tokens to a running stream. Only the stream sender may call this. */ export const topUpStreamHandler = async (req: Request, res: Response) => { - const streamId = parseInt( + const streamId = parseStreamId( Array.isArray(req.params.streamId) - ? req.params.streamId[0]! - : (req.params.streamId ?? ""), - 10, + ? req.params.streamId[0] + : req.params.streamId, ); - if (isNaN(streamId)) { + if (streamId === null) { return res.status(400).json({ error: "Invalid streamId" }); } @@ -769,9 +766,8 @@ export const pauseStream = async (req: Request, res: Response) => { const streamIdParam = Array.isArray(req.params.streamId) ? req.params.streamId[0] : req.params.streamId; - const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10); - - if (!Number.isFinite(parsedStreamId)) { + const parsedStreamId = parseStreamId(streamIdParam); + if (parsedStreamId === null) { return res.status(400).json({ error: "Invalid streamId parameter" }); } @@ -861,9 +857,8 @@ export const resumeStream = async (req: Request, res: Response) => { const streamIdParam = Array.isArray(req.params.streamId) ? req.params.streamId[0] : req.params.streamId; - const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10); - - if (!Number.isFinite(parsedStreamId)) { + const parsedStreamId = parseStreamId(streamIdParam); + if (parsedStreamId === null) { return res.status(400).json({ error: "Invalid streamId parameter" }); } diff --git a/backend/src/controllers/stream/cancel.ts b/backend/src/controllers/stream/cancel.ts index dfcef63c..eb16d62a 100644 --- a/backend/src/controllers/stream/cancel.ts +++ b/backend/src/controllers/stream/cancel.ts @@ -4,6 +4,7 @@ import logger from '../../logger.js'; import * as sorobanService from '../../services/sorobanService.js'; import type { AuthenticatedRequest } from '../../types/auth.types.js'; import * as streamRepository from '../../repositories/stream.repository.js'; +import { parseStreamId } from '../../lib/stream-id.js'; /** * @openapi @@ -55,7 +56,10 @@ export const cancelStreamHandler = async (req: AuthenticatedRequest, res: Respon return res.status(400).json({ error: 'Missing streamId parameter' }); } - const parsedStreamId = parseInt(streamId, 10); + const parsedStreamId = parseStreamId(streamId); + if (parsedStreamId === null) { + return res.status(400).json({ error: 'Invalid streamId parameter' }); + } // 1. Fetch stream from DB const stream = await prisma.stream.findUnique({ diff --git a/backend/src/lib/stream-id.ts b/backend/src/lib/stream-id.ts new file mode 100644 index 00000000..1a105969 --- /dev/null +++ b/backend/src/lib/stream-id.ts @@ -0,0 +1,53 @@ +/** + * Helpers for on-chain stream IDs (Soroban u64). + * + * DB columns are Prisma BigInt / Postgres bigint so values above int4 max + * (2_147_483_647) round-trip without overflow. Prefer bigint in application + * code; never use Number()/parseInt for identifiers that may exceed 2^53-1. + */ + +const U64_DECIMAL = /^\d+$/; + +/** + * Parse a path/query/body streamId into a bigint. + * Accepts decimal strings and non-negative integers / bigints. + */ +export function parseStreamId(raw: unknown): bigint | null { + if (typeof raw === 'bigint') { + return raw >= 0n ? raw : null; + } + if (typeof raw === 'number') { + if (!Number.isInteger(raw) || raw < 0 || !Number.isSafeInteger(raw)) { + return null; + } + return BigInt(raw); + } + if (typeof raw === 'string') { + const trimmed = raw.trim(); + if (!U64_DECIMAL.test(trimmed)) return null; + try { + return BigInt(trimmed); + } catch { + return null; + } + } + return null; +} + +/** + * JSON-safe encoding: numbers stay numbers within Number.MAX_SAFE_INTEGER + * (covers all historical int4 IDs and the >2^31 cases the bug cares about + * until 2^53); larger u64 values become decimal strings. + */ +export function streamIdToJson(streamId: bigint): number | string { + const asNumber = Number(streamId); + return Number.isSafeInteger(asNumber) ? asNumber : streamId.toString(); +} + +// Ensure Express / SSE JSON.stringify can serialize Prisma BigInt fields. +const bigIntProto = BigInt.prototype as unknown as { toJSON?: () => number | string }; +if (typeof bigIntProto.toJSON !== 'function') { + bigIntProto.toJSON = function bigIntToJSON(this: bigint) { + return streamIdToJson(this); + }; +} diff --git a/backend/src/repositories/stream.repository.ts b/backend/src/repositories/stream.repository.ts index 4f017c18..27181f4b 100644 --- a/backend/src/repositories/stream.repository.ts +++ b/backend/src/repositories/stream.repository.ts @@ -3,7 +3,7 @@ import { prisma } from '../lib/prisma.js'; /** * Update the status and active flag of a stream in the database. */ -export const updateStatus = async (streamId: number, status: 'ACTIVE' | 'CANCELLED' | 'COMPLETED' | 'PAUSED') => { +export const updateStatus = async (streamId: bigint, status: 'ACTIVE' | 'CANCELLED' | 'COMPLETED' | 'PAUSED') => { return prisma.stream.update({ where: { streamId }, data: { diff --git a/backend/src/routes/v1/streams/withdraw.ts b/backend/src/routes/v1/streams/withdraw.ts index 9abc268c..26cdb875 100644 --- a/backend/src/routes/v1/streams/withdraw.ts +++ b/backend/src/routes/v1/streams/withdraw.ts @@ -4,6 +4,7 @@ import logger from '../../../logger.js'; import { claimableAmountService } from '../../../services/claimable.service.js'; import { withdraw as sorobanWithdraw } from '../../../services/sorobanService.js'; import type { AuthenticatedRequest } from '../../../types/auth.types.js'; +import { parseStreamId } from '../../../lib/stream-id.js'; /** * @openapi @@ -56,9 +57,9 @@ export const withdrawHandler = async (req: AuthenticatedRequest, res: Response) const streamIdParam = Array.isArray(req.params.streamId) ? req.params.streamId[0] : req.params.streamId; - const parsedStreamId = Number.parseInt(streamIdParam ?? '', 10); + const parsedStreamId = parseStreamId(streamIdParam); - if (!Number.isFinite(parsedStreamId)) { + if (parsedStreamId === null) { return res.status(400).json({ error: 'Invalid streamId parameter' }); } diff --git a/backend/src/services/claimable.service.ts b/backend/src/services/claimable.service.ts index 78b75839..88058202 100644 --- a/backend/src/services/claimable.service.ts +++ b/backend/src/services/claimable.service.ts @@ -4,7 +4,7 @@ const I128_MAX = (1n << 127n) - 1n; const I128_MIN = -(1n << 127n); export interface ClaimableStreamState { - streamId: number; + streamId: bigint; ratePerSecond: string; depositedAmount: string; withdrawnAmount: string; @@ -18,7 +18,7 @@ export interface ClaimableStreamState { } export interface ClaimableAmountResult { - streamId: number; + streamId: bigint; claimableAmount: string; actionable: boolean; calculatedAt: number; diff --git a/backend/src/services/soroban-indexer.service.ts b/backend/src/services/soroban-indexer.service.ts index 9e2f7299..40ec7714 100644 --- a/backend/src/services/soroban-indexer.service.ts +++ b/backend/src/services/soroban-indexer.service.ts @@ -116,12 +116,18 @@ export class SorobanIndexerService { return null; } - private parseStreamId(record: JsonRecord): number | null { + private parseStreamId(record: JsonRecord): bigint | null { const raw = record.stream_id ?? record.streamId; - if (typeof raw === 'number' && Number.isInteger(raw)) return raw; - if (typeof raw === 'string' && raw.trim()) { - const parsed = Number(raw); - if (Number.isInteger(parsed)) return parsed; + if (typeof raw === 'bigint' && raw >= 0n) return raw; + if (typeof raw === 'number' && Number.isInteger(raw) && raw >= 0 && Number.isSafeInteger(raw)) { + return BigInt(raw); + } + if (typeof raw === 'string' && /^\d+$/.test(raw.trim())) { + try { + return BigInt(raw.trim()); + } catch { + return null; + } } return null; } diff --git a/backend/src/services/sorobanService.ts b/backend/src/services/sorobanService.ts index 30895ace..366f4f19 100644 --- a/backend/src/services/sorobanService.ts +++ b/backend/src/services/sorobanService.ts @@ -51,7 +51,7 @@ export function resetServer(): void { } export interface ChainStream { - streamId: number; + streamId: bigint; sender: string; recipient: string; tokenAddress: string; @@ -157,7 +157,7 @@ export async function submitContractCall(method: string, args: xdr.ScVal[], send return response.hash; } -export async function getStreamFromChain(streamId: number): Promise { +export async function getStreamFromChain(streamId: bigint): Promise { if (!getContractId()) return null; try { @@ -189,7 +189,7 @@ export async function getStreamFromChain(streamId: number): Promise { +export async function getClaimableFromChain(streamId: bigint): Promise { if (!getContractId()) return null; try { @@ -204,13 +204,13 @@ export async function getClaimableFromChain(streamId: number): Promise { +export async function cancelStream(streamId: bigint, senderSecret: string): Promise { return submitContractCall('cancel_stream', [ nativeToScVal(streamId, { type: 'u64' }), ], senderSecret); } -export async function topUpStream(streamId: number, amount: bigint, callerAddress: string): Promise { +export async function topUpStream(streamId: bigint, amount: bigint, callerAddress: string): Promise { const keeperSecret = getKeeperSecret(); if (!keeperSecret) throw new Error('KEEPER_SECRET_KEY not configured'); return submitContractCall('top_up_stream', [ @@ -236,7 +236,7 @@ export interface PauseResumeResult { */ export async function pauseStream( senderAddress: string, - streamId: number + streamId: bigint ): Promise { if (!getContractId()) { throw new Error('Stream contract ID not configured'); @@ -270,7 +270,7 @@ export async function pauseStream( */ export async function resumeStream( senderAddress: string, - streamId: number + streamId: bigint ): Promise { if (!getContractId()) { throw new Error('Stream contract ID not configured'); @@ -302,7 +302,7 @@ export async function resumeStream( * matching the current pause/resume backend pattern. */ export async function withdraw( - streamId: number, + streamId: bigint, recipientAddress: string, ): Promise { if (!getContractId()) { diff --git a/backend/src/services/sse.service.ts b/backend/src/services/sse.service.ts index 88fd702d..ff1b9e80 100644 --- a/backend/src/services/sse.service.ts +++ b/backend/src/services/sse.service.ts @@ -4,29 +4,106 @@ import { isRedisAvailable, getPublisher, getSubscriber } from '../lib/redis.js'; const HEARTBEAT_INTERVAL_MS = 30_000; const MAX_WRITABLE_BUFFER = 64 * 1024; +const MAX_CONNECTIONS_PER_IP = 5; +const RETRY_AFTER_SECONDS = 60; interface SSEClient { id: string; res: Response; subscriptions: Set; paused: boolean; + ip: string; +} + +interface SSECapacityCheckResult { + allowed: boolean; + status?: number; + retryAfterSeconds?: number; + message?: string; } export class SSEService { private clients: Map = new Map(); + private readonly ipConnectionCounts: Map = new Map(); private heartbeatTimer: ReturnType | null = null; private slowClientsDropped = 0; + private shuttingDown = false; + private perIpPeakConnections = 0; + + private readonly maxConnections: number = (() => { + const parsed = Number.parseInt(process.env.MAX_SSE_CONNECTIONS ?? '10000', 10); + if (!Number.isFinite(parsed) || parsed <= 0) return 10000; + return parsed; + })(); + + isShuttingDown(): boolean { + return this.shuttingDown; + } + + async initRedisSubscription(): Promise { + const sub = getSubscriber(); + if (!sub) return; + + await sub.psubscribe('sse:stream:*', 'sse:user:*'); + sub.on('pmessage', (_pattern: string, channel: string, message: string) => { + try { + const { event, data } = JSON.parse(message) as { event: string; data: unknown }; + if (channel.startsWith('sse:stream:')) { + this._localBroadcastToStream(channel.slice('sse:stream:'.length), event, data); + } else if (channel.startsWith('sse:user:')) { + this._localBroadcastToUser(channel.slice('sse:user:'.length), event, data); + } + } catch (err) { + logger.warn('[Redis SSE] Failed to handle pub/sub message:', err); + } + }); + + logger.info('[SSEService] Redis pub/sub subscription active.'); + } + + checkCapacity(ip: string): SSECapacityCheckResult { + if (this.clients.size >= this.maxConnections) { + return { + allowed: false, + status: 503, + message: 'SSE capacity reached. Please try again shortly.', + }; + } + + const currentIpConnections = this.ipConnectionCounts.get(ip) ?? 0; + if (currentIpConnections >= MAX_CONNECTIONS_PER_IP) { + return { + allowed: false, + status: 429, + retryAfterSeconds: RETRY_AFTER_SECONDS, + message: `Too many SSE connections from this IP. Max ${MAX_CONNECTIONS_PER_IP}.`, + }; + } + + return { allowed: true }; + } + + addClient( + clientId: string, + res: Response, + subscriptions: string[] = [], + ip = 'unknown', + ): void { + const nextIpCount = (this.ipConnectionCounts.get(ip) ?? 0) + 1; + this.ipConnectionCounts.set(ip, nextIpCount); + this.perIpPeakConnections = Math.max(this.perIpPeakConnections, nextIpCount); const client: SSEClient = { id: clientId, res, subscriptions: new Set(subscriptions), paused: false, + ip, }; this.clients.set(clientId, client); logger.info( - `[SSEService] Connection opened: ${clientId}, ip: ${ip}, subscriptions: ${subscriptions.join(', ')}` + `[SSEService] Connection opened: ${clientId}, ip: ${ip}, subscriptions: ${subscriptions.join(', ')}`, ); res.on('close', () => { @@ -36,6 +113,15 @@ export class SSEService { this.ensureHeartbeat(); } + sendReconnectToAll(): void { + this.shuttingDown = true; + const message = 'event: reconnect\ndata: {}\n\n'; + for (const client of this.clients.values()) { + this.writeToClient(client, message); + } + logger.info(`[SSEService] Sent reconnect to ${this.clients.size} client(s).`); + } + sendHeartbeat(): void { const message = ': heartbeat\n\n'; @@ -55,14 +141,37 @@ export class SSEService { } broadcastToStream(streamId: string, event: string, data: unknown): void { + if (isRedisAvailable()) { + getPublisher()?.publish(`sse:stream:${streamId}`, JSON.stringify({ event, data })); + } else { + this._localBroadcastToStream(streamId, event, data); + } + } + + broadcastToUser(publicKey: string, event: string, data: unknown): void { + if (isRedisAvailable()) { + getPublisher()?.publish(`sse:user:${publicKey}`, JSON.stringify({ event, data })); + } else { + this._localBroadcastToUser(publicKey, event, data); + } + } + + broadcastToAdmin(event: string, data: unknown): void { + const adminKey = process.env.ADMIN_PUBLIC_KEY; + if (adminKey) { + this.broadcastToUser(adminKey, event, data); + } + } + + private _localBroadcastToStream(streamId: string, event: string, data: unknown): void { this.broadcast(event, data, (client) => - client.subscriptions.has(streamId) || client.subscriptions.has('*') + client.subscriptions.has(streamId) || client.subscriptions.has('*'), ); } - broadcastToUser(publicKey: string, event: string, data: unknown): void { + private _localBroadcastToUser(publicKey: string, event: string, data: unknown): void { this.broadcast(event, data, (client) => - client.subscriptions.has(`user:${publicKey}`) || client.subscriptions.has('*') + client.subscriptions.has(`user:${publicKey}`) || client.subscriptions.has('*'), ); } @@ -74,6 +183,18 @@ export class SSEService { return this.slowClientsDropped; } + getMaxConnections(): number { + return this.maxConnections; + } + + getPerIpPeakConnections(): number { + return this.perIpPeakConnections; + } + + getActiveIpCount(): number { + return this.ipConnectionCounts.size; + } + stopHeartbeat(): void { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); @@ -108,6 +229,13 @@ export class SSEService { this.clients.delete(clientId); + const currentIpCount = this.ipConnectionCounts.get(client.ip) ?? 0; + if (currentIpCount <= 1) { + this.ipConnectionCounts.delete(client.ip); + } else { + this.ipConnectionCounts.set(client.ip, currentIpCount - 1); + } + try { if (!client.res.writableEnded) { client.res.end(); @@ -118,6 +246,8 @@ export class SSEService { if (reason) { logger.warn(`SSE client removed (${reason}): ${clientId}`); + } else { + logger.info(`SSE client disconnected: ${clientId}, ip: ${client.ip}`); } } diff --git a/backend/src/validators/stream.validator.ts b/backend/src/validators/stream.validator.ts index 7a165389..9d54a138 100644 --- a/backend/src/validators/stream.validator.ts +++ b/backend/src/validators/stream.validator.ts @@ -1,7 +1,20 @@ import { z } from 'zod'; +import { parseStreamId } from '../lib/stream-id.js'; export const createStreamSchema = z.object({ - streamId: z.union([z.number().int().nonnegative(), z.string().regex(/^\d+$/).transform(v => parseInt(v))]), + streamId: z + .union([z.number(), z.string(), z.bigint()]) + .transform((v, ctx) => { + const parsed = parseStreamId(v); + if (parsed === null) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'streamId must be a non-negative integer', + }); + return z.NEVER; + } + return parsed; + }), sender: z.string().min(1, 'Sender address is required'), recipient: z.string().min(1, 'Recipient address is required'), tokenAddress: z.string().min(1, 'Token address is required'), diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index c2ac91ab..7ce4d750 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -4,6 +4,7 @@ import { INDEXER_STATE_ID } from "../lib/indexer-state.js"; import { sseService } from "../services/sse.service.js"; import logger from "../logger.js"; import { Prisma } from "../generated/prisma/index.js"; +import "../lib/stream-id.js"; // ─── Config ────────────────────────────────────────────────────────────────── @@ -152,9 +153,9 @@ export class SorobanEventWorker { update: {}, }); await tx.stream.upsert({ - where: { streamId: 0 }, + where: { streamId: 0n }, create: { - streamId: 0, + streamId: 0n, sender: systemUser, recipient: systemUser, tokenAddress: @@ -364,7 +365,7 @@ export class SorobanEventWorker { }, }, create: { - streamId: 0, + streamId: 0n, eventType: "FEE_CONFIG_UPDATED", transactionHash: event.txHash, ledgerSequence: event.ledger, @@ -417,7 +418,7 @@ export class SorobanEventWorker { }, }, create: { - streamId: 0, + streamId: 0n, eventType: "ADMIN_TRANSFERRED", transactionHash: event.txHash, ledgerSequence: event.ledger, @@ -447,7 +448,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if ( @@ -564,7 +565,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if (!body["amount"] || !body["new_deposited_amount"]) { @@ -642,7 +643,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if (!body["recipient"] || !body["amount"] || !body["timestamp"]) { @@ -711,7 +712,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if (!body["amount_withdrawn"] || !body["refunded_amount"]) { @@ -781,7 +782,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if (!body["recipient"] || !body["total_withdrawn"]) { @@ -851,7 +852,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if (!body["treasury"] || !body["fee_amount"] || !body["token"]) { @@ -912,7 +913,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if (!body["sender"] || !body["paused_at"]) { @@ -981,7 +982,7 @@ export class SorobanEventWorker { event: rpc.Api.EventResponse, streamIdTopic: xdr.ScVal, ): Promise { - const streamId = Number(decodeU64(streamIdTopic)); + const streamId = decodeU64(streamIdTopic); const body = decodeMap(event.value); if (!body["sender"] || !body["new_end_time"]) { diff --git a/backend/tests/cancel.controller.test.ts b/backend/tests/cancel.controller.test.ts index 8c8902c3..9ba145eb 100644 --- a/backend/tests/cancel.controller.test.ts +++ b/backend/tests/cancel.controller.test.ts @@ -77,8 +77,8 @@ describe('Cancel Stream Controller', () => { await cancelStreamHandler(req as AuthenticatedRequest, res as Response); - expect(sorobanService.cancelStream).toHaveBeenCalledWith(123, 'SABC123'); - expect(streamRepository.updateStatus).toHaveBeenCalledWith(123, 'CANCELLED'); + expect(sorobanService.cancelStream).toHaveBeenCalledWith(123n, 'SABC123'); + expect(streamRepository.updateStatus).toHaveBeenCalledWith(123n, 'CANCELLED'); expect(res.status).toHaveBeenCalledWith(200); expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ status: 'CANCELLED', txHash: 'tx_hash_123' })); }); diff --git a/backend/tests/integration/indexer-worker.test.ts b/backend/tests/integration/indexer-worker.test.ts index aaf85fc7..35d0bf9d 100644 --- a/backend/tests/integration/indexer-worker.test.ts +++ b/backend/tests/integration/indexer-worker.test.ts @@ -160,7 +160,7 @@ describe('Indexer worker integration (mocked DB)', () => { expect(mockPrisma.stream.update).toHaveBeenCalledWith( expect.objectContaining({ - where: { streamId }, + where: { streamId: BigInt(streamId) }, data: expect.objectContaining({ isPaused: true }), }), ); @@ -197,7 +197,7 @@ describe('Indexer worker integration (mocked DB)', () => { expect(mockPrisma.stream.update).toHaveBeenCalledWith( expect.objectContaining({ - where: { streamId }, + where: { streamId: BigInt(streamId) }, data: expect.objectContaining({ isPaused: false, }), @@ -231,7 +231,7 @@ describe('Indexer worker integration (mocked DB)', () => { expect(mockPrisma.stream.update).toHaveBeenCalledWith( expect.objectContaining({ - where: { streamId }, + where: { streamId: BigInt(streamId) }, data: expect.objectContaining({ isActive: false }), }), ); @@ -295,7 +295,7 @@ describe('Indexer worker integration (mocked DB)', () => { expect(mockPrisma.stream.upsert).toHaveBeenCalledWith( expect.objectContaining({ - where: { streamId: 0 }, + where: { streamId: 0n }, }), ); @@ -308,7 +308,7 @@ describe('Indexer worker integration (mocked DB)', () => { }, }, create: expect.objectContaining({ - streamId: 0, + streamId: 0n, eventType: 'FEE_CONFIG_UPDATED', transactionHash: 'hash-fee-config', ledgerSequence: 105, @@ -361,7 +361,7 @@ describe('Indexer worker integration (mocked DB)', () => { }, }, create: expect.objectContaining({ - streamId: 0, + streamId: 0n, eventType: 'ADMIN_TRANSFERRED', transactionHash: 'hash-admin-transfer', ledgerSequence: 106, diff --git a/backend/tests/integration/stream-actions.test.ts b/backend/tests/integration/stream-actions.test.ts index afd128ff..933fb2db 100644 --- a/backend/tests/integration/stream-actions.test.ts +++ b/backend/tests/integration/stream-actions.test.ts @@ -123,7 +123,7 @@ describe('stream action routes', () => { streamId: 7, txHash: 'pause-tx-hash', }); - expect(mockPauseStream).toHaveBeenCalledWith(sender.publicKey(), 7); + expect(mockPauseStream).toHaveBeenCalledWith(sender.publicKey(), 7n); }); it('rejects a raw signed transaction bearer token without a JWT', async () => { @@ -173,7 +173,7 @@ describe('stream action routes', () => { streamId: 9, txHash: 'resume-tx-hash', }); - expect(mockResumeStream).toHaveBeenCalledWith(sender.publicKey(), 9); + expect(mockResumeStream).toHaveBeenCalledWith(sender.publicKey(), 9n); }); it('POST /v1/streams/:streamId/withdraw withdraws the claimable amount for the recipient', async () => { @@ -213,7 +213,7 @@ describe('stream action routes', () => { txHash: 'withdraw-tx-hash', amount: '100', }); - expect(mockWithdraw).toHaveBeenCalledWith(11, recipient.publicKey()); + expect(mockWithdraw).toHaveBeenCalledWith(11n, recipient.publicKey()); expect(mockPrisma.streamEvent.create).toHaveBeenCalledWith( expect.objectContaining({ data: expect.objectContaining({ diff --git a/backend/tests/integration/streams/cancel.test.ts b/backend/tests/integration/streams/cancel.test.ts index 27f81a2a..fc3c918b 100644 --- a/backend/tests/integration/streams/cancel.test.ts +++ b/backend/tests/integration/streams/cancel.test.ts @@ -86,9 +86,9 @@ describe('POST /v1/streams/:streamId/cancel', () => { status: 'CANCELLED', }); - expect(sorobanService.cancelStream).toHaveBeenCalledWith(streamId, 'S_SECRET_123'); + expect(sorobanService.cancelStream).toHaveBeenCalledWith(BigInt(streamId), 'S_SECRET_123'); expect(prisma.stream.update).toHaveBeenCalledWith({ - where: { streamId }, + where: { streamId: BigInt(streamId) }, data: { isActive: false }, }); }); diff --git a/backend/tests/integration/streams/withdraw.test.ts b/backend/tests/integration/streams/withdraw.test.ts index d4e77d34..44fdb4cc 100644 --- a/backend/tests/integration/streams/withdraw.test.ts +++ b/backend/tests/integration/streams/withdraw.test.ts @@ -103,12 +103,12 @@ describe('POST /api/v1/streams/:streamId/withdraw', () => { }); // Verify service call with new signature (streamId, recipientAddress) - expect(mockWithdraw).toHaveBeenCalledWith(streamId, recipient.publicKey()); + expect(mockWithdraw).toHaveBeenCalledWith(BigInt(streamId), recipient.publicKey()); // Verify DB update expect(mockPrisma.stream.update).toHaveBeenCalledWith( expect.objectContaining({ - where: { streamId }, + where: { streamId: BigInt(streamId) }, data: expect.objectContaining({ withdrawnAmount: expect.any(String), }), @@ -120,7 +120,7 @@ describe('POST /api/v1/streams/:streamId/withdraw', () => { expect.objectContaining({ data: expect.objectContaining({ eventType: 'WITHDRAWN', - streamId, + streamId: BigInt(streamId), transactionHash: 'withdraw-tx-hash', }), }) diff --git a/backend/tests/integration/top-up.test.ts b/backend/tests/integration/top-up.test.ts index 9c42e2af..9ff26e78 100644 --- a/backend/tests/integration/top-up.test.ts +++ b/backend/tests/integration/top-up.test.ts @@ -97,7 +97,7 @@ describe('POST /v1/streams/:streamId/top-up', () => { expect(res.status).toBe(200); expect(res.body.txHash).toBe('abc123txhash'); expect(res.body.streamId).toBe(42); - expect(topUpStream).toHaveBeenCalledWith(42, 1000n, SENDER); + expect(topUpStream).toHaveBeenCalledWith(42n, 1000n, SENDER); }); it('returns 400 when amount is missing', async () => { @@ -156,7 +156,7 @@ describe('POST /v1/streams/:streamId/top-up', () => { expect(mockPrisma.stream.update).toHaveBeenCalledWith( expect.objectContaining({ - where: { streamId: 42 }, + where: { streamId: 42n }, data: expect.objectContaining({ depositedAmount: '87400' }), }), ); diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 772017f8..3e0deba7 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -326,7 +326,7 @@ describe('SorobanEventWorker', () => { expect(mockTx.streamEvent.upsert).toHaveBeenCalledWith( expect.objectContaining({ create: expect.objectContaining({ - streamId: 0, + streamId: 0n, eventType: 'FEE_CONFIG_UPDATED', transactionHash: txHash, ledgerSequence: 1005, @@ -490,7 +490,7 @@ describe('SorobanEventWorker', () => { await expect((worker as any).handleTokensWithdrawn(mockEvent, mockEvent.topic![1])).resolves.not.toThrow(); expect(mockTx.stream.update).toHaveBeenCalledTimes(1); expect(mockTx.stream.update).toHaveBeenCalledWith({ - where: { streamId }, + where: { streamId: BigInt(streamId) }, data: { withdrawnAmount: '1500', lastUpdateTime: 1700002000 }, }); expect(mockTx.streamEvent.upsert).toHaveBeenCalledTimes(1); @@ -604,7 +604,7 @@ describe('SorobanEventWorker', () => { const mockTx = { user: { upsert: vi.fn().mockResolvedValue({}) }, - stream: { upsert: vi.fn().mockResolvedValue({ streamId: 0, isActive: false }) }, + stream: { upsert: vi.fn().mockResolvedValue({ streamId: 0n, isActive: false }) }, streamEvent: { findUnique: vi.fn().mockResolvedValue(null), upsert: vi.fn().mockResolvedValue({ id: 'event-admin-transferred' }), @@ -622,7 +622,7 @@ describe('SorobanEventWorker', () => { expect(mockTx.streamEvent.upsert).toHaveBeenCalledWith( expect.objectContaining({ create: expect.objectContaining({ - streamId: 0, + streamId: 0n, eventType: 'ADMIN_TRANSFERRED', transactionHash: txHash, ledgerSequence: 1006, @@ -630,5 +630,71 @@ describe('SorobanEventWorker', () => { }) ); }); + + it('persists a u64 streamId above int4 max (2^31-1) without Number coercion (#829)', async () => { + // int4 max is 2_147_483_647; this value would previously fail with + // "value out of range for type integer" on insert. + const streamId = 3_000_000_000n; + const txHash = 'large-u64-stream-id-tx'; + + const mockEvent: rpc.Api.EventResponse = { + id: 'large-u64-event-1', + type: 'contract', + ledger: 5000, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash, + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'stream_created' } as any, + { switch: () => ({ value: 1 }), u64: () => ({ toString: () => streamId.toString() }) } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [ + { key: () => ({ sym: () => 'sender' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + { key: () => ({ sym: () => 'recipient' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + { key: () => ({ sym: () => 'token_address' }), val: () => ({ address: () => ({ switch: () => ({ value: 1 }), contractId: () => Buffer.alloc(32) }) }) }, + { key: () => ({ sym: () => 'rate_per_second' }), val: () => ({ i128: () => ({ hi: () => ({ toString: () => '0' }), lo: () => ({ toString: () => '100' }) }) }) }, + { key: () => ({ sym: () => 'deposited_amount' }), val: () => ({ i128: () => ({ hi: () => ({ toString: () => '0' }), lo: () => ({ toString: () => '86400' }) }) }) }, + { key: () => ({ sym: () => 'start_time' }), val: () => ({ u64: () => ({ toString: () => '1700000000' }) }) }, + ] as any, + } as any, + }; + + let capturedStreamUpsert: any = null; + let capturedEventUpsert: any = null; + const mockTx = { + user: { upsert: vi.fn().mockResolvedValue({}) }, + stream: { + upsert: vi.fn().mockImplementation((args) => { + capturedStreamUpsert = args; + return Promise.resolve({ streamId, isActive: true }); + }), + }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue(null), + upsert: vi.fn().mockImplementation((args) => { + capturedEventUpsert = args; + return Promise.resolve({ id: 'event-large-u64' }); + }), + }, + }; + + (prisma.$transaction as ReturnType).mockImplementation((cb) => cb(mockTx)); + + await expect( + (worker as any).handleStreamCreated(mockEvent, mockEvent.topic![1]), + ).resolves.not.toThrow(); + + expect(capturedStreamUpsert?.where?.streamId).toBe(streamId); + expect(capturedStreamUpsert?.create?.streamId).toBe(streamId); + expect(typeof capturedStreamUpsert?.create?.streamId).toBe('bigint'); + expect(capturedStreamUpsert.create.streamId > 2_147_483_647n).toBe(true); + + expect(capturedEventUpsert?.create?.streamId).toBe(streamId); + expect(typeof capturedEventUpsert?.create?.streamId).toBe('bigint'); + }); }); }); diff --git a/backend/tests/stream-id.test.ts b/backend/tests/stream-id.test.ts new file mode 100644 index 00000000..19a8b5ba --- /dev/null +++ b/backend/tests/stream-id.test.ts @@ -0,0 +1,30 @@ +import { describe, it, expect } from 'vitest'; +import { parseStreamId, streamIdToJson } from '../src/lib/stream-id.js'; + +describe('parseStreamId (#829)', () => { + it('parses decimal strings and numbers into bigint', () => { + expect(parseStreamId('42')).toBe(42n); + expect(parseStreamId(42)).toBe(42n); + expect(parseStreamId(42n)).toBe(42n); + }); + + it('parses values above int4 max without losing precision', () => { + expect(parseStreamId('3000000000')).toBe(3_000_000_000n); + expect(parseStreamId(3_000_000_000)).toBe(3_000_000_000n); + }); + + it('rejects negatives, non-integers, and non-decimal input', () => { + expect(parseStreamId(-1)).toBeNull(); + expect(parseStreamId('abc')).toBeNull(); + expect(parseStreamId('12.5')).toBeNull(); + expect(parseStreamId(1.5)).toBeNull(); + expect(parseStreamId(undefined)).toBeNull(); + }); + + it('serializes safe integers as numbers and larger values as strings', () => { + expect(streamIdToJson(42n)).toBe(42); + expect(streamIdToJson(3_000_000_000n)).toBe(3_000_000_000); + const huge = (1n << 60n); + expect(streamIdToJson(huge)).toBe(huge.toString()); + }); +}); diff --git a/backend/tests/stream.repository.test.ts b/backend/tests/stream.repository.test.ts index 86996616..2d05844f 100644 --- a/backend/tests/stream.repository.test.ts +++ b/backend/tests/stream.repository.test.ts @@ -16,33 +16,33 @@ describe('Stream Repository', () => { }); it('should update isActive to false for CANCELLED', async () => { - await updateStatus(123, 'CANCELLED'); + await updateStatus(123n, 'CANCELLED'); expect(prisma.stream.update).toHaveBeenCalledWith({ - where: { streamId: 123 }, + where: { streamId: 123n }, data: { isActive: false }, }); }); it('should update isActive to false for COMPLETED', async () => { - await updateStatus(123, 'COMPLETED'); + await updateStatus(123n, 'COMPLETED'); expect(prisma.stream.update).toHaveBeenCalledWith({ - where: { streamId: 123 }, + where: { streamId: 123n }, data: { isActive: false }, }); }); it('should update isActive to true for ACTIVE', async () => { - await updateStatus(123, 'ACTIVE'); + await updateStatus(123n, 'ACTIVE'); expect(prisma.stream.update).toHaveBeenCalledWith({ - where: { streamId: 123 }, + where: { streamId: 123n }, data: { isActive: true }, }); }); it('should update isActive to true for PAUSED', async () => { - await updateStatus(123, 'PAUSED'); + await updateStatus(123n, 'PAUSED'); expect(prisma.stream.update).toHaveBeenCalledWith({ - where: { streamId: 123 }, + where: { streamId: 123n }, data: { isActive: true }, }); }); diff --git a/backend/tests/stream.validator.test.ts b/backend/tests/stream.validator.test.ts index ff3b443f..29970798 100644 --- a/backend/tests/stream.validator.test.ts +++ b/backend/tests/stream.validator.test.ts @@ -14,6 +14,25 @@ describe('Stream Validator', () => { }; const result = createStreamSchema.safeParse(validData); expect(result.success).toBe(true); + if (result.success) { + expect(result.data.streamId).toBe(123n); + } + }); + + it('accepts a u64 streamId above int4 max (#829)', () => { + const result = createStreamSchema.safeParse({ + streamId: '3000000000', + sender: 'GSENDER', + recipient: 'GRECIPIENT', + tokenAddress: 'TABC', + ratePerSecond: '100', + depositedAmount: '1000', + startTime: 1622505600, + }); + expect(result.success).toBe(true); + if (result.success) { + expect(result.data.streamId).toBe(3_000_000_000n); + } }); it('should fail on invalid stream data', () => {