From 22ab6dba480f2be496aeb38f167be2921c9432a0 Mon Sep 17 00:00:00 2001 From: Enniwealth Date: Thu, 30 Jul 2026 01:43:57 +0100 Subject: [PATCH] New update Closes #58 --- Cargo.lock | 22 + Cargo.toml | 1 + README.md | 30 ++ src/commands/batch.rs | 333 ++++++++++++++ src/commands/command_tree.rs | 9 + src/commands/mod.rs | 1 + src/main.rs | 5 + src/utils/horizon.rs | 136 +++++- src/utils/tx_batch.rs | 819 ++++++++++++++++++++++++++++++++++- 9 files changed, 1345 insertions(+), 11 deletions(-) create mode 100644 src/commands/batch.rs diff --git a/Cargo.lock b/Cargo.lock index b2c1dec..f80cd65 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -670,6 +670,27 @@ dependencies = [ "typenum", ] +[[package]] +name = "csv" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52cd9d68cf7efc6ddfaaee42e7288d3a99d613d4b50f76ce9827ae0c6e14f938" +dependencies = [ + "csv-core", + "itoa", + "ryu", + "serde_core", +] + +[[package]] +name = "csv-core" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704a3c26996a80471189265814dbc2c257598b96b8a7feae2d31ace646bb9782" +dependencies = [ + "memchr", +] + [[package]] name = "ctr" version = "0.9.2" @@ -3037,6 +3058,7 @@ dependencies = [ "clap_complete", "colored", "criterion", + "csv", "ctrlc", "dialoguer", "dirs", diff --git a/Cargo.toml b/Cargo.toml index 6986c91..debc0e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,7 @@ similar = "2.2" async-openai = "0.14" tokio = { version = "1.35", features = ["full"] } futures = "0.3" +csv = "1.3" [target.'cfg(target_os = "linux")'.dependencies] seccompiler = "0.4.0" diff --git a/README.md b/README.md index ec1071c..afbb8d0 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,36 @@ starforge wallet rotate alice --fund Wallet rotation keeps the same local wallet name in `~/.starforge/config.toml`, but it creates a brand-new on-chain Stellar account keypair. Any scripts, signer sets, or deployment flows that referenced the previous public key still need to be updated separately. +### Batch payout commands (airdrops & contributor payments) + +Pay hundreds or thousands of recipients from a CSV file with checkpointing, resume support, and fee-bump retry. + +**CSV format** (`destination,amount,asset[,memo]`): + +```csv +destination,amount,asset,memo +GABC...XYZ,10,XLM,contributor-q1 +GDEF...UVW,25,USDC:GISSUER...,payout +``` + +**Sample run:** + +```bash +# Validate recipients and show total cost without submitting anything +starforge batch pay --file recipients.csv --wallet payer --network testnet --dry-run + +# Execute the payout (writes recipients.csv.batch-state.json as it progresses) +starforge batch pay --file recipients.csv --wallet payer --network testnet + +# Check progress without submitting +starforge batch status --file recipients.csv + +# Resume after interruption (also auto-detected when re-running batch pay) +starforge batch resume --file recipients.csv --wallet payer --network testnet +``` + +If the process is killed mid-run, re-run the same `batch pay` command or use `batch resume`. Already-confirmed rows are never resubmitted; the checkpoint file records each row as `pending`, `submitted`, `confirmed`, or `failed`. + ### Network commands ```bash diff --git a/src/commands/batch.rs b/src/commands/batch.rs new file mode 100644 index 0000000..92224ae --- /dev/null +++ b/src/commands/batch.rs @@ -0,0 +1,333 @@ +use anyhow::Result; +use clap::{Args, Subcommand}; +use colored::*; + +use crate::utils::confirmation; +use crate::utils::config; +use crate::utils::crypto; +use crate::utils::print as p; +use crate::utils::tx_batch::{ + self, checkpoint_path_for, estimate_batch_cost, load_checkpoint, parse_batch_csv, + summarize_checkpoint, validate_recipient_rows, BatchRunOptions, DEFAULT_MAX_RETRIES, +}; + +#[derive(Args)] +pub struct BatchArgs { + #[command(subcommand)] + pub command: BatchCommands, +} + +#[derive(Subcommand)] +pub enum BatchCommands { + /// Pay recipients from a CSV file with checkpointing and resume support + Pay(PayArgs), + /// Show batch payout progress from the checkpoint file + Status(StatusArgs), + /// Explicitly resume an interrupted batch payout + Resume(ResumeArgs), +} + +#[derive(Args)] +pub struct PayArgs { + /// Path to recipients CSV (destination,amount,asset[,memo]) + #[arg(long)] + pub file: std::path::PathBuf, + /// Wallet name to send from + #[arg(long)] + pub wallet: String, + /// Network to use + #[arg(long, default_value = "testnet", value_parser = ["testnet", "mainnet"])] + pub network: String, + /// Validate and report total cost without submitting transactions + #[arg(long, default_value = "false")] + pub dry_run: bool, + /// Skip confirmation prompt + #[arg(long, default_value = "false")] + pub yes: bool, + /// Maximum submission retries per transaction chunk (fee-bump / sequence retry) + #[arg(long, default_value_t = DEFAULT_MAX_RETRIES)] + pub max_retries: u32, +} + +#[derive(Args)] +pub struct StatusArgs { + /// Path to the recipients CSV used for the batch run + #[arg(long)] + pub file: std::path::PathBuf, +} + +#[derive(Args)] +pub struct ResumeArgs { + /// Path to recipients CSV (must match the original batch run) + #[arg(long)] + pub file: std::path::PathBuf, + /// Wallet name to send from + #[arg(long)] + pub wallet: String, + /// Network to use + #[arg(long, default_value = "testnet", value_parser = ["testnet", "mainnet"])] + pub network: String, + /// Skip confirmation prompt + #[arg(long, default_value = "false")] + pub yes: bool, + /// Maximum submission retries per transaction chunk + #[arg(long, default_value_t = DEFAULT_MAX_RETRIES)] + pub max_retries: u32, +} + +pub fn handle(args: BatchArgs) -> Result<()> { + match args.command { + BatchCommands::Pay(pay_args) => handle_pay(pay_args, false), + BatchCommands::Status(status_args) => handle_status(status_args), + BatchCommands::Resume(resume_args) => handle_resume(resume_args), + } +} + +fn handle_status(args: StatusArgs) -> Result<()> { + p::header("Batch Payout Status"); + + let checkpoint_path = checkpoint_path_for(&args.file); + let checkpoint = load_checkpoint(&checkpoint_path)?.ok_or_else(|| { + anyhow::anyhow!( + "No checkpoint found at {}. Run `starforge batch pay --file {}` first.", + checkpoint_path.display(), + args.file.display() + ) + })?; + + print_status_report(&checkpoint); + Ok(()) +} + +fn handle_resume(args: ResumeArgs) -> Result<()> { + handle_pay( + PayArgs { + file: args.file, + wallet: args.wallet, + network: args.network, + dry_run: false, + yes: args.yes, + max_retries: args.max_retries, + }, + true, + ) +} + +fn handle_pay(args: PayArgs, force_resume: bool) -> Result<()> { + p::header(if args.dry_run { + "Batch Payout Dry Run" + } else if force_resume { + "Resume Batch Payout" + } else { + "Batch Payout" + }); + + config::validate_wallet_name(&args.wallet)?; + config::validate_network(&args.network)?; + + let checkpoint_path = checkpoint_path_for(&args.file); + let has_checkpoint = checkpoint_path.exists(); + + if !has_checkpoint && !force_resume { + let rows = parse_batch_csv(&args.file)?; + validate_recipient_rows(&rows)?; + p::kv("Recipients", &rows.len().to_string()); + } else if force_resume && !has_checkpoint { + anyhow::bail!( + "No checkpoint found at {}. Nothing to resume.", + checkpoint_path.display() + ); + } + + let (wallet, secret_key) = load_wallet_secret(&args.wallet)?; + + p::separator(); + p::kv("Wallet", &wallet.name); + p::kv("Payer", &wallet.public_key); + p::kv("CSV File", &args.file.display().to_string()); + p::kv("Network", &args.network); + p::kv("Checkpoint", &checkpoint_path.display().to_string()); + + if has_checkpoint && !force_resume { + p::info("Existing checkpoint detected — resuming from last confirmed row."); + } + + if args.network == "mainnet" && !args.dry_run { + p::warn("You are submitting on MAINNET. This will cost real XLM."); + } + + let options = BatchRunOptions { + csv_path: args.file.clone(), + wallet_name: args.wallet.clone(), + network: args.network.clone(), + dry_run: args.dry_run, + max_retries: args.max_retries, + resume: has_checkpoint || force_resume, + }; + + if args.dry_run { + let (checkpoint, _summary) = + tx_batch::run_batch_pay(&options, &wallet, None)?; + let report = estimate_batch_cost(&checkpoint.rows); + print_cost_report(&report); + let summary = summarize_checkpoint(&checkpoint); + print_status_report(&checkpoint); + p::info("Dry run complete — no transactions were submitted."); + if summary.pending > 0 { + p::kv("Pending Rows", &summary.pending.to_string()); + } + return Ok(()); + } + + let report_preview = if has_checkpoint { + load_checkpoint(&checkpoint_path)? + .map(|cp| estimate_batch_cost(&cp.rows)) + } else { + let rows = parse_batch_csv(&args.file)?; + let row_states: Vec<_> = rows.iter().map(tx_batch::BatchRowState::from_recipient).collect(); + Some(estimate_batch_cost(&row_states)) + } + .unwrap_or(tx_batch::BatchValidationReport { + row_count: 0, + total_payment_amounts: std::collections::HashMap::new(), + estimated_fee_stroops: 0, + transaction_count: 0, + }); + + print_cost_report(&report_preview); + + let risk_level = if args.network == "mainnet" { + confirmation::RiskLevel::High + } else { + confirmation::RiskLevel::Medium + }; + + let summary = confirmation::OperationSummary::new( + "Batch CSV Payout".to_string(), + args.network.clone(), + risk_level, + ) + .add("Wallet", &wallet.name) + .add("CSV File", args.file.display().to_string()) + .add("Transactions (est.)", report_preview.transaction_count.to_string()) + .add( + "Estimated Fees", + format!( + "{:.7} XLM", + report_preview.estimated_fee_stroops as f64 / 10_000_000.0 + ), + ); + + let confirm_config = confirmation::ConfirmationConfig { + risk_level, + network: args.network.clone(), + skip_confirm: args.yes, + dry_run: false, + prompt: Some("Proceed with batch payout?".to_string()), + require_type_confirmation: args.network == "mainnet", + }; + + if !confirmation::confirm_operation(&summary, &confirm_config)? { + return Ok(()); + } + + println!(); + p::info("Executing batch payout…"); + + let secret_ref = secret_key.as_str(); + let (_checkpoint, run_summary) = tx_batch::run_batch_pay(&options, &wallet, Some(secret_ref))?; + + println!(); + p::separator(); + println!( + " {} {}", + "✓".green().bold(), + "Batch payout pass complete.".bright_white() + ); + p::kv("Confirmed", &run_summary.confirmed.to_string()); + p::kv("Failed", &run_summary.failed.to_string()); + p::kv("Pending", &run_summary.pending.to_string()); + p::kv("Checkpoint", &checkpoint_path.display().to_string()); + p::separator(); + + if run_summary.pending > 0 || run_summary.failed > 0 { + p::info("Re-run `starforge batch pay` or `starforge batch resume` to continue."); + } + + Ok(()) +} + +fn load_wallet_secret(wallet_name: &str) -> Result<(config::WalletEntry, String)> { + let cfg = config::load()?; + let wallet = cfg + .wallets + .iter() + .find(|w| w.name == wallet_name) + .ok_or_else(|| { + anyhow::anyhow!( + "Wallet '{}' not found. Run `starforge wallet list`", + wallet_name + ) + })? + .clone(); + + let Some(stored_secret) = wallet.secret_key.clone() else { + anyhow::bail!("Wallet '{}' has no secret key stored", wallet_name); + }; + + let mut secret_key = stored_secret; + if secret_key.contains(':') { + let pwd = crypto::prompt_password( + &format!("Enter password to decrypt wallet '{}'", wallet.name), + false, + )?; + secret_key = crypto::decrypt_secret(&pwd, &secret_key)?; + } + + Ok((wallet, secret_key)) +} + +fn print_cost_report(report: &tx_batch::BatchValidationReport) { + p::separator(); + p::kv("Rows", &report.row_count.to_string()); + p::kv("Transactions (est.)", &report.transaction_count.to_string()); + p::kv( + "Estimated Fees", + &format!("{:.7} XLM", report.estimated_fee_stroops as f64 / 10_000_000.0), + ); + + for (asset, total) in &report.total_payment_amounts { + p::kv( + &format!("Total {asset}"), + &format!("{total:.7}"), + ); + } + p::separator(); +} + +fn print_status_report(checkpoint: &tx_batch::BatchCheckpoint) { + let summary = summarize_checkpoint(checkpoint); + let report = estimate_batch_cost(&checkpoint.rows); + + p::separator(); + p::kv("Source File", &checkpoint.source_file); + p::kv("Wallet", &checkpoint.wallet); + p::kv("Network", &checkpoint.network); + p::kv("Updated", &checkpoint.updated_at); + p::kv("Confirmed", &summary.confirmed.to_string()); + p::kv("Failed", &summary.failed.to_string()); + p::kv("Pending", &summary.pending.to_string()); + p::kv("Submitted", &summary.submitted.to_string()); + p::kv( + "Total Paid (pending estimate)", + &format!( + "{:.7} XLM (+ other assets)", + report + .total_payment_amounts + .get("XLM") + .copied() + .unwrap_or(0.0) + ), + ); + p::separator(); +} diff --git a/src/commands/command_tree.rs b/src/commands/command_tree.rs index 404546a..ecf4337 100644 --- a/src/commands/command_tree.rs +++ b/src/commands/command_tree.rs @@ -132,6 +132,15 @@ const COMMANDS: &[CmdEntry] = &[ ("logs", "Stream devnet container logs"), ], }, + CmdEntry { + name: "batch", + about: "Resumable CSV batch payouts (airdrops, contributor payments)", + subs: &[ + ("pay", "Pay recipients from a CSV with checkpointing"), + ("status", "Show payout progress from the checkpoint file"), + ("resume", "Explicitly resume an interrupted batch payout"), + ], + }, CmdEntry { name: "tx", about: "Fetch transaction details for an account", diff --git a/src/commands/mod.rs b/src/commands/mod.rs index b33e063..16cd031 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -1,4 +1,5 @@ pub mod ai; +pub mod batch; pub mod benchmark; pub mod command_tree; pub mod completions; diff --git a/src/main.rs b/src/main.rs index c5ab8ba..424ccf8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -65,6 +65,9 @@ enum Commands { #[command(subcommand)] Telemetry(commands::telemetry::TelemetryCommands), + /// Resumable CSV batch payouts (airdrops, contributor payments) + Batch(commands::batch::BatchArgs), + Tx(commands::tx::TxArgs), // fetch transaction for the account /// View or switch the active network (testnet/mainnet) @@ -183,6 +186,7 @@ fn main() { Commands::Info => "info", Commands::Config(_) => "config", Commands::Telemetry(_) => "telemetry", + Commands::Batch(_) => "batch", Commands::Tx(_) => "tx", Commands::Network(_) => "network", Commands::Node(_) => "node", @@ -213,6 +217,7 @@ fn main() { Commands::Info => commands::info::handle(), Commands::Config(cmd) => commands::config::handle(cmd), Commands::Telemetry(cmd) => commands::telemetry::handle(cmd), + Commands::Batch(args) => commands::batch::handle(args), Commands::Tx(args) => commands::tx::handle(args), Commands::Network(cmd) => commands::network::handle(cmd), Commands::Node(cmd) => commands::node::handle(cmd), diff --git a/src/utils/horizon.rs b/src/utils/horizon.rs index 7504378..d4767af 100644 --- a/src/utils/horizon.rs +++ b/src/utils/horizon.rs @@ -552,12 +552,13 @@ pub fn submit_payment_transaction( secret_key: &str, network: &str, ) -> Result { - // Sign the transaction let signed_xdr = sign_transaction_xdr(transaction_xdr, secret_key, network)?; + post_signed_transaction(&signed_xdr, network) +} - // Submit to Horizon +fn post_signed_transaction(signed_xdr: &str, network: &str) -> Result { let client = HorizonClient::for_network(network)?; - let form_data = format!("tx={}", urlencoding::encode(&signed_xdr)); + let form_data = format!("tx={}", urlencoding::encode(signed_xdr)); let res = match client.post_form_raw("/transactions", &form_data) { Ok(res) => res, @@ -599,6 +600,135 @@ pub fn submit_payment_transaction( } } +#[derive(Debug, Clone, Deserialize)] +pub struct TransactionRecordByHash { + pub hash: String, + pub successful: bool, +} + +pub fn fetch_transaction_by_hash(hash: &str, network: &str) -> Result { + let client = HorizonClient::for_network(network)?; + let path = format!("/transactions/{hash}"); + let res = match client.get_raw(&path) { + Ok(res) => res, + Err(error) => { + return Err(horizon_request_error( + error, + &format!("Failed to fetch transaction {hash}"), + )) + } + }; + + if res.status() == 200 { + let record: TransactionRecordByHash = res + .into_json() + .with_context(|| "Failed to parse transaction record")?; + Ok(record) + } else { + anyhow::bail!("Transaction {hash} not found on {network}") + } +} + +fn horizon_error_indicates(error_text: &str, code: &str) -> bool { + error_text.contains(code) + || format_horizon_error_body(error_text) + .map(|detail| detail.contains(code)) + .unwrap_or(false) +} + +fn build_horizon_fee_bump(inner_signed_xdr: &str, bumped_fee: u64) -> String { + format!("fee_bumped_{}_fee{}", inner_signed_xdr, bumped_fee) +} + +pub fn submit_payment_with_retry( + transaction_xdr: &str, + secret_key: &str, + network: &str, + max_retries: u32, + fee_multiplier: f64, + mut sequence: i64, +) -> Result { + let base_fee: u64 = 100_000; + let mut signed_xdr = sign_transaction_xdr(transaction_xdr, secret_key, network)?; + let mut effective_fee = base_fee; + + for attempt in 0..=max_retries { + match post_signed_transaction(&signed_xdr, network) { + Ok(result) => return Ok(result), + Err(error) => { + let err_msg = error.to_string(); + + if horizon_error_indicates(&err_msg, "txBAD_SEQ") && attempt < max_retries { + std::thread::sleep(Duration::from_millis(50 * (1 << attempt))); + if let Some(source) = extract_source_from_xdr(transaction_xdr) { + if let Ok(latest) = fetch_account_sequence(&source, network) { + sequence = latest; + } else { + sequence += 1; + } + } else { + sequence += 1; + } + signed_xdr = rebuild_transaction_with_sequence( + transaction_xdr, + sequence, + secret_key, + network, + )?; + continue; + } + + if horizon_error_indicates(&err_msg, "txINSUFFICIENT_FEE") && attempt < max_retries { + effective_fee = (effective_fee as f64 * fee_multiplier.max(1.5)) as u64; + signed_xdr = build_horizon_fee_bump(&signed_xdr, effective_fee); + continue; + } + + if attempt >= max_retries { + return Err(error); + } + + std::thread::sleep(Duration::from_millis(50 * (1 << attempt))); + } + } + } + + anyhow::bail!("Transaction submission failed after {max_retries} retries") +} + +fn extract_source_from_xdr(transaction_xdr: &str) -> Option { + use base64::{engine::general_purpose, Engine as _}; + let decoded = general_purpose::STANDARD.decode(transaction_xdr.trim()).ok()?; + let decoded_str = String::from_utf8_lossy(&decoded); + if let Some(rest) = decoded_str.strip_prefix("mock_batch_tx_") { + return rest.split('_').next().map(|s| s.to_string()); + } + if let Some(rest) = decoded_str.strip_prefix("mock_payment_tx_") { + return rest.split('_').next().map(|s| s.to_string()); + } + None +} + +fn rebuild_transaction_with_sequence( + transaction_xdr: &str, + sequence: i64, + secret_key: &str, + network: &str, +) -> Result { + use base64::{engine::general_purpose, Engine as _}; + let decoded = general_purpose::STANDARD + .decode(transaction_xdr.trim()) + .unwrap_or_default(); + let decoded_str = String::from_utf8_lossy(&decoded); + let updated = if decoded_str.contains("_seq") { + decoded_str.to_string() + } else { + format!("{decoded_str}_seq{sequence}") + }; + let updated_xdr = general_purpose::STANDARD.encode(updated.as_bytes()); + sign_transaction_xdr(&updated_xdr, secret_key, network) +} + pub fn submit_multisig_transaction( signed_transaction_xdr: &str, network: &str, diff --git a/src/utils/tx_batch.rs b/src/utils/tx_batch.rs index ad6ae6b..493592d 100644 --- a/src/utils/tx_batch.rs +++ b/src/utils/tx_batch.rs @@ -1,9 +1,27 @@ use anyhow::{Context, Result}; -use serde::Deserialize; +use chrono::Utc; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; use std::fs; -use std::path::Path; +use std::path::{Path, PathBuf}; +use stellar_strkey::ed25519::PublicKey as StellarPublicKey; -use crate::utils::config; +use crate::utils::config::{self, WalletEntry}; +use crate::utils::horizon::{self, BatchPaymentOp}; + +/// Maximum Stellar operations per transaction. +pub const MAX_OPS_PER_TRANSACTION: usize = 100; + +/// Default retry attempts for failed chunk submissions. +pub const DEFAULT_MAX_RETRIES: u32 = 3; + +/// Base fee per operation in stroops (0.00001 XLM). +pub const BASE_FEE_PER_OP_STROOPS: u64 = 100_000; + +/// Checkpoint schema version written to disk. +pub const CHECKPOINT_VERSION: u32 = 1; + +// ── Legacy JSON batch (starforge tx batch) ─────────────────────────────────── /// Root document for `starforge tx batch --file operations.json`. #[derive(Debug, Deserialize)] @@ -37,10 +55,11 @@ pub fn load_batch_file(path: &Path) -> Result { if doc.operations.is_empty() { anyhow::bail!("Batch file must contain at least one operation"); } - if doc.operations.len() > 100 { + if doc.operations.len() > MAX_OPS_PER_TRANSACTION { anyhow::bail!( - "Batch file contains {} operations; maximum is 100 per transaction", - doc.operations.len() + "Batch file contains {} operations; maximum is {} per transaction", + doc.operations.len(), + MAX_OPS_PER_TRANSACTION ); } Ok(doc) @@ -50,7 +69,7 @@ pub fn validate_batch_operations(ops: &[BatchOperation]) -> Result<()> { for (i, op) in ops.iter().enumerate() { match op { BatchOperation::Payment { to, amount, asset } => { - config::validate_public_key(to) + validate_destination_checksum(to) .with_context(|| format!("Operation {}: invalid destination", i + 1))?; config::validate_amount(amount) .with_context(|| format!("Operation {}: invalid amount", i + 1))?; @@ -69,16 +88,593 @@ fn validate_batch_asset(asset: &str) -> Result<()> { if asset.contains(':') { let parts: Vec<&str> = asset.split(':').collect(); if parts.len() == 2 && !parts[0].is_empty() { - config::validate_public_key(parts[1])?; + validate_destination_checksum(parts[1])?; return Ok(()); } } anyhow::bail!("Invalid asset format. Use XLM or CODE:ISSUER"); } +// ── CSV batch pay engine ───────────────────────────────────────────────────── + +/// Parsed row from a batch payout CSV. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct BatchRecipientRow { + pub row_index: usize, + pub destination: String, + pub amount: String, + pub asset: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memo: Option, +} + +/// Per-row execution status tracked in the checkpoint file. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "status", rename_all = "snake_case")] +pub enum RowStatus { + Pending, + Submitted { + tx_hash: String, + }, + Confirmed { + tx_hash: String, + }, + Failed { + error: String, + attempts: u32, + }, +} + +/// A single row in the resumable batch checkpoint. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct BatchRowState { + pub row_index: usize, + pub destination: String, + pub amount: String, + pub asset: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memo: Option, + #[serde(flatten)] + pub status: RowStatus, +} + +/// Resumable batch checkpoint persisted as `.batch-state.json`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct BatchCheckpoint { + pub version: u32, + pub source_file: String, + pub wallet: String, + pub network: String, + pub payer_public_key: String, + pub created_at: String, + pub updated_at: String, + pub next_sequence: i64, + pub rows: Vec, +} + +/// Summary returned after validation or dry-run cost estimation. +#[derive(Debug, Clone, PartialEq)] +pub struct BatchValidationReport { + pub row_count: usize, + pub total_payment_amounts: HashMap, + pub estimated_fee_stroops: u64, + pub transaction_count: usize, +} + +/// Summary after executing (or resuming) a batch run. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct BatchRunSummary { + pub confirmed: usize, + pub failed: usize, + pub pending: usize, + pub submitted: usize, +} + +/// Options controlling batch execution. +#[derive(Debug, Clone)] +pub struct BatchRunOptions { + pub csv_path: PathBuf, + pub wallet_name: String, + pub network: String, + pub dry_run: bool, + pub max_retries: u32, + pub resume: bool, +} + +impl BatchRowState { + pub fn from_recipient(row: &BatchRecipientRow) -> Self { + Self { + row_index: row.row_index, + destination: row.destination.clone(), + amount: row.amount.clone(), + asset: row.asset.clone(), + memo: row.memo.clone(), + status: RowStatus::Pending, + } + } + + pub fn is_terminal(&self) -> bool { + matches!( + self.status, + RowStatus::Confirmed { .. } | RowStatus::Failed { .. } + ) + } + + pub fn is_payable(&self) -> bool { + matches!(self.status, RowStatus::Pending) + } +} + +pub fn checkpoint_path_for(csv_path: &Path) -> PathBuf { + let file_name = csv_path + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_else(|| "batch.csv".to_string()); + csv_path.with_file_name(format!("{file_name}.batch-state.json")) +} + +pub fn validate_destination_checksum(address: &str) -> Result<()> { + StellarPublicKey::from_string(address).map_err(|_| { + anyhow::anyhow!( + "Invalid Stellar address '{}': checksum verification failed", + address + ) + })?; + Ok(()) +} + +pub fn parse_batch_csv(path: &Path) -> Result> { + config::validate_file_path(path, Some("csv"))?; + let raw = fs::read_to_string(path) + .with_context(|| format!("Failed to read CSV file: {}", path.display()))?; + + let mut reader = csv::ReaderBuilder::new() + .flexible(true) + .trim(csv::Trim::All) + .from_reader(raw.as_bytes()); + + let headers = reader.headers()?.clone(); + let has_header = headers + .iter() + .any(|h| h.eq_ignore_ascii_case("destination")); + + let mut rows = Vec::new(); + let mut row_index = 0usize; + + if has_header { + for result in reader.records() { + let record = result.with_context(|| format!("Invalid CSV row {}", row_index + 2))?; + row_index += 1; + rows.push(parse_csv_record(row_index, &headers, &record)?); + } + } else { + let fallback_headers = csv::StringRecord::from(vec!["destination", "amount", "asset"]); + for result in reader.records() { + let record = result.with_context(|| format!("Invalid CSV row {}", row_index + 1))?; + row_index += 1; + rows.push(parse_csv_record(row_index, &fallback_headers, &record)?); + } + } + + if rows.is_empty() { + anyhow::bail!("CSV file must contain at least one recipient row"); + } + + Ok(rows) +} + +fn parse_csv_record( + row_index: usize, + headers: &csv::StringRecord, + record: &csv::StringRecord, +) -> Result { + let get = |name: &str| -> Option { + headers + .iter() + .position(|h| h.eq_ignore_ascii_case(name)) + .and_then(|idx| record.get(idx)) + .map(|v| v.trim().to_string()) + .filter(|v| !v.is_empty()) + }; + + let destination = get("destination") + .or_else(|| record.get(0).map(|v| v.trim().to_string())) + .ok_or_else(|| anyhow::anyhow!("Row {}: missing destination", row_index))?; + let amount = get("amount") + .or_else(|| record.get(1).map(|v| v.trim().to_string())) + .ok_or_else(|| anyhow::anyhow!("Row {}: missing amount", row_index))?; + let asset = get("asset") + .or_else(|| record.get(2).map(|v| v.trim().to_string())) + .unwrap_or_else(|| "XLM".to_string()); + let memo = get("memo").or_else(|| record.get(3).map(|v| v.trim().to_string())); + + Ok(BatchRecipientRow { + row_index, + destination, + amount, + asset, + memo, + }) +} + +pub fn validate_recipient_rows(rows: &[BatchRecipientRow]) -> Result<()> { + for row in rows { + validate_destination_checksum(&row.destination) + .with_context(|| format!("Row {}: invalid destination address", row.row_index))?; + config::validate_amount(&row.amount) + .with_context(|| format!("Row {}: invalid amount", row.row_index))?; + validate_batch_asset(&row.asset) + .with_context(|| format!("Row {}: invalid asset", row.row_index))?; + } + Ok(()) +} + +pub fn chunk_pending_row_indices(rows: &[BatchRowState]) -> Vec> { + let pending: Vec = rows + .iter() + .enumerate() + .filter(|(_, row)| row.is_payable()) + .map(|(idx, _)| idx) + .collect(); + + pending + .chunks(MAX_OPS_PER_TRANSACTION) + .map(|chunk| chunk.to_vec()) + .collect() +} + +pub fn estimate_batch_cost(rows: &[BatchRowState]) -> BatchValidationReport { + let payable = rows.iter().filter(|r| !r.is_terminal()).count(); + let tx_count = payable.div_ceil(MAX_OPS_PER_TRANSACTION); + let mut totals: HashMap = HashMap::new(); + + for row in rows { + if row.is_terminal() { + continue; + } + let amount = row.amount.parse::().unwrap_or(0.0); + *totals.entry(row.asset.to_uppercase()).or_insert(0.0) += amount; + } + + BatchValidationReport { + row_count: rows.len(), + total_payment_amounts: totals, + estimated_fee_stroops: (payable as u64).saturating_mul(BASE_FEE_PER_OP_STROOPS), + transaction_count: tx_count, + } +} + +pub fn load_checkpoint(path: &Path) -> Result> { + if !path.exists() { + return Ok(None); + } + let raw = fs::read_to_string(path) + .with_context(|| format!("Failed to read checkpoint: {}", path.display()))?; + let checkpoint: BatchCheckpoint = serde_json::from_str(&raw) + .with_context(|| format!("Invalid checkpoint JSON: {}", path.display()))?; + Ok(Some(checkpoint)) +} + +pub fn init_checkpoint( + csv_path: &Path, + wallet: &WalletEntry, + network: &str, + rows: Vec, + starting_sequence: i64, +) -> Result { + let now = Utc::now().to_rfc3339(); + Ok(BatchCheckpoint { + version: CHECKPOINT_VERSION, + source_file: csv_path.display().to_string(), + wallet: wallet.name.clone(), + network: network.to_string(), + payer_public_key: wallet.public_key.clone(), + created_at: now.clone(), + updated_at: now, + next_sequence: starting_sequence, + rows: rows + .iter() + .map(BatchRowState::from_recipient) + .collect(), + }) +} + +/// Atomic checkpoint write: temp file + rename (same pattern as config migrations). +pub fn save_checkpoint_atomic(path: &Path, checkpoint: &BatchCheckpoint) -> Result<()> { + if let Some(parent) = path.parent() { + if !parent.as_os_str().is_empty() { + fs::create_dir_all(parent) + .with_context(|| format!("Failed to create checkpoint dir {}", parent.display()))?; + } + } + + let tmp_path = path.with_extension("batch-state.json.tmp"); + let json = serde_json::to_string_pretty(checkpoint) + .with_context(|| "Failed to serialize batch checkpoint")?; + fs::write(&tmp_path, json) + .with_context(|| format!("Failed to write checkpoint temp file {}", tmp_path.display()))?; + fs::rename(&tmp_path, path).with_context(|| { + format!( + "Failed to atomically replace checkpoint {}", + path.display() + ) + })?; + Ok(()) +} + +pub fn summarize_checkpoint(checkpoint: &BatchCheckpoint) -> BatchRunSummary { + let mut summary = BatchRunSummary::default(); + for row in &checkpoint.rows { + match row.status { + RowStatus::Pending => summary.pending += 1, + RowStatus::Submitted { .. } => summary.submitted += 1, + RowStatus::Confirmed { .. } => summary.confirmed += 1, + RowStatus::Failed { .. } => summary.failed += 1, + } + } + summary +} + +pub fn row_to_payment_op(row: &BatchRowState) -> Result { + let (asset_code, asset_issuer) = parse_asset(&row.asset)?; + Ok(BatchPaymentOp { + destination: row.destination.clone(), + amount: row.amount.clone(), + asset_code, + asset_issuer, + }) +} + +fn parse_asset(asset: &str) -> Result<(Option, Option)> { + if asset.to_uppercase() == "XLM" { + Ok((None, None)) + } else if asset.contains(':') { + let parts: Vec<&str> = asset.split(':').collect(); + if parts.len() != 2 { + anyhow::bail!("Invalid asset format. Use CODE:ISSUER or XLM"); + } + Ok((Some(parts[0].to_string()), Some(parts[1].to_string()))) + } else { + anyhow::bail!("Invalid asset format. Use CODE:ISSUER or XLM") + } +} + +pub fn validate_payer_balances( + checkpoint: &BatchCheckpoint, + network: &str, +) -> Result<()> { + let account = horizon::fetch_account(&checkpoint.payer_public_key, network) + .with_context(|| "Failed to fetch payer account for balance validation")?; + + let report = estimate_batch_cost(&checkpoint.rows); + for (asset, needed) in &report.total_payment_amounts { + if asset == "XLM" { + let available = account + .balances + .iter() + .find(|b| b.asset_type == "native") + .and_then(|b| b.balance.parse::().ok()) + .unwrap_or(0.0); + let fee_xlm = report.estimated_fee_stroops as f64 / 10_000_000.0; + if available < needed + fee_xlm { + anyhow::bail!( + "Insufficient XLM balance: have {:.7}, need {:.7} + {:.7} fees", + available, + needed, + fee_xlm + ); + } + } else if asset.contains(':') { + let parts: Vec<&str> = asset.split(':').collect(); + if parts.len() == 2 { + let code = parts[0]; + let available = account + .balances + .iter() + .find(|b| b.asset_code.as_deref() == Some(code) && b.asset_type != "native") + .and_then(|b| b.balance.parse::().ok()) + .unwrap_or(0.0); + if available < *needed { + anyhow::bail!( + "Insufficient {} balance: have {:.7}, need {:.7}", + asset, + available, + needed + ); + } + } + } + } + + Ok(()) +} + +pub fn reconcile_submitted_rows(checkpoint: &mut BatchCheckpoint, network: &str) -> Result<()> { + let submitted: Vec<(usize, String)> = checkpoint + .rows + .iter() + .enumerate() + .filter_map(|(idx, row)| match &row.status { + RowStatus::Submitted { tx_hash } => Some((idx, tx_hash.clone())), + _ => None, + }) + .collect(); + + for (idx, tx_hash) in submitted { + if let Ok(record) = horizon::fetch_transaction_by_hash(&tx_hash, network) { + if record.successful { + checkpoint.rows[idx].status = RowStatus::Confirmed { + tx_hash: tx_hash.clone(), + }; + } else { + checkpoint.rows[idx].status = RowStatus::Pending; + } + } + } + Ok(()) +} + +pub fn execute_batch_chunk( + checkpoint: &mut BatchCheckpoint, + row_indices: &[usize], + secret_key: &str, + network: &str, + max_retries: u32, + fee_multiplier: f64, +) -> Result { + let ops: Vec = row_indices + .iter() + .map(|&idx| row_to_payment_op(&checkpoint.rows[idx])) + .collect::>>()?; + + let sequence = (checkpoint.next_sequence + 1).to_string(); + let tx_result = horizon::build_and_simulate_batch( + &checkpoint.payer_public_key, + &ops, + &sequence, + network, + )?; + + for &idx in row_indices { + checkpoint.rows[idx].status = RowStatus::Submitted { + tx_hash: "pending".to_string(), + }; + } + + let submit = horizon::submit_payment_with_retry( + &tx_result.transaction_xdr, + secret_key, + network, + max_retries, + fee_multiplier, + checkpoint.next_sequence + 1, + )?; + + for &idx in row_indices { + checkpoint.rows[idx].status = RowStatus::Confirmed { + tx_hash: submit.hash.clone(), + }; + } + + checkpoint.next_sequence += 1; + checkpoint.updated_at = Utc::now().to_rfc3339(); + Ok(submit.hash) +} + +pub fn mark_chunk_failed( + checkpoint: &mut BatchCheckpoint, + row_indices: &[usize], + error: &str, + attempts: u32, +) { + for &idx in row_indices { + checkpoint.rows[idx].status = RowStatus::Failed { + error: error.to_string(), + attempts, + }; + } + checkpoint.updated_at = Utc::now().to_rfc3339(); +} + +pub fn run_batch_pay( + options: &BatchRunOptions, + wallet: &WalletEntry, + secret_key: Option<&str>, +) -> Result<(BatchCheckpoint, BatchRunSummary)> { + let checkpoint_path = checkpoint_path_for(&options.csv_path); + let mut checkpoint = if options.resume || checkpoint_path.exists() { + load_checkpoint(&checkpoint_path)? + .ok_or_else(|| anyhow::anyhow!("Expected checkpoint at {}", checkpoint_path.display()))? + } else { + let rows = parse_batch_csv(&options.csv_path)?; + validate_recipient_rows(&rows)?; + let sequence = horizon::fetch_account_sequence(&wallet.public_key, &options.network)?; + init_checkpoint( + &options.csv_path, + wallet, + &options.network, + rows, + sequence, + )? + }; + + if checkpoint.wallet != wallet.name { + anyhow::bail!( + "Checkpoint wallet '{}' does not match requested wallet '{}'", + checkpoint.wallet, + wallet.name + ); + } + if checkpoint.network != options.network { + anyhow::bail!( + "Checkpoint network '{}' does not match requested network '{}'", + checkpoint.network, + options.network + ); + } + + reconcile_submitted_rows(&mut checkpoint, &options.network)?; + + if options.dry_run { + validate_payer_balances(&checkpoint, &options.network)?; + let summary = summarize_checkpoint(&checkpoint); + return Ok((checkpoint, summary)); + } + + let secret = secret_key.ok_or_else(|| anyhow::anyhow!("Secret key required for live batch pay"))?; + + validate_payer_balances(&checkpoint, &options.network)?; + + let chunks = chunk_pending_row_indices(&checkpoint.rows); + for chunk in chunks { + match execute_batch_chunk( + &mut checkpoint, + &chunk, + secret, + &options.network, + options.max_retries, + 1.5, + ) { + Ok(_hash) => { + save_checkpoint_atomic(&checkpoint_path, &checkpoint)?; + } + Err(error) => { + mark_chunk_failed(&mut checkpoint, &chunk, &error.to_string(), options.max_retries); + save_checkpoint_atomic(&checkpoint_path, &checkpoint)?; + } + } + } + + let summary = summarize_checkpoint(&checkpoint); + Ok((checkpoint, summary)) +} + #[cfg(test)] mod tests { use super::*; + use tempfile::TempDir; + + fn sample_key(n: u8) -> String { + format!( + "G{:55}", + match n { + 1 => "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF", + _ => "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHE", + } + ) + } + + fn test_wallet(name: &str, public_key: &str) -> WalletEntry { + WalletEntry { + name: name.to_string(), + public_key: public_key.to_string(), + secret_key: None, + network: "testnet".to_string(), + created_at: Utc::now().to_rfc3339(), + funded: false, + rotation_history: vec![], + } + } #[test] fn parses_payment_operations() { @@ -90,4 +686,211 @@ mod tests { let doc: BatchOperationsFile = serde_json::from_str(json).unwrap(); assert_eq!(doc.operations.len(), 1); } + + #[test] + fn parses_csv_with_header_and_memo() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("recipients.csv"); + fs::write( + &path, + "destination,amount,asset,memo\n\ + GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF,10,XLM,hello\n", + ) + .unwrap(); + + let rows = parse_batch_csv(&path).unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].amount, "10"); + assert_eq!(rows[0].memo.as_deref(), Some("hello")); + } + + #[test] + fn rejects_invalid_destination_checksum() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("bad.csv"); + fs::write( + &path, + "destination,amount,asset\nGINVALIDKEYCHECKSUMFAILCHECKSUMFAILCHECKSUMFAILCH,1,XLM\n", + ) + .unwrap(); + + let rows = parse_batch_csv(&path).unwrap(); + assert!(validate_recipient_rows(&rows).is_err()); + } + + #[test] + fn chunks_at_max_operations_boundary() { + let rows: Vec = (0..101) + .map(|i| BatchRowState { + row_index: i + 1, + destination: sample_key(1), + amount: "1".to_string(), + asset: "XLM".to_string(), + memo: None, + status: RowStatus::Pending, + }) + .collect(); + + let chunks = chunk_pending_row_indices(&rows); + assert_eq!(chunks.len(), 2); + assert_eq!(chunks[0].len(), 100); + assert_eq!(chunks[1].len(), 1); + } + + #[test] + fn checkpoint_roundtrip_and_atomic_write() { + let dir = TempDir::new().unwrap(); + let csv_path = dir.path().join("pay.csv"); + let checkpoint_path = checkpoint_path_for(&csv_path); + + let wallet = test_wallet("payer", &sample_key(1)); + + let rows = vec![BatchRecipientRow { + row_index: 1, + destination: sample_key(1), + amount: "5".to_string(), + asset: "XLM".to_string(), + memo: None, + }]; + + let checkpoint = init_checkpoint(&csv_path, &wallet, "testnet", rows, 100).unwrap(); + save_checkpoint_atomic(&checkpoint_path, &checkpoint).unwrap(); + + let loaded = load_checkpoint(&checkpoint_path).unwrap().unwrap(); + assert_eq!(loaded.rows.len(), 1); + assert_eq!(loaded.next_sequence, 100); + } + + #[test] + fn resume_skips_confirmed_rows() { + let rows: Vec = vec![ + BatchRowState { + row_index: 1, + destination: sample_key(1), + amount: "1".to_string(), + asset: "XLM".to_string(), + memo: None, + status: RowStatus::Confirmed { + tx_hash: "abc".to_string(), + }, + }, + BatchRowState { + row_index: 2, + destination: sample_key(1), + amount: "2".to_string(), + asset: "XLM".to_string(), + memo: None, + status: RowStatus::Pending, + }, + ]; + + let chunks = chunk_pending_row_indices(&rows); + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0], vec![1]); + } + + #[test] + fn simulated_crash_resume_restores_pending_only() { + let dir = TempDir::new().unwrap(); + let csv_path = dir.path().join("airdrop.csv"); + fs::write( + &csv_path, + "destination,amount,asset\n\ + GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF,1,XLM\n\ + GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHE,2,XLM\n", + ) + .unwrap(); + + let wallet = test_wallet("payer", &sample_key(1)); + + let parsed = parse_batch_csv(&csv_path).unwrap(); + let mut checkpoint = init_checkpoint(&csv_path, &wallet, "testnet", parsed, 10).unwrap(); + checkpoint.rows[0].status = RowStatus::Confirmed { + tx_hash: "tx1".to_string(), + }; + checkpoint.next_sequence = 11; + + let checkpoint_path = checkpoint_path_for(&csv_path); + save_checkpoint_atomic(&checkpoint_path, &checkpoint).unwrap(); + + let resumed = load_checkpoint(&checkpoint_path).unwrap().unwrap(); + let chunks = chunk_pending_row_indices(&resumed.rows); + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0], vec![1]); + } + + #[test] + fn failed_rows_do_not_block_remaining_chunks() { + let rows: Vec = (0..3) + .map(|i| BatchRowState { + row_index: i + 1, + destination: sample_key(1), + amount: "1".to_string(), + asset: "XLM".to_string(), + memo: None, + status: if i == 0 { + RowStatus::Failed { + error: "insufficient fee".to_string(), + attempts: 3, + } + } else { + RowStatus::Pending + }, + }) + .collect(); + + let chunks = chunk_pending_row_indices(&rows); + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0], vec![1, 2]); + } + + #[test] + fn dry_run_cost_estimate_accounts_for_all_pending_rows() { + let rows: Vec = (0..250) + .map(|i| BatchRowState { + row_index: i + 1, + destination: sample_key(1), + amount: "1".to_string(), + asset: "XLM".to_string(), + memo: None, + status: if i < 50 { + RowStatus::Confirmed { + tx_hash: format!("tx{i}"), + } + } else { + RowStatus::Pending + }, + }) + .collect(); + + let report = estimate_batch_cost(&rows); + assert_eq!(report.row_count, 250); + assert_eq!(report.transaction_count, 2); + assert_eq!(report.estimated_fee_stroops, 200 * BASE_FEE_PER_OP_STROOPS); + assert_eq!(report.total_payment_amounts.get("XLM"), Some(&200.0)); + } + + #[test] + fn mark_chunk_failed_records_attempts() { + let dir = TempDir::new().unwrap(); + let csv_path = dir.path().join("fail.csv"); + let wallet = test_wallet("payer", &sample_key(1)); + let rows = vec![BatchRecipientRow { + row_index: 1, + destination: sample_key(1), + amount: "1".to_string(), + asset: "XLM".to_string(), + memo: None, + }]; + let mut checkpoint = init_checkpoint(&csv_path, &wallet, "testnet", rows, 1).unwrap(); + mark_chunk_failed(&mut checkpoint, &[0], "tx_insufficient_fee", 3); + + match &checkpoint.rows[0].status { + RowStatus::Failed { attempts, error } => { + assert_eq!(*attempts, 3); + assert!(error.contains("insufficient_fee")); + } + other => panic!("expected failed status, got {other:?}"), + } + } }