diff --git a/.github/workflows/testnet-e2e.yml b/.github/workflows/testnet-e2e.yml
new file mode 100644
index 00000000..29c022b2
--- /dev/null
+++ b/.github/workflows/testnet-e2e.yml
@@ -0,0 +1,208 @@
+# Scheduled end-to-end integration suite against the live Stellar Testnet.
+#
+# Mocked RPC tests in CI cannot detect testnet drift: protocol upgrades,
+# ledger close timing, RPC behaviour changes, fee surges, Friendbot outages,
+# or indexer ingestion regressions. This workflow:
+#   1. Builds the current stream_contract and deploys a fresh instance to
+#      testnet (or reuses a contract ID passed via workflow_dispatch).
+#   2. Runs backend/e2e/testnet against the live network and a real Postgres,
+#      exercising create_stream → withdraw → cancel_stream and verifying the
+#      SorobanEventWorker ingests those transactions.
+#   3. On scheduled failures, opens (or comments on) a tracking issue.
+name: Testnet E2E
+
+on:
+  schedule:
+    # Every 6 hours, offset from the hour to avoid peak Friendbot load.
+    - cron: "17 */6 * * *"
+  workflow_dispatch:
+    inputs:
+      contract_id:
+        description: "Existing testnet stream contract ID (C...). Leave empty to deploy a fresh one."
+        required: false
+        default: ""
+        type: string
+
+concurrency:
+  group: testnet-e2e
+  cancel-in-progress: false
+
+permissions:
+  contents: read
+
+env:
+  SOROBAN_RPC_URL: https://soroban-testnet.stellar.org
+  STELLAR_NETWORK_PASSPHRASE: "Test SDF Network ; September 2015"
+
+jobs:
+  deploy-contract:
+    name: Deploy stream_contract to testnet
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    outputs:
+      contract_id: ${{ steps.resolve.outputs.contract_id }}
+    steps:
+      - name: Checkout code
+        if: inputs.contract_id == ''
+        uses: actions/checkout@v4
+
+      - name: Setup Rust toolchain
+        if: inputs.contract_id == ''
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: stable
+          targets: wasm32-unknown-unknown
+
+      - name: Rust Cache
+        if: inputs.contract_id == ''
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: "contracts -> target"
+
+      - name: Install Stellar CLI
+        if: inputs.contract_id == ''
+        run: |
+          curl -fsSL https://github.com/stellar/stellar-cli/raw/main/install.sh | sh -s -- --install-deps
+          echo "$HOME/.stellar-cli/bin" >> "$GITHUB_PATH"
+
+      - name: Build & optimize WASM
+        if: inputs.contract_id == ''
+        working-directory: contracts
+        run: |
+          set -euo pipefail
+          cargo build --package stream_contract --target wasm32-unknown-unknown --release
+          stellar contract optimize \
+            --wasm target/wasm32-unknown-unknown/release/stream_contract.wasm \
+            --wasm-out target/wasm32-unknown-unknown/release/stream_contract.optimized.wasm
+
+      - name: Fund ephemeral deployer & deploy
+        id: deploy
+        if: inputs.contract_id == ''
+        run: |
+          set -euo pipefail
+          # Ephemeral identity funded by Friendbot — no long-lived secrets needed.
+          for i in 1 2 3 4 5; do
+            stellar keys generate e2e-deployer --network testnet --fund --overwrite && break
+            echo "Friendbot funding failed (attempt $i), retrying…"; sleep $((i * 5))
+          done
+          CONTRACT_ID=""
+          for i in 1 2 3; do
+            CONTRACT_ID=$(stellar contract deploy \
+              --wasm contracts/target/wasm32-unknown-unknown/release/stream_contract.optimized.wasm \
+              --source-account e2e-deployer \
+              --network testnet) && break
+            echo "Deploy failed (attempt $i), retrying…"; sleep $((i * 10))
+          done
+          if [[ ! "$CONTRACT_ID" =~ ^C[A-Z2-7]{55}$ ]]; then
+            echo "::error::Contract deployment did not return a valid contract ID: '$CONTRACT_ID'"
+            exit 1
+          fi
+          echo "contract_id=$CONTRACT_ID" >> "$GITHUB_OUTPUT"
+
+      - name: Resolve contract ID
+        id: resolve
+        run: |
+          set -euo pipefail
+          ID="${{ inputs.contract_id || steps.deploy.outputs.contract_id }}"
+          echo "contract_id=$ID" >> "$GITHUB_OUTPUT"
+          echo "### Testnet contract under test: \`$ID\`" >> "$GITHUB_STEP_SUMMARY"
+
+  e2e:
+    name: Testnet E2E suite
+    runs-on: ubuntu-latest
+    needs: deploy-contract
+    timeout-minutes: 30
+    services:
+      postgres:
+        image: postgres:16-alpine@sha256:e013e867e712fec275706a6c51c966f0bb0c93cfa8f51000f85a15f9865a28cb
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: password
+          POSTGRES_DB: flowfi_e2e
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      DATABASE_URL: postgresql://postgres:password@127.0.0.1:5432/flowfi_e2e
+      E2E_STREAM_CONTRACT_ID: ${{ needs.deploy-contract.outputs.contract_id }}
+      E2E_SOROBAN_RPC_URL: https://soroban-testnet.stellar.org
+      E2E_HORIZON_URL: https://horizon-testnet.stellar.org
+      E2E_FRIENDBOT_URL: https://friendbot.stellar.org
+      NODE_ENV: test
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+          cache: "npm"
+          cache-dependency-path: package-lock.json
+
+      - name: Install dependencies
+        run: npm ci --include=optional
+
+      - name: Setup Database
+        working-directory: backend
+        run: |
+          npx prisma generate --schema=prisma/schema.prisma
+          npx prisma db push --accept-data-loss --schema=prisma/schema.prisma
+
+      - name: Run Testnet E2E suite
+        working-directory: backend
+        run: npm run test:e2e:testnet
+
+      - name: Upload JUnit report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: testnet-e2e-results
+          path: backend/e2e-results/
+          if-no-files-found: ignore
+
+  report-failure:
+    name: Report scheduled failure
+    runs-on: ubuntu-latest
+    needs: [deploy-contract, e2e]
+    if: failure() && github.event_name == 'schedule'
+    permissions:
+      issues: write
+    steps:
+      - name: Open or update tracking issue
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const label = 'testnet-e2e-failure';
+            const runUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
+            const body = [
+              `Scheduled Stellar Testnet E2E run failed: ${runUrl}`,
+              '',
+              'Possible causes: testnet protocol upgrade, RPC regression, ledger timing drift,',
+              'fee surge, Friendbot outage, or indexer ingestion drift. See the job logs and',
+              'the `testnet-e2e-results` artifact for details.',
+            ].join('\n');
+
+            try {
+              await github.rest.issues.getLabel({ ...context.repo, name: label });
+            } catch {
+              await github.rest.issues.createLabel({ ...context.repo, name: label, color: 'd73a4a' });
+            }
+
+            const { data: open } = await github.rest.issues.listForRepo({
+              ...context.repo, state: 'open', labels: label, per_page: 1,
+            });
+            if (open.length > 0) {
+              await github.rest.issues.createComment({ ...context.repo, issue_number: open[0].number, body });
+            } else {
+              await github.rest.issues.create({
+                ...context.repo,
+                title: 'Stellar Testnet E2E suite failing',
+                labels: [label],
+                body,
+              });
+            }
diff --git a/.gitignore b/.gitignore
index c92a2f56..e30d3413 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,6 +4,7 @@ target
 .env
 .DS_Store
 coverage
