conduit-contracts

Soroban smart contracts powering the Conduit streaming payments protocol.

Three contracts. One protocol.

---

Component Status

This repository now spans more than the original three streaming contracts. The table below marks each component's intended track so newcomers don't have to infer it from source layout.

Component	Type	Status	Notes
DripStream	Contract	Production-track	Core protocol — per-stream escrow, see Contracts below
DripFactory	Contract	Production-track	Core protocol — deploys and registers streams
DripGovernor	Contract	Production-track	Core protocol — protocol configuration authority
BatchTransferProcessor	Contract	Production-track	Optional batch-transfer execution boundary — see Contracts below and ADR-007
TwapOracle	Contract	Production-track	Independent price-oracle service; not in the stream settlement path — see docs/architecture.md
TokenVault	Contract	Independent / experimental	Standalone token vault; not part of the streaming protocol — no protocol call path reaches it, see docs/architecture.md
indexer/	Service (TypeScript)	Scaffold	Off-chain event poller into Postgres; not wired to a live RPC endpoint yet — see indexer/README.md
frontend/	Application (TypeScript)	Scaffold	Components-only library, not a standalone app — no bundler config of its own, see frontend/README.md

Audit status: none of the above has undergone an external audit yet — see docs/security.md. Do not deploy any of them to Mainnet with real funds.

---

Contracts

DripStream

The core contract. One instance is deployed per payment stream. Holds the token balance and enforces the release schedule.

Storage:

Key	Type	Description
Sender	Address	Who created and funded the stream
Recipient	Address	Who receives the stream
Token	Address	Stellar asset contract address
RatePerSecond	i128	Tokens released per second (in stroops)
StartTime	u64	Unix timestamp — stream begins
EndTime	u64	Unix timestamp — stream ends (0 = open-ended)
Withdrawn	i128	Total withdrawn by recipient so far
Flags	u32	Bit-packed state flags (see below)
PausedAt	u64	Timestamp when stream was last paused

StreamInfo.flags bit layout:

Bit	Mask	Name	Meaning
0	0x01	FLAG_PAUSED	Stream is currently paused
1	0x02	FLAG_CLAWBACK_ENABLED	Sender can reclaim unstreamed tokens
2	0x04	FLAG_CANCELLED	Stream has been cancelled

Off-chain callers should use the is_*() getters on StreamInfo rather than reading the bit values directly:

info.is_paused()           // (info.flags & 0x01) != 0
info.is_clawback_enabled() // (info.flags & 0x02) != 0
info.is_cancelled()        // (info.flags & 0x04) != 0
``` |

**Public functions:**

```rust
fn withdraw(env: Env, amount: i128) -> Result<i128, Error>   // recipient-only

// caller must be the sender or the delegated operator, if any (see below)
fn cancel(env: Env, caller: Address) -> Result<(), Error>
fn pause(env: Env, caller: Address) -> Result<(), Error>
fn resume(env: Env, caller: Address) -> Result<(), Error>
fn top_up(env: Env, caller: Address, amount: i128) -> Result<(), Error>
fn clawback(env: Env, caller: Address) -> Result<i128, Error> // rejected while paused; resume() first

// Extend end_time by extra_time_seconds, pulling the exact rate-implied deposit from the sender
fn extend_duration(env: Env, caller: Address, extra_time_seconds: u64) -> Result<(), Error>

// Combines top_up(amount) + extend_duration(extra_time_seconds) in one call; neither works on an open-ended stream (end_time == 0)
fn top_up_and_extend(env: Env, caller: Address, amount: i128, extra_time_seconds: u64) -> Result<(), Error>

fn withdrawable(env: Env) -> i128
fn info(env: Env) -> StreamInfo
fn clawback_enabled(env: Env) -> bool

// Recipient-initiated escape hatch — see docs/architecture.md
fn force_cancel(env: Env) -> Result<(), Error>

// Recipient reassigns their claim to a new address; withdrawable balance carries over
fn transfer_recipient(env: Env, new_recipient: Address) -> Result<(), Error>

// Operator delegation — sender-only; see "Operator delegation" below
fn set_operator(env: Env, caller: Address, operator: Address) -> Result<(), Error>
fn revoke_operator(env: Env, caller: Address) -> Result<(), Error>
fn operator(env: Env) -> Option<Address>

