diff --git a/backend/README.md b/backend/README.md index ff6eda52..0a18b6d9 100644 --- a/backend/README.md +++ b/backend/README.md @@ -32,6 +32,23 @@ We use Prisma as our ORM to interact with PostgreSQL. - Configuration (schema path, migrations path, datasource URL) is defined in `prisma.config.ts`. - Run `npx prisma studio` to view the database through a web UI. +## Database Seeding + +Run the seed script to populate the database with demo data for local development: + +```bash +npx prisma db seed +``` + +**Idempotency guarantee** – The seed script (`prisma/seed.ts`) is safe to run multiple times against the same database. Every record is written with `upsert()` rather than `create()`: + +| Model | Unique key used for upsert | +|---|---| +| `User` | `publicKey` | +| `Stream` | `streamId` | +| `StreamEvent` | `transactionHash` + `eventType` | + +Re-running the seed against a populated database will leave existing rows unchanged and will not throw unique-constraint errors or create duplicate rows. ### Seeding the database `prisma/seed.ts` populates the database with demo fixtures for local development. Run it with: diff --git a/backend/prisma/seed.ts b/backend/prisma/seed.ts index ccf856aa..a1d7f0fd 100644 --- a/backend/prisma/seed.ts +++ b/backend/prisma/seed.ts @@ -54,13 +54,22 @@ async function main() { console.log({ stream1 }); - // Create an example event - const event1 = await prisma.streamEvent.create({ - data: { + // Upsert an example event – keyed on the @@unique([transactionHash, eventType]) + // constraint so re-running the seed never creates duplicate rows. + const DEMO_TX_HASH = '6f7e8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r'; + const event1 = await prisma.streamEvent.upsert({ + where: { + transactionHash_eventType: { + transactionHash: DEMO_TX_HASH, + eventType: 'CREATED', + }, + }, + update: {}, + create: { streamId: stream1.streamId, eventType: 'CREATED', amount: '1000000000000', - transactionHash: '6f7e8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r', + transactionHash: DEMO_TX_HASH, ledgerSequence: 123456, timestamp: Math.floor(Date.now() / 1000), metadata: JSON.stringify({ memo: 'Seed data' }), diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts index e02ca46d..ac9f7e5f 100644 --- a/backend/src/controllers/stream.controller.ts +++ b/backend/src/controllers/stream.controller.ts @@ -211,7 +211,7 @@ export const createStream = async (req: Request, res: Response) => { }, }); - return res.status(201).json(stream); + return res.status(201).json(JSON.parse(JSON.stringify(stream, (_key, value) => typeof value === "bigint" ? value.toString() : value))); } catch (error) { if ( error instanceof RangeError || diff --git a/backend/src/lib/indexer-state.ts b/backend/src/lib/indexer-state.ts index e6568647..92b3ae79 100644 --- a/backend/src/lib/indexer-state.ts +++ b/backend/src/lib/indexer-state.ts @@ -7,7 +7,6 @@ export interface IndexerStateRow { id: string; lastLedger: number; lastCursor: string | null; - createdAt: Date; updatedAt: Date; } diff --git a/backend/src/middleware/admin-rate-limiter.middleware.ts b/backend/src/middleware/admin-rate-limiter.middleware.ts index 1f665ef7..09160f03 100644 --- a/backend/src/middleware/admin-rate-limiter.middleware.ts +++ b/backend/src/middleware/admin-rate-limiter.middleware.ts @@ -15,7 +15,7 @@ export const adminRateLimiter = rateLimit({ // Use x-forwarded-for or remote address as key const forwarded = req.headers['x-forwarded-for']; if (typeof forwarded === 'string') { - return forwarded.split(',')[0].trim(); + return forwarded.split(',')[0]?.trim() ?? 'unknown'; } return req.ip ?? 'unknown'; }, diff --git a/backend/src/services/soroban-indexer.service.ts b/backend/src/services/soroban-indexer.service.ts index 53fd1a4c..a232ce81 100644 --- a/backend/src/services/soroban-indexer.service.ts +++ b/backend/src/services/soroban-indexer.service.ts @@ -179,8 +179,13 @@ export class SorobanIndexerService { const ratePerSecond = this.readString(value, 'rate_per_second', 'ratePerSecond'); const depositedAmount = this.readString(value, 'deposited_amount', 'depositedAmount'); const startTimeRaw = value.start_time ?? value.startTime ?? timestamp; - const startTime = BigInt(startTimeRaw ?? timestamp); - const timestampBigInt = BigInt(timestamp); + const startTime = BigInt( + typeof startTimeRaw === 'string' || + typeof startTimeRaw === 'number' || + typeof startTimeRaw === 'bigint' + ? startTimeRaw + : timestamp, + ); if (!sender || !recipient || !tokenAddress || !ratePerSecond || !depositedAmount) return; diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index d53ebf7e..cc322941 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -1005,7 +1005,7 @@ export class SorobanEventWorker { } const sender = decodeAddress(body["sender"]); - const newEndTime = Number(decodeU64(body["new_end_time"])); + const newEndTime = decodeU64(body["new_end_time"]); const timestamp = Math.floor(Date.now() / 1000); await prisma.$transaction(async (tx: Prisma.TransactionClient) => { @@ -1018,7 +1018,7 @@ export class SorobanEventWorker { // Calculate the duration of this pause interval let additionalPausedDuration = 0; if (currentStream.pausedAt) { - additionalPausedDuration = timestamp - currentStream.pausedAt; + additionalPausedDuration = timestamp - Number(currentStream.pausedAt); } const newTotalPausedDuration = @@ -1064,7 +1064,7 @@ export class SorobanEventWorker { timestamp, metadata: JSON.stringify({ sender, - newEndTime, + newEndTime: newEndTime.toString(), pausedDuration: additionalPausedDuration, totalPausedDuration: newTotalPausedDuration, }), @@ -1080,7 +1080,7 @@ export class SorobanEventWorker { sseService.broadcastToStream(String(streamId), "stream.resumed", { streamId, sender, - newEndTime, + newEndTime: newEndTime.toString(), transactionHash: event.txHash, ledger: event.ledger, timestamp, diff --git a/backend/tests/events-wire-format.test.ts b/backend/tests/events-wire-format.test.ts index e7cd12fb..acdacfd3 100644 --- a/backend/tests/events-wire-format.test.ts +++ b/backend/tests/events-wire-format.test.ts @@ -91,7 +91,7 @@ describe('event wire format', () => { expect(decodedKeys).toEqual([...allFields].sort()); - for (const field of HANDLER_READ_FIELDS[eventName]) { + for (const field of HANDLER_READ_FIELDS[eventName] ?? []) { expect(decoded).toHaveProperty(field); } }, diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 772017f8..d541e6ba 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -575,7 +575,7 @@ describe('SorobanEventWorker', () => { // Sanity check: depositedAmount/endTime from the (only) applied update // match what a single application should produce. expect(firstUpdateArgs.data.depositedAmount).toBe('1200'); - expect(expectedEndTime).toBe(1700000000 + Math.floor(1200 / 10) + 0); + expect(expectedEndTime).toBe(1700000000n + BigInt(Math.floor(1200 / 10))); }); it('should process admin_transferred events successfully', async () => { diff --git a/backend/tests/stream.test.ts b/backend/tests/stream.test.ts index cad05ac1..a0dbe230 100644 --- a/backend/tests/stream.test.ts +++ b/backend/tests/stream.test.ts @@ -65,8 +65,8 @@ describe('POST /v1/streams', () => { ratePerSecond: '100', depositedAmount: '86400', withdrawnAmount: '0', - startTime: 1700000000, - lastUpdateTime: 1700000000, + startTime: 1700000000n, + lastUpdateTime: 1700000000n, isPaused: false, pausedAt: null, totalPausedDuration: 0, @@ -247,8 +247,8 @@ describe('GET /v1/users/:address/summary', () => { ratePerSecond: '10', depositedAmount: '100', withdrawnAmount: '30', - startTime: 1000, - lastUpdateTime: 2000, + startTime: 1000n, + lastUpdateTime: 2000n, isPaused: false, endTime: null, pausedAt: null, @@ -266,8 +266,8 @@ describe('GET /v1/users/:address/summary', () => { ratePerSecond: '20', depositedAmount: '200', withdrawnAmount: '20', - startTime: 1000, - lastUpdateTime: 2000, + startTime: 1000n, + lastUpdateTime: 2000n, isPaused: false, endTime: null, pausedAt: null, @@ -287,8 +287,8 @@ describe('GET /v1/users/:address/summary', () => { ratePerSecond: '10', depositedAmount: '1000', withdrawnAmount: '100', - startTime: 1000, - lastUpdateTime: 0, + startTime: 1000n, + lastUpdateTime: 0n, isPaused: false, endTime: null, pausedAt: null, @@ -306,8 +306,8 @@ describe('GET /v1/users/:address/summary', () => { ratePerSecond: '5', depositedAmount: '500', withdrawnAmount: '0', - startTime: 1000, - lastUpdateTime: 0, + startTime: 1000n, + lastUpdateTime: 0n, isPaused: false, endTime: null, pausedAt: null, @@ -344,8 +344,8 @@ describe('GET /v1/users/:address/summary', () => { ratePerSecond: '1', depositedAmount: '100', withdrawnAmount: '1', - startTime: 1000, - lastUpdateTime: 2000, + startTime: 1000n, + lastUpdateTime: 2000n, isPaused: false, endTime: null, pausedAt: null, diff --git a/backend/tests/stream.validator.test.ts b/backend/tests/stream.validator.test.ts index f546a5d0..64bef076 100644 --- a/backend/tests/stream.validator.test.ts +++ b/backend/tests/stream.validator.test.ts @@ -42,7 +42,7 @@ describe('Stream Validator', () => { }; const result = createStreamSchema.safeParse(data); expect(result.success).toBe(false); - expect(result.error?.issues[0].message).toBe('Rate exceeds maximum allowed value'); + expect(result.error?.issues[0]?.message).toBe('Rate exceeds maximum allowed value'); }); it('should accept ratePerSecond at i128 max', () => { diff --git a/contracts/stream_contract/src/test.rs b/contracts/stream_contract/src/test.rs index 194d74be..74a87b12 100644 --- a/contracts/stream_contract/src/test.rs +++ b/contracts/stream_contract/src/test.rs @@ -433,8 +433,8 @@ fn test_backdated_start_time_would_immediately_vest_full_amount() { // env.ledger().timestamp() — but it demonstrates the risk that would exist // if a caller-supplied start_time were ever added. let mut stream = client.get_stream(&stream_id).unwrap(); - stream.start_time = 0; // backdated far into the past - stream.last_update_time = 0; // sync anchor to match + stream.start_time = 0; // backdated far into the past + stream.last_update_time = 0; // sync anchor to match env.as_contract(&client.address, || { env.storage() .persistent() @@ -2709,11 +2709,8 @@ fn test_cancel_state_committed_before_transfers_prevents_double_cancel() { fn event_field_names(env: &Env, payload: &soroban_sdk::Val) -> std::vec::Vec { let map = soroban_sdk::Map::::try_from_val(env, payload) .expect("event data is not a Map"); - let mut names: std::vec::Vec = map - .keys() - .iter() - .map(|sym| sym.to_string()) - .collect(); + let mut names: std::vec::Vec = + map.keys().iter().map(|sym| sym.to_string()).collect(); names.sort(); names } diff --git a/frontend/src/hooks/useIncomingStreams.test.ts b/frontend/src/hooks/useIncomingStreams.test.tsx similarity index 89% rename from frontend/src/hooks/useIncomingStreams.test.ts rename to frontend/src/hooks/useIncomingStreams.test.tsx index 39a9b9d8..ac223766 100644 --- a/frontend/src/hooks/useIncomingStreams.test.ts +++ b/frontend/src/hooks/useIncomingStreams.test.tsx @@ -1,3 +1,4 @@ +/* eslint-disable @typescript-eslint/no-explicit-any */ import { renderHook, waitFor, act } from "@testing-library/react"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { vi, describe, it, expect, beforeEach, afterEach } from "vitest"; @@ -31,6 +32,7 @@ describe("useIncomingStreams hooks", () => { }); afterEach(() => { + vi.useRealTimers(); queryClient.clear(); }); @@ -89,6 +91,7 @@ describe("useIncomingStreams hooks", () => { ); const invalidateSpy = vi.spyOn(queryClient, "invalidateQueries"); + vi.useFakeTimers(); await act(async () => { await result.current.mutateAsync({ @@ -102,12 +105,14 @@ describe("useIncomingStreams hooks", () => { } as any); }); - // Wait for pollIndexerForWithdraw to complete and call invalidateQueries - await waitFor(() => { - expect(invalidateSpy).toHaveBeenCalledWith({ - queryKey: incomingStreamsQueryKey("pubkey"), - }); - }, { timeout: 10000 }); + // Advance the retry delays so the fallback invalidation runs immediately. + await act(async () => { + await vi.advanceTimersByTimeAsync(63_000); + }); + + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: incomingStreamsQueryKey("pubkey"), + }); }); }); });