diff --git a/backend/src/services/indexer.ts b/backend/src/services/indexer.ts index 0e0914e..021fe13 100644 --- a/backend/src/services/indexer.ts +++ b/backend/src/services/indexer.ts @@ -202,6 +202,14 @@ async function indexEvents(): Promise { /** * Processes a single contract event and records it in history. * Note: This is now synchronous to support database transactions. + * + * All contract events share three base fields emitted by the contract: + * stream_id – identifies the stream + * actor – on-chain address that triggered the event + * timestamp – ledger close time (Unix seconds) from the contract + * + * We use event.ledgerClosedAt as the authoritative wall-clock timestamp for + * storage, and pass actor / amount fields as appropriate for each event type. */ function processEvent(db: any, event: rpc.Api.EventResponse): void { try { @@ -221,7 +229,8 @@ function processEvent(db: any, event: rpc.Api.EventResponse): void { value.stream_id.toString(), "created", timestamp, - value.sender, + // actor == sender for Created events + value.actor ?? value.sender, value.total_amount, { recipient: value.recipient, @@ -239,8 +248,22 @@ function processEvent(db: any, event: rpc.Api.EventResponse): void { value.stream_id.toString(), "claimed", timestamp, - value.recipient, + // actor == recipient for Claimed events + value.actor ?? value.recipient, value.amount, + { claimed_amount: value.claimed_amount }, + event.ledger, + ); + break; + + case "Completed": + recordEventWithDb( + db, + value.stream_id.toString(), + "completed", + timestamp, + value.actor, + value.total_amount, undefined, event.ledger, ); @@ -252,8 +275,9 @@ function processEvent(db: any, event: rpc.Api.EventResponse): void { value.stream_id.toString(), "canceled", timestamp, - value.sender, - undefined, + // actor == sender for Canceled events + value.actor ?? value.sender, + value.refunded_amount, undefined, event.ledger, ); @@ -265,9 +289,10 @@ function processEvent(db: any, event: rpc.Api.EventResponse): void { value.stream_id.toString(), "paused", timestamp, - value.sender, - undefined, + // actor == sender for Paused events + value.actor ?? value.sender, undefined, + { paused_at: value.paused_at }, event.ledger, ); break; @@ -278,9 +303,10 @@ function processEvent(db: any, event: rpc.Api.EventResponse): void { value.stream_id.toString(), "resumed", timestamp, - value.sender, - undefined, + // actor == sender for Resumed events + value.actor ?? value.sender, undefined, + { resumed_at: value.resumed_at }, event.ledger, ); break; @@ -291,12 +317,31 @@ function processEvent(db: any, event: rpc.Api.EventResponse): void { value.stream_id.toString(), "transferred", timestamp, - value.old_recipient, + // actor == old_recipient (the one who authorized the transfer) + value.actor ?? value.old_recipient, undefined, { new_recipient: value.new_recipient }, event.ledger, ); break; + + case "Clawback": + recordEventWithDb( + db, + value.stream_id.toString(), + "clawback", + timestamp, + // actor == admin address + value.actor, + value.amount, + { recipient: value.recipient }, + event.ledger, + ); + break; + + default: + logger.warn({ eventName }, "unknown contract event type — skipped"); + break; } } catch (err) { logger.error({ err }, "failed to process event"); diff --git a/contracts/src/lib.rs b/contracts/src/lib.rs index 8277761..77c35b7 100644 --- a/contracts/src/lib.rs +++ b/contracts/src/lib.rs @@ -46,12 +46,26 @@ pub enum DataKey { // --------------------------------------------------------------------------- // Events +// +// All events share three mandatory fields: +// stream_id – identifies the stream this event belongs to +// actor – the on-chain address that triggered the event +// timestamp – ledger close time (Unix seconds) at the moment of emission +// +// Additional fields carry event-specific data (amounts, addresses, etc.). // --------------------------------------------------------------------------- +/// Emitted once when a new stream is created via `create_stream` or as a +/// child record inside `create_split_stream`. #[contracttype] #[derive(Clone, Debug, PartialEq, Eq)] pub struct StreamCreated { + // --- mandatory base fields --- pub stream_id: u64, + /// The sender who funded and created the stream. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- pub sender: Address, pub recipient: Address, pub token: Address, @@ -63,51 +77,103 @@ pub struct StreamCreated { pub metadata: Option>, } +/// Emitted each time a recipient successfully claims vested tokens. #[contracttype] #[derive(Clone, Debug, PartialEq, Eq)] pub struct StreamClaimed { + // --- mandatory base fields --- pub stream_id: u64, + /// The recipient who performed the claim. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- pub recipient: Address, pub amount: i128, + /// Cumulative amount claimed after this operation. + pub claimed_amount: i128, +} + +/// Emitted when a stream is fully claimed (claimed_amount == total_amount). +/// Always follows a `StreamClaimed` event in the same transaction. +#[contracttype] +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct StreamCompleted { + // --- mandatory base fields --- + pub stream_id: u64, + /// The recipient whose final claim completed the stream. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- + pub total_amount: i128, } +/// Emitted when a sender cancels an active stream before it ends. #[contracttype] #[derive(Clone, Debug, PartialEq, Eq)] pub struct StreamCanceled { + // --- mandatory base fields --- pub stream_id: u64, + /// The sender who canceled the stream. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- pub sender: Address, + /// Amount refunded to the sender (unvested tokens). + pub refunded_amount: i128, } +/// Emitted when a sender pauses an active stream. #[contracttype] #[derive(Clone, Debug, PartialEq, Eq)] - pub struct StreamPaused { + // --- mandatory base fields --- pub stream_id: u64, + /// The sender who paused the stream. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- pub sender: Address, pub paused_at: u64, } +/// Emitted when a sender resumes a previously paused stream. #[contracttype] #[derive(Clone, Debug, PartialEq, Eq)] pub struct StreamResumed { + // --- mandatory base fields --- pub stream_id: u64, + /// The sender who resumed the stream. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- pub sender: Address, pub resumed_at: u64, - } +/// Emitted when an admin executes a clawback of unclaimed vested tokens. #[contracttype] #[derive(Clone, Debug, PartialEq, Eq)] pub struct ClawbackExecuted { + // --- mandatory base fields --- pub stream_id: u64, + /// The admin address that performed the clawback. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- pub amount: i128, pub recipient: Address, } +/// Emitted when the current recipient transfers their stream rights to a new address. #[contracttype] #[derive(Clone, Debug, PartialEq, Eq)] pub struct StreamTransferred { + // --- mandatory base fields --- pub stream_id: u64, + /// The previous recipient who authorized the transfer. + pub actor: Address, + pub timestamp: u64, + // --- event-specific fields --- pub old_recipient: Address, pub new_recipient: Address, } @@ -224,10 +290,13 @@ impl StellarStreamContract { .persistent() .set(&DataKey::Stream(next_id), &stream); + let now = env.ledger().timestamp(); env.events().publish( (symbol_short!("Stream"), symbol_short!("Created")), StreamCreated { stream_id: next_id, + actor: sender.clone(), + timestamp: now, sender, recipient, token: token.clone(), @@ -326,6 +395,8 @@ impl StellarStreamContract { (symbol_short!("Stream"), symbol_short!("Created")), StreamCreated { stream_id: child_stream_id, + actor: sender.clone(), + timestamp: env.ledger().timestamp(), sender: sender.clone(), recipient, token: token.clone(), @@ -448,11 +519,34 @@ impl StellarStreamContract { .persistent() .set(&DataKey::Stream(stream_id), &stream); + let now = env.ledger().timestamp(); + let new_claimed_total = stream.claimed_amount; + env.events().publish( (symbol_short!("Stream"), symbol_short!("Claimed")), - StreamClaimed { stream_id, recipient, amount }, + StreamClaimed { + stream_id, + actor: recipient.clone(), + timestamp: now, + recipient: recipient.clone(), + amount, + claimed_amount: new_claimed_total, + }, ); + // If the stream is now fully claimed, also emit StreamCompleted. + if stream.claimed_amount >= stream.total_amount { + env.events().publish( + (symbol_short!("Stream"), symbol_short!("Completed")), + StreamCompleted { + stream_id, + actor: recipient, + timestamp: now, + total_amount: stream.total_amount, + }, + ); + } + amount } @@ -498,7 +592,13 @@ impl StellarStreamContract { env.events().publish( (symbol_short!("Stream"), symbol_short!("Canceled")), - StreamCanceled { stream_id, sender }, + StreamCanceled { + stream_id, + actor: sender.clone(), + timestamp: now, + sender, + refunded_amount: sender_refund, + }, ); } @@ -513,10 +613,13 @@ impl StellarStreamContract { .persistent() .set(&DataKey::Stream(stream_id), &stream); + let now = env.ledger().timestamp(); env.events().publish( (symbol_short!("Stream"), symbol_short!("Transfer")), StreamTransferred { stream_id, + actor: old_recipient.clone(), + timestamp: now, old_recipient, new_recipient, }, @@ -548,6 +651,8 @@ impl StellarStreamContract { (symbol_short!("Stream"), symbol_short!("Paused")), StreamPaused { stream_id, + actor: sender.clone(), + timestamp: now, sender, paused_at: now, }, @@ -583,6 +688,8 @@ impl StellarStreamContract { (symbol_short!("Stream"), symbol_short!("Resumed")), StreamResumed { stream_id, + actor: sender.clone(), + timestamp: now, sender, resumed_at: now, }, @@ -639,6 +746,8 @@ impl StellarStreamContract { (symbol_short!("Stream"), symbol_short!("Clawback")), ClawbackExecuted { stream_id, + actor: admin.clone(), + timestamp: env.ledger().timestamp(), amount: actual_clawback, recipient: admin, }, diff --git a/docs/CONTRACT_EVENTS.md b/docs/CONTRACT_EVENTS.md new file mode 100644 index 0000000..84fc14a --- /dev/null +++ b/docs/CONTRACT_EVENTS.md @@ -0,0 +1,236 @@ +# StellarStream Contract Event Schema + +This document describes every event emitted by the StellarStream Soroban contract +(`contracts/src/lib.rs`). The indexer worker (`backend/src/services/indexer.ts`) +reads these events from Stellar RPC and writes them to the SQLite `stream_events` +table for the frontend to display. + +--- + +## Overview + +### Event Topics + +Every event is published with a two-symbol topic tuple: + +``` +(Symbol("Stream"), Symbol("")) +``` + +The indexer matches on `topic[1]` (the event name) to dispatch to the correct +handler. + +### Mandatory Base Fields + +**All** StellarStream events carry these three fields regardless of type: + +| Field | Type | Description | +|-------------|-----------|-------------------------------------------------------------------| +| `stream_id` | `u64` | Numeric ID of the stream this event belongs to. | +| `actor` | `Address` | On-chain address of the party who triggered the event. | +| `timestamp` | `u64` | Ledger close time in Unix seconds at the moment of emission. | + +Additional fields are event-specific and documented in each section below. + +--- + +## Events + +### StreamCreated + +**Topic:** `("Stream", "Created")` +**Triggered by:** `create_stream()`, `create_split_stream()` +**Actor:** The sender who funded the stream. + +| Field | Type | Description | +|-----------------|------------------------------|-----------------------------------------------------------| +| `stream_id` | `u64` | Unique stream identifier. | +| `actor` | `Address` | Sender address (same as `sender` field). | +| `timestamp` | `u64` | Ledger close time when the stream was created. | +| `sender` | `Address` | Account that funded the stream. | +| `recipient` | `Address` | Account entitled to claim tokens. | +| `token` | `Address` | Contract address of the streamed token. | +| `token_symbol` | `String` | Symbol of the token (e.g. `"USDC"`). | +| `total_amount` | `i128` | Total tokens locked in the stream (in stroops). | +| `start_time` | `u64` | Unix timestamp when vesting begins. | +| `end_time` | `u64` | Unix timestamp when vesting ends. | +| `cliff_seconds` | `u64` | Seconds after `start_time` before any tokens vest. | +| `metadata` | `Option>` | Optional key-value metadata attached to the stream. | + +**Indexer mapping:** event type `"created"`, amount = `total_amount`, metadata includes `recipient`, `token`, `startTime`, `endTime`. + +--- + +### StreamClaimed + +**Topic:** `("Stream", "Claimed")` +**Triggered by:** `claim()` +**Actor:** The recipient performing the claim. + +| Field | Type | Description | +|------------------|-----------|---------------------------------------------------------------| +| `stream_id` | `u64` | Stream identifier. | +| `actor` | `Address` | Recipient address (same as `recipient` field). | +| `timestamp` | `u64` | Ledger close time when the claim occurred. | +| `recipient` | `Address` | Account that received the tokens. | +| `amount` | `i128` | Tokens transferred in this claim (in stroops). | +| `claimed_amount` | `i128` | Cumulative total claimed after this operation (in stroops). | + +**Indexer mapping:** event type `"claimed"`, amount = `amount`, metadata includes `claimed_amount`. + +> **Note:** When a claim causes `claimed_amount >= total_amount`, a `StreamCompleted` +> event is also emitted in the same transaction immediately after `StreamClaimed`. + +--- + +### StreamCompleted + +**Topic:** `("Stream", "Completed")` +**Triggered by:** `claim()` — emitted only when the final claim fully drains the stream. +**Actor:** The recipient whose claim completed the stream. + +| Field | Type | Description | +|----------------|-----------|-----------------------------------------------------------| +| `stream_id` | `u64` | Stream identifier. | +| `actor` | `Address` | Recipient who made the completing claim. | +| `timestamp` | `u64` | Ledger close time when completion was reached. | +| `total_amount` | `i128` | Total amount that was streamed (in stroops). | + +**Indexer mapping:** event type `"completed"`, amount = `total_amount`. + +--- + +### StreamCanceled + +**Topic:** `("Stream", "Canceled")` +**Triggered by:** `cancel()` +**Actor:** The sender who canceled the stream. + +| Field | Type | Description | +|-------------------|-----------|--------------------------------------------------------------------| +| `stream_id` | `u64` | Stream identifier. | +| `actor` | `Address` | Sender address (same as `sender` field). | +| `timestamp` | `u64` | Ledger close time when the cancellation was recorded. | +| `sender` | `Address` | Account that canceled the stream. | +| `refunded_amount` | `i128` | Unvested tokens refunded to the sender (in stroops). May be `0`. | + +**Indexer mapping:** event type `"canceled"`, amount = `refunded_amount`. + +--- + +### StreamPaused + +**Topic:** `("Stream", "Paused")` +**Triggered by:** `pause_stream()` +**Actor:** The sender who paused the stream. + +| Field | Type | Description | +|-------------|-----------|--------------------------------------------------------------| +| `stream_id` | `u64` | Stream identifier. | +| `actor` | `Address` | Sender address (same as `sender` field). | +| `timestamp` | `u64` | Ledger close time when the pause was recorded. | +| `sender` | `Address` | Account that paused the stream. | +| `paused_at` | `u64` | Ledger close time at which vesting was frozen (Unix seconds).| + +**Indexer mapping:** event type `"paused"`, metadata includes `paused_at`. + +--- + +### StreamResumed + +**Topic:** `("Stream", "Resumed")` +**Triggered by:** `resume_stream()` +**Actor:** The sender who resumed the stream. + +| Field | Type | Description | +|--------------|-----------|----------------------------------------------------------------| +| `stream_id` | `u64` | Stream identifier. | +| `actor` | `Address` | Sender address (same as `sender` field). | +| `timestamp` | `u64` | Ledger close time when the resume was recorded. | +| `sender` | `Address` | Account that resumed the stream. | +| `resumed_at` | `u64` | Ledger close time at which vesting restarted (Unix seconds). | + +**Indexer mapping:** event type `"resumed"`, metadata includes `resumed_at`. + +--- + +### StreamTransferred *(bonus — emitted by `transfer_stream`)* + +**Topic:** `("Stream", "Transfer")` +**Triggered by:** `transfer_stream()` +**Actor:** The previous recipient who authorized the transfer. + +| Field | Type | Description | +|-----------------|-----------|------------------------------------------------------| +| `stream_id` | `u64` | Stream identifier. | +| `actor` | `Address` | Old recipient address (same as `old_recipient`). | +| `timestamp` | `u64` | Ledger close time when the transfer was recorded. | +| `old_recipient` | `Address` | Account transferring away the stream rights. | +| `new_recipient` | `Address` | Account receiving the stream rights. | + +**Indexer mapping:** event type `"transferred"`, metadata includes `new_recipient`. + +--- + +### ClawbackExecuted *(bonus — emitted by `clawback`)* + +**Topic:** `("Stream", "Clawback")` +**Triggered by:** `clawback()` +**Actor:** The admin address that authorized the clawback. + +| Field | Type | Description | +|-------------|-----------|---------------------------------------------------------------| +| `stream_id` | `u64` | Stream identifier. | +| `actor` | `Address` | Admin address (same as `recipient` field). | +| `timestamp` | `u64` | Ledger close time when the clawback was executed. | +| `amount` | `i128` | Tokens clawed back from the stream (in stroops). | +| `recipient` | `Address` | Admin account that received the clawed-back tokens. | + +**Indexer mapping:** event type `"clawback"`, amount = `amount`, metadata includes `recipient`. + +--- + +## Event Ordering Guarantees + +Within a single transaction: + +- `create_stream` → exactly one `StreamCreated` +- `create_split_stream` → exactly one `StreamCreated` per child stream, in allocation order +- `claim` → exactly one `StreamClaimed`, followed by at most one `StreamCompleted` (only when the stream is fully drained) +- `cancel` → exactly one `StreamCanceled` +- `pause_stream` → exactly one `StreamPaused` +- `resume_stream` → exactly one `StreamResumed` +- `transfer_stream` → exactly one `StreamTransferred` +- `clawback` → exactly one `ClawbackExecuted` (only when `actual_clawback > 0`) + +--- + +## Indexer Event Type Mapping + +The table below maps contract event names to the `eventType` strings stored in the +`stream_events` SQLite table and returned by `GET /api/streams/:id/history`. + +| Contract event topic | `eventType` in DB | `actor` field source | +|----------------------|--------------------|---------------------------------| +| `Created` | `created` | `sender` | +| `Claimed` | `claimed` | `recipient` | +| `Completed` | `completed` | `recipient` (final claimant) | +| `Canceled` | `canceled` | `sender` | +| `Paused` | `paused` | `sender` | +| `Resumed` | `resumed` | `sender` | +| `Transfer` | `transferred` | `old_recipient` | +| `Clawback` | `clawback` | `admin` | + +--- + +## Adding New Events + +To add a new event type: + +1. Define a `#[contracttype]` struct in `contracts/src/lib.rs` with the three + mandatory base fields (`stream_id`, `actor`, `timestamp`) plus any + event-specific fields. +2. Call `env.events().publish((symbol_short!("Stream"), symbol_short!("")), { ... })` inside the relevant contract function. +3. Add a `case "":` branch in `processEvent()` inside + `backend/src/services/indexer.ts` and call `recordEventWithDb`. +4. Document the new event in this file.