// Read-only: total streamed so far, regardless of what's been withdrawn
fn streamed_total(env: Env) -> i128

// Read-only: latest committed event sequence, for detecting a gap after reconnecting
fn event_sequence(env: Env) -> u64

// Read-only: storage layout version this instance was initialized with
fn storage_version(env: Env) -> u32

Not yet in the SDK. force_cancel, transfer_recipient, streamed_total, extend_duration, top_up_and_extend, set_operator, revoke_operator, operator, event_sequence, and storage_version exist in the contract but aren't wrapped by conduit-sdk yet — callers need to invoke them directly until the SDK catches up.

Operator delegation:

A sender can delegate a subset of sender-level actions to another address via set_operator, without handing over sender itself. Only the sender may call set_operator/revoke_operator; the operator cannot re-delegate.

Action	Caller allowed
pause	sender or operator
resume	sender or operator
cancel	sender or operator
top_up	sender or operator (funds come from the caller)
extend_duration	sender or operator (funds come from the caller)
top_up_and_extend	sender or operator (funds come from the caller)
clawback	sender or operator
set_operator	sender only
revoke_operator	sender only

withdraw, force_cancel, and transfer_recipient are recipient-level actions and are never available to the operator. See docs/architecture.md for the full write-up.

Events emitted:

Event	Topics	Data
stream_withdrawn	[recipient]	{ amount, total_withdrawn, remaining }
stream_cancelled	[sender]	{ refund_amount, withdrawn_so_far }
stream_paused	[sender]	{ paused_at, withdrawable }
stream_resumed	[sender]	{ resumed_at }
stream_topped_up	[sender]	{ amount, new_balance }
stream_clawback	[sender]	{ amount }
xfer_rec	[old_recipient]	new_recipient

force_cancel reuses the stream_cancelled event — from the chain's perspective it settles the same way cancel() does.

Validation on initialize (called once at stream creation — by the factory, or by a direct deployer per ADR-001):

• Re-initialization rejected (AlreadyInitialized)
• rate_per_second > 0 — a zero/negative rate would create an "empty stream" that escrows tokens but never releases any; rejected with InvalidAmount

---

DripFactory

The singleton protocol entry point. Deploys new DripStream contracts, assigns them a monotonically incrementing stream_id, and maintains the global stream registry.

Public functions:

fn create_stream(
    env:           Env,
    sender:        Address,   // creator / funder — must require_auth
    recipient:     Address,
    token:         Address,
    deposit:       i128,
    rate_per_sec:  i128,
    start_time:    u64,
    end_time:      u64,
    clawback:      bool,
) -> Result<u64, Error>    // returns stream_id

fn stream_address(env: Env, stream_id: u64) -> Option<Address>
fn streams_by_sender(env: Env, sender: Address, offset: u32, limit: u32) -> StreamPage   // { ids: Vec<u64>, total: u32 } — limit is capped at 100; compare offset + ids.len() against total to detect truncation
fn streams_by_recipient(env: Env, recipient: Address, offset: u32, limit: u32) -> StreamPage
fn stream_count(env: Env) -> u64
fn protocol_fee_bps(env: Env) -> u32   // basis points, e.g. 30 = 0.3%; reads live from DripGovernor

// Governor-only: point future create_stream calls at a new DripStream WASM version.
// Existing streams are unaffected — each is an independently deployed contract.
fn upgrade_stream_wasm(env: Env, new_wasm_hash: BytesN<32>) -> Result<(), Error>

// Governor-only: emergency halt. While paused, create_stream reverts with
// ContractPaused before pulling any deposit. Already-deployed streams keep
// running — front-ends and the stream contract can gate withdrawals on is_paused.
fn pause(env: Env) -> Result<(), Error>
fn unpause(env: Env) -> Result<(), Error>
fn is_paused(env: Env) -> bool

Validation on create_stream:

All checks run before any state mutation (fail early — invalid calls neither touch storage nor extend TTL):

