From 85a103a8a60b55c83da384e255f5beec62a10dba Mon Sep 17 00:00:00 2001 From: victor-134 Date: Fri, 24 Jul 2026 13:50:06 +0100 Subject: [PATCH] feat(backend): wire Soroban create_stream contract call with simulation, signing, and SQLite fallback --- backend/src/services/streamStore.test.ts | 109 ++++++++++++++++++++++- backend/src/services/streamStore.ts | 88 +++++++++++------- 2 files changed, 162 insertions(+), 35 deletions(-) diff --git a/backend/src/services/streamStore.test.ts b/backend/src/services/streamStore.test.ts index 29fba0a9..411c3160 100644 --- a/backend/src/services/streamStore.test.ts +++ b/backend/src/services/streamStore.test.ts @@ -113,13 +113,38 @@ vi.mock("@stellar/stellar-sdk", () => { }; } + if (operation.method === "create_stream") { + return { + kind: "success", + result: { retval: 101 }, + }; + } + throw new Error(`Unexpected contract method: ${operation.method}`); } + + async prepareTransaction(tx: any) { + return { + ...tx, + sign: vi.fn(), + }; + } + + async sendTransaction(_tx: any) { + return { status: "PENDING", hash: "mock-hash-123" }; + } + + async getTransaction(_hash: string) { + return { status: "SUCCESS", returnValue: 101 }; + } } return { Keypair: { - fromSecret: vi.fn(), + fromSecret: vi.fn(() => ({ + publicKey: () => "GB3K5Z74Z76QZ3ZZZ3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3", + sign: vi.fn(), + })), }, rpc: { Server: MockServer, @@ -130,7 +155,10 @@ vi.mock("@stellar/stellar-sdk", () => { Contract: MockContract, nativeToScVal: (value: any) => value, scValToNative: (value: any) => value, - Address: class MockAddress {}, + Address: class MockAddress { + constructor(private addr: string) {} + toScVal() { return this.addr; } + }, TimeoutInfinite: {}, TransactionBuilder: MockTransactionBuilder, Networks: { @@ -788,3 +816,80 @@ describe("metadata round-trip", () => { expect(parsed).toEqual({ purpose: "salary", project: "apollo" }); }); }); + +describe("createStream", () => { + beforeEach(() => { + vi.resetModules(); + vi.clearAllMocks(); + + mockState.nextId = 1; + mockState.existingStreamIds = new Set(); + mockState.chainStreams = new Map(); + mockState.upsertedStreams = []; + mockState.createdEventIds = new Set(); + + dbMocks.initDb.mockImplementation(() => undefined); + dbMocks.getDb.mockReturnValue(createDbMock()); + + delete process.env.SOROBAN_DISABLED; + delete process.env.SOROBAN_ENABLED; + process.env.CONTRACT_ID = "C1234567890ABCDEF1234567890ABCDEF1234567890ABCDEF1234567890"; + process.env.RPC_URL = "https://soroban-testnet.stellar.org:443"; + process.env.STELLAR_SECRET_KEY = "SDUMMYSECRETKEY12345678901234567890123456789012345678901"; + }); + + it("builds, simulates, signs, and submits Soroban tx when configured", async () => { + const { initSoroban, createStream } = await import("./streamStore"); + await initSoroban(); + + const stream = await createStream({ + sender: "GB3K5Z74Z76QZ3ZZZ3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3", + recipient: "GDESTINATION1234567890123456789012345678901234567890123", + assetCode: "USDC", + totalAmount: 500, + durationSeconds: 3600, + }); + + expect(stream.id).toBe("101"); + expect(stream.totalAmount).toBe(500); + expect(mockState.upsertedStreams).toHaveLength(1); + expect(mockState.upsertedStreams[0].id).toBe("101"); + }); + + it("uses SQLite fallback when SOROBAN_ENABLED=false or SOROBAN_DISABLED=true", async () => { + process.env.SOROBAN_ENABLED = "false"; + const { initSoroban, createStream } = await import("./streamStore"); + await initSoroban(); + + const dbMockWithMax = { + prepare(sql: string) { + if (sql.includes("SELECT MAX(CAST(id AS INTEGER))")) { + return { get: () => ({ maxId: 41 }) }; + } + if (sql.includes("INSERT INTO streams")) { + return { + run: (params: any) => { + mockState.upsertedStreams.push(params); + return { changes: 1 }; + }, + }; + } + throw new Error(`Unexpected SQL: ${sql}`); + }, + transaction any>(callback: T): T { + return ((...args: Parameters) => callback(...args)) as T; + }, + }; + dbMocks.getDb.mockReturnValue(dbMockWithMax); + + const stream = await createStream({ + sender: "GB3K5Z74Z76QZ3ZZZ3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3Z3", + recipient: "GDESTINATION1234567890123456789012345678901234567890123", + assetCode: "XLM", + totalAmount: 100, + durationSeconds: 1800, + }); + + expect(stream.id).toBe("42"); + }); +}); diff --git a/backend/src/services/streamStore.ts b/backend/src/services/streamStore.ts index ce2a29f2..bece6d93 100644 --- a/backend/src/services/streamStore.ts +++ b/backend/src/services/streamStore.ts @@ -192,11 +192,12 @@ export async function initSoroban() { process.env.RPC_URL || "https://soroban-testnet.stellar.org:443"; rpcServer = new rpc.Server(rpcUrl); - if (process.env.SERVER_PRIVATE_KEY) { - serverKeypair = Keypair.fromSecret(process.env.SERVER_PRIVATE_KEY); + const secretKey = process.env.STELLAR_SECRET_KEY || process.env.SERVER_PRIVATE_KEY; + if (secretKey) { + serverKeypair = Keypair.fromSecret(secretKey); } else { logger.warn( - "SERVER_PRIVATE_KEY missing. Creating streams on-chain will fail.", + "SERVER_PRIVATE_KEY / STELLAR_SECRET_KEY missing. Creating streams on-chain will fail.", ); } } @@ -812,50 +813,71 @@ export async function reconcileMissingStreams(): Promise { */ export async function createStream(input: StreamInput): Promise { const startAt = input.startAt ?? nowInSeconds(); + const sorobanDisabled = + process.env.SOROBAN_DISABLED?.toLowerCase() === "true" || + process.env.SOROBAN_ENABLED?.toLowerCase() === "false"; + const contractId = process.env.CONTRACT_ID; const netPass = process.env.NETWORK_PASSPHRASE || "Test SDF Network ; September 2015"; - if (!contractId || !rpcServer || !serverKeypair) { - throw new Error("Backend not configured for Soroban."); - } + let streamIdStr: string; - const sourceAccount = await rpcServer.getAccount(serverKeypair.publicKey()); - const tx = createStreamOperation(contractId, input, startAt); + if (sorobanDisabled || !contractId || !rpcServer || !serverKeypair) { + if (!sorobanDisabled && (!contractId || !rpcServer || !serverKeypair)) { + logger.warn( + "Soroban configuration incomplete or serverKeypair missing, falling back to local SQLite creation.", + ); + } + // Fallback SQLite-only path (e.g., local dev or SOROBAN_ENABLED=false) + const db = getDb(); + const row = db + .prepare("SELECT MAX(CAST(id AS INTEGER)) as maxId FROM streams") + .get() as { maxId: number | null } | undefined; + const nextNumericId = (row?.maxId ?? 0) + 1; + streamIdStr = nextNumericId.toString(); + } else { + // Soroban path: build, simulate, sign, and submit transaction + const sourceAccount = await rpcServer.getAccount(serverKeypair.publicKey()); + const op = createStreamOperation(contractId, input, startAt); - // We have to build and send this tx. Wait, doing this properly via building is long: - const built = await rpcServer.prepareTransaction( - new TransactionBuilder(sourceAccount, { + const txToSimulate = new TransactionBuilder(sourceAccount, { fee: "1000", networkPassphrase: netPass, }) - .addOperation(tx) + .addOperation(op) .setTimeout(30) - .build(), - ); + .build(); - built.sign(serverKeypair); + const simRes = await rpcServer.simulateTransaction(txToSimulate); + if (!rpc.Api.isSimulationSuccess(simRes)) { + throw new Error("Soroban RPC simulation failed: " + JSON.stringify(simRes)); + } - const sendRes = await retryWithBackoff(() => rpcServer!.sendTransaction(built)); - if (sendRes.status !== "PENDING") { - throw new Error("Failed to send transaction: " + JSON.stringify(sendRes)); - } + const preparedTx = await rpcServer.prepareTransaction(txToSimulate); + preparedTx.sign(serverKeypair); - 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++; - } + const sendRes = await retryWithBackoff(() => rpcServer!.sendTransaction(preparedTx)); + if (sendRes.status !== "PENDING") { + throw new Error("Failed to send transaction: " + JSON.stringify(sendRes)); + } - if (txResult?.status !== "SUCCESS" || !txResult.returnValue) { - throw new Error("Tx failed on chain: " + JSON.stringify(txResult)); - } + 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" || !txResult.returnValue) { + throw new Error("Tx failed on chain: " + JSON.stringify(txResult)); + } - const streamIdVal = scValToNative(txResult.returnValue); - const streamIdStr = streamIdVal.toString(); + const streamIdVal = scValToNative(txResult.returnValue); + streamIdStr = streamIdVal.toString(); + } const stream: StreamRecord = { id: streamIdStr, @@ -896,7 +918,7 @@ export async function createStream(input: StreamInput): Promise { resetStatsCache(); resetStreamMetricsCache(); - // Webhook fires after the transaction commits — a webhook failure + // Webhook fires after the transaction commits — a webhook failure // must never roll back an already-persisted stream. triggerWebhook("created", stream); return stream;