quantus-cli 2.1.1

//! Collect Rewards Library Functions
//!
//! This module provides library-friendly functions for collecting miner rewards
//! that can be used by external crates (like quantus-sdk) via FFI.
//!
//! The main function `collect_rewards` handles the entire flow:
//! 1. Query Subsquid for pending transfers
//! 2. Generate ZK proofs for each transfer
//! 3. Aggregate proofs into batches
//! 4. Submit withdrawal transactions to chain
//!
//! This is designed to be called from the SDK without needing CLI-specific features.

use crate::{
	chain::{
		client::QuantusClient,
		quantus_subxt::{self as quantus_node, api::wormhole},
	},
	cli::wormhole::{
		compute_merkle_positions, parse_secret_hex as parse_secret_hex_str, ZkMerkleProofRpc,
	},
	subsquid::{
		compute_address_hash, get_hash_prefix, SubsquidClient, Transfer, TransferQueryParams,
	},
	wormhole_lib,
	wormhole_lib::{compute_output_amount, NATIVE_ASSET_ID, VOLUME_FEE_BPS},
};

use plonky2::plonk::proof::ProofWithPublicInputs;
use qp_rusty_crystals_hdwallet::{derive_wormhole_from_mnemonic, QUANTUS_WORMHOLE_CHAIN_ID};
use qp_wormhole_aggregator::{
	config::CircuitBinsConfig, private_batch::prover::PrivateBatchProver,
};
use qp_zk_circuits_common::circuit::{C, D, F};
use sp_core::crypto::{AccountId32, Ss58Codec};
use std::path::Path;
use subxt::{
	ext::{
		codec::Encode,
		jsonrpsee::{core::client::ClientT, rpc_params},
	},
	tx::TxStatus,
};

/// Result type for collect rewards operations
pub type Result<T> = std::result::Result<T, CollectRewardsError>;

const MAX_PRE_SUBMISSION_NULLIFIER_QUERIES: usize = 7;

/// Error type for collect rewards operations
#[derive(Debug, Clone)]
pub struct CollectRewardsError {
	pub message: String,
}

impl std::fmt::Display for CollectRewardsError {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		write!(f, "{}", self.message)
	}
}

impl std::error::Error for CollectRewardsError {}

impl From<String> for CollectRewardsError {
	fn from(message: String) -> Self {
		Self { message }
	}
}

impl From<crate::error::QuantusError> for CollectRewardsError {
	fn from(e: crate::error::QuantusError) -> Self {
		Self { message: e.to_string() }
	}
}

/// Information about a pending transfer found via Subsquid
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct PendingTransfer {
	/// Block height where the transfer was included
	pub block_height: i64,
	/// Block hash (hex encoded)
	pub block_hash: String,
	/// Transfer amount in planck
	pub amount: u128,
	/// Leaf index in the ZK tree
	pub leaf_index: u64,
	/// Transfer count (for nullifier computation)
	pub transfer_count: u64,
	/// Wormhole address that received the transfer (SS58)
	pub wormhole_address: String,
	/// Funding account that sent the transfer (SS58)
	pub funding_account: String,
}

/// Result of querying pending transfers
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct QueryPendingTransfersResult {
	/// The wormhole address that was queried (SS58)
	pub wormhole_address: String,
	/// List of pending transfers
	pub transfers: Vec<PendingTransfer>,
	/// Total amount available for withdrawal (in planck)
	pub total_available: u128,
}

/// Progress callback trait for reporting status during collect_rewards
#[allow(dead_code)]
pub trait ProgressCallback: Send + Sync {
	/// Called when a step starts
	fn on_step(&self, step: &str, details: &str);
	/// Called when a proof is generated
	fn on_proof_generated(&self, index: usize, total: usize);
	/// Called when a batch is submitted
	fn on_batch_submitted(&self, batch_index: usize, total_batches: usize, amount_withdrawn: u128);
	/// Called on error
	fn on_error(&self, message: &str);
}

/// No-op progress callback for when caller doesn't need updates
#[allow(dead_code)]
pub struct NoOpProgress;
impl ProgressCallback for NoOpProgress {
	fn on_step(&self, _step: &str, _details: &str) {}
	fn on_proof_generated(&self, _index: usize, _total: usize) {}
	fn on_batch_submitted(&self, _batch_index: usize, _total_batches: usize, _amount: u128) {}
	fn on_error(&self, _message: &str) {}
}

/// Information about a completed withdrawal batch
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct WithdrawalBatch {
	/// Block hash where the transaction was included
	pub block_hash: String,
	/// Transaction/extrinsic hash
	pub tx_hash: String,
	/// Amount withdrawn in this batch (planck)
	pub amount_withdrawn: u128,
	/// Number of proofs in this batch
	pub num_proofs: usize,
}

/// Result of the full collect_rewards operation
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct CollectRewardsResult {
	/// The wormhole address that was withdrawn from
	pub wormhole_address: String,
	/// Destination address that received the funds
	pub destination_address: String,
	/// Total amount withdrawn (planck)
	pub total_withdrawn: u128,
	/// Information about each batch submitted
	pub batches: Vec<WithdrawalBatch>,
	/// Number of transfers processed
	pub transfers_processed: usize,
}

/// Wormhole credential - either a mnemonic for HD derivation or a direct secret
#[derive(Debug, Clone)]
pub enum WormholeCredential {
	/// BIP39 mnemonic with HD derivation
	Mnemonic {
		/// 24-word BIP39 mnemonic phrase
		phrase: String,
		/// Wormhole address index for HD derivation (usually 0)
		wormhole_index: usize,
	},
	/// Direct wormhole secret (no HD derivation)
	Secret {
		/// 32-byte hex-encoded secret (with or without 0x prefix)
		hex: String,
	},
}

/// Resolve a `WormholeCredential` into `(ss58_address, address_bytes, secret_bytes)`.
pub fn resolve_credential(credential: &WormholeCredential) -> Result<(String, [u8; 32], [u8; 32])> {
	match credential {
		WormholeCredential::Mnemonic { phrase, wormhole_index } => {
			let path = format!("m/44'/{}/0'/0'/{}'", QUANTUS_WORMHOLE_CHAIN_ID, wormhole_index);
			let wormhole_pair = derive_wormhole_from_mnemonic(phrase, None, &path)
				.map_err(|e| CollectRewardsError::from(format!("HD derivation failed: {:?}", e)))?;
			let address_bytes: [u8; 32] = *wormhole_pair.address();
			let secret_bytes: [u8; 32] = *wormhole_pair.secret().as_bytes();
			Ok((AccountId32::from(address_bytes).to_ss58check(), address_bytes, secret_bytes))
		},
		WormholeCredential::Secret { hex } => {
			let secret_bytes = parse_secret_hex_str(hex)
				.map_err(|e| CollectRewardsError::from(format!("Invalid secret: {}", e)))?;
			let address_bytes = wormhole_lib::compute_wormhole_address(&secret_bytes)
				.map_err(|e| CollectRewardsError::from(e.message))?;
			Ok((AccountId32::from(address_bytes).to_ss58check(), address_bytes, secret_bytes))
		},
	}
}