• Factory not under emergency pause (is_paused() == false, else ContractPaused)
• deposit > 0
• rate_per_sec > 0
• deposit >= rate_per_sec (must fund at least 1 second)
• end_time == 0 || end_time > start_time
• start_time >= env.ledger().timestamp() (no backdated streams)
• end_time == 0 || deposit >= rate_per_sec × (end_time - start_time) (must fund the entire declared duration)
• rate_per_sec <= DripGovernor::config().max_rate_per_second
• end_time == 0 || (end_time - start_time) >= DripGovernor::config().min_duration_seconds
• Token must be a valid Stellar asset contract

---

DripGovernor

Protocol configuration and upgrade authority. Holds mutable parameters that DripFactory reads at stream creation time. Access is governed by role-based access control (RBAC), so independent wallets can own distinct slices of protocol administration.

Configurable parameters:

Parameter	Default	Description
fee_bps	30	Protocol fee in basis points (30 = 0.3%)
fee_recipient	treasury	Address that receives protocol fees
min_duration_seconds	3600	Minimum stream duration (1 hour)
max_rate_per_second	10^15	Maximum rate cap
factory_address	set at init	The DripFactory this governor controls

Roles:

Role	Governs	Granted at init to
Admin	grant_role / revoke_role (including Admin itself)	the deploy authority
FeeManager	set_fee_bps, set_fee_recipient	the deploy authority
RateManager	set_max_rate, set_min_duration	the deploy authority

A role may be held by any number of accounts, and one account may hold any combination of roles. The deploy authority starts with all three, so it can bootstrap the protocol and then delegate fee and rate management to separate wallets. The final Admin cannot be revoked (LastAdmin), so governance can never be permanently frozen.

Public functions:

fn config(env: Env) -> GovernorConfig                 // read-only: full config struct
fn has_role(env: Env, role: Role, account: Address) -> bool

// Role administration — caller must hold Admin
fn grant_role(env: Env, caller: Address, role: Role, account: Address) -> Result<(), Error>
fn revoke_role(env: Env, caller: Address, role: Role, account: Address) -> Result<(), Error>  // LastAdmin if it drops the final Admin
fn transfer_authority(env: Env, caller: Address, new_authority: Address) -> Result<(), Error>  // grant Admin to new, revoke from caller

// Parameter setters — caller must hold the gating role; return InvalidParam on bad input
fn set_fee_bps(env: Env, caller: Address, fee_bps: u32) -> Result<(), Error>              // FeeManager;  0..=10_000
fn set_fee_recipient(env: Env, caller: Address, recipient: Address) -> Result<(), Error>  // FeeManager
fn set_min_duration(env: Env, caller: Address, seconds: u64) -> Result<(), Error>         // RateManager; > 0
fn set_max_rate(env: Env, caller: Address, max_rate: i128) -> Result<(), Error>           // RateManager; > 0

Each caller must require_auth() and hold the role gating the call, otherwise the call reverts with NotAuthorized.

---

BatchTransferProcessor

Optional batch-transfer execution boundary: pulls sum(amounts) from a single funder in one authorised transfer, then fans the funds out to up to 100 recipients. It is stateless — it stores no protocol state, holds no governor coupling, and charges no protocol fee (fee-exempt by design, see ADR-007).

Storage:

Key	Type	Description
(none)	—	No contract keys. The only ledger record touched is the contract's own instance entry, whose TTL process_batch renews (keep-alive) exactly like the other three contracts.

Public functions:

// Argument order matters: funder first, then token — both are bare Address,
// so a transposed pair compiles. It fails at runtime with InvalidToken
// (check 5 below), before require_auth and before any transfer.
fn process_batch(
    env:         Env,
    funder:      Address,        // pays; must require_auth
    token:       Address,        // SEP-41 asset the batch is denominated in
    recipients:  Vec<Address>,
    amounts:     Vec<i128>,
) -> Result<i128, Error>         // returns sum(amounts) transferred

// Read-only: the same checks 1–4 as process_batch, returning the total
// without auth, without a token, and without moving funds. Show users
// "this batch costs X" before they sign — never re-derive the total
// client-side, or a disagreement stays invisible until the tx fails.
fn preview_batch(
    env:         Env,
    recipients:  Vec<Address>,
    amounts:     Vec<i128>,
) -> Result<i128, Error>

fn max_batch_size(env: Env) -> u32   // 100 — the cap process_batch enforces
fn version(env: Env) -> u32          // behaviour version of this deployment (currently 2)

