Skip to content

feat(backend): wire Soroban create_stream contract call with simulati… #650

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ritik4ever merged 2 commits into from
Jul 31, 2026
Merged

feat(backend): wire Soroban create_stream contract call with simulati… #650

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
85a103a
feat(backend): wire Soroban create_stream contract call with simulati…
Jul 24, 2026
4bcd872
Merge branch 'main' into feat/wire-create-stream-soroban
fredericklamar342-prog Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 99 additions & 3 deletions backend/src/services/streamStore.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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: {
Expand Down Expand Up @@ -789,6 +817,10 @@ describe("metadata round-trip", () => {
});
});

describe("createStream", () => {
beforeEach(() => {
vi.resetModules();
vi.clearAllMocks();
describe("getStreamById", () => {
const frozenTime = Math.floor(Date.now() / 1000);

Expand All @@ -804,6 +836,70 @@ describe("getStreamById", () => {
mockState.createdEventIds = new Set<string>();

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<T extends (...args: any[]) => any>(callback: T): T {
return ((...args: Parameters<T>) => 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");
});
});
});

afterEach(() => {
Expand Down Expand Up @@ -925,4 +1021,4 @@ describe("getStreamById", () => {
expect(result?.archived_at).not.toBeNull();
expect(result?.status).toBe("completed");
});
});
});
85 changes: 56 additions & 29 deletions backend/src/services/streamStore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,11 +194,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.",
);
}
}
Expand Down Expand Up @@ -779,49 +780,73 @@ export async function reconcileMissingStreams(): Promise<number> {

export async function createStream(input: StreamInput): Promise<StreamRecord> {
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();
Comment on lines +793 to +805

@coderabbitai coderabbitai Bot Jul 24, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift

SQLite fallback IDs and Soroban on-chain IDs are independent namespaces — collisions will silently overwrite unrelated streams.

streamIdStr in the fallback branch is MAX(CAST(id AS INTEGER)) + 1 computed purely from the local streams table, while the Soroban branch derives streamIdStr from the contract's own on-chain counter (scValToNative(txResult.returnValue), seen as a sequential integer in the mock server's get_next_stream_id). These two counters are not coordinated. Any time the fallback path is used (RPC outage, SOROBAN_ENABLED=false toggled temporarily, etc.) and later the Soroban path resumes, the on-chain contract can hand back an id that already exists locally (or vice versa). Since upsertStream does INSERT ... ON CONFLICT(id) DO UPDATE, a colliding id doesn't error — it silently overwrites a different, unrelated stream's row with the new stream's data.

Additionally, even within the fallback path alone, SELECT MAX(...) and the later INSERT are two separate statements not wrapped in one transaction; across multiple app instances/processes sharing the same SQLite file this is a TOCTOU race that can also produce duplicate ids.

Consider allocating stream ids from a single authoritative source regardless of path (e.g., a dedicated counter table incremented atomically, or a UUID), and/or storing the on-chain id in a separate column from the local primary key so the two are never conflated.

Also applies to: 839-880

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/src/services/streamStore.ts` around lines 826 - 838, Replace the
numeric streamIdStr allocation in the fallback branch and the Soroban-derived ID
path with a single authoritative local identifier strategy used by both paths,
such as an atomically incremented counter or UUID. Ensure allocation and
persistence are transaction-safe across processes, and update upsertStream and
related lookups so on-chain IDs are stored separately from the local primary key
rather than being conflated.

} else {
// Soroban path: build, simulate, sign, and submit transaction
const sourceAccount = await rpcServer.getAccount(serverKeypair.publicKey());
const op = createStreamOperation(contractId, input, startAt);

const txToSimulate = new TransactionBuilder(sourceAccount, {
const built = await rpcServer.prepareTransaction(
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));
}
Comment on lines +843 to +845

@coderabbitai coderabbitai Bot Jul 24, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Falsy check rejects a valid returnValue of 0.

!txResult.returnValue treats a legitimate stream id of 0 as a failure, throwing "Tx failed on chain" even though the transaction succeeded. Use an explicit nullish check instead.

🐛 Proposed fix 
-    if (txResult?.status !== "SUCCESS" || !txResult.returnValue) {
+    if (txResult?.status !== "SUCCESS" || txResult.returnValue === undefined || txResult.returnValue === null) {
       throw new Error("Tx failed on chain: " + JSON.stringify(txResult));
     }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
if (txResult?.status !== "SUCCESS" || !txResult.returnValue) {
throw new Error("Tx failed on chain: " + JSON.stringify(txResult));
}
if (txResult?.status !== "SUCCESS" || txResult.returnValue === undefined || txResult.returnValue === null) {
throw new Error("Tx failed on chain: " + JSON.stringify(txResult));
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/src/services/streamStore.ts` around lines 874 - 876, Update the
transaction validation condition around txResult in the stream-store flow to
reject only nullish return values, not valid falsy values such as 0. Preserve
the SUCCESS status check and existing error message for failed transactions.


const streamIdVal = scValToNative(txResult.returnValue);
const streamIdStr = streamIdVal.toString();
const streamIdVal = scValToNative(txResult.returnValue);
streamIdStr = streamIdVal.toString();
}

const stream: StreamRecord = {
id: streamIdStr,
Expand Down Expand Up @@ -860,6 +885,8 @@ export async function createStream(input: StreamInput): Promise<StreamRecord> {
resetStatsCache();
resetStreamMetricsCache();

// Webhook fires after the transaction commits — a webhook failure
// must never roll back an already-persisted stream.
triggerWebhook("created", stream);
return stream;
}
Expand Down