/// Configuration for collect_rewards
#[derive(Debug, Clone)]
pub struct CollectRewardsConfig {
	/// Wormhole credential - either mnemonic or direct secret
	pub credential: WormholeCredential,
	/// Destination address (SS58) to receive withdrawn funds
	pub destination_address: String,
	/// Subsquid GraphQL endpoint URL
	pub subsquid_url: String,
	/// Chain RPC node URL
	pub node_url: String,
	/// Path to circuit binary directory (verifier/common + private/public-batch bins).
	pub bins_dir: String,
	/// Optional: specific amount to withdraw (None = withdraw all)
	pub amount: Option<u128>,
	/// If true, only query and return info without submitting transactions
	pub dry_run: bool,
	/// Optional: specific block number to use for proofs (None = use latest)
	pub at_block: Option<u32>,
}

/// Collect miner rewards by querying Subsquid, generating proofs, and submitting withdrawals.
///
/// This is the main entry point for the SDK to collect rewards. It handles the entire flow:
/// 1. Derives wormhole address from mnemonic
/// 2. Queries Subsquid for pending transfers to that address
/// 3. Filters spent nullifiers against the current best chain state
/// 4. Generates ZK proofs for selected transfers
/// 5. Aggregates proofs into batches (size determined by circuit config)
/// 6. Submits withdrawal transactions to chain
///
/// # Arguments
/// * `config` - Configuration for the operation
/// * `progress` - Optional callback for progress updates
///
/// # Returns
/// Information about the withdrawal including total amount and batch details
pub async fn collect_rewards<P: ProgressCallback>(
	config: CollectRewardsConfig,
	progress: &P,
) -> Result<CollectRewardsResult> {
	let (wormhole_address, wormhole_address_bytes, wormhole_secret_bytes) =
		resolve_credential(&config.credential)?;
	progress.on_step("derive", &format!("Derived wormhole address: {}", wormhole_address));

	// Parse destination address
	let destination_bytes = parse_ss58_address(&config.destination_address)?;

	// Step 2: Query Subsquid for pending transfers
	progress.on_step("query", "Querying Subsquid for pending transfers");

	let subsquid_client = SubsquidClient::new(config.subsquid_url.clone())?;
	let address_hash = compute_address_hash(&wormhole_address_bytes);
	let prefix = get_hash_prefix(&address_hash, 8);

	let params = TransferQueryParams::new();
	let transfers = subsquid_client
		.query_all_transfers_by_prefix(Some(vec![prefix]), None, params)
		.await?;

	// Filter to only transfers TO our wormhole address
	let incoming_transfers: Vec<_> =
		transfers.into_iter().filter(|t| t.to_hash == address_hash).collect();

	if incoming_transfers.is_empty() {
		return Ok(CollectRewardsResult {
			wormhole_address,
			destination_address: config.destination_address,
			total_withdrawn: 0,
			batches: vec![],
			transfers_processed: 0,
		});
	}

	// Step 2b: Connect to chain for authoritative nullifier checks and proof submission
	progress.on_step("connect", "Connecting to chain");

	let quantus_client = QuantusClient::new(&config.node_url)
		.await
		.map_err(|e| CollectRewardsError::from(format!("Failed to connect to node: {}", e)))?;

	// Step 2c: Filter out already-spent transfers against a pinned best-chain snapshot
	progress.on_step("nullifiers", "Checking current on-chain nullifiers");

	let unspent_transfers = filter_unspent_transfers_onchain(
		&incoming_transfers,
		&wormhole_secret_bytes,
		&quantus_client,
	)
	.await?;

	if unspent_transfers.is_empty() {
		progress
			.on_step("complete", "All transfers have already been withdrawn (nullifiers spent)");
		return Ok(CollectRewardsResult {
			wormhole_address,
			destination_address: config.destination_address,
			total_withdrawn: 0,
			batches: vec![],
			transfers_processed: 0,
		});
	}

	// Calculate total available (only unspent)
	let mut total_available: u128 = 0;
	for t in &unspent_transfers {
		let amount = parse_transfer_amount(&t.amount, &format!("transfer {}", t.id))?;
		total_available = checked_add_amount(total_available, amount, "total available transfers")?;
	}

	// Determine amount to withdraw
	let withdraw_amount = config.amount.unwrap_or(total_available);
	if withdraw_amount > total_available {
		return Err(CollectRewardsError::from(format!(
			"Requested {} but only {} available (after filtering spent nullifiers)",
			withdraw_amount, total_available
		)));
	}

	// Parse amounts for sorting (fail early if any are invalid)
	let mut transfers_with_amounts: Vec<(Transfer, u128)> = Vec::new();
	for t in unspent_transfers {
		let amt = parse_transfer_amount(&t.amount, &format!("transfer {}", t.id))?;
		transfers_with_amounts.push((t, amt));
	}

	// Sort by amount descending (largest first)
	transfers_with_amounts.sort_by_key(|k| std::cmp::Reverse(k.1));

	let mut selected_transfers = Vec::new();
	let mut selected_total: u128 = 0;
	for (t, amt) in transfers_with_amounts {
		if selected_total >= withdraw_amount {
			break;
		}
		selected_transfers.push(t);
		selected_total = checked_add_amount(selected_total, amt, "selected transfers")?;
	}

	if config.dry_run {
		return Ok(CollectRewardsResult {
			wormhole_address,
			destination_address: config.destination_address,
			total_withdrawn: 0,
			batches: vec![],
			transfers_processed: selected_transfers.len(),
		});
	}

	// Get block for proofs - either specific block or latest
	let proof_block = if let Some(block_num) = config.at_block {
		// Fetch block hash for the specified block number
		use subxt::ext::jsonrpsee::{core::client::ClientT, rpc_params};
		let block_hash: Option<subxt::utils::H256> = quantus_client
			.rpc_client()
			.request("chain_getBlockHash", rpc_params![block_num])
			.await
			.map_err(|e| {
				CollectRewardsError::from(format!(
					"Failed to get block hash for block {}: {}",
					block_num, e
				))
			})?;
		let block_hash = block_hash
			.ok_or_else(|| CollectRewardsError::from(format!("Block {} not found", block_num)))?;
		quantus_client
			.client()
			.blocks()
			.at(block_hash)
			.await
			.map_err(|e| CollectRewardsError::from(format!("Failed to get block: {}", e)))?
	} else {
		// Prove against the latest finalized block. Best-block proofs can be
		// invalidated by reorgs before finality (same class as recursive flows).
		use subxt::ext::jsonrpsee::{core::client::ClientT, rpc_params};
		let finalized_hash: subxt::utils::H256 = quantus_client
			.rpc_client()
			.request("chain_getFinalizedHead", rpc_params![])
			.await
			.map_err(|e| {
				CollectRewardsError::from(format!("Failed to get finalized block hash: {}", e))
			})?;
		quantus_client.client().blocks().at(finalized_hash).await.map_err(|e| {
			CollectRewardsError::from(format!("Failed to get finalized block: {}", e))
		})?
	};
	let proof_block_hash = proof_block.hash();

	// Step 4: Generate proofs
	progress.on_step("proofs", &format!("Generating {} proofs", selected_transfers.len()));

	let bins_dir = Path::new(&config.bins_dir);
	let num_transfers = selected_transfers.len();
	let mut proof_bytes_list: Vec<Vec<u8>> = Vec::new();

	for (i, transfer) in selected_transfers.iter().enumerate() {
		progress.on_proof_generated(i + 1, num_transfers);

		let leaf_index: u64 = transfer.leaf_index.parse().map_err(|_| {
			CollectRewardsError::from(format!("Invalid leaf_index: {}", transfer.leaf_index))
		})?;

		// Fetch ZK Merkle proof from chain
		let proof_params = rpc_params![leaf_index, proof_block_hash];
		let zk_proof: Option<ZkMerkleProofRpc> = quantus_client
			.rpc_client()
			.request("zkTree_getMerkleProof", proof_params)
			.await
			.map_err(|e| {
				CollectRewardsError::from(format!(
					"Failed to get ZK Merkle proof for leaf {}: {}",
					leaf_index, e
				))
			})?;

		let zk_proof = zk_proof.ok_or_else(|| {
			CollectRewardsError::from(format!(
				"No ZK Merkle proof found for leaf_index {}",
				leaf_index
			))
		})?;

		// Decode transfer data from leaf
		let (leaf_to_account, transfer_count, _leaf_asset_id, _leaf_raw_amount) =
			decode_full_leaf_data(&zk_proof.leaf_data)?;
		let input_amount = decode_input_amount_from_leaf(&zk_proof.leaf_data)?;
		let output_amount = compute_output_amount(input_amount, VOLUME_FEE_BPS);

		// Compute sorted siblings and positions from unsorted RPC data
		let (sorted_siblings, positions) =
			compute_merkle_positions(&zk_proof.siblings, zk_proof.leaf_hash);

		// Get block header data
		let header = proof_block.header();
		let parent_hash: [u8; 32] = header.parent_hash.0;
		let state_root: [u8; 32] = header.state_root.0;
		let extrinsics_root: [u8; 32] = header.extrinsics_root.0;
		let digest = header.digest.encode();
		let block_number = header.number;

		if leaf_to_account != wormhole_address_bytes {
			return Err(CollectRewardsError::from(format!(
				"Leaf to_account mismatch: expected 0x{}, got 0x{}",
				hex::encode(wormhole_address_bytes),
				hex::encode(leaf_to_account)
			)));
		}

		// generate_proof zeroizes input.secret before returning; each iteration
		// rebuilds the input from wormhole_secret_bytes.
		let mut input = wormhole_lib::ProofGenerationInput {
			secret: wormhole_secret_bytes,
			transfer_count,
			wormhole_address: wormhole_address_bytes,
			input_amount,
			block_hash: proof_block_hash.0,
			block_number,
			parent_hash,
			state_root,
			extrinsics_root,
			digest: digest.clone(),
			zk_tree_root: zk_proof.root,
			zk_merkle_siblings: sorted_siblings,
			zk_merkle_positions: positions,
			exit_account_1: destination_bytes,
			exit_account_2: [0u8; 32],
			output_amount_1: output_amount,
			output_amount_2: 0,
			volume_fee_bps: VOLUME_FEE_BPS,
			asset_id: NATIVE_ASSET_ID,
		};

		// Generate proof
		let prover_path = bins_dir.join("prover.bin");
		let common_path = bins_dir.join("common.bin");
		let result = wormhole_lib::generate_proof(&mut input, &prover_path, &common_path)
			.map_err(|e| CollectRewardsError::from(e.message))?;

		proof_bytes_list.push(result.proof_bytes);
	}

	// Step 5: Aggregate and submit in batches
	// Load circuit config to get actual batch size (may differ from DEFAULT_NUM_LEAF_PROOFS
	// if QP_NUM_LEAF_PROOFS was set during build)
	let agg_config = CircuitBinsConfig::load(bins_dir).map_err(|e| {
		CollectRewardsError::from(format!("Failed to load circuit bins config: {}", e))
	})?;
	let batch_size = agg_config.num_leaf_proofs;

	let batches: Vec<Vec<Vec<u8>>> =
		proof_bytes_list.chunks(batch_size).map(|chunk| chunk.to_vec()).collect();

	progress.on_step("submit", &format!("Submitting {} batch(es)", batches.len()));

	let mut total_withdrawn: u128 = 0;
	let mut batch_results: Vec<WithdrawalBatch> = Vec::new();

	for (batch_idx, batch_proofs) in batches.iter().enumerate() {
		// Aggregate proofs
		let aggregated_proof = aggregate_proof_bytes(batch_proofs, bins_dir)?;

		// Submit to chain
		let (block_hash, tx_hash, transfer_events) =
			submit_and_get_events(&quantus_client, aggregated_proof, bins_dir).await?;

		let batch_amount = transfer_events.iter().try_fold(0_u128, |acc, e| {
			checked_add_amount(acc, e.amount, "batch withdrawal events")
		})?;
		total_withdrawn = checked_add_amount(total_withdrawn, batch_amount, "total withdrawn")?;

		progress.on_batch_submitted(batch_idx + 1, batches.len(), batch_amount);

		batch_results.push(WithdrawalBatch {
			block_hash: hex::encode(block_hash.0),
			tx_hash: hex::encode(tx_hash.0),
			amount_withdrawn: batch_amount,
			num_proofs: batch_proofs.len(),
		});
	}

	Ok(CollectRewardsResult {
		wormhole_address,
		destination_address: config.destination_address,
		total_withdrawn,
		batches: batch_results,
		transfers_processed: selected_transfers.len(),
	})
}

