Skip to content

feat(contracts): add structured event schema with actor+timestamp #827

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 1 commit into from
Jul 31, 2026
Merged

feat(contracts): add structured event schema with actor+timestamp #827

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
63 changes: 54 additions & 9 deletions backend/src/services/indexer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,6 +202,14 @@ async function indexEvents(): Promise<void> {
/**
* 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 {
Expand All @@ -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,
Expand All @@ -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 },
Comment on lines +251 to +254

@coderabbitai coderabbitai Bot Jul 29, 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 | 🟠 Major | ⚡ Quick win

Persist claimed_amount in the canonical amount column.

Line 253 passes value.amount, while the new claim schema provides value.claimed_amount on Line 254. Because recordEventWithDb stores the positional amount field directly and converts undefined to NULL, claim history will lose its canonical amount even though metadata contains it.

Proposed fix 
-          value.amount,
+          value.claimed_amount,
📝 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
// actor == recipient for Claimed events
value.actor ?? value.recipient,
value.amount,
{ claimed_amount: value.claimed_amount },
// actor == recipient for Claimed events
value.actor ?? value.recipient,
value.claimed_amount,
{ claimed_amount: value.claimed_amount },
🤖 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/indexer.ts` around lines 251 - 254, Update the Claimed
event handling near the recordEventWithDb call to pass value.claimed_amount as
the canonical positional amount instead of value.amount, while preserving
value.claimed_amount in the claimed_amount metadata field and the existing
actor/recipient fallback.

event.ledger,
);
break;

case "Completed":
recordEventWithDb(
db,
value.stream_id.toString(),
"completed",
timestamp,
value.actor,
value.total_amount,
undefined,
event.ledger,
);
Expand All @@ -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,
);
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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");
Expand Down
117 changes: 113 additions & 4 deletions contracts/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -63,51 +77,103 @@ pub struct StreamCreated {
pub metadata: Option<Map<String, String>>,
}

/// 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,
}
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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,
},
);
}

Expand All @@ -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,
},
Expand Down Expand Up @@ -548,6 +651,8 @@ impl StellarStreamContract {
(symbol_short!("Stream"), symbol_short!("Paused")),
StreamPaused {
stream_id,
actor: sender.clone(),
timestamp: now,
sender,
paused_at: now,
},
Expand Down Expand Up @@ -583,6 +688,8 @@ impl StellarStreamContract {
(symbol_short!("Stream"), symbol_short!("Resumed")),
StreamResumed {
stream_id,
actor: sender.clone(),
timestamp: now,
sender,
resumed_at: now,
},
Expand Down Expand Up @@ -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,
},
Expand Down
Loading