Skip to content

[indexer] - refactoring and minor bugfix #406

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
tonylee08 merged 4 commits into from
Sep 22, 2025
Merged

[indexer] - refactoring and minor bugfix #406

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c680a07
refactoring and minor bugfix
patrickkuo Jun 26, 2025
bc469ec
revert unwanted changes
patrickkuo Jun 26, 2025
0b127ce
fix fmt
patrickkuo Jun 26, 2025
4077e7d
Merge branch 'main' into pat/indexer-refactoring
tonylee08 Sep 22, 2025
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
73 changes: 35 additions & 38 deletions crates/indexer/src/handlers/balances_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,46 +30,43 @@ impl Processor for BalancesHandler {
type Value = Balances;

fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
checkpoint
.transactions
.iter()
.try_fold(vec![], |result, tx| {
if !is_deepbook_tx(tx) {
return Ok(result);
}
let Some(events) = &tx.events else {
return Ok(result);
};
let mut results = vec![];

for tx in &checkpoint.transactions {
if !is_deepbook_tx(&tx) {
continue;
}
let Some(events) = &tx.events else {
continue;
};

let package = try_extract_move_call_package(tx).unwrap_or_default();
let checkpoint_timestamp_ms = checkpoint.checkpoint_summary.timestamp_ms as i64;
let checkpoint = checkpoint.checkpoint_summary.sequence_number as i64;
let digest = tx.transaction.digest();
let package = try_extract_move_call_package(tx).unwrap_or_default();
let checkpoint_timestamp_ms = checkpoint.checkpoint_summary.timestamp_ms as i64;
let checkpoint = checkpoint.checkpoint_summary.sequence_number as i64;
let digest = tx.transaction.digest();

return events
.data
.iter()
.filter(|ev| ev.type_ == self.event_type)
.enumerate()
.try_fold(result, |mut result, (index, ev)| {
let event: BalanceEvent = bcs::from_bytes(&ev.contents)?;
let data = Balances {
digest: digest.to_string(),
event_digest: format!("{digest}{index}"),
sender: tx.transaction.sender_address().to_string(),
checkpoint,
checkpoint_timestamp_ms,
package: package.clone(),
balance_manager_id: event.balance_manager_id.to_string(),
asset: event.asset.to_string(),
amount: event.amount as i64,
deposit: event.deposit,
};
debug!("Observed Deepbook Balance Event {:?}", data);
result.push(data);
Ok(result)
});
})
for (index, ev) in events.data.iter().enumerate() {
if ev.type_ != self.event_type {
continue;
}
let event: BalanceEvent = bcs::from_bytes(&ev.contents)?;
let data = Balances {
digest: digest.to_string(),
event_digest: format!("{digest}{index}"),
sender: tx.transaction.sender_address().to_string(),
checkpoint,
checkpoint_timestamp_ms,
package: package.clone(),
balance_manager_id: event.balance_manager_id.to_string(),
asset: event.asset.to_string(),
amount: event.amount as i64,
deposit: event.deposit,
};
debug!("Observed Deepbook Balance Event {:?}", data);
results.push(data);
}
}
Ok(results)
}
}