/// Query pending transfers for a wormhole address via Subsquid.
///
/// This function queries the Subsquid indexer for transfers that were sent
/// to the wormhole address derived from the given mnemonic. This allows
/// discovery of mining rewards even if they weren't tracked locally.
///
/// # Arguments
/// * `mnemonic` - The 24-word BIP39 mnemonic phrase
/// * `wormhole_index` - The wormhole address index (default: 0)
/// * `subsquid_url` - The Subsquid GraphQL endpoint URL
///
/// # Returns
/// The wormhole address, list of pending transfers, and total available balance.
#[allow(dead_code)]
pub async fn query_pending_transfers(
	mnemonic: &str,
	wormhole_index: usize,
	subsquid_url: &str,
) -> Result<QueryPendingTransfersResult> {
	// Derive wormhole secret using HD path for miner rewards (purpose = 1)
	let path = format!("m/44'/{}/0'/1'/{}'", QUANTUS_WORMHOLE_CHAIN_ID, wormhole_index);
	let wormhole_secret = derive_wormhole_from_mnemonic(mnemonic, None, &path)
		.map_err(|e| CollectRewardsError::from(format!("HD derivation failed: {:?}", e)))?;

	let wormhole_address = AccountId32::from(*wormhole_secret.address()).to_ss58check();

	// Query Subsquid using privacy-preserving hash prefix
	let subsquid_client = SubsquidClient::new(subsquid_url.to_string())?;
	let address_hash = compute_address_hash(wormhole_secret.address());
	let prefix = get_hash_prefix(&address_hash, 8); // 8 hex chars for good privacy

	let params = TransferQueryParams::new();
	let transfers = subsquid_client
		.query_all_transfers_by_prefix(Some(vec![prefix]), None, params)
		.await?;

	// Filter to only transfers TO our wormhole address
	let incoming_transfers: Vec<_> =
		transfers.into_iter().filter(|t| t.to_hash == address_hash).collect();

	let secret_bytes: [u8; 32] = *wormhole_secret.secret().as_bytes();
	let unspent_transfers =
		filter_unspent_transfers_by_indexer(&incoming_transfers, &secret_bytes, &subsquid_client)
			.await?;

	let mut total_available: u128 = 0;
	let mut pending = Vec::new();

	for t in &unspent_transfers {
		let ctx = format!("transfer {}", t.id);
		let amount = parse_transfer_amount(&t.amount, &ctx)?;
		let leaf_index = parse_leaf_index(&t.leaf_index, &ctx)?;
		let transfer_count = parse_transfer_count(&t.transfer_count, &ctx)?;

		total_available = checked_add_amount(total_available, amount, "pending transfers")?;

		pending.push(PendingTransfer {
			block_height: t.block_height,
			block_hash: t.block_id.clone(),
			amount,
			leaf_index,
			transfer_count,
			wormhole_address: wormhole_address.clone(),
			funding_account: t.from_id.clone(),
		});
	}

	Ok(QueryPendingTransfersResult { wormhole_address, transfers: pending, total_available })
}