+e2e-results
 *.log
 
 # Soroban test runner — auto-generated, never commit
diff --git a/backend/e2e/testnet/README.md b/backend/e2e/testnet/README.md
new file mode 100644
index 00000000..d7b05ac4
--- /dev/null
+++ b/backend/e2e/testnet/README.md
@@ -0,0 +1,50 @@
+# Stellar Testnet E2E suite
+
+Live end-to-end checks against the public Stellar Testnet. Unlike the mocked
+suites under `tests/`, these hit the real network to catch drift early:
+
+| Suite | Detects |
+| --- | --- |
+| `network-health.e2e.test.ts` | RPC outages, passphrase/protocol changes, ledger close-time drift, RPC↔Horizon skew, fee surges, Friendbot regressions |
+| `stream-lifecycle.e2e.test.ts` | Contract regressions on the live protocol (`create_stream` → `withdraw` → `cancel_stream`, `getEvents`) |
+| `stream-lifecycle.e2e.test.ts` › indexer | Indexer drift — the real `SorobanEventWorker` must ingest those txs into Postgres with matching hashes, ledgers and amounts |
+
+It is **not** part of `npm test`. It runs on a schedule (every 6h) via
+`.github/workflows/testnet-e2e.yml`, which deploys a fresh contract from the
+current `contracts/` source, runs the suite, and opens/updates a
+`testnet-e2e-failure` issue when a scheduled run fails. It can also be run
+manually from the Actions tab, optionally against an existing contract ID.
+
+## Running locally
+
+```bash
+cd backend
+# Network health only (no contract / DB needed):
+npm run test:e2e:testnet -- network-health
+
+# Full suite: deploy a contract, then point the suite at it and a Postgres DB
+export E2E_STREAM_CONTRACT_ID=C...
+export DATABASE_URL=postgresql://flowfi:flowfi_dev_password@127.0.0.1:5433/flowfi
+npx prisma db push --schema=prisma/schema.prisma
+npm run test:e2e:testnet
+```
+
+The indexer test **resets the `IndexerState` cursor** in the target database —
+never point `DATABASE_URL` at a shared or production database.
+
+Contract tests are skipped when `E2E_STREAM_CONTRACT_ID` is unset; indexer
+tests are skipped when `DATABASE_URL` is unset.
+
+## Configuration
+
+| Variable | Default |
+| --- | --- |
+| `E2E_SOROBAN_RPC_URL` | `https://soroban-testnet.stellar.org` |
+| `E2E_HORIZON_URL` | `https://horizon-testnet.stellar.org` |
+| `E2E_FRIENDBOT_URL` | `https://friendbot.stellar.org` |
+| `E2E_NETWORK_PASSPHRASE` | Testnet passphrase |
+| `E2E_STREAM_CONTRACT_ID` | _(unset → contract suites skipped)_ |
+| `E2E_MAX_LEDGER_CLOSE_SECONDS` | `15` |
+| `E2E_MAX_P90_INCLUSION_FEE` | `100000` stroops |
+| `E2E_TX_TIMEOUT_MS` | `90000` |
+| `E2E_INDEXER_TIMEOUT_MS` | `120000` |
diff --git a/backend/e2e/testnet/helpers/config.ts b/backend/e2e/testnet/helpers/config.ts
new file mode 100644
index 00000000..e5b13241
--- /dev/null
+++ b/backend/e2e/testnet/helpers/config.ts
@@ -0,0 +1,29 @@
+import { Networks } from '@stellar/stellar-sdk';
+
+/**
+ * Runtime configuration for the Stellar Testnet E2E suite.
+ * Every value can be overridden via env so the suite can also target a
+ * local `stellar/quickstart` container or a futurenet deployment.
+ */
+export const testnetConfig = {
+  rpcUrl: process.env.E2E_SOROBAN_RPC_URL ?? 'https://soroban-testnet.stellar.org',
+  horizonUrl: process.env.E2E_HORIZON_URL ?? 'https://horizon-testnet.stellar.org',
+  friendbotUrl: process.env.E2E_FRIENDBOT_URL ?? 'https://friendbot.stellar.org',
+  networkPassphrase: process.env.E2E_NETWORK_PASSPHRASE ?? Networks.TESTNET,
+  /** Deployed stream contract under test (C...). Required for contract/indexer suites. */
+  contractId: process.env.E2E_STREAM_CONTRACT_ID ?? '',
+  /** Upper bound on average ledger close time before we flag timing drift. */
+  maxLedgerCloseSeconds: Number(process.env.E2E_MAX_LEDGER_CLOSE_SECONDS ?? '15'),
+  /** Max time to wait for a submitted transaction to reach a final status. */
+  txTimeoutMs: Number(process.env.E2E_TX_TIMEOUT_MS ?? '90000'),
+  /** Max time to wait for the indexer to ingest on-chain events into Postgres. */
+  indexerTimeoutMs: Number(process.env.E2E_INDEXER_TIMEOUT_MS ?? '120000'),
+};
+
+export function hasContract(): boolean {
+  return testnetConfig.contractId.length > 0;
+}
+
+export function hasDatabase(): boolean {
+  return Boolean(process.env.DATABASE_URL);
+}
diff --git a/backend/e2e/testnet/helpers/soroban.ts b/backend/e2e/testnet/helpers/soroban.ts
new file mode 100644
index 00000000..ff6006d4
--- /dev/null
+++ b/backend/e2e/testnet/helpers/soroban.ts
@@ -0,0 +1,170 @@
+import {
+  Asset,
+  BASE_FEE,
+  Contract,
+  Keypair,
+  TransactionBuilder,
+  rpc,
+  scValToNative,
+  type xdr,
+} from '@stellar/stellar-sdk';
+import { testnetConfig } from './config.js';
+
+export const server = new rpc.Server(testnetConfig.rpcUrl, {
+  allowHttp: testnetConfig.rpcUrl.startsWith('http://'),
+});
+
+export const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+/** Retry `fn` with exponential backoff; used for flaky public endpoints. */
+export async function withRetry<T>(
+  label: string,
+  fn: () => Promise<T>,
+  { attempts = 5, baseMs = 1_000 } = {},
+): Promise<T> {
+  let lastErr: unknown;
+  for (let i = 0; i < attempts; i++) {
+    try {
+      return await fn();
+    } catch (err) {
+      lastErr = err;
+      if (i < attempts - 1) await sleep(baseMs * 2 ** i);
+    }
+  }
+  throw new Error(
+    `${label} failed after ${attempts} attempts: ${lastErr instanceof Error ? lastErr.message : String(lastErr)}`,
+  );
+}
+
+/** Poll `check` until it returns a non-undefined value or the timeout elapses. */
+export async function waitFor<T>(
+  label: string,
+  check: () => Promise<T | undefined>,
+  { timeoutMs, intervalMs = 2_000 }: { timeoutMs: number; intervalMs?: number },
+): Promise<T> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    const value = await check();
+    if (value !== undefined) return value;
+    await sleep(intervalMs);
+  }
+  throw new Error(`Timed out after ${timeoutMs}ms waiting for: ${label}`);
+}
+
+/** Fund a fresh account via Friendbot and return its keypair. */
+export async function createFundedAccount(): Promise<Keypair> {
+  const kp = Keypair.random();
+  await withRetry('Friendbot funding', async () => {
+    const res = await fetch(
+      `${testnetConfig.friendbotUrl}?addr=${encodeURIComponent(kp.publicKey())}`,
+    );
+    // Friendbot returns 400 "createAccountAlreadyExist" on a retry after a
+    // successful-but-timed-out first attempt; treat that as success.
+    if (!res.ok) {
+      const body = await res.text();
+      if (!body.includes('createAccountAlreadyExist')) {
+        throw new Error(`Friendbot HTTP ${res.status}: ${body.slice(0, 300)}`);
+      }
+    }
+  });
+  // Wait until the RPC node can see the new account (ledger ingestion lag).
+  await waitFor(
+    `account ${kp.publicKey()} visible on RPC`,
+    async () => {
+      try {
+        return await server.getAccount(kp.publicKey());
+      } catch {
+        return undefined;
+      }
+    },
+    { timeoutMs: 60_000 },
+  );
+  return kp;
+}
+
+/** Stellar Asset Contract address for native XLM on the configured network. */
+export function nativeTokenContractId(): string {
+  return Asset.native().contractId(testnetConfig.networkPassphrase);
+}
+
+export interface InvokeResult {
+  hash: string;
+  ledger: number;
+  returnValue: unknown;
+}
+
+/**
+ * Build, simulate, sign, submit and await a Soroban contract invocation.
+ * Throws with the full result XDR on simulation or execution failure so the
+ * CI log pinpoints protocol/RPC regressions.
+ */
+export async function invokeContract(
+  signer: Keypair,
+  contractId: string,
+  method: string,
+  args: xdr.ScVal[],
+): Promise<InvokeResult> {
+  const source = await server.getAccount(signer.publicKey());
+  const tx = new TransactionBuilder(source, {
+    fee: BASE_FEE,
+    networkPassphrase: testnetConfig.networkPassphrase,
+  })
+    .addOperation(new Contract(contractId).call(method, ...args))
+    .setTimeout(60)
+    .build();
+
+  // prepareTransaction simulates, applies the footprint/resource fee and auth.
+  const prepared = await server.prepareTransaction(tx);
+  prepared.sign(signer);
+
+  const sent = await server.sendTransaction(prepared);
+  if (sent.status === 'ERROR' || sent.status === 'TRY_AGAIN_LATER') {
+    throw new Error(
+      `${method}: sendTransaction returned ${sent.status}: ${sent.errorResult?.toXDR('base64') ?? 'no errorResult'}`,
+    );
+  }
+
+  const final = await waitFor(
+    `${method} tx ${sent.hash} to finalize`,
+    async () => {
+      const res = await server.getTransaction(sent.hash);
+      return res.status === rpc.Api.GetTransactionStatus.NOT_FOUND ? undefined : res;
+    },
+    { timeoutMs: testnetConfig.txTimeoutMs, intervalMs: 1_500 },
+  );
+
+  if (final.status !== rpc.Api.GetTransactionStatus.SUCCESS) {
+    throw new Error(
+      `${method}: transaction ${sent.hash} ${final.status}: ${final.resultXdr?.toXDR('base64') ?? ''}`,
+    );
+  }
+
+  return {
+    hash: sent.hash,
+    ledger: final.ledger,
+    returnValue: final.returnValue ? scValToNative(final.returnValue) : undefined,
+  };
+}
+
+/** Read-only contract call via simulation (no submission, no fees). */
+export async function simulateRead(
+  source: Keypair,
+  contractId: string,
+  method: string,
+  args: xdr.ScVal[],
+): Promise<unknown> {
+  const account = await server.getAccount(source.publicKey());
+  const tx = new TransactionBuilder(account, {
+    fee: BASE_FEE,
+    networkPassphrase: testnetConfig.networkPassphrase,
+  })
+    .addOperation(new Contract(contractId).call(method, ...args))
+    .setTimeout(30)
+    .build();
+  const sim = await server.simulateTransaction(tx);
+  if (rpc.Api.isSimulationError(sim)) {
+    throw new Error(`${method} simulation failed: ${sim.error}`);
+  }
+  const retval = sim.result?.retval;
+  return retval ? scValToNative(retval) : undefined;
+}
diff --git a/backend/e2e/testnet/network-health.e2e.test.ts b/backend/e2e/testnet/network-health.e2e.test.ts
new file mode 100644
index 00000000..2d8293cc
--- /dev/null
+++ b/backend/e2e/testnet/network-health.e2e.test.ts
@@ -0,0 +1,85 @@
+/**
+ * Live Stellar Testnet protocol-health checks.
+ *
+ * Catches drift that mocked RPC tests cannot: RPC availability, network
+ * passphrase / protocol version changes, ledger close timing, fee surges,
+ * and Friendbot funding regressions.
+ */
+import { describe, it, expect } from 'vitest';
+import { Horizon } from '@stellar/stellar-sdk';
+import { testnetConfig } from './helpers/config.js';
+import { createFundedAccount, server, sleep, withRetry } from './helpers/soroban.js';
+
+describe('Stellar Testnet: RPC & network health', () => {
+  it('RPC reports healthy', async () => {
+    const health = await withRetry('getHealth', () => server.getHealth());
+    expect(health.status).toBe('healthy');
+  });
+
+  it('network passphrase matches and protocol version is supported', async () => {
+    const network = await withRetry('getNetwork', () => server.getNetwork());
+    expect(network.passphrase).toBe(testnetConfig.networkPassphrase);
+    // Soroban requires protocol >= 20. Log the version so upgrades are
+    // visible in the scheduled-run history even when nothing breaks.
+    console.info(`[testnet-e2e] protocolVersion=${network.protocolVersion}`);
+    expect(network.protocolVersion).toBeGreaterThanOrEqual(20);
+  });
+
+  it('ledgers advance within the expected close-time budget', async () => {
+    const first = await withRetry('getLatestLedger', () => server.getLatestLedger());
+    const startedAt = Date.now();
+    const targetAdvance = 3;
+
+    let latest = first;
+    while (latest.sequence < first.sequence + targetAdvance) {
+      if (Date.now() - startedAt > testnetConfig.maxLedgerCloseSeconds * targetAdvance * 1_000) {
+        throw new Error(
+          `Ledger stalled: advanced ${latest.sequence - first.sequence}/${targetAdvance} ledgers in ${
+            (Date.now() - startedAt) / 1_000
+          }s (from ${first.sequence})`,
+        );
+      }
+      await sleep(1_000);
+      latest = await withRetry('getLatestLedger', () => server.getLatestLedger());
+    }
+
+    const avgClose = (Date.now() - startedAt) / 1_000 / (latest.sequence - first.sequence);
+    console.info(`[testnet-e2e] avgLedgerClose=${avgClose.toFixed(2)}s`);
+    expect(avgClose).toBeLessThanOrEqual(testnetConfig.maxLedgerCloseSeconds);
+  });
+
+  it('RPC and Horizon agree on the latest ledger (within tolerance)', async () => {
+    const horizon = new Horizon.Server(testnetConfig.horizonUrl);
+    const [rpcLedger, horizonLedgers] = await Promise.all([
+      withRetry('getLatestLedger', () => server.getLatestLedger()),
+      withRetry('horizon ledgers', () => horizon.ledgers().order('desc').limit(1).call()),
+    ]);
+    const horizonSeq = horizonLedgers.records[0]?.sequence ?? 0;
+    // Allow a few ledgers of ingestion skew between the two services.
+    expect(Math.abs(rpcLedger.sequence - horizonSeq)).toBeLessThanOrEqual(10);
+  });
+
+  it('fee stats are available and not in an extreme surge', async () => {
+    const stats = await withRetry('getFeeStats', () => server.getFeeStats());
+    const p90 = Number(stats.sorobanInclusionFee.p90);
+    console.info(
+      `[testnet-e2e] sorobanInclusionFee p50=${stats.sorobanInclusionFee.p50} p90=${p90} maxFee=${stats.sorobanInclusionFee.max}`,
+    );
+    expect(Number.isFinite(p90)).toBe(true);
+    // BASE_FEE-priced transactions (what the app submits) stop landing if
+    // inclusion fees surge by orders of magnitude; flag that early.
+    const surgeCeiling = Number(process.env.E2E_MAX_P90_INCLUSION_FEE ?? '100000');
+    expect(p90).toBeLessThanOrEqual(surgeCeiling);
+  });
+
+  it('Friendbot funds a fresh account that is visible on RPC', async () => {
+    const kp = await createFundedAccount();
+    const account = await server.getAccount(kp.publicKey());
+    expect(account.accountId()).toBe(kp.publicKey());
+
+    const horizon = new Horizon.Server(testnetConfig.horizonUrl);
+    const loaded = await withRetry('horizon loadAccount', () => horizon.loadAccount(kp.publicKey()));
+    const native = loaded.balances.find((b) => b.asset_type === 'native');
+    expect(Number(native?.balance ?? '0')).toBeGreaterThan(0);
+  });
+});
diff --git a/backend/e2e/testnet/stream-lifecycle.e2e.test.ts b/backend/e2e/testnet/stream-lifecycle.e2e.test.ts
new file mode 100644
index 00000000..431872c5
--- /dev/null
+++ b/backend/e2e/testnet/stream-lifecycle.e2e.test.ts
@@ -0,0 +1,225 @@
+/**
+ * Live Stellar Testnet end-to-end: stream contract lifecycle + indexer ingestion.
+ *
+ * 1. Executes create_stream → withdraw → cancel_stream against the deployed
+ *    contract (E2E_STREAM_CONTRACT_ID) using Friendbot-funded accounts and
+ *    the native XLM Stellar Asset Contract.
+ * 2. Runs the real SorobanEventWorker against the live RPC and a real
+ *    Postgres (DATABASE_URL) and asserts the on-chain transactions were
+ *    ingested with matching amounts, ledgers and tx hashes.
+ */
+import { describe, it, expect, beforeAll, afterAll } from 'vitest';
+import { Address, Keypair, nativeToScVal } from '@stellar/stellar-sdk';
+import { hasContract, hasDatabase, testnetConfig } from './helpers/config.js';
+import {
+  createFundedAccount,
+  invokeContract,
+  nativeTokenContractId,
+  server,
+  simulateRead,
+  waitFor,
+  type InvokeResult,
+} from './helpers/soroban.js';
+
+const DEPOSIT = 100_000_000n; // 10 XLM in stroops
+const DURATION_SECS = 3_600n;
+
+interface OnChainStream {
+  sender: string;
+  recipient: string;
+  token_address: string;
+  rate_per_second: bigint;
+  deposited_amount: bigint;
+  withdrawn_amount: bigint;
+  is_active: boolean;
+  status: unknown;
+}
+
+const u64 = (v: bigint | number) => nativeToScVal(v, { type: 'u64' });
+const addr = (pk: string) => new Address(pk).toScVal();
+
+/** Shared state across the ordered tests in this file. */
+const run: {
+  sender?: Keypair;
+  recipient?: Keypair;
+  token?: string;
+  startLedger?: number;
+  streamId?: bigint;
+  created?: InvokeResult;
+  withdrawn?: InvokeResult;
+  withdrawnAmount?: bigint;
+  cancelled?: InvokeResult;
+} = {};
+
+describe.skipIf(!hasContract())('Stellar Testnet: stream contract lifecycle', () => {
+  const contractId = testnetConfig.contractId;
+
+  beforeAll(async () => {
+    // Sequential to stay under Friendbot's per-IP rate limit.
+    run.sender = await createFundedAccount();
+    run.recipient = await createFundedAccount();
+    run.token = nativeTokenContractId();
+    run.startLedger = (await server.getLatestLedger()).sequence;
+  });
+
+  it('contract is deployed and reachable', async () => {
+    const res = await server.getContractWasmByContractId(contractId);
+    expect(res.length).toBeGreaterThan(0);
+  });
+
+  it('create_stream succeeds and returns a stream id', async () => {
+    const { sender, recipient, token } = run as Required<typeof run>;
+    run.created = await invokeContract(sender, contractId, 'create_stream', [
+      addr(sender.publicKey()),
+      addr(recipient.publicKey()),
+      addr(token),
+      nativeToScVal(DEPOSIT, { type: 'i128' }),
+      u64(DURATION_SECS),
+    ]);
+    expect(typeof run.created.returnValue).toBe('bigint');
+    run.streamId = run.created.returnValue as bigint;
+    console.info(
+      `[testnet-e2e] stream_created id=${run.streamId} tx=${run.created.hash} ledger=${run.created.ledger}`,
+    );
+  });
+
+  it('get_stream reflects the created stream', async () => {
+    const { sender, recipient, token, streamId } = run as Required<typeof run>;
+    const stream = (await simulateRead(sender, contractId, 'get_stream', [u64(streamId)])) as OnChainStream;
+    expect(stream.sender).toBe(sender.publicKey());
+    expect(stream.recipient).toBe(recipient.publicKey());
+    expect(stream.token_address).toBe(token);
+    expect(stream.is_active).toBe(true);
+    // deposited_amount is net of protocol fee (if a fee config is set).
+    expect(stream.deposited_amount).toBeGreaterThan(0n);
+    expect(stream.deposited_amount).toBeLessThanOrEqual(DEPOSIT);
+    expect(stream.rate_per_second).toBe(stream.deposited_amount / DURATION_SECS);
+  });
+
+  it('claimable amount accrues as ledgers close', async () => {
+    const { sender, streamId } = run as Required<typeof run>;
+    const claimable = await waitFor(
+      'claimable amount > 0',
+      async () => {
+        const v = (await simulateRead(sender, contractId, 'get_claimable_amount', [u64(streamId)])) as
+          | bigint
+          | undefined;
+        return v !== undefined && v > 0n ? v : undefined;
+      },
+      { timeoutMs: 60_000, intervalMs: 3_000 },
+    );
+    expect(claimable).toBeGreaterThan(0n);
+  });
+
+  it('recipient can withdraw accrued tokens', async () => {
+    const { recipient, streamId } = run as Required<typeof run>;
+    run.withdrawn = await invokeContract(recipient, contractId, 'withdraw', [
+      addr(recipient.publicKey()),
+      u64(streamId),
+    ]);
+    run.withdrawnAmount = run.withdrawn.returnValue as bigint;
+    expect(run.withdrawnAmount).toBeGreaterThan(0n);
+  });
+
+  it('sender can cancel the stream', async () => {
+    const { sender, streamId } = run as Required<typeof run>;
+    run.cancelled = await invokeContract(sender, contractId, 'cancel_stream', [
+      addr(sender.publicKey()),
+      u64(streamId),
+    ]);
+    const stream = (await simulateRead(sender, contractId, 'get_stream', [u64(streamId)])) as OnChainStream;
+    expect(stream.is_active).toBe(false);
+    expect(stream.status).toEqual(['Cancelled']);
+    expect(stream.withdrawn_amount).toBeGreaterThanOrEqual(run.withdrawnAmount!);
+  });
+
+  it('contract events are queryable via getEvents', async () => {
+    const { startLedger, created, withdrawn, cancelled } = run as Required<typeof run>;
+    const hashes = new Set([created.hash, withdrawn.hash, cancelled.hash]);
+    const found = await waitFor(
+      'contract events for lifecycle txs',
+      async () => {
+        const res = await server.getEvents({
+          startLedger,
+          filters: [{ type: 'contract', contractIds: [contractId] }],
+          limit: 200,
+        });
+        const mine = res.events.filter((e) => hashes.has(e.txHash));
+        return mine.length >= 3 ? mine : undefined;
+      },
+      { timeoutMs: 60_000, intervalMs: 3_000 },
+    );
+    const txHashes = new Set(found.map((e) => e.txHash));
+    expect(txHashes).toEqual(hashes);
+  });
+
+  // ─── Indexer drift ─────────────────────────────────────────────────────────
+
+  describe.skipIf(!hasDatabase())('indexer ingestion into Postgres', () => {
+    type Worker = import('../../src/workers/soroban-event-worker.js').SorobanEventWorker;
+    type Prisma = typeof import('../../src/lib/prisma.js').prisma;
+    let worker: Worker;
+    let prisma: Prisma;
+
+    beforeAll(async () => {
+      // The worker reads its config in the constructor, so set env before import.
+      process.env.STREAM_CONTRACT_ID = contractId;
+      process.env.SOROBAN_RPC_URL = testnetConfig.rpcUrl;
+      process.env.INDEXER_START_LEDGER = String(run.startLedger);
+      // Drive polls manually; keep the background timer out of the way.
+      process.env.INDEXER_POLL_INTERVAL_MS = String(60 * 60 * 1_000);
+
+      ({ prisma } = await import('../../src/lib/prisma.js'));
+      const { SorobanEventWorker } = await import('../../src/workers/soroban-event-worker.js');
+
+      // Fresh cursor so ingestion starts at this run's first ledger.
+      await prisma.indexerState.deleteMany({});
+      worker = new SorobanEventWorker();
+      await worker.start();
+    });
+
+    afterAll(async () => {
+      worker?.stop();
+      await worker?.waitForDrain();
+      await prisma?.$disconnect();
+    });
+
+    it('ingests the stream and its CREATED / WITHDRAWN / CANCELLED events', async () => {
+      const { streamId, created, withdrawn, cancelled, sender, recipient, token } =
+        run as Required<typeof run>;
+      expect(streamId, 'lifecycle tests must pass before indexer checks').toBeDefined();
+
+      const events = await waitFor(
+        `indexer to ingest stream ${streamId}`,
+        async () => {
+          await worker.triggerPoll();
+          const rows = await prisma.streamEvent.findMany({ where: { streamId } });
+          const types = new Set(rows.map((r) => r.eventType));
+          return ['CREATED', 'WITHDRAWN', 'CANCELLED'].every((t) => types.has(t)) ? rows : undefined;
+        },
+        { timeoutMs: testnetConfig.indexerTimeoutMs, intervalMs: 3_000 },
+      );
+
+      const byType = Object.fromEntries(events.map((e) => [e.eventType, e]));
+      expect(byType.CREATED!.transactionHash).toBe(created.hash);
+      expect(byType.CREATED!.ledgerSequence).toBe(created.ledger);
+      expect(byType.WITHDRAWN!.transactionHash).toBe(withdrawn.hash);
+      expect(byType.WITHDRAWN!.amount).toBe(run.withdrawnAmount!.toString());
+      expect(byType.CANCELLED!.transactionHash).toBe(cancelled.hash);
+
+      const stream = await prisma.stream.findUniqueOrThrow({ where: { streamId } });
+      expect(stream.sender).toBe(sender.publicKey());
+      expect(stream.recipient).toBe(recipient.publicKey());
+      expect(stream.tokenAddress).toBe(token);
+      expect(stream.isActive).toBe(false);
+      expect(BigInt(stream.withdrawnAmount)).toBeGreaterThanOrEqual(run.withdrawnAmount!);
+
+      // No events from this run should have been dead-lettered.
+      const deadLetters = await prisma.indexerDeadLetterEvent.findMany({
+        where: { transactionHash: { in: [created.hash, withdrawn.hash, cancelled.hash] } },
+      });
+      expect(deadLetters).toEqual([]);
+      expect(worker.getEventCounters().eventsFailed).toBe(0);
+    });
+  });
+});
diff --git a/backend/e2e/testnet/vitest.config.ts b/backend/e2e/testnet/vitest.config.ts
new file mode 100644
index 00000000..10b38cd1
--- /dev/null
+++ b/backend/e2e/testnet/vitest.config.ts
@@ -0,0 +1,30 @@
+import { defineConfig } from 'vitest/config';
+
+/**
+ * Vitest config for the live Stellar Testnet E2E suite.
+ *
+ * Kept separate from the default config so `npm test` never touches the
+ * network. Run with `npm run test:e2e:testnet` (see e2e/testnet/README.md).
+ */
+export default defineConfig({
+  test: {
+    environment: 'node',
+    globals: true,
+    root: new URL('../..', import.meta.url).pathname,
+    include: ['e2e/testnet/**/*.e2e.test.ts'],
+    env: {
+      JWT_SECRET: 'flowfi-testnet-e2e-secret-do-not-use-in-production',
+    },
+    // Live network: ledger close is ~5s and Friendbot/RPC can be slow.
+    testTimeout: 180_000,
+    hookTimeout: 240_000,
+    // Tests within a file share on-chain state (accounts, stream IDs) and
+    // must run in order; files run sequentially to avoid Friendbot rate limits.
+    sequence: { concurrent: false },
+    fileParallelism: false,
+    pool: 'forks',
+    coverage: { enabled: false },
+    reporters: ['default', 'junit'],
+    outputFile: { junit: './e2e-results/testnet-junit.xml' },
+  },
+});
diff --git a/backend/package.json b/backend/package.json
index 7e837a87..aa6334a4 100644
--- a/backend/package.json
+++ b/backend/package.json
@@ -11,6 +11,7 @@
     "test:unit": "vitest run --exclude='tests/integration/**'",
     "test:integration": "vitest run tests/integration",
     "test:integration:docker": "docker compose up -d postgres && vitest run tests/integration/stream-lifecycle.test.ts; docker compose stop postgres",
+    "test:e2e:testnet": "vitest run --config e2e/testnet/vitest.config.ts",
     "dev": "nodemon",
     "build": "tsc",
     "start": "node dist/index.js",