Expand Down
72 changes: 34 additions & 38 deletions crates/indexer/src/handlers/flash_loan_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,46 +30,42 @@ impl Processor for FlashLoanHandler {
type Value = Flashloan;

fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
checkpoint
.transactions
.iter()
.try_fold(vec![], |result, tx| {
if !is_deepbook_tx(tx) {
return Ok(result);
}
let Some(events) = &tx.events else {
return Ok(result);
};
let mut results = vec![];
for tx in &checkpoint.transactions {
if !is_deepbook_tx(tx) {
continue;
}
let Some(events) = &tx.events else {
continue;
};

let package = try_extract_move_call_package(tx).unwrap_or_default();
let checkpoint_timestamp_ms = checkpoint.checkpoint_summary.timestamp_ms as i64;
let checkpoint = checkpoint.checkpoint_summary.sequence_number as i64;
let digest = tx.transaction.digest();
let package = try_extract_move_call_package(tx).unwrap_or_default();
let checkpoint_timestamp_ms = checkpoint.checkpoint_summary.timestamp_ms as i64;
let checkpoint = checkpoint.checkpoint_summary.sequence_number as i64;
let digest = tx.transaction.digest();

return events
.data
.iter()
.filter(|ev| ev.type_ == self.event_type)
.enumerate()
.try_fold(result, |mut result, (index, ev)| {
let event: FlashLoanBorrowed = bcs::from_bytes(&ev.contents)?;
let data = Flashloan {
digest: digest.to_string(),
event_digest: format!("{digest}{index}"),
sender: tx.transaction.sender_address().to_string(),
checkpoint,
checkpoint_timestamp_ms,
package: package.clone(),
pool_id: event.pool_id.to_string(),
borrow_quantity: event.borrow_quantity as i64,
borrow: true,
type_name: event.type_name.to_string(),
};
debug!("Observed Deepbook Flash Loan Borrowed {:?}", data);
result.push(data);
Ok(result)
});
})
for (index, ev) in events.data.iter().enumerate() {
if ev.type_ != self.event_type {
continue;
}
let event: FlashLoanBorrowed = bcs::from_bytes(&ev.contents)?;
let data = Flashloan {
digest: digest.to_string(),
event_digest: format!("{digest}{index}"),
sender: tx.transaction.sender_address().to_string(),
checkpoint,
checkpoint_timestamp_ms,
package: package.clone(),
pool_id: event.pool_id.to_string(),
borrow_quantity: event.borrow_quantity as i64,
borrow: true,
type_name: event.type_name.to_string(),
};
debug!("Observed Deepbook Flash Loan Borrowed {:?}", data);
results.push(data);
}
}
Ok(results)
}
}

Expand Down
97 changes: 47 additions & 50 deletions crates/indexer/src/handlers/order_fill_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,58 +30,55 @@ impl Processor for OrderFillHandler {
type Value = OrderFill;

fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
checkpoint
.transactions
.iter()
.try_fold(vec![], |result, tx| {
if !is_deepbook_tx(tx) {
return Ok(result);
}
let Some(events) = &tx.events else {
return Ok(result);
};
let mut results = vec![];

for tx in &checkpoint.transactions {
if !is_deepbook_tx(tx) {
continue;
}
let Some(events) = &tx.events else {
continue;
};

let package = try_extract_move_call_package(tx).unwrap_or_default();
let checkpoint_timestamp_ms = checkpoint.checkpoint_summary.timestamp_ms as i64;
let checkpoint = checkpoint.checkpoint_summary.sequence_number as i64;
let digest = tx.transaction.digest();
let package = try_extract_move_call_package(tx).unwrap_or_default();
let checkpoint_timestamp_ms = checkpoint.checkpoint_summary.timestamp_ms as i64;
let checkpoint = checkpoint.checkpoint_summary.sequence_number as i64;
let digest = tx.transaction.digest();

return events
.data
.iter()
.filter(|ev| ev.type_ == self.event_type)
.enumerate()
.try_fold(result, |mut result, (index, ev)| {
let event: OrderFilled = bcs::from_bytes(&ev.contents)?;
let data = OrderFill {
digest: digest.to_string(),
event_digest: format!("{digest}{index}"),
sender: tx.transaction.sender_address().to_string(),
checkpoint,
checkpoint_timestamp_ms,
package: package.clone(),
pool_id: event.pool_id.to_string(),
maker_order_id: event.maker_order_id.to_string(),
taker_order_id: event.taker_order_id.to_string(),
maker_client_order_id: event.maker_client_order_id as i64,
taker_client_order_id: event.taker_client_order_id as i64,
price: event.price as i64,
taker_is_bid: event.taker_is_bid,
taker_fee: event.taker_fee as i64,
taker_fee_is_deep: event.taker_fee_is_deep,
maker_fee: event.maker_fee as i64,
maker_fee_is_deep: event.maker_fee_is_deep,
base_quantity: event.base_quantity as i64,
quote_quantity: event.quote_quantity as i64,
maker_balance_manager_id: event.maker_balance_manager_id.to_string(),
taker_balance_manager_id: event.taker_balance_manager_id.to_string(),
onchain_timestamp: event.timestamp as i64,
};
debug!("Observed Deepbook Order Filled {:?}", data);
result.push(data);
Ok(result)
});
})
for (index, ev) in events.data.iter().enumerate() {
if ev.type_ != self.event_type {
continue;
}
let event: OrderFilled = bcs::from_bytes(&ev.contents)?;
let data = OrderFill {
digest: digest.to_string(),
event_digest: format!("{digest}{index}"),
sender: tx.transaction.sender_address().to_string(),
checkpoint,
checkpoint_timestamp_ms,
package: package.clone(),
pool_id: event.pool_id.to_string(),
maker_order_id: event.maker_order_id.to_string(),
taker_order_id: event.taker_order_id.to_string(),
maker_client_order_id: event.maker_client_order_id as i64,
taker_client_order_id: event.taker_client_order_id as i64,
price: event.price as i64,
taker_is_bid: event.taker_is_bid,
taker_fee: event.taker_fee as i64,
taker_fee_is_deep: event.taker_fee_is_deep,
maker_fee: event.maker_fee as i64,
maker_fee_is_deep: event.maker_fee_is_deep,
base_quantity: event.base_quantity as i64,
quote_quantity: event.quote_quantity as i64,
maker_balance_manager_id: event.maker_balance_manager_id.to_string(),
taker_balance_manager_id: event.taker_balance_manager_id.to_string(),
onchain_timestamp: event.timestamp as i64,
};
debug!("Observed Deepbook Order Filled {:?}", data);
results.push(data);
}
}
Ok(results)
}
}