/// Query pending transfers for an already-known wormhole address.
///
/// Use this when you already have the wormhole address and don't need to derive it.
///
/// Address-only discovery cannot reconcile spent nullifiers. When any incoming
/// transfers are present, this returns an error directing callers to use a
/// secret-bearing API (`query_pending_transfers` / `collect_rewards`).
///
/// # Arguments
/// * `wormhole_address_bytes` - The 32-byte wormhole address
/// * `subsquid_url` - The Subsquid GraphQL endpoint URL
#[allow(dead_code)]
pub async fn query_pending_transfers_for_address(
	wormhole_address_bytes: &[u8; 32],
	subsquid_url: &str,
) -> Result<QueryPendingTransfersResult> {
	let wormhole_address = AccountId32::from(*wormhole_address_bytes).to_ss58check();

	// Query Subsquid using privacy-preserving hash prefix
	let subsquid_client = SubsquidClient::new(subsquid_url.to_string())?;
	let address_hash = compute_address_hash(wormhole_address_bytes);
	let prefix = get_hash_prefix(&address_hash, 8);

	let params = TransferQueryParams::new();
	let transfers = subsquid_client
		.query_all_transfers_by_prefix(Some(vec![prefix]), None, params)
		.await?;

	// Filter to only transfers TO our wormhole address
	let incoming_transfers: Vec<_> =
		transfers.into_iter().filter(|t| t.to_hash == address_hash).collect();

	if !incoming_transfers.is_empty() {
		return Err(CollectRewardsError::from(
			"Cannot determine available withdrawals for an address without the wormhole secret; use query_pending_transfers or collect_rewards so spent nullifiers can be reconciled"
				.to_string(),
		));
	}

	Ok(QueryPendingTransfersResult { wormhole_address, transfers: vec![], total_available: 0 })
}

// ============================================================================
// Internal Helper Functions
// ============================================================================

fn checked_add_amount(acc: u128, amount: u128, context: &str) -> Result<u128> {
	acc.checked_add(amount).ok_or_else(|| {
		CollectRewardsError::from(format!(
			"Transfer amount overflow while accumulating {}",
			context
		))
	})
}

/// Parse a transfer amount string to u128
fn parse_transfer_amount(amount_str: &str, context: &str) -> Result<u128> {
	amount_str.parse::<u128>().map_err(|e| {
		CollectRewardsError::from(format!(
			"Invalid transfer amount '{}' in {}: {}",
			amount_str, context, e
		))
	})
}

/// Parse a leaf index string to u64
fn parse_leaf_index(leaf_index_str: &str, context: &str) -> Result<u64> {
	leaf_index_str.parse::<u64>().map_err(|e| {
		CollectRewardsError::from(format!(
			"Invalid leaf_index '{}' in {}: {}",
			leaf_index_str, context, e
		))
	})
}

/// Parse a transfer count string to u64
fn parse_transfer_count(transfer_count_str: &str, context: &str) -> Result<u64> {
	transfer_count_str.parse::<u64>().map_err(|e| {
		CollectRewardsError::from(format!(
			"Invalid transfer_count '{}' in {}: {}",
			transfer_count_str, context, e
		))
	})
}

/// Parse an SS58 address to 32 bytes
fn parse_ss58_address(address: &str) -> Result<[u8; 32]> {
	let account = AccountId32::from_ss58check(address)
		.map_err(|e| CollectRewardsError::from(format!("Invalid SS58 address: {:?}", e)))?;
	Ok(account.into())
}

/// Decode all fields from SCALE-encoded ZkLeaf data.
/// Returns (to_account, transfer_count, asset_id, raw_amount_u128)
fn decode_full_leaf_data(leaf_data: &[u8]) -> Result<([u8; 32], u64, u32, u128)> {
	if leaf_data.len() < 60 {
		return Err(CollectRewardsError::from(format!(
			"Invalid leaf data length: expected at least 60 bytes, got {}",
			leaf_data.len()
		)));
	}

	let to_account: [u8; 32] = leaf_data[0..32]
		.try_into()
		.map_err(|_| CollectRewardsError::from("Failed to extract to_account".to_string()))?;

	let transfer_count =
		u64::from_le_bytes(leaf_data[32..40].try_into().map_err(|_| {
			CollectRewardsError::from("Failed to extract transfer_count".to_string())
		})?);

	let asset_id = u32::from_le_bytes(
		leaf_data[40..44]
			.try_into()
			.map_err(|_| CollectRewardsError::from("Failed to extract asset_id".to_string()))?,
	);

	let amount = u128::from_le_bytes(
		leaf_data[44..60]
			.try_into()
			.map_err(|_| CollectRewardsError::from("Failed to extract amount".to_string()))?,
	);

	Ok((to_account, transfer_count, asset_id, amount))
}

/// Decode the input amount from SCALE-encoded ZkLeaf data (quantized).
fn decode_input_amount_from_leaf(leaf_data: &[u8]) -> Result<u32> {
	if leaf_data.len() < 60 {
		return Err(CollectRewardsError::from(format!(
			"Invalid leaf data length: expected at least 60 bytes, got {}",
			leaf_data.len()
		)));
	}

	let amount_bytes: [u8; 16] = leaf_data[44..60]
		.try_into()
		.map_err(|_| CollectRewardsError::from("Failed to extract amount bytes".to_string()))?;

	let raw_amount = u128::from_le_bytes(amount_bytes);

	// Quantize: divide by 10^10 to get 2 decimal places
	const AMOUNT_SCALE_DOWN_FACTOR: u128 = 10_000_000_000;
	let quantized = (raw_amount / AMOUNT_SCALE_DOWN_FACTOR) as u32;

	Ok(quantized)
}