Validation order:

All checks run before funder.require_auth() and before any token movement; the first failure wins (so an oversized batch containing a zero amount reports BatchTooLarge, never InvalidAmount):

1. recipients.len() == amounts.len() — else LengthMismatch
2. amounts.len() <= max_batch_size() (100) — else BatchTooLarge
3. every amount > 0 — else InvalidAmount
4. sum(amounts) fits in i128 — else ArithmeticOverflow
5. token behaves like a SEP-41 asset — non-zero, not a G... wallet, and answering the balance probe — else InvalidToken

Checks 1–4 are shared verbatim with preview_batch; check 5 needs token and therefore only exists on process_batch. An empty batch short-circuits to Ok(0) after check 5, before auth.

Error codes (this contract only — see docs/contract-errors.md):

Code	Name	Fires when
1	LengthMismatch	recipients and amounts differ in length
2	BatchTooLarge	the batch exceeds max_batch_size() (100)
3	InvalidAmount	an individual amount is zero or negative
4	ArithmeticOverflow	the checked sum of the amounts overflows i128
5	InvalidToken	token is the all-zero address, a G... wallet, or a contract that does not implement SEP-41 — also what a transposed funder/token call returns

Events emitted:

Event	Topics	Data
batch_transferred	[funder]	{ token, recipient_count, total }

Emitted only after the fan-out succeeds — a reverted transfer rolls the whole call back, so the event never describes a batch that did not move funds. Validation failures and preview_batch emit nothing.

---

Error Codes

Each contract defines its own Error enum — the same numeric code means something different in each one (e.g. code 1 is NotAuthorized in DripStream and DripGovernor, but NotInitialized in DripFactory). Match errors against the enum for the contract you called, not by number alone.

A complete, authoritative list of every error variant across all contracts — with the numeric codes and a short description of when each fires — lives in docs/contract-errors.md.

---

Development

Requirements

Tool	Version
Rust	≥ 1.78
wasm32-unknown-unknown target	via rustup
Stellar CLI	≥ 20.0

Setup

git clone https://github.com/conduit-protocol/conduit-contracts
cd conduit-contracts

# Add WASM target
rustup target add wasm32-unknown-unknown

# Build
cargo build --target wasm32-unknown-unknown --release

# Test
cargo test --all

# Lint
cargo clippy --all-targets -- -D warnings
cargo fmt --all --check

Deploy to local network

# Start local Stellar node (Docker required)
stellar network start local

# Deploy all contracts
./scripts/deploy.sh local

# Output: contract IDs written to .contract-ids/local.json

Indexer

The indexer/ service tails Soroban getEvents and projects stream state into Postgres.

# Database — versioned migrations (replaces one-shot psql)
# Legacy: psql "$DATABASE_URL" -f db/schema.sql  (deprecated — no history)
DATABASE_URL=postgres://user:pass@localhost:5432/streamfi npm run --prefix indexer migrate
# Check pending/applied:  DATABASE_URL=... npm run --prefix indexer migrate -- status
# Alternative runner:      DATABASE_URL=... node db/migrate.js up

# Start the worker (exposes /healthz and /metrics)
PORT=3000 START_LEDGER=1 POLL_INTERVAL_MS=5000 npm run --prefix indexer start
# Health probe:   curl http://localhost:3000/healthz  # { lastSuccessfulPollTimestamp, currentCursor }
# Metrics:        curl http://localhost:3000/metrics   # Prometheus format: pages_processed, events_folded, fold_failures
# Readiness file alternative (if HEALTHZ_DISABLE=1):  HEALTHZ_FILE=/tmp/indexer.ready npm run --prefix indexer start

Deploy to testnet

# Set up a funded testnet identity
stellar keys generate dev --network testnet --fund

# Deploy
./scripts/deploy.sh testnet

---

Directory Structure