Expand Down
80 changes: 38 additions & 42 deletions crates/indexer/src/handlers/order_update_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,50 +37,46 @@ impl Processor for OrderUpdateHandler {
const NAME: &'static str = "order_update";
type Value = OrderUpdate;
fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
checkpoint
.transactions
.iter()
.try_fold(vec![], |result, tx| {
if !is_deepbook_tx(tx) {
return Ok(result);
}
let Some(events) = &tx.events else {
return Ok(result);
};
let mut results = vec![];

for tx in &checkpoint.transactions {
if !is_deepbook_tx(tx) {
continue;
}
let Some(events) = &tx.events else {
continue;
};

let package = try_extract_move_call_package(tx).unwrap_or_default();
let metadata = (
tx.transaction.sender_address().to_string(),
checkpoint.checkpoint_summary.sequence_number,
checkpoint.checkpoint_summary.timestamp_ms,
tx.transaction.digest().to_string(),
package.clone(),
);
let package = try_extract_move_call_package(tx).unwrap_or_default();
let metadata = (
tx.transaction.sender_address().to_string(),
checkpoint.checkpoint_summary.sequence_number,
checkpoint.checkpoint_summary.timestamp_ms,
tx.transaction.digest().to_string(),
package.clone(),
);

return events.data.iter().enumerate().try_fold(
result,
|mut result, (index, ev)| {
if ev.type_ == self.order_placed_type {
let event = bcs::from_bytes(&ev.contents)?;
result.push(process_order_placed(event, metadata.clone(), index));
debug!("Observed Deepbook Order Placed {:?}", tx);
} else if ev.type_ == self.order_modified_type {
let event = bcs::from_bytes(&ev.contents)?;
result.push(process_order_modified(event, metadata.clone(), index));
debug!("Observed Deepbook Order Modified {:?}", tx);
} else if ev.type_ == self.order_canceled_type {
let event = bcs::from_bytes(&ev.contents)?;
result.push(process_order_canceled(event, metadata.clone(), index));
debug!("Observed Deepbook Order Canceled {:?}", tx);
} else if ev.type_ == self.order_expired_type {
let event = bcs::from_bytes(&ev.contents)?;
result.push(process_order_expired(event, metadata.clone(), index));
debug!("Observed Deepbook Order Expired {:?}", tx);
}
Ok(result)
},
);
})
for (index, ev) in events.data.iter().enumerate() {
if ev.type_ == self.order_placed_type {
let event = bcs::from_bytes(&ev.contents)?;
results.push(process_order_placed(event, metadata.clone(), index));
debug!("Observed Deepbook Order Placed {:?}", tx);
} else if ev.type_ == self.order_modified_type {
let event = bcs::from_bytes(&ev.contents)?;
results.push(process_order_modified(event, metadata.clone(), index));
debug!("Observed Deepbook Order Modified {:?}", tx);
} else if ev.type_ == self.order_canceled_type {
let event = bcs::from_bytes(&ev.contents)?;
results.push(process_order_canceled(event, metadata.clone(), index));
debug!("Observed Deepbook Order Canceled {:?}", tx);
} else if ev.type_ == self.order_expired_type {
let event = bcs::from_bytes(&ev.contents)?;
results.push(process_order_expired(event, metadata.clone(), index));
debug!("Observed Deepbook Order Expired {:?}", tx);
}
}
}
Ok(results)
}
}

Expand Down
Loading