/// Aggregate proof bytes into a single aggregated proof
fn aggregate_proof_bytes(proof_bytes_list: &[Vec<u8>], bins_dir: &Path) -> Result<Vec<u8>> {
	// Load config to validate
	let agg_config = CircuitBinsConfig::load(bins_dir).map_err(|e| {
		CollectRewardsError::from(format!("Failed to load circuit bins config: {}", e))
	})?;

	if proof_bytes_list.len() > agg_config.num_leaf_proofs {
		return Err(CollectRewardsError::from(format!(
			"Too many proofs: {} provided, max {} supported",
			proof_bytes_list.len(),
			agg_config.num_leaf_proofs
		)));
	}

	let common_bytes = std::fs::read(bins_dir.join("common.bin"))
		.map_err(|e| CollectRewardsError::from(format!("Failed to read common.bin: {}", e)))?;
	let verifier_bytes = std::fs::read(bins_dir.join("verifier.bin"))
		.map_err(|e| CollectRewardsError::from(format!("Failed to read verifier.bin: {}", e)))?;
	let leaf = qp_wormhole_aggregator::common::utils::load_canonical_leaf_verifier_data(
		&common_bytes,
		&verifier_bytes,
	)
	.map_err(|e| CollectRewardsError::from(format!("Failed to load leaf circuit data: {}", e)))?;

	let mut proofs = Vec::with_capacity(proof_bytes_list.len());
	for proof_bytes in proof_bytes_list {
		let proof = ProofWithPublicInputs::<F, C, D>::from_bytes(proof_bytes.clone(), &leaf.common)
			.map_err(|e| {
				CollectRewardsError::from(format!("Failed to deserialize proof: {:?}", e))
			})?;
		proofs.push(proof);
	}

	let prover = PrivateBatchProver::new_from_binaries_dir(bins_dir).map_err(|e| {
		CollectRewardsError::from(format!("Failed to load private-batch prover: {}", e))
	})?;

	let aggregated_proof = prover
		.aggregate(proofs)
		.map_err(|e| CollectRewardsError::from(format!("Aggregation failed: {}", e)))?;

	Ok(aggregated_proof.to_bytes())
}

/// Submit aggregated proof and collect events
async fn submit_and_get_events(
	quantus_client: &QuantusClient,
	proof_bytes: Vec<u8>,
	bins_dir: &Path,
) -> Result<(subxt::utils::H256, subxt::utils::H256, Vec<wormhole::events::NativeTransferred>)> {
	// Verify locally first

	let verifier = crate::batch_verifier::load_private_batch_verifier(bins_dir)
		.map_err(|e| CollectRewardsError::from(format!("Failed to load verifier: {}", e)))?;

	let proof = qp_wormhole_verifier::ProofWithPublicInputs::<
		qp_wormhole_verifier::F,
		qp_wormhole_verifier::C,
		{ qp_wormhole_verifier::D },
	>::from_bytes(proof_bytes.clone(), &verifier.circuit_data.common)
	.map_err(|e| CollectRewardsError::from(format!("Failed to deserialize proof: {}", e)))?;

	verifier
		.verify(proof.clone())
		.map_err(|e| CollectRewardsError::from(format!("Local verification failed: {}", e)))?;

	// Parse public inputs to do pre-submission validation
	let inputs = qp_wormhole_verifier::parse_private_batch_public_inputs(&proof).map_err(|e| {
		CollectRewardsError::from(format!("Failed to parse public inputs: {:?}", e))
	})?;
	validate_pre_submission_nullifier_count(inputs.nullifiers.len())?;

	// Check asset_id (must be 0 for native)
	if inputs.asset_id != 0 {
		return Err(CollectRewardsError::from(format!(
			"Pre-validation failed: NonNativeAssetNotSupported (asset_id = {}, expected 0)",
			inputs.asset_id
		)));
	}

	// Must match on-chain VolumeFeeRateBps (currently 4 bps).
	const EXPECTED_VOLUME_FEE_BPS: u32 = crate::wormhole_lib::VOLUME_FEE_BPS;
	if inputs.volume_fee_bps != EXPECTED_VOLUME_FEE_BPS {
		return Err(CollectRewardsError::from(format!(
			"Pre-validation failed: InvalidVolumeFeeRate (got {}, expected {})",
			inputs.volume_fee_bps, EXPECTED_VOLUME_FEE_BPS
		)));
	}

	// Check block hash exists and matches
	let proof_block_number = inputs.block_data.block_number;
	use subxt::ext::jsonrpsee::{core::client::ClientT, rpc_params};
	let chain_block_hash: Option<subxt::utils::H256> = quantus_client
		.rpc_client()
		.request("chain_getBlockHash", rpc_params![proof_block_number])
		.await
		.map_err(|e| {
			CollectRewardsError::from(format!(
				"Failed to query block hash for block {}: {}",
				proof_block_number, e
			))
		})?;

	match chain_block_hash {
		None => {
			return Err(CollectRewardsError::from(format!(
				"Pre-validation failed: BlockNotFound - Block {} does not exist on chain (may be pruned or too old)",
				proof_block_number
			)));
		},
		Some(chain_hash) => {
			let proof_hash: [u8; 32] = *inputs.block_data.block_hash;
			if chain_hash.0 != proof_hash {
				return Err(CollectRewardsError::from(format!(
					"Pre-validation failed: InvalidPublicInputs - Block hash mismatch!\n  Chain has: 0x{}\n  Proof has: 0x{}",
					hex::encode(chain_hash.0),
					hex::encode(proof_hash)
				)));
			}
		},
	}

	// Check nullifiers against one fresh best-chain snapshot immediately before submission
	let best_block = quantus_client.get_latest_block().await.map_err(|e| {
		CollectRewardsError::from(format!(
			"Failed to get best block for pre-submission nullifier check: {}",
			e
		))
	})?;
	let storage = quantus_client.client().storage().at(best_block);

	for (i, nullifier) in inputs.nullifiers.iter().enumerate() {
		let nullifier_bytes: [u8; 32] = (*nullifier).as_ref().try_into().map_err(|_| {
			CollectRewardsError::from(format!("Failed to convert nullifier {} to bytes", i))
		})?;

		let storage_key = quantus_node::api::storage().wormhole().used_nullifiers(nullifier_bytes);
		let is_used = storage.fetch(&storage_key).await.map_err(|e| {
			CollectRewardsError::from(format!(
				"Failed to query nullifier {} at best block {:?}: {}",
				i, best_block, e
			))
		})?;

		if matches!(is_used, Some(true)) {
			return Err(CollectRewardsError::from(format!(
				"Pre-validation failed: NullifierAlreadyUsed - Nullifier {} (0x{}) has already been spent",
				i,
				hex::encode(nullifier_bytes)
			)));
		}
	}

	// Submit unsigned tx
	let verify_tx = quantus_node::api::tx().wormhole().verify_private_batch(proof_bytes);

	let unsigned_tx =
		quantus_client.client().tx().create_unsigned(&verify_tx).map_err(|e| {
			CollectRewardsError::from(format!("Failed to create unsigned tx: {}", e))
		})?;

	let mut tx_progress = unsigned_tx.submit_and_watch().await.map_err(|e| {
		CollectRewardsError::from(format!("Failed to submit wormhole withdrawal: {}", e))
	})?;

	// Wait for inclusion
	let (block_hash, tx_hash) = loop {
		match tx_progress.next().await {
			Some(Ok(TxStatus::InBestBlock(tx_in_block))) => {
				break (tx_in_block.block_hash(), tx_in_block.extrinsic_hash());
			},
			Some(Ok(TxStatus::InFinalizedBlock(tx_in_block))) => {
				break (tx_in_block.block_hash(), tx_in_block.extrinsic_hash());
			},
			Some(Ok(TxStatus::Error { message })) | Some(Ok(TxStatus::Invalid { message })) => {
				return Err(CollectRewardsError::from(format!(
					"Wormhole withdrawal was rejected before inclusion: {}",
					message
				)));
			},
			Some(Ok(_)) => continue,
			Some(Err(e)) => {
				return Err(CollectRewardsError::from(format!(
					"Wormhole withdrawal status error: {}",
					e
				)));
			},
			None => {
				return Err(CollectRewardsError::from(
					"Transaction stream ended unexpectedly".to_string(),
				));
			},
		}
	};

	// Collect events
	let block = quantus_client
		.client()
		.blocks()
		.at(block_hash)
		.await
		.map_err(|e| CollectRewardsError::from(format!("Failed to get block: {}", e)))?;

	let events = block
		.events()
		.await
		.map_err(|e| CollectRewardsError::from(format!("Failed to get events: {}", e)))?;

	let extrinsics = block
		.extrinsics()
		.await
		.map_err(|e| CollectRewardsError::from(format!("Failed to get extrinsics: {}", e)))?;

	let our_ext_idx = extrinsics
		.iter()
		.enumerate()
		.find(|(_, ext)| ext.hash() == tx_hash)
		.map(|(idx, _)| idx as u32)
		.ok_or_else(|| {
			CollectRewardsError::from("Could not find extrinsic in block".to_string())
		})?;

	use crate::chain::quantus_subxt::api::system::events::ExtrinsicFailed;

	let mut transfer_events = Vec::new();
	let mut found_proof_verified = false;
	let mut dispatch_error = None;

	for event_result in events.iter() {
		let event = event_result
			.map_err(|e| CollectRewardsError::from(format!("Failed to decode event: {}", e)))?;

		if let subxt::events::Phase::ApplyExtrinsic(ext_idx) = event.phase() {
			if ext_idx == our_ext_idx {
				if let Ok(Some(ExtrinsicFailed { dispatch_error: error, .. })) =
					event.as_event::<ExtrinsicFailed>()
				{
					dispatch_error = Some(crate::cli::common::format_dispatch_error(
						&error,
						&quantus_client.client().metadata(),
					));
				}
				if let Ok(Some(_)) = event.as_event::<wormhole::events::ProofVerified>() {
					found_proof_verified = true;
				}
				if let Ok(Some(transfer)) = event.as_event::<wormhole::events::NativeTransferred>()
				{
					transfer_events.push(transfer);
				}
			}
		}
	}

	if !found_proof_verified {
		if let Some(error) = dispatch_error {
			return Err(CollectRewardsError::from(format!(
				"Wormhole withdrawal failed on-chain: {}",
				error
			)));
		}
		return Err(CollectRewardsError::from(
			"Wormhole withdrawal was included but emitted neither Wormhole::ProofVerified nor System::ExtrinsicFailed"
				.to_string(),
		));
	}

	Ok((block_hash, tx_hash, transfer_events))
}