conduit-contracts/
├── Cargo.toml                  # workspace
├── contracts/
│   ├── stream/
│   │   ├── Cargo.toml
│   │   └── src/
│   │       ├── lib.rs          # contract entry points (thin — delegates to the modules below)
│   │       ├── state.rs        # load/save StreamInfo, cancelled-state guard
│   │       ├── storage.rs      # storage key definitions + StreamInfo struct
│   │       ├── errors.rs       # Error enum
│   │       ├── math.rs         # withdrawable calculation
│   │       ├── events.rs       # event helpers
│   │       └── ttl.rs          # instance TTL extension
│   ├── factory/
│   │   ├── Cargo.toml
│   │   └── src/
│   │       ├── lib.rs          # contract entry points (thin — delegates to the modules below)
│   │       ├── storage.rs      # DataKey enum
│   │       ├── errors.rs       # Error enum
│   │       ├── deploy.rs       # WASM hash + deploy logic
│   │       ├── governance.rs   # cross-contract calls into DripGovernor + bounds checks
│   │       ├── query.rs        # pagination helper for streams_by_sender/recipient
│   │       └── ttl.rs          # instance + persistent-entry TTL extension
│   └── governor/
│       ├── Cargo.toml
│       └── src/
│           ├── lib.rs          # contract entry points (thin — delegates to the modules below)
│           ├── storage.rs      # DataKey enum
│           ├── errors.rs       # Error enum
│           ├── config.rs       # GovernorConfig struct + load helper
│           ├── auth.rs         # authority-gate shared by every write
│           └── ttl.rs          # instance TTL extension
├── indexer/
│   ├── src/
│   │   ├── indexer/
│   │   │   ├── types.ts        # SorobanEventSource contract + pagination & fields docs
│   │   │   ├── poller.ts       # 105-line poller with counters (pages/events/failures)
│   │   │   ├── eventSource.ts  # stub SorobanEventSource (replace with RPC)
│   │   │   └── fold.ts         # per-event fold logic
│   │   ├── worker.ts           # bare process + /healthz + /metrics (+ readiness file alt)
│   │   ├── metrics.ts          # pages_processed / events_folded / fold_failures
│   │   ├── health.ts           # lastSuccessfulPoll + cursor for /healthz
│   │   └── db/
│   │       ├── migrate.ts      # hand-rolled numbered-file runner (up/status/down)
│   │       └── index.ts        # cursor load/save helpers
│   ├── package.json
│   └── tsconfig.json
├── db/
│   ├── schema.sql              # legacy one-shot (deprecated)
│   ├── migrate.js              # hand-rolled runner (node db/migrate.js up)
│   └── migrations/
│       └── 001_initial.sql     # converted schema.sql — first versioned migration
├── tests/
│   ├── stream_lifecycle.rs     # create → withdraw → cancel
│   ├── stream_clawback.rs
│   ├── stream_pause_resume.rs
│   ├── factory_deploy.rs
│   └── governor_config.rs
├── scripts/
│   ├── deploy.sh               # deploy to local / testnet / mainnet
│   ├── upgrade.sh              # upgrade factory/governor WASM
│   └── query.sh                # read stream state from CLI
└── docs/
    ├── architecture.md
    ├── contract-errors.md       # per-contract Error enums (single source of truth)
    ├── security.md              # threat model
    └── adr/                     # Architecture Decision Records

---

Off-chain Indexer

indexer/ holds a scaffold for polling contract events into Postgres (raw event log plus derived tables). It's not wired to a live RPC endpoint yet — see indexer/README.md for setup and known gaps (single-instance only, non-idempotent derived-table folds).

---

Security Considerations

• All auth checks use address.require_auth() — no manual signature verification.
• Arithmetic uses checked operations throughout; overflow returns Error::ArithmeticOverflow.
• The withdrawable() calculation is read-only and cannot modify state.
• Paused time does not count toward streamed balance (pause freezes the clock).
• Clawback can only be called by the sender and only if enabled at creation time.
• Re-entrancy is prevented by Soroban's execution model (no external calls mid-state-mutation).
• initialize() on all three contracts rejects a second call — a stream/factory/governor can't be re-initialized post-deployment to hijack its stored addresses.
• withdraw/top_up reject non-positive amounts.
• Every state-mutating call extends storage TTL (instance storage, plus the factory's BySender/ByRecipient/StreamAddr persistent entries) — see docs/security.md Known Limitation #1.

Audit status: Not yet audited. Do not use on Mainnet with real funds.

---

Contributing

See CONTRIBUTING.md. For contract-specific guidance, see docs/architecture.md.

---

License

MIT — see LICENSE.