From e7b41e1a2c802503227d30d59e0c11f82878219f Mon Sep 17 00:00:00 2001 From: p4uld4vid016-code Date: Sun, 18 Oct 2026 02:24:44 +0000 Subject: [PATCH] fix: resolve stream cancel visibility and wizard tx/stream matching issues - #1219: write a provisional CANCELLED StreamEvent atomically with the API cancel status update so cancelled streams immediately appear under the 'cancelled' status filter; the indexer reconciles the provisional event with authoritative ledger data instead of dropping it. - #1204: wizard post-submit polling now matches the newly created stream by recipient, deposit amount and most-recent createdAt instead of blindly taking streams[0]. - #1205: cover /streams/create client-side recipient validation with an integration test asserting an inline error and no createStream call. - #1203: verify the wizard surfaces the real transaction hash (already typed/returned correctly, with regression tests). Closes #1219 Closes #1204 Closes #1205 Closes #1203 --- backend/src/controllers/stream/cancel.ts | 6 +- backend/src/repositories/stream.repository.ts | 64 +++++ backend/src/workers/soroban-event-worker.ts | 48 ++-- backend/tests/cancel.controller.test.ts | 5 +- .../integration/cancel-status-filter.test.ts | 265 ++++++++++++++++++ .../tests/integration/streams/cancel.test.ts | 19 ++ backend/tests/stream.repository.test.ts | 42 ++- .../__tests__/create-stream-content.test.tsx | 79 +++++- .../stream-creation/StreamCreationWizard.tsx | 48 +++- .../__tests__/StreamCreationWizard.test.tsx | 108 ++++++- 10 files changed, 633 insertions(+), 51 deletions(-) create mode 100644 backend/tests/integration/cancel-status-filter.test.ts diff --git a/backend/src/controllers/stream/cancel.ts b/backend/src/controllers/stream/cancel.ts index baefce72..8fd7cbb3 100644 --- a/backend/src/controllers/stream/cancel.ts +++ b/backend/src/controllers/stream/cancel.ts @@ -89,8 +89,10 @@ export const cancelStreamHandler = async (req: AuthenticatedRequest, res: Respon const txHash = await sorobanService.cancelStream(parsedStreamId, senderSecret); - // 5. Update DB record status using repository helper - await streamRepository.updateStatus(parsedStreamId, 'CANCELLED'); + // 5. Update DB record status and write a provisional CANCELLED StreamEvent + // atomically (issue #1219) so the stream shows up under the "cancelled" + // filter before the indexer has processed the on-chain event. + await streamRepository.markCancelledProvisionally(parsedStreamId, txHash); logger.info(`[CancelStream] Stream ${parsedStreamId} cancelled by ${callerAddress}. Tx: ${txHash}`); diff --git a/backend/src/repositories/stream.repository.ts b/backend/src/repositories/stream.repository.ts index 6546d22f..ba5ce9ee 100644 --- a/backend/src/repositories/stream.repository.ts +++ b/backend/src/repositories/stream.repository.ts @@ -1,4 +1,68 @@ import { prisma, withReplicaFallback } from '../lib/prisma.js'; +import type { Prisma } from '../generated/prisma/index.js'; + +/** + * Transaction client passed to interactive Prisma transactions. + */ +export type PrismaTxClient = Prisma.TransactionClient; + +/** + * Atomically mark a stream as cancelled (via the API) and write a provisional + * CANCELLED StreamEvent in the same transaction. + * + * The indexer's handleStreamCancelled writes an authoritative event for the + * real on-chain StreamCancelled event later; both variants share the + * (transactionHash, eventType) unique key, so the indexer reconciles (updates) + * the provisional record instead of creating a duplicate. + * + * Issue #1219: without the provisional event the cancelled stream matched + * neither the "active" filter (isActive=false) nor the "cancelled" filter + * (requires a CANCELLED event) and disappeared from all status views until + * the on-chain event was indexed. + */ +export const markCancelledProvisionally = async ( + streamId: bigint, + txHash: string, + txClient?: PrismaTxClient, +) => { + const timestamp = Math.floor(Date.now() / 1000); + const run = async (tx: PrismaTxClient) => { + await tx.stream.update({ + where: { streamId }, + data: { + isActive: false, + }, + }); + + await tx.streamEvent.upsert({ + where: { + transactionHash_eventType: { + transactionHash: txHash, + eventType: 'CANCELLED', + }, + }, + create: { + streamId, + eventType: 'CANCELLED', + amount: null, + transactionHash: txHash, + // Ledger details are unknown until the indexer sees the real event; + // ledgerSequence 0 + provisional metadata marks this as API-written. + ledgerSequence: 0, + timestamp, + metadata: JSON.stringify({ provisional: true, source: 'api-cancel' }), + }, + // If another request already wrote the provisional event for this + // transaction, don't clobber it with a second provisional copy. + update: {}, + }); + }; + + if (txClient) { + return run(txClient); + } + return prisma.$transaction(run); +}; /** * Update the status and active flag of a stream in the database. diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index b631bc43..be4df164 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -1166,31 +1166,35 @@ export class SorobanEventWorker { }); if (existingEvent) { logger.warn( - `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=CANCELLED`, + `[SorobanWorker] Reconciling existing StreamEvent: txHash=${event.txHash} type=CANCELLED`, ); - } else { - await tx.streamEvent.upsert({ - where: { - transactionHash_eventType: { - transactionHash: event.txHash, - eventType: "CANCELLED", - }, - }, - create: { - streamId, - eventType: "CANCELLED", - amount: refundedAmount, + } + await tx.streamEvent.upsert({ + where: { + transactionHash_eventType: { transactionHash: event.txHash, - ledgerSequence: event.ledger, - timestamp, - metadata: JSON.stringify({ amountWithdrawn, refundedAmount }), - }, - update: { - ledgerSequence: event.ledger, - timestamp, + eventType: "CANCELLED", }, - }); - } + }, + create: { + streamId, + eventType: "CANCELLED", + amount: refundedAmount, + transactionHash: event.txHash, + ledgerSequence: event.ledger, + timestamp, + metadata: JSON.stringify({ amountWithdrawn, refundedAmount }), + }, + // When an API-cancelled stream already wrote a provisional CANCELLED + // event (issue #1219), replace the provisional fields with the + // authoritative ledger details from the indexed on-chain event. + update: { + ledgerSequence: event.ledger, + timestamp, + amount: refundedAmount, + metadata: JSON.stringify({ amountWithdrawn, refundedAmount }), + }, + }); }); sseService.broadcastToStream(String(streamId), "stream.cancelled", { diff --git a/backend/tests/cancel.controller.test.ts b/backend/tests/cancel.controller.test.ts index ce1a457e..a41709f8 100644 --- a/backend/tests/cancel.controller.test.ts +++ b/backend/tests/cancel.controller.test.ts @@ -20,6 +20,7 @@ vi.mock('../src/services/sorobanService.js', () => ({ vi.mock('../src/repositories/stream.repository.js', () => ({ updateStatus: vi.fn(), + markCancelledProvisionally: vi.fn(), })); vi.mock('../src/logger.js', () => ({ @@ -79,7 +80,7 @@ describe('Cancel Stream Controller', () => { await cancelStreamHandler(req as AuthenticatedRequest, res as Response); expect(sorobanService.cancelStream).toHaveBeenCalledWith(123n, 'SABC123'); - expect(streamRepository.updateStatus).toHaveBeenCalledWith(123n, 'CANCELLED'); + expect(streamRepository.markCancelledProvisionally).toHaveBeenCalledWith(123n, 'tx_hash_123'); expect(res.status).toHaveBeenCalledWith(200); expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ status: 'CANCELLED', txHash: 'tx_hash_123' })); }); @@ -103,7 +104,7 @@ describe('Cancel Stream Controller', () => { await cancelStreamHandler(req as AuthenticatedRequest, res as Response); expect(sorobanService.cancelStream).toHaveBeenCalledWith(123n, 'SABC123'); - expect(streamRepository.updateStatus).not.toHaveBeenCalled(); + expect(streamRepository.markCancelledProvisionally).not.toHaveBeenCalled(); expect(res.status).toHaveBeenCalledWith(500); }); }); diff --git a/backend/tests/integration/cancel-status-filter.test.ts b/backend/tests/integration/cancel-status-filter.test.ts new file mode 100644 index 00000000..94d0b824 --- /dev/null +++ b/backend/tests/integration/cancel-status-filter.test.ts @@ -0,0 +1,265 @@ +/** + * Integration test for issue #1219. + * + * Cancelling a stream through the API must write a provisional CANCELLED + * StreamEvent together with the isActive=false update, so the stream + * immediately matches the "cancelled" status filter. The indexer later + * reconciles the provisional record with the authoritative on-chain event. + * + * Uses a real Postgres database (skipped cleanly when DATABASE_URL is not + * reachable, mirroring stream-lifecycle.test.ts). + */ +import { + describe, + it, + expect, + vi, + beforeAll, + beforeEach, + afterEach, + afterAll, +} from "vitest"; +import request from "supertest"; +import pg from "pg"; +import { PrismaPg } from "@prisma/adapter-pg"; +import { xdr, nativeToScVal } from "@stellar/stellar-sdk"; +import type { PrismaClient } from "../../src/generated/prisma/index.js"; +import app from "../../src/app.js"; +import { SorobanEventWorker } from "../../src/workers/soroban-event-worker.js"; +import { + resolveDbReadiness, + resolveTestDatabaseUrl, + explainSkipReason, +} from "./_db.js"; + +const { SENDER, RECIPIENT, TOKEN, API_CANCEL_TX } = vi.hoisted(() => ({ + SENDER: "GAV4A377RAEV6YVAWZVHXF4VZD5ZBXGIKEMNHV5YIMV5LIKSNQVYUBR7", + RECIPIENT: "GBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB", + TOKEN: "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + API_CANCEL_TX: "api_cancel_tx_hash", +})); + +vi.mock("../../src/services/sorobanService.js", () => ({ + cancelStream: vi.fn().mockResolvedValue(API_CANCEL_TX), + getStreamFromChain: vi.fn(), + getClaimableFromChain: vi.fn(), + isStale: vi.fn().mockReturnValue(false), +})); + +// Bypass real Stellar signature verification; identify the caller via header. +vi.mock("../../src/middleware/auth.js", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + requireAuth: (req: any, _res: any, next: any) => { + req.user = { publicKey: req.headers["x-test-caller"] ?? SENDER }; + next(); + }, + requireAdmin: (_req: any, res: any, _next: any) => { + res.status(403).json({ error: "Forbidden" }); + }, + }; +}); + +let testPrisma!: PrismaClient; +let testPool!: pg.Pool; +let lastDbReadiness: Awaited> | null = + null; + +function getDb(): PrismaClient { + if (!testPrisma) { + throw new Error("testPrisma accessed before initialization"); + } + return testPrisma; +} + +function scvU64(n: bigint): xdr.ScVal { + return nativeToScVal(n, { type: "u64" }); +} +function scvI128(n: bigint): xdr.ScVal { + return nativeToScVal(n, { type: "i128" }); +} +function scvSymbol(s: string): xdr.ScVal { + return xdr.ScVal.scvSymbol(s); +} +function scvMap(entries: [string, xdr.ScVal][]): xdr.ScVal { + return xdr.ScVal.scvMap( + entries.map(([k, v]) => new xdr.ScMapEntry({ key: scvSymbol(k), val: v })), + ); +} + +function createStreamCancelledEvent( + streamId: number, + amountWithdrawn: number, + refundedAmount: number, + txHash: string, + ledger = 9001, +) { + return { + id: `evt-stream-cancelled-${streamId}`, + type: "contract" as const, + ledger, + ledgerClosedAt: "2023-01-01T00:00:00Z", + transactionIndex: 0, + operationIndex: 0, + txHash, + topic: [scvSymbol("stream_cancelled"), scvU64(BigInt(streamId))], + value: scvMap([ + ["amount_withdrawn", scvI128(BigInt(amountWithdrawn))], + ["refunded_amount", scvI128(BigInt(refundedAmount))], + ]), + inSuccessfulContractCall: true, + }; +} + +async function cleanupDatabase() { + const db = getDb(); + await db.streamEvent.deleteMany(); + await db.stream.deleteMany(); + await db.user.deleteMany(); +} + +const STREAM_ID = 900; + +async function seedActiveStream() { + const db = getDb(); + await db.user.createMany({ + data: [{ publicKey: SENDER }, { publicKey: RECIPIENT }], + skipDuplicates: true, + }); + await db.stream.create({ + data: { + streamId: BigInt(STREAM_ID), + sender: SENDER, + recipient: RECIPIENT, + tokenAddress: TOKEN, + ratePerSecond: "10", + depositedAmount: "86400", + withdrawnAmount: "0", + startTime: 1700000000, + endTime: 1700086400, + lastUpdateTime: 1700000000, + isActive: true, + isPaused: false, + }, + }); +} + +async function cancelViaApi() { + return request(app) + .post(`/v1/streams/${STREAM_ID}/cancel`) + .set("Authorization", "Bearer dummy_token") + .set("x-test-caller", SENDER) + .send({ senderSecret: "S_SECRET_123" }); +} + +describe("API cancel → cancelled status filter (issue #1219)", () => { + beforeAll(async () => { + const readiness = await resolveDbReadiness(); + lastDbReadiness = readiness; + if (!readiness.ready) { + // eslint-disable-next-line no-console + console.warn(`\n${explainSkipReason(readiness)}\n`); + return; + } + + const { PrismaClient: PrismaClientCtor } = await import( + "../../src/generated/prisma/index.js" + ); + const { createPgPool } = await import("../../src/lib/pg-pool.js"); + testPool = createPgPool({ connectionString: resolveTestDatabaseUrl() }); + const adapter = new PrismaPg(testPool); + testPrisma = new PrismaClientCtor({ adapter, log: ["error"] }); + }); + + beforeEach(async (ctx) => { + if (!lastDbReadiness?.ready) { + ctx.skip(); + return; + } + vi.clearAllMocks(); + await cleanupDatabase(); + await seedActiveStream(); + }); + + afterEach(async () => { + if (!lastDbReadiness?.ready) return; + await cleanupDatabase(); + }); + + afterAll(async () => { + if (!lastDbReadiness?.ready || !testPrisma) return; + await testPrisma.$disconnect(); + await testPool.end().catch(() => undefined); + }); + + it("writes a provisional CANCELLED event so the stream appears under status=cancelled", async () => { + const res = await cancelViaApi(); + + expect(res.status).toBe(200); + expect(res.body).toMatchObject({ txHash: API_CANCEL_TX, status: "CANCELLED" }); + + // The stream is now inactive... + const stream = await getDb().stream.findUnique({ + where: { streamId: BigInt(STREAM_ID) }, + }); + expect(stream?.isActive).toBe(false); + + // ...and immediately matches the "cancelled" status filter. + const cancelledRes = await request(app) + .get(`/v1/streams?status=cancelled&sender=${SENDER}`) + .expect(200); + const cancelledIds = cancelledRes.body.data.map((s: any) => + String(s.streamId), + ); + expect(cancelledIds).toContain(String(STREAM_ID)); + + // It no longer matches the "active" filter. + const activeRes = await request(app) + .get(`/v1/streams?status=active&sender=${SENDER}`) + .expect(200); + const activeIds = activeRes.body.data.map((s: any) => String(s.streamId)); + expect(activeIds).not.toContain(String(STREAM_ID)); + + // A provisional CANCELLED event was persisted with the cancel tx hash. + const event = await getDb().streamEvent.findFirst({ + where: { streamId: BigInt(STREAM_ID), eventType: "CANCELLED" }, + }); + expect(event).not.toBeNull(); + expect(event?.transactionHash).toBe(API_CANCEL_TX); + expect(JSON.parse(event?.metadata ?? "{}")).toMatchObject({ + provisional: true, + }); + }); + + it("reconciles the provisional event once the indexer sees the same transaction", async () => { + await cancelViaApi(); + + const worker = new SorobanEventWorker(); + await worker.processEvent( + createStreamCancelledEvent(STREAM_ID, 300, 700, API_CANCEL_TX, 9001), + ); + + // No duplicate: the provisional row is updated in place. + const events = await getDb().streamEvent.findMany({ + where: { streamId: BigInt(STREAM_ID), eventType: "CANCELLED" }, + }); + expect(events).toHaveLength(1); + expect(events[0]?.ledgerSequence).toBe(9001); + expect(events[0]?.amount).toBe("700"); + + // The authoritative on-chain amounts are applied to the stream. + const stream = await getDb().stream.findUnique({ + where: { streamId: BigInt(STREAM_ID) }, + }); + expect(stream?.withdrawnAmount).toBe("300"); + expect(stream?.isActive).toBe(false); + + // Still exactly one entry under the cancelled filter. + const cancelledRes = await request(app) + .get(`/v1/streams?status=cancelled&sender=${SENDER}`) + .expect(200); + expect(cancelledRes.body.data).toHaveLength(1); + }); +}); diff --git a/backend/tests/integration/streams/cancel.test.ts b/backend/tests/integration/streams/cancel.test.ts index e71ddff2..c0fa50d2 100644 --- a/backend/tests/integration/streams/cancel.test.ts +++ b/backend/tests/integration/streams/cancel.test.ts @@ -29,7 +29,9 @@ vi.mock('../../../src/lib/prisma.js', () => { }, streamEvent: { create: vi.fn(), + upsert: vi.fn(), }, + $transaction: vi.fn(async (fn: any) => fn(mockPrisma)), }; return { prisma: mockPrisma, @@ -91,6 +93,23 @@ describe('POST /v1/streams/:streamId/cancel', () => { where: { streamId: BigInt(streamId) }, data: { isActive: false }, }); + // Issue #1219: a provisional CANCELLED StreamEvent is written atomically + // so the stream immediately matches the "cancelled" status filter. + expect(prisma.streamEvent.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { + transactionHash_eventType: { + transactionHash: 'tx_hash_123', + eventType: 'CANCELLED', + }, + }, + create: expect.objectContaining({ + streamId: BigInt(streamId), + eventType: 'CANCELLED', + transactionHash: 'tx_hash_123', + }), + }), + ); }); it('returns 400 if senderSecret is not provided in the request body', async () => { diff --git a/backend/tests/stream.repository.test.ts b/backend/tests/stream.repository.test.ts index 2f07ac5a..513b57f8 100644 --- a/backend/tests/stream.repository.test.ts +++ b/backend/tests/stream.repository.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; -import { updateStatus, findStreams } from '../src/repositories/stream.repository.js'; +import { updateStatus, findStreams, markCancelledProvisionally } from '../src/repositories/stream.repository.js'; import { prisma } from '../src/lib/prisma.js'; vi.mock('../src/lib/prisma.js', () => ({ @@ -10,6 +10,10 @@ vi.mock('../src/lib/prisma.js', () => ({ findMany: vi.fn(), count: vi.fn(), }, + streamEvent: { + upsert: vi.fn(), + }, + $transaction: vi.fn((fn: any) => fn(prisma)), }, })); @@ -52,6 +56,42 @@ describe('Stream Repository', () => { }); }); + describe('markCancelledProvisionally', () => { + it('atomically marks the stream inactive and writes a provisional CANCELLED event', async () => { + await markCancelledProvisionally(123n, 'tx_hash_abc'); + + expect(prisma.stream.update).toHaveBeenCalledWith({ + where: { streamId: 123n }, + data: { isActive: false }, + }); + + expect(prisma.streamEvent.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { + transactionHash_eventType: { + transactionHash: 'tx_hash_abc', + eventType: 'CANCELLED', + }, + }, + create: expect.objectContaining({ + streamId: 123n, + eventType: 'CANCELLED', + transactionHash: 'tx_hash_abc', + ledgerSequence: 0, + }), + }), + ); + + const createArg = (prisma.streamEvent.upsert as any).mock.calls[0][0].create; + expect(JSON.parse(createArg.metadata)).toMatchObject({ provisional: true }); + }); + + it('runs inside a transaction so the status and event are written together', async () => { + await markCancelledProvisionally(123n, 'tx_hash_abc'); + expect(prisma.$transaction).toHaveBeenCalledTimes(1); + }); + }); + describe('findStreams', () => { const baseParams = { limit: 10, offset: 0 }; diff --git a/frontend/src/app/streams/create/__tests__/create-stream-content.test.tsx b/frontend/src/app/streams/create/__tests__/create-stream-content.test.tsx index 2e25b723..7992ad61 100644 --- a/frontend/src/app/streams/create/__tests__/create-stream-content.test.tsx +++ b/frontend/src/app/streams/create/__tests__/create-stream-content.test.tsx @@ -33,6 +33,14 @@ if (typeof window !== "undefined") { const searchParamsMock = { get: vi.fn() }; const push = vi.fn(); +// Overridable wallet state so individual tests can simulate a connected wallet. +const mockUseWallet = vi.hoisted(() => + vi.fn(() => ({ + status: "disconnected", + session: null as null | { publicKey: string }, + })), +); + vi.mock("next/navigation", () => ({ useRouter: () => ({ push }), useSearchParams: () => searchParamsMock, @@ -43,15 +51,18 @@ vi.mock("react-hot-toast", () => ({ })); vi.mock("@/context/wallet-context", () => ({ - useWallet: () => ({ status: "disconnected", session: null }), + useWallet: mockUseWallet, })); vi.mock("@/lib/soroban", () => ({ createStream: vi.fn(), - toBaseUnits: vi.fn(), - toDurationSeconds: vi.fn(), - getTokenAddress: vi.fn(), - toSorobanErrorMessage: vi.fn(), + toBaseUnits: vi.fn((value: string) => + BigInt(Math.round(parseFloat(value || "0") * 10 ** 7)), + ), + toDurationSeconds: vi.fn(() => 0n), + getTokenAddress: vi.fn(() => "token-address"), + toSorobanErrorMessage: vi.fn(() => "error"), + fetchTokenBalanceDisplay: vi.fn().mockResolvedValue("1000"), TOKEN_ADDRESSES: { XLM: "xlm-address", USDC: "usdc-address" }, })); @@ -60,6 +71,7 @@ vi.mock("@/lib/logger", () => ({ })); import CreateStreamContent from "../create-stream-content"; +import { createStream } from "@/lib/soroban"; function getRecipientInput() { return screen.getByLabelText(/recipient/i) as HTMLInputElement; @@ -78,6 +90,7 @@ describe("CreateStreamContent sessionStorage draft persistence", () => { vi.clearAllMocks(); sessionStorageMock.clear(); searchParamsMock.get.mockReturnValue(null); + mockUseWallet.mockReturnValue({ status: "disconnected", session: null }); }); it("saves form data to sessionStorage when fields change", async () => { @@ -272,3 +285,59 @@ describe("CreateStreamContent sessionStorage draft persistence", () => { expect(getDurationInput().value).toBe("30"); }); }); + +describe("CreateStreamContent recipient validation (issue #1205)", () => { + beforeEach(() => { + vi.clearAllMocks(); + sessionStorageMock.clear(); + searchParamsMock.get.mockReturnValue(null); + mockUseWallet.mockReturnValue({ + status: "connected", + session: { + publicKey: "GAV4A377RAEV6YVAWZVHXF4VZD5ZBXGIKEMNHV5YIMV5LIKSNQVYUBR7", + }, + }); + }); + + it("shows an inline error and never calls createStream for a malformed address", async () => { + render(); + + await act(async () => { + fireEvent.change(getRecipientInput(), { + target: { value: "not-a-valid-stellar-key" }, + }); + fireEvent.change(getAmountInput(), { target: { value: "100" } }); + }); + + await act(async () => { + fireEvent.click(screen.getByRole("button", { name: /start streaming/i })); + }); + + expect( + await screen.findByText(/invalid stellar public key/i), + ).toBeInTheDocument(); + expect(createStream).not.toHaveBeenCalled(); + }); + + it("calls createStream for a valid address", async () => { + (createStream as ReturnType).mockResolvedValue({ + success: true, + txHash: "tx_hash_valid", + }); + + render(); + + await act(async () => { + fireEvent.change(getRecipientInput(), { + target: { value: "GAV4A377RAEV6YVAWZVHXF4VZD5ZBXGIKEMNHV5YIMV5LIKSNQVYUBR7" }, + }); + fireEvent.change(getAmountInput(), { target: { value: "100" } }); + }); + + await act(async () => { + fireEvent.click(screen.getByRole("button", { name: /start streaming/i })); + }); + + expect(createStream).toHaveBeenCalledTimes(1); + }); +}); diff --git a/frontend/src/components/stream-creation/StreamCreationWizard.tsx b/frontend/src/components/stream-creation/StreamCreationWizard.tsx index 10f2c7da..f6b259cf 100644 --- a/frontend/src/components/stream-creation/StreamCreationWizard.tsx +++ b/frontend/src/components/stream-creation/StreamCreationWizard.tsx @@ -12,7 +12,7 @@ import { TemplateStep } from "./TemplateStep"; import { transactionSuccessToast } from "@/lib/transaction-feedback"; import { useRouter } from "next/navigation"; import { getApiBaseUrl } from "@/lib/api/_shared"; -import { MultisigRequiredError } from "@/lib/soroban"; +import { MultisigRequiredError, toBaseUnits } from "@/lib/soroban"; import type { MultisigRoutingResult } from "@/lib/stellar-multisig"; import { MultisigSignModal } from "../wallet/MultisigSignModal"; import { @@ -145,24 +145,50 @@ export const StreamCreationWizard: React.FC = ({ } }; - const startPolling = async (senderAddress: string) => { + const startPolling = async (senderAddress: string, form: StreamFormData) => { const startTime = Date.now(); const TIMEOUT_MS = 30000; // 30 seconds const POLL_INTERVAL = 2000; // 2 seconds const baseUrl = getApiBaseUrl(); + // Identify the stream the user just created instead of assuming the API's + // first result is it (issue #1204). We match on the submitted recipient and + // deposit amount, then prefer the most recently created match. + const expectedRecipient = form.recipient.trim().toLowerCase(); + const expectedDeposit = toBaseUnits(form.amount); + while (Date.now() - startTime < TIMEOUT_MS) { try { const response = await fetch(`${baseUrl}/v1/streams?sender=${senderAddress}`); const payload = await response.json(); - const streams = Array.isArray(payload) ? payload : (payload.data ?? []); - - // Assuming the latest stream is what we want - if (streams && streams.length > 0) { - // Found! - const newStream = streams[0]; // Simplification + const streams: Array> = Array.isArray(payload) + ? payload + : (payload.data ?? []); + + const candidates = streams.filter((stream) => { + if (typeof stream.recipient !== "string") return false; + if (stream.recipient.toLowerCase() !== expectedRecipient) return false; + // depositedAmount is stored in base units. Streams surfaced from the + // live RPC fallback may omit it, in which case recipient + recency + // decides the match. + if (typeof stream.depositedAmount === "string") { + try { + return BigInt(stream.depositedAmount) === expectedDeposit; + } catch { + return false; + } + } + return true; + }); + + if (candidates.length > 0) { + const newStream = candidates.reduce((latest, stream) => { + const latestTime = Date.parse(String(latest.createdAt)) || 0; + const streamTime = Date.parse(String(stream.createdAt)) || 0; + return streamTime > latestTime ? stream : latest; + }); transactionSuccessToast("Stream indexed and confirmed!"); - router.push(`/streams/${newStream.streamId}`); // Updated path to match new structure + router.push(`/streams/${String(newStream.streamId)}`); return; } } catch (e) { @@ -194,7 +220,7 @@ export const StreamCreationWizard: React.FC = ({ // Step 2: Start Polling for Indexer setIsPolling(true); - await startPolling(walletPublicKey || ""); + await startPolling(walletPublicKey || "", formData); } catch (error) { // Multisig accounts: instead of surfacing a raw broadcast failure, @@ -474,7 +500,7 @@ export const StreamCreationWizard: React.FC = ({ setMultisigProposal(null); setTxHash(hash); setIsPolling(true); - void startPolling(walletPublicKey || ""); + void startPolling(walletPublicKey || "", formData); }} /> )} diff --git a/frontend/src/components/stream-creation/__tests__/StreamCreationWizard.test.tsx b/frontend/src/components/stream-creation/__tests__/StreamCreationWizard.test.tsx index d17b7ddd..20faf1f0 100644 --- a/frontend/src/components/stream-creation/__tests__/StreamCreationWizard.test.tsx +++ b/frontend/src/components/stream-creation/__tests__/StreamCreationWizard.test.tsx @@ -29,6 +29,9 @@ vi.mock("@/lib/soroban", () => { return { fetchTokenBalanceDisplay: vi.fn().mockResolvedValue("1000"), MultisigRequiredError, + toBaseUnits: vi.fn((value: string) => + BigInt(Math.round(parseFloat(value) * 10 ** 7)), + ), }; }); @@ -193,6 +196,23 @@ import { fetchTokenBalanceDisplay } from "@/lib/soroban"; // Valid Stellar Ed25519 public key (correct StrKey checksum) const VALID_KEY = "GAV4A377RAEV6YVAWZVHXF4VZD5ZBXGIKEMNHV5YIMV5LIKSNQVYUBR7"; +// A different recipient, used to make sure polling does NOT pick it. +const OTHER_KEY = "GBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"; + +// toBaseUnits("5000") with the default 7 decimals — matches the wizard's +// default form amount, which is what the polled streams must match on. +const EXPECTED_DEPOSIT = "50000000000"; + +function indexedStream(overrides: Record = {}) { + return { + streamId: "123", + recipient: VALID_KEY, + depositedAmount: EXPECTED_DEPOSIT, + createdAt: new Date().toISOString(), + ...overrides, + }; +} + function renderWizard(overrides: Partial> = {}) { const onClose = vi.fn(); const onSubmit = vi.fn().mockResolvedValue({ txHash: "abc123hash" }); @@ -362,7 +382,7 @@ describe("StreamCreationWizard", () => { advanceToStep5(); vi.stubGlobal("fetch", vi.fn().mockResolvedValue( - new Response(JSON.stringify({ data: [{ streamId: "123" }] })) + new Response(JSON.stringify({ data: [indexedStream()] })) )); await act(async () => { @@ -385,7 +405,7 @@ describe("StreamCreationWizard", () => { advanceToStep5(); vi.stubGlobal("fetch", vi.fn().mockResolvedValue( - new Response(JSON.stringify({ data: [{ streamId: "123" }] })) + new Response(JSON.stringify({ data: [indexedStream()] })) )); await act(async () => { @@ -402,7 +422,7 @@ describe("StreamCreationWizard", () => { advanceToStep5(); vi.stubGlobal("fetch", vi.fn().mockResolvedValue( - new Response(JSON.stringify({ data: [{ streamId: "123" }] })) + new Response(JSON.stringify({ data: [indexedStream()] })) )); await act(async () => { @@ -420,7 +440,7 @@ describe("StreamCreationWizard", () => { advanceToStep5(); vi.stubGlobal("fetch", vi.fn().mockResolvedValue( - new Response(JSON.stringify({ data: [{ streamId: "123" }] })) + new Response(JSON.stringify({ data: [indexedStream()] })) )); await act(async () => { @@ -437,7 +457,7 @@ describe("StreamCreationWizard", () => { advanceToStep5(); vi.stubGlobal("fetch", vi.fn().mockResolvedValue( - new Response(JSON.stringify({ data: [{ streamId: "123" }] })) + new Response(JSON.stringify({ data: [indexedStream()] })) )); await act(async () => { @@ -454,7 +474,7 @@ describe("StreamCreationWizard", () => { advanceToStep5(); vi.stubGlobal("fetch", vi.fn().mockResolvedValue( - new Response(JSON.stringify([{ streamId: "789" }])) + new Response(JSON.stringify([indexedStream({ streamId: "789" })])) )); await act(async () => { @@ -471,7 +491,7 @@ describe("StreamCreationWizard", () => { advanceToStep5(); vi.stubGlobal("fetch", vi.fn().mockResolvedValue( - new Response(JSON.stringify({ data: [{ streamId: "999" }] })) + new Response(JSON.stringify({ data: [indexedStream({ streamId: "999" })] })) )); await act(async () => { @@ -483,6 +503,78 @@ describe("StreamCreationWizard", () => { }); }); + // ── Correct stream selection (issue #1204) ───────────────────────────────── + + it("does not redirect to an unrelated stream that appears first", async () => { + const { push } = renderWizard(); + advanceToStep5(); + + vi.stubGlobal("fetch", vi.fn().mockResolvedValue( + new Response(JSON.stringify({ + data: [ + // A newer stream for the same sender but a different recipient — + // the old streams[0] heuristic would have redirected here. + indexedStream({ streamId: "unrelated", recipient: OTHER_KEY }), + indexedStream({ streamId: "correct" }), + ], + })) + )); + + await act(async () => { + clickCreate(); + }); + + await waitFor(() => { + expect(push).toHaveBeenCalledWith("/streams/correct"); + }); + expect(push).not.toHaveBeenCalledWith("/streams/unrelated"); + }); + + it("ignores a stream whose deposit amount does not match the submitted form", async () => { + const { push } = renderWizard(); + advanceToStep5(); + + vi.stubGlobal("fetch", vi.fn().mockResolvedValue( + new Response(JSON.stringify({ + data: [ + indexedStream({ streamId: "wrong-amount", depositedAmount: "1" }), + indexedStream({ streamId: "matching" }), + ], + })) + )); + + await act(async () => { + clickCreate(); + }); + + await waitFor(() => { + expect(push).toHaveBeenCalledWith("/streams/matching"); + }); + expect(push).not.toHaveBeenCalledWith("/streams/wrong-amount"); + }); + + it("picks the most recently created matching stream", async () => { + const { push } = renderWizard(); + advanceToStep5(); + + vi.stubGlobal("fetch", vi.fn().mockResolvedValue( + new Response(JSON.stringify({ + data: [ + indexedStream({ streamId: "older", createdAt: "2023-01-01T00:00:00Z" }), + indexedStream({ streamId: "newer", createdAt: "2023-06-01T00:00:00Z" }), + ], + })) + )); + + await act(async () => { + clickCreate(); + }); + + await waitFor(() => { + expect(push).toHaveBeenCalledWith("/streams/newer"); + }); + }); + it("retries on fetch error during polling", async () => { vi.useFakeTimers(); const { push } = renderWizard(); @@ -490,7 +582,7 @@ describe("StreamCreationWizard", () => { const fetchMock = vi.fn() .mockRejectedValueOnce(new Error("network error")) - .mockResolvedValueOnce(new Response(JSON.stringify({ data: [{ streamId: "retry-ok" }] }))); + .mockResolvedValueOnce(new Response(JSON.stringify({ data: [indexedStream({ streamId: "retry-ok" })] }))); vi.stubGlobal("fetch", fetchMock); await act(async () => {