fn validate_pre_submission_nullifier_count(count: usize) -> Result<()> {
	if count > MAX_PRE_SUBMISSION_NULLIFIER_QUERIES {
		return Err(CollectRewardsError::from(format!(
			"Pre-validation failed: proof contains {} nullifiers; maximum supported per batch is {}",
			count, MAX_PRE_SUBMISSION_NULLIFIER_QUERIES
		)));
	}

	Ok(())
}

/// Filter transfers against spent nullifiers reported by the indexer.
async fn filter_unspent_transfers_by_indexer(
	transfers: &[Transfer],
	secret_bytes: &[u8; 32],
	subsquid_client: &SubsquidClient,
) -> Result<Vec<Transfer>> {
	if transfers.is_empty() {
		return Ok(vec![]);
	}

	let mut seen_nullifiers = std::collections::HashSet::new();
	let mut transfers_with_nullifiers = Vec::new();
	let mut nullifier_pairs = Vec::new();

	for transfer in transfers {
		let ctx = format!("transfer {}", transfer.id);
		let transfer_count = parse_transfer_count(&transfer.transfer_count, &ctx)?;
		let nullifier =
			wormhole_lib::compute_nullifier(secret_bytes, transfer_count).map_err(|e| {
				CollectRewardsError::from(format!("Failed to compute nullifier: {}", e.message))
			})?;

		if !seen_nullifiers.insert(nullifier) {
			continue;
		}

		let nullifier_hex = hex::encode(nullifier);
		let nullifier_hash = compute_address_hash(&nullifier);
		nullifier_pairs.push((nullifier_hex.clone(), nullifier_hash));
		transfers_with_nullifiers.push((transfer.clone(), nullifier_hex));
	}

	let spent = subsquid_client.check_nullifiers_spent(&nullifier_pairs, 8).await?;

	Ok(transfers_with_nullifiers
		.into_iter()
		.filter_map(|(transfer, nullifier_hex)| {
			(!spent.contains(&nullifier_hex)).then_some(transfer)
		})
		.collect())
}

/// Filter transfers against `UsedNullifiers` at one pinned best-chain snapshot.
async fn filter_unspent_transfers_onchain(
	transfers: &[Transfer],
	secret_bytes: &[u8; 32],
	quantus_client: &QuantusClient,
) -> Result<Vec<Transfer>> {
	if transfers.is_empty() {
		return Ok(vec![]);
	}

	let best_block = quantus_client.get_latest_block().await.map_err(|e| {
		CollectRewardsError::from(format!("Failed to get best block for nullifier check: {}", e))
	})?;
	let storage = quantus_client.client().storage().at(best_block);
	let mut seen_nullifiers = std::collections::HashSet::new();
	let mut unspent = Vec::new();

	for transfer in transfers {
		let ctx = format!("transfer {}", transfer.id);
		let transfer_count = parse_transfer_count(&transfer.transfer_count, &ctx)?;

		let nullifier =
			wormhole_lib::compute_nullifier(secret_bytes, transfer_count).map_err(|e| {
				CollectRewardsError::from(format!("Failed to compute nullifier: {}", e.message))
			})?;

		if !seen_nullifiers.insert(nullifier) {
			continue;
		}

		let storage_key = quantus_node::api::storage().wormhole().used_nullifiers(nullifier);
		let is_used = storage.fetch(&storage_key).await.map_err(|e| {
			CollectRewardsError::from(format!(
				"Failed to query nullifier for transfer_count {} at best block {:?}: {}",
				transfer_count, best_block, e
			))
		})?;

		if !matches!(is_used, Some(true)) {
			unspent.push(transfer.clone());
		}
	}

	Ok(unspent)
}

