From 2c5b87d177a703a83fa8fd7fb1f871718e96e592 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 | 4 +- 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/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 +++++ 24 files changed, 288 insertions(+), 87 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 664105b7..d0caba15 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 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 e02ca46d..c453e67d 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 ea0bf849..f517597a 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 9c6e9846..5988eb05 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 7907fed6..b46e914a 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 53fd1a4c..62e7a0fd 100644 --- a/backend/src/services/soroban-indexer.service.ts +++ b/backend/src/services/soroban-indexer.service.ts @@ -122,12 +122,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 d13b5843..02691fe8 100644 --- a/backend/src/services/sorobanService.ts +++ b/backend/src/services/sorobanService.ts @@ -135,7 +135,7 @@ export function resetServer(): void { } export interface ChainStream { - streamId: number; + streamId: bigint; sender: string; recipient: string; tokenAddress: string; @@ -245,7 +245,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 { @@ -277,7 +277,7 @@ export async function getStreamFromChain(streamId: number): Promise { +export async function getClaimableFromChain(streamId: bigint): Promise { if (!getContractId()) return null; try { @@ -292,13 +292,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', [ @@ -324,7 +324,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'); @@ -358,7 +358,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'); @@ -390,7 +390,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/validators/stream.validator.ts b/backend/src/validators/stream.validator.ts index 7e132fc8..e0aef700 100644 --- a/backend/src/validators/stream.validator.ts +++ b/backend/src/validators/stream.validator.ts @@ -1,9 +1,22 @@ import { z } from 'zod'; +import { parseStreamId } from '../lib/stream-id.js'; const MAX_I128 = BigInt('170141183460469231731687303715884105727'); 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 d53ebf7e..9f19b315 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, ensureIndexerState } 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: @@ -356,7 +357,7 @@ export class SorobanEventWorker { }, }, create: { - streamId: 0, + streamId: 0n, eventType: "FEE_CONFIG_UPDATED", transactionHash: event.txHash, ledgerSequence: event.ledger, @@ -412,7 +413,7 @@ export class SorobanEventWorker { }, }, create: { - streamId: 0, + streamId: 0n, eventType: "ADMIN_TRANSFERRED", transactionHash: event.txHash, ledgerSequence: event.ledger, @@ -445,7 +446,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 ( @@ -565,7 +566,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"]) { @@ -646,7 +647,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"]) { @@ -718,7 +719,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"]) { @@ -791,7 +792,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"]) { @@ -864,7 +865,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"]) { @@ -925,7 +926,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"]) { @@ -997,7 +998,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 bdc386bf..05fec5a9 100644 --- a/backend/tests/stream.repository.test.ts +++ b/backend/tests/stream.repository.test.ts @@ -19,33 +19,33 @@ describe('Stream Repository', () => { describe('updateStatus', () => { 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 f546a5d0..01a39124 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', () => {