From ab84d75f32bdfbb42f1ef6f00a65c77dd90c3a93 Mon Sep 17 00:00:00 2001 From: dev__abby Date: Tue, 28 Jul 2026 14:05:53 +0000 Subject: [PATCH 1/6] feat(backend): add transferStream service function and schema - Add transferStream() to streamStore: validates stream state, submits Soroban transfer_stream tx, updates recipient in SQLite, records 'transferred' event, triggers webhook - Add transferStreamSchema: validates sender and newRecipient as Stellar account IDs - Add 'transferred' to VALID_EVENT_TYPES in schemas - Fix duplicate elapsed/ratio declarations in calculateProgress - Fix indentation in cancelStream triggerWebhook call --- backend/src/services/streamStore.ts | 117 +++++++++++++++++++++++++++- backend/src/validation/schemas.ts | 7 +- 2 files changed, 120 insertions(+), 4 deletions(-) diff --git a/backend/src/services/streamStore.ts b/backend/src/services/streamStore.ts index ce2a29f2..cd840825 100644 --- a/backend/src/services/streamStore.ts +++ b/backend/src/services/streamStore.ts @@ -453,9 +453,6 @@ export function calculateProgress( const effectiveAt = stream.pausedAt !== undefined ? Math.min(at, stream.pausedAt) : at; - const elapsed = Math.max(0, Math.min(effectiveAt - stream.startAt - stream.pausedDuration, stream.durationSeconds)); - - const ratio = Math.min(1, elapsed / stream.durationSeconds); const elapsed = Math.max(0, Math.max(0, effectiveAt - stream.startAt) - stream.pausedDuration); const ratio = stream.durationSeconds <= 0 ? 1 : Math.min(1, elapsed / stream.durationSeconds); const elapsedSeconds = stream.durationSeconds <= 0 ? 0 : Math.min(elapsed, stream.durationSeconds); @@ -1197,6 +1194,120 @@ export async function cancelStream( return stream; } +/** + * Transfers a stream to a new recipient on-chain and updates the local DB. + * Submits a Soroban transfer_stream transaction, then updates the recipient + * field in SQLite and records a stream_transferred event. + * + * Only the sender may transfer. The stream must not be finalized + * (canceled/completed). + * + * @param id - Stream ID + * @param newRecipient - New Stellar recipient address + * @returns Promise resolving to the updated stream record + * @throws Error with statusCode if the stream cannot be transferred + */ +export async function transferStream( + id: string, + newRecipient: string, +): Promise { + const stream = getStream(id); + if (!stream) { + const err: any = new Error("Stream not found."); + err.statusCode = 404; + throw err; + } + + const status = computeStatus(stream, nowInSeconds()); + if (status === "canceled" || status === "completed") { + const err: any = new Error("Cannot transfer a finalized stream."); + err.statusCode = 400; + throw err; + } + + const oldRecipient = stream.recipient; + if (oldRecipient === newRecipient) { + const err: any = new Error("New recipient must differ from the current recipient."); + err.statusCode = 400; + throw err; + } + + // Submit on-chain transfer_stream transaction + try { + const sorobanContext = getSorobanContext(); + if (sorobanContext && rpcServer && serverKeypair) { + const contractId = process.env.CONTRACT_ID; + if (contractId) { + const sourceAccount = await rpcServer.getAccount(serverKeypair.publicKey()); + const contract = new Contract(contractId); + const tx = contract.call( + "transfer_stream", + nativeToScVal(parseInt(id), { type: "u64" }), + new Address(newRecipient).toScVal(), + ); + + const built = await rpcServer.prepareTransaction( + new TransactionBuilder(sourceAccount, { + fee: "1000", + networkPassphrase: process.env.NETWORK_PASSPHRASE || Networks.TESTNET, + }) + .addOperation(tx) + .setTimeout(30) + .build(), + ); + + built.sign(serverKeypair); + const sendRes = await retryWithBackoff(() => rpcServer!.sendTransaction(built)); + if (sendRes.status === "PENDING") { + let txResult; + let attempts = 0; + while (attempts < 10) { + txResult = await retryWithBackoff(() => + rpcServer!.getTransaction(sendRes.hash), + ); + if (txResult.status !== "NOT_FOUND") break; + await new Promise((r) => setTimeout(r, 1000)); + attempts++; + } + + if (txResult?.status !== "SUCCESS") { + throw new Error("On-chain transfer_stream transaction failed: " + JSON.stringify(txResult)); + } + } else { + throw new Error("Failed to send transfer_stream transaction: " + JSON.stringify(sendRes)); + } + } + } + } catch (err: any) { + if (err.statusCode) throw err; + logger.warn({ err, streamId: id }, "on-chain transfer_stream failed"); + throw new Error("On-chain transfer failed: " + (err.message || JSON.stringify(err))); + } + + const now = nowInSeconds(); + stream.recipient = newRecipient; + + // Invalidate cache + await invalidateCache(`stream:${id}`); + await invalidateCache("streams:list:"); + await invalidateCache("streams:export:"); + resetStatsCache(); + resetStreamMetricsCache(); + + // Atomically write the updated stream row and the transfer event. + const db = getDb(); + db.transaction(() => { + upsertStream(stream); + recordEventWithDb(db, stream.id, "transferred", now, stream.sender, undefined, { + oldRecipient, + newRecipient, + }); + })(); + + // Webhook fires after the transaction commits. + triggerWebhook("transferred", stream); + return stream; +} /** * Updates the start time of a scheduled stream. diff --git a/backend/src/validation/schemas.ts b/backend/src/validation/schemas.ts index cd03cffb..3ec4d845 100644 --- a/backend/src/validation/schemas.ts +++ b/backend/src/validation/schemas.ts @@ -128,7 +128,7 @@ export const updateStreamStartAtSchema = z.object({ } }); -const VALID_EVENT_TYPES = ["created", "claimed", "canceled", "start_time_updated", "paused", "resumed", "completed"] as const; +const VALID_EVENT_TYPES = ["created", "claimed", "canceled", "start_time_updated", "paused", "resumed", "completed", "transferred"] as const; export const webhookRegistrationSchema = z.object({ url: z @@ -217,6 +217,11 @@ export const bulkCancelStreamsSchema = z.object({ sender: stellarAccountIdSchema, }); +export const transferStreamSchema = z.object({ + sender: stellarAccountIdSchema, + newRecipient: stellarAccountIdSchema, +}); + export type CreateStreamPayload = z.infer; export type ValidationIssue = { From 5b4f30b5aa5733d3444dbbcd7660c897eaeb3533 Mon Sep 17 00:00:00 2001 From: dev__abby Date: Tue, 28 Jul 2026 14:06:03 +0000 Subject: [PATCH 2/6] feat(backend): add POST /api/streams/:id/transfer route and integration tests - Add transfer route: validates stream ID, sender auth, body schema, calls transferStream, returns updated stream - Add 8 integration tests covering: success, wrong sender, sender mismatch, same recipient, canceled stream, not found, invalid ID, no auth - Fix duplicate query/data declarations in recipient streams route handler --- backend/src/index.ts | 102 +++++++++++++------ backend/src/integration.test.ts | 167 ++++++++++++++++++++++++++++++++ 2 files changed, 238 insertions(+), 31 deletions(-) diff --git a/backend/src/index.ts b/backend/src/index.ts index a7d07158..2c0071b2 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -69,6 +69,7 @@ import { StreamRecord, StreamStatus, syncStreams, + transferStream, updateStreamStartAt, getOnChainStreamCount, } from "./services/streamStore"; @@ -89,6 +90,7 @@ import { recipientAccountIdSchema, senderAccountIdSchema, streamIdSchema, + transferStreamSchema, updateStreamStartAtSchema, } from "./validation/schemas"; import { validateEnv } from "./config/validateEnv"; @@ -1044,19 +1046,6 @@ app.get( progress: calculateProgress(stream, now), })); - const parsedQuery = listStreamsQuerySchema.safeParse(req.query); - if (!parsedQuery.success) { - sendValidationError(req, res, parsedQuery.error.issues); - return; - } - const query = parsedQuery.data; - - let data = listStreamsByRecipient(accountId) - .map((stream) => ({ - ...stream, - progress: calculateProgress(stream), - })); - if (query.status) { data = data.filter((stream) => stream.progress.status === query.status); } @@ -1084,24 +1073,8 @@ app.get( stream.recipient.toLowerCase().includes(searchTerm) || stream.assetCode.toLowerCase().includes(searchTerm) ); - } - if (query.asset) { - data = data.filter( - (stream) => - stream.assetCode.toLowerCase() === query.asset!.toLowerCase(), - ); - } - if (query.q && query.q.length > 0) { - const searchTerm = query.q.toLowerCase(); - data = data.filter((stream) => { - return ( - stream.id.toLowerCase().includes(searchTerm) || - stream.sender.toLowerCase().includes(searchTerm) || - stream.recipient.toLowerCase().includes(searchTerm) || - stream.assetCode.toLowerCase().includes(searchTerm) - ); - }); - } + }); + } const hasPage = req.query.page !== undefined; const hasLimit = req.query.limit !== undefined; @@ -1403,6 +1376,73 @@ app.post( }, ); +// POST /api/streams/:id/transfer — sender transfers stream to a new recipient +app.post( + "/api/streams/:id/transfer", + mutationLimiter, + authMiddleware, + async (req: Request, res: Response) => { + const parsedId = parseStreamId(req.params.id); + if (!parsedId.ok) { + sendValidationError(req, res, parsedId.issues); + return; + } + + const stream = getStream(parsedId.value); + if (!stream) { + sendApiError(req, res, 404, "Stream not found.", { code: "NOT_FOUND" }); + return; + } + + const user = (req as any).user; + if (stream.sender !== user.accountId) { + sendApiError(req, res, 403, "Only the sender can transfer this stream.", { + code: "FORBIDDEN", + }); + return; + } + + const parsedBody = transferStreamSchema.safeParse(req.body); + if (!parsedBody.success) { + sendValidationError(req, res, parsedBody.error.issues); + return; + } + + // Validate that the sender in the body matches the authenticated user + if (parsedBody.data.sender !== user.accountId) { + sendApiError(req, res, 403, "Sender in request body does not match authenticated user.", { + code: "FORBIDDEN", + }); + return; + } + + try { + const updated = await transferStream(parsedId.value, parsedBody.data.newRecipient); + res.json({ + data: { + ...updated, + progress: calculateProgress(updated), + }, + }); + } catch (error: any) { + logger.error({ err: error, streamId: parsedId.value }, "failed to transfer stream"); + const normalizedError = normalizeUnknownApiError( + error, + "Failed to transfer stream.", + ); + sendApiError( + req, + res, + normalizedError.statusCode, + normalizedError.message, + { + code: normalizedError.code ?? "INTERNAL_ERROR", + }, + ); + } + }, +); + // POST /api/streams/:id/pause — sender pauses an active stream app.post( "/api/streams/:id/pause", diff --git a/backend/src/integration.test.ts b/backend/src/integration.test.ts index b53a96a8..891c9aca 100644 --- a/backend/src/integration.test.ts +++ b/backend/src/integration.test.ts @@ -1425,6 +1425,173 @@ describe("Backend Integration Tests", () => { expect(response.body.code).toBe("RATE_LIMIT_EXCEEDED"); }); }); + + describe("POST /api/streams/:id/transfer", () => { + let senderKeypair: ReturnType; + let newRecipientKeypair: ReturnType; + let transferStreamId: string; + let senderToken: string; + let testCounter = 300; + + beforeEach(() => { + senderKeypair = Keypair.random(); + newRecipientKeypair = Keypair.random(); + const now = Math.floor(Date.now() / 1000); + transferStreamId = String(testCounter++); + + const db = getDb(); + db.prepare(` + INSERT INTO streams (id, sender, recipient, asset_code, total_amount, duration_seconds, start_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `).run( + transferStreamId, + senderKeypair.publicKey(), + Keypair.random().publicKey(), + "USDC", + 1000, + 3600, + now - 1800, + now - 1800, + ); + + senderToken = jwt.sign( + { accountId: senderKeypair.publicKey() }, + getJwtSecret(), + { expiresIn: "1h" }, + ); + }); + + it("should transfer an active stream to a new recipient", async () => { + const response = await request(app) + .post(`/api/streams/${transferStreamId}/transfer`) + .set("Authorization", `Bearer ${senderToken}`) + .send({ + sender: senderKeypair.publicKey(), + newRecipient: newRecipientKeypair.publicKey(), + }); + + expect(response.status).toBe(200); + expect(response.body.data.recipient).toBe(newRecipientKeypair.publicKey()); + expect(response.body.data.progress).toBeDefined(); + + // Verify SQLite was updated + const db = getDb(); + const updatedStream = db + .prepare(`SELECT * FROM streams WHERE id = ?`) + .get(transferStreamId) as any; + expect(updatedStream.recipient).toBe(newRecipientKeypair.publicKey()); + + // Verify transfer event recorded + const events = db + .prepare(`SELECT * FROM stream_events WHERE stream_id = ? AND event_type = 'transferred'`) + .all(transferStreamId) as any[]; + expect(events).toHaveLength(1); + expect(events[0].metadata).toContain(newRecipientKeypair.publicKey()); + }); + + it("should return 403 when a non-sender tries to transfer", async () => { + const nonSenderKeypair = Keypair.random(); + const nonSenderToken = jwt.sign( + { accountId: nonSenderKeypair.publicKey() }, + getJwtSecret(), + { expiresIn: "1h" }, + ); + + const response = await request(app) + .post(`/api/streams/${transferStreamId}/transfer`) + .set("Authorization", `Bearer ${nonSenderToken}`) + .send({ + sender: nonSenderKeypair.publicKey(), + newRecipient: newRecipientKeypair.publicKey(), + }); + + expect(response.status).toBe(403); + expect(response.body.code).toBe("FORBIDDEN"); + }); + + it("should return 403 when sender in body does not match authenticated user", async () => { + const otherKeypair = Keypair.random(); + + const response = await request(app) + .post(`/api/streams/${transferStreamId}/transfer`) + .set("Authorization", `Bearer ${senderToken}`) + .send({ + sender: otherKeypair.publicKey(), + newRecipient: newRecipientKeypair.publicKey(), + }); + + expect(response.status).toBe(403); + expect(response.body.code).toBe("FORBIDDEN"); + }); + + it("should return 400 when transferring to same recipient", async () => { + const db = getDb(); + const stream = db + .prepare(`SELECT recipient FROM streams WHERE id = ?`) + .get(transferStreamId) as any; + + const response = await request(app) + .post(`/api/streams/${transferStreamId}/transfer`) + .set("Authorization", `Bearer ${senderToken}`) + .send({ + sender: senderKeypair.publicKey(), + newRecipient: stream.recipient, + }); + + expect(response.status).toBe(400); + }); + + it("should return 400 for canceled stream", async () => { + const db = getDb(); + db.prepare(`UPDATE streams SET canceled_at = ? WHERE id = ?`) + .run(Math.floor(Date.now() / 1000), transferStreamId); + + const response = await request(app) + .post(`/api/streams/${transferStreamId}/transfer`) + .set("Authorization", `Bearer ${senderToken}`) + .send({ + sender: senderKeypair.publicKey(), + newRecipient: newRecipientKeypair.publicKey(), + }); + + expect(response.status).toBe(400); + }); + + it("should return 404 for non-existent stream", async () => { + const response = await request(app) + .post(`/api/streams/99999/transfer`) + .set("Authorization", `Bearer ${senderToken}`) + .send({ + sender: senderKeypair.publicKey(), + newRecipient: newRecipientKeypair.publicKey(), + }); + + expect(response.status).toBe(404); + }); + + it("should return 400 for invalid stream ID", async () => { + const response = await request(app) + .post(`/api/streams/invalid/transfer`) + .set("Authorization", `Bearer ${senderToken}`) + .send({ + sender: senderKeypair.publicKey(), + newRecipient: newRecipientKeypair.publicKey(), + }); + + expect(response.status).toBe(400); + }); + + it("should return 401 when no auth token is provided", async () => { + const response = await request(app) + .post(`/api/streams/${transferStreamId}/transfer`) + .send({ + sender: senderKeypair.publicKey(), + newRecipient: newRecipientKeypair.publicKey(), + }); + + expect(response.status).toBe(401); + }); + }); }); describe("Stream History", () => { From 365422c8c3d4bd9b524f5f9084a97934e8f0fae0 Mon Sep 17 00:00:00 2001 From: dev__abby Date: Tue, 28 Jul 2026 14:06:13 +0000 Subject: [PATCH 3/6] feat(frontend): add Transfer Stream button in StreamDetailDrawer - Add transferStream() API function with auth token support - Add 'transferred' to StreamEvent eventType union - Add Transfer Stream button in StreamDetailDrawer for sender with inline Stellar address input and validation - Add handleTransfer callback in DashboardPage - Add CSS styles for transfer input group with dark mode --- .../src/components/StreamDetailDrawer.tsx | 101 +++++++++++++++++- frontend/src/index.css | 48 +++++++++ frontend/src/pages/DashboardPage.tsx | 21 ++++ frontend/src/services/api.ts | 19 +++- 4 files changed, 187 insertions(+), 2 deletions(-) diff --git a/frontend/src/components/StreamDetailDrawer.tsx b/frontend/src/components/StreamDetailDrawer.tsx index f688b4d5..fc0e55b8 100644 --- a/frontend/src/components/StreamDetailDrawer.tsx +++ b/frontend/src/components/StreamDetailDrawer.tsx @@ -36,6 +36,8 @@ interface StreamDetailDrawerProps { onPause?: (streamId: string) => Promise; /** Called when resume action is triggered from the drawer */ onResume?: (streamId: string) => Promise; + /** Called when transfer action is triggered from the drawer */ + onTransfer?: (streamId: string, newRecipient: string) => Promise; /** * Sign an arbitrary action payload before mutating actions. * Receives { action, streamId, timestamp } and returns the signature. @@ -69,6 +71,7 @@ function eventIcon(type: StreamEvent["eventType"]): string { start_time_updated: "✎", paused: "⏸", resumed: "▶", + transferred: "⇄", }; return icons[type] ?? "•"; } @@ -81,6 +84,7 @@ function eventLabel(type: StreamEvent["eventType"]): string { start_time_updated: "Start time updated", paused: "Stream paused", resumed: "Stream resumed", + transferred: "Stream transferred", }; return labels[type] ?? type; } @@ -102,6 +106,7 @@ export function StreamDetailDrawer({ onCancel, onPause, onResume, + onTransfer, signAction, walletAddress, }: StreamDetailDrawerProps) { @@ -114,6 +119,10 @@ export function StreamDetailDrawer({ const [pausing, setPausing] = useState(false); const [resuming, setResuming] = useState(false); const [actionError, setActionError] = useState(null); + const [transferring, setTransferring] = useState(false); + const [showTransferInput, setShowTransferInput] = useState(false); + const [newRecipient, setNewRecipient] = useState(""); + const [transferError, setTransferError] = useState(null); // Abort controller to avoid race conditions on rapid open/close const abortRef = useRef(null); @@ -224,6 +233,31 @@ export function StreamDetailDrawer({ } } + async function handleTransfer() { + if (!stream || !onTransfer) return; + const recipient = newRecipient.trim(); + if (!recipient) { + setTransferError("Please enter a new recipient address."); + return; + } + if (!/^G[A-Z2-7]{55}$/.test(recipient)) { + setTransferError("Please enter a valid Stellar account ID (starts with G, 56 chars)."); + return; + } + setTransferring(true); + setTransferError(null); + try { + await onTransfer(stream.id, recipient); + setShowTransferInput(false); + setNewRecipient(""); + await fetchData(stream.id); + } catch (err) { + setTransferError(err instanceof Error ? err.message : "Transfer failed."); + } finally { + setTransferring(false); + } + } + const isFinalised = stream ? stream.progress.status === "completed" || stream.progress.status === "canceled" : false; @@ -246,7 +280,12 @@ export function StreamDetailDrawer({ !!signAction && stream?.progress.status === "paused"; - const hasActions = !!onCancel || showPause || showResume; + const showTransfer = + isSender && + !!onTransfer && + !isFinalised; + + const hasActions = !!onCancel || showPause || showResume || showTransfer; return (
{actionError}

)} + {transferError && ( +

{transferError}

+ )}
{/* Pause — active streams, sender only */} @@ -466,6 +508,63 @@ export function StreamDetailDrawer({ )} + {/* Transfer — active/scheduled/paused streams, sender only */} + {showTransfer && ( + <> + {!showTransferInput ? ( + + ) : ( +
+ { + setNewRecipient(e.target.value); + setTransferError(null); + }} + disabled={transferring} + aria-label="New recipient address" + autoFocus + /> +
+ + +
+
+ )} + + )} + {/* Cancel */} {onCancel && (