#[cfg(test)]
mod tests {
	use super::*;

	#[test]
	fn test_collect_rewards_error_from_string() {
		let err = CollectRewardsError::from("test error".to_string());
		assert_eq!(err.message, "test error");
		assert_eq!(format!("{}", err), "test error");
	}

	#[test]
	fn checked_add_amount_rejects_indexer_overflow() {
		let err = checked_add_amount(u128::MAX, 2, "pending transfers")
			.expect_err("untrusted transfer totals must not wrap on overflow");
		assert!(err.message.contains("overflow"), "unexpected overflow error: {}", err.message);
		assert_eq!(checked_add_amount(10, 5, "pending transfers").unwrap(), 15);
	}

	#[test]
	fn test_pre_submission_nullifier_query_count_is_bounded() {
		assert!(
			validate_pre_submission_nullifier_count(MAX_PRE_SUBMISSION_NULLIFIER_QUERIES).is_ok()
		);

		let err = validate_pre_submission_nullifier_count(MAX_PRE_SUBMISSION_NULLIFIER_QUERIES + 1)
			.unwrap_err();
		assert!(err.message.contains("maximum supported per batch is 7"));
	}

	#[test]
	fn test_parse_ss58_address() {
		// Valid Quantus address (SS58 prefix 189)
		let result = parse_ss58_address("5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY");
		assert!(result.is_ok());
	}

	#[test]
	fn test_decode_input_amount_from_leaf() {
		// Create a mock leaf data (60 bytes minimum)
		let mut leaf_data = vec![0u8; 60];
		// Set amount at bytes 44-60 (1 QTM = 10^12 planck)
		let amount: u128 = 1_000_000_000_000; // 1 QTM
		leaf_data[44..60].copy_from_slice(&amount.to_le_bytes());

		let result = decode_input_amount_from_leaf(&leaf_data).unwrap();
		// 1 QTM = 10^12 planck, quantized = 10^12 / 10^10 = 100
		assert_eq!(result, 100);
	}

	const TEST_SECRET_HEX: &str =
		"0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20";
	const TEST_MNEMONIC: &str = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon art";

	#[test]
	fn test_resolve_credential_secret() {
		let cred = WormholeCredential::Secret { hex: TEST_SECRET_HEX.to_string() };
		let (address, address_bytes, secret_bytes) = resolve_credential(&cred).unwrap();

		let expected_secret: [u8; 32] = hex::decode(TEST_SECRET_HEX).unwrap().try_into().unwrap();
		let expected_address_bytes =
			wormhole_lib::compute_wormhole_address(&expected_secret).unwrap();

		assert_eq!(secret_bytes, expected_secret);
		assert_eq!(address_bytes, expected_address_bytes);
		assert_eq!(address, AccountId32::from(expected_address_bytes).to_ss58check());
	}

	#[test]
	fn test_resolve_credential_secret_accepts_0x_prefix() {
		let cred_plain = WormholeCredential::Secret { hex: TEST_SECRET_HEX.to_string() };
		let cred_prefixed = WormholeCredential::Secret { hex: format!("0x{}", TEST_SECRET_HEX) };
		assert_eq!(
			resolve_credential(&cred_plain).unwrap(),
			resolve_credential(&cred_prefixed).unwrap()
		);
	}

	#[test]
	fn test_resolve_credential_secret_invalid() {
		let too_short = WormholeCredential::Secret { hex: "0102".to_string() };
		assert!(resolve_credential(&too_short).unwrap_err().message.contains("Invalid secret"));

		let bad_hex = WormholeCredential::Secret { hex: "zz".repeat(32) };
		assert!(resolve_credential(&bad_hex).unwrap_err().message.contains("Invalid secret"));
	}

	#[test]
	fn test_resolve_credential_mnemonic() {
		let cred =
			WormholeCredential::Mnemonic { phrase: TEST_MNEMONIC.to_string(), wormhole_index: 0 };
		let (address, address_bytes, secret_bytes) = resolve_credential(&cred).unwrap();

		assert_ne!(secret_bytes, [0u8; 32]);
		assert_ne!(address_bytes, [0u8; 32]);
		assert_eq!(address, AccountId32::from(address_bytes).to_ss58check());
		assert_eq!(address_bytes, wormhole_lib::compute_wormhole_address(&secret_bytes).unwrap());
	}

	#[test]
	fn test_resolve_credential_mnemonic_pinned_derivation_path() {
		// Regression guard for the HD path `m/44'/CHAIN/0'/0'/index'` (fixed in #93).
		// If this breaks, the derivation path or the underlying HD library changed.
		let cred =
			WormholeCredential::Mnemonic { phrase: TEST_MNEMONIC.to_string(), wormhole_index: 0 };
		let (_, address_bytes, secret_bytes) = resolve_credential(&cred).unwrap();
		assert_eq!(
			hex::encode(address_bytes),
			"b8a7c11fc57b36fbad44e437ec05d91c44231974c058ded1fed66cb7baa41973",
		);
		assert_eq!(
			hex::encode(secret_bytes),
			"110684de72bc884f854accf8bc6ba724dcc1cc2f99932a4d28bdf85fc6f28ccf",
		);
	}

	#[test]
	fn test_resolve_credential_mnemonic_index_changes_output() {
		let cred_0 =
			WormholeCredential::Mnemonic { phrase: TEST_MNEMONIC.to_string(), wormhole_index: 0 };
		let cred_1 =
			WormholeCredential::Mnemonic { phrase: TEST_MNEMONIC.to_string(), wormhole_index: 1 };
		assert_ne!(resolve_credential(&cred_0).unwrap(), resolve_credential(&cred_1).unwrap());
	}

	#[test]
	fn test_resolve_credential_mnemonic_invalid_phrase() {
		let cred = WormholeCredential::Mnemonic {
			phrase: "not a real mnemonic".to_string(),
			wormhole_index: 0,
		};
		assert!(resolve_credential(&cred).unwrap_err().message.contains("HD derivation"));
	}

	#[test]
	fn test_resolve_credential_mnemonic_and_secret_equivalence() {
		let mnemonic_cred =
			WormholeCredential::Mnemonic { phrase: TEST_MNEMONIC.to_string(), wormhole_index: 0 };
		let (m_address, m_address_bytes, m_secret_bytes) =
			resolve_credential(&mnemonic_cred).unwrap();

		let secret_cred = WormholeCredential::Secret { hex: hex::encode(m_secret_bytes) };
		let (s_address, s_address_bytes, s_secret_bytes) =
			resolve_credential(&secret_cred).unwrap();

		assert_eq!(m_address, s_address);
		assert_eq!(m_address_bytes, s_address_bytes);
		assert_eq!(m_secret_bytes, s_secret_bytes);
	}

	fn read_http_request(stream: &mut std::net::TcpStream) -> String {
		use std::{io::Read, time::Duration};

		stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
		let mut buf = Vec::new();
		let mut tmp = [0u8; 1024];
		let mut header_end = None;
		let mut content_len = 0usize;

		loop {
			let n = stream.read(&mut tmp).unwrap();
			assert_ne!(n, 0, "mock indexer connection closed before request was complete");
			buf.extend_from_slice(&tmp[..n]);

			if header_end.is_none() {
				if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
					header_end = Some(pos + 4);
					let headers = String::from_utf8_lossy(&buf[..pos]);
					for line in headers.lines() {
						if let Some((name, value)) = line.split_once(':') {
							if name.eq_ignore_ascii_case("content-length") {
								content_len = value.trim().parse().unwrap();
							}
						}
					}
				}
			}

			if let Some(end) = header_end {
				if buf.len() >= end + content_len {
					break;
				}
			}
		}

		String::from_utf8(buf).unwrap()
	}

	fn write_json_response(stream: &mut std::net::TcpStream, body: serde_json::Value) {
		use std::io::Write;
		let body = body.to_string();
		write!(
			stream,
			"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
			body.len(),
			body
		)
		.unwrap();
	}

	/// #159890: address-only queries must not report availability without secret reconciliation.
	#[tokio::test]
	async fn pending_transfer_query_for_address_refuses_without_secret() {
		use serde_json::json;
		use std::{net::TcpListener, thread};

		let secret = [7u8; 32];
		let wormhole_address = wormhole_lib::compute_wormhole_address(&secret).unwrap();
		let address_hash = compute_address_hash(&wormhole_address);

		let listener = TcpListener::bind("127.0.0.1:0").unwrap();
		let mock_url = format!("http://{}", listener.local_addr().unwrap());
		thread::spawn(move || {
			let (mut stream, _) = listener.accept().unwrap();
			let _ = read_http_request(&mut stream);
			write_json_response(
				&mut stream,
				json!({
					"data": {
						"transfers": [{
							"id": "spent-transfer",
							"block_id": "0xspentblock",
							"block": { "height": 11 },
							"timestamp": "2026-01-01T00:00:00Z",
							"extrinsic_id": "0xspentextrinsic",
							"from_id": "miner-a",
							"to_id": "wormhole",
							"amount": "100",
							"fee": "0",
							"from_hash": "from-hash-a",
							"to_hash": address_hash,
							"leaf_index": "40",
							"transfer_count": "5"
						}],
						"meta": { "aggregate": { "count": 1 } }
					}
				}),
			);
		});

		let err = query_pending_transfers_for_address(&wormhole_address, &mock_url)
			.await
			.expect_err("address-only query must refuse when transfers exist");
		assert!(
			err.message.contains("without the wormhole secret"),
			"unexpected error: {}",
			err.message
		);
	}

	/// #159890: mnemonic pending-transfer query excludes spent nullifiers via indexer.
	#[tokio::test]
	async fn query_pending_transfers_excludes_spent_nullifiers() {
		use serde_json::json;
		use std::{net::TcpListener, thread};

		let path = format!("m/44'/{}/0'/1'/0'", QUANTUS_WORMHOLE_CHAIN_ID);
		let wormhole_secret = derive_wormhole_from_mnemonic(TEST_MNEMONIC, None, &path).unwrap();
		let secret_bytes: [u8; 32] = *wormhole_secret.secret().as_bytes();
		let address_hash = compute_address_hash(wormhole_secret.address());

		let spent_transfer_count = 5u64;
		let spent_nullifier =
			wormhole_lib::compute_nullifier(&secret_bytes, spent_transfer_count).unwrap();
		let spent_nullifier_hex = hex::encode(spent_nullifier);
		let spent_nullifier_hash = compute_address_hash(&spent_nullifier);

		let listener = TcpListener::bind("127.0.0.1:0").unwrap();
		let mock_url = format!("http://{}", listener.local_addr().unwrap());
		let address_hash_for_server = address_hash.clone();
		let spent_hex_for_server = spent_nullifier_hex.clone();
		let spent_hash_for_server = spent_nullifier_hash.clone();

		thread::spawn(move || {
			for _ in 0..2 {
				let (mut stream, _) = listener.accept().unwrap();
				let request = read_http_request(&mut stream);
				let body = request.split("\r\n\r\n").nth(1).unwrap_or_default();
				if body.contains("TransfersByHashPrefix") {
					write_json_response(
						&mut stream,
						json!({
							"data": {
								"transfers": [
									{
										"id": "spent-transfer",
										"block_id": "0xspentblock",
										"block": { "height": 11 },
										"timestamp": "2026-01-01T00:00:00Z",
										"extrinsic_id": "0xspentextrinsic",
										"from_id": "miner-a",
										"to_id": "wormhole",
										"amount": "100",
										"fee": "0",
										"from_hash": "from-hash-a",
										"to_hash": address_hash_for_server,
										"leaf_index": "40",
										"transfer_count": "5"
									},
									{
										"id": "unspent-transfer",
										"block_id": "0xunspentblock",
										"block": { "height": 12 },
										"timestamp": "2026-01-01T00:00:01Z",
										"extrinsic_id": "0xunspentextrinsic",
										"from_id": "miner-b",
										"to_id": "wormhole",
										"amount": "7",
										"fee": "0",
										"from_hash": "from-hash-b",
										"to_hash": address_hash_for_server,
										"leaf_index": "41",
										"transfer_count": "6"
									}
								],
								"meta": { "aggregate": { "count": 2 } }
							}
						}),
					);
				} else if body.contains("NullifiersByPrefix") {
					write_json_response(
						&mut stream,
						json!({
							"data": {
								"nullifiers": [{
									"nullifier": spent_hex_for_server,
									"nullifier_hash": spent_hash_for_server,
									"block": { "height": 20 },
									"timestamp": "2026-01-01T00:00:02Z",
									"wormholeExtrinsic": { "extrinsic_id": "0xwithdrawal" }
								}]
							}
						}),
					);
				} else {
					panic!("unexpected GraphQL request body: {body}");
				}
			}
		});

		let reported = query_pending_transfers(TEST_MNEMONIC, 0, &mock_url)
			.await
			.expect("mnemonic pending query should reconcile nullifiers");

		assert_eq!(reported.transfers.len(), 1);
		assert_eq!(reported.transfers[0].transfer_count, 6);
		assert_eq!(reported.total_available, 7);
		assert!(!reported.transfers.iter().any(|t| t.transfer_count == spent_transfer_count));
	}
}