diff --git a/apps/web/src/components/modules/payment-stream/CreatePaymentStream.tsx b/apps/web/src/components/modules/payment-stream/CreatePaymentStream.tsx index 50666593..d26f36ee 100644 --- a/apps/web/src/components/modules/payment-stream/CreatePaymentStream.tsx +++ b/apps/web/src/components/modules/payment-stream/CreatePaymentStream.tsx @@ -11,7 +11,7 @@ import { PaymentStreamConfirmationModal } from "./PaymentStreamConfirmationModal import { capitalizeWord } from "@/lib/utils"; import { SUPPORTED_TOKENS, PaymentStreamFormData } from "@/lib/validations"; import { StellarService } from "@/lib/stellar"; -import { validateEndTime } from "@/lib/stream-validation"; +import { validateEndTime, validateContractId } from "@/lib/stream-validation"; import { useDebouncedCallback } from "@/hooks/use-debounce-callback"; import { useBalanceValidation } from "@/hooks/use-balance-validation"; import { useUnsavedChanges } from "@/hooks/use-unsaved-changes"; @@ -158,6 +158,14 @@ const CreatePaymentStream = () => { toast.error("Invalid Stellar address"); return; } + + // Validate token contract ID (must be 'native' for XLM or a valid StrKey contract address) + const selectedTokenMeta = SUPPORTED_TOKENS.find((t) => t.value === streamData.token); + const tokenAddress = selectedTokenMeta?.address; + if (!tokenAddress || (tokenAddress !== "native" && !validateContractId(tokenAddress))) { + toast.error("Invalid token: contract address is not a valid Stellar contract ID"); + return; + } if (!streamData.amount || parseFloat(streamData.amount) <= 0) { toast.error("Amount must be greater than 0"); return; diff --git a/apps/web/src/hooks/use-distribute.ts b/apps/web/src/hooks/use-distribute.ts index 4b314956..f59b93d7 100644 --- a/apps/web/src/hooks/use-distribute.ts +++ b/apps/web/src/hooks/use-distribute.ts @@ -2,7 +2,7 @@ import { QueryClient, QueryKey, useMutation, useQueryClient } from '@tanstack/re import toast from 'react-hot-toast'; import { distribute } from '@/lib/api'; import { useWallet } from '@/providers/StellarWalletProvider'; -import { createBatches } from '../../../../packages/sdk/src/utils/batchDistribution'; +import { createBatches } from '@fundable/sdk'; type DistributeInput = Parameters[0]; diff --git a/apps/web/src/hooks/use-distribution-transaction.ts b/apps/web/src/hooks/use-distribution-transaction.ts index ef68372a..ae6ef126 100644 --- a/apps/web/src/hooks/use-distribution-transaction.ts +++ b/apps/web/src/hooks/use-distribution-transaction.ts @@ -2,7 +2,7 @@ import { useState, useCallback } from 'react'; import { Horizon } from '@stellar/stellar-sdk'; -import { DistributorClient } from '../../../../packages/sdk/src/DistributorClient'; +import { DistributorClient } from '@fundable/sdk'; import { useWallet } from '@/providers/StellarWalletProvider'; import { notify } from '@/utils/notification'; import { DISTRIBUTOR_CONTRACT_ID, SOROBAN_RPC_URL, NETWORK_PASSPHRASE } from '@/lib/constants'; diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index f1c07f95..c7fa46c4 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -3,8 +3,7 @@ import { PAYMENT_STREAM_CONTRACT_ID, DISTRIBUTOR_CONTRACT_ID, SOROBAN_RPC_URL, N import { env } from '@/lib/env'; import { throwIfAborted } from '@/utils/retry'; import { StellarService, type Stream as ServiceStream, type AccountInfo } from '@/services'; -import { PaymentStreamClient } from '../../../../packages/sdk/src/PaymentStreamClient'; -import { DistributorClient } from '../../../../packages/sdk/src/DistributorClient'; +import { PaymentStreamClient, DistributorClient, createBatches } from '@fundable/sdk'; import { Stream, StreamStatus } from '../types'; type WalletSigner = (xdr: string) => Promise; @@ -151,8 +150,6 @@ export async function withdraw(params: { await signAndSendTx(tx, params.signTransaction); } -import { createBatches } from '../../../../packages/sdk/src/utils/batchDistribution'; - export async function distribute(params: { sender: string; token: string; diff --git a/apps/web/src/lib/stream-validation.ts b/apps/web/src/lib/stream-validation.ts index 0429a849..55291f1f 100644 --- a/apps/web/src/lib/stream-validation.ts +++ b/apps/web/src/lib/stream-validation.ts @@ -2,6 +2,22 @@ * Stream validation utilities for payment stream forms */ +import { StrKey } from "@stellar/stellar-sdk"; + +/** + * Validate that a string is a valid Stellar contract ID (StrKey C... format) + * @param contractId - The contract address to validate + * @returns true if the address is a valid contract StrKey, false otherwise + */ +export function validateContractId(contractId: string): boolean { + if (!contractId || typeof contractId !== "string") return false; + try { + return StrKey.isValidContract(contractId); + } catch { + return false; + } +} + /** * Duration unit multipliers in seconds */ diff --git a/apps/web/src/lib/validations.ts b/apps/web/src/lib/validations.ts index dde18a0d..15043266 100644 --- a/apps/web/src/lib/validations.ts +++ b/apps/web/src/lib/validations.ts @@ -1,5 +1,6 @@ import { z } from "zod" import { StellarService } from "./stellar" +import { validateContractId } from "./stream-validation" // Stream record type for display export interface StreamRecord { @@ -26,7 +27,11 @@ export const paymentStreamSchema = z.object({ token: z .string() - .min(1, "Token selection is required"), + .min(1, "Token selection is required") + .refine( + (val) => val === "native" || validateContractId(val), + "Invalid token: must be 'native' or a valid Stellar contract ID" + ), totalAmount: z .string() diff --git a/contracts/payment-stream/src/lib.rs b/contracts/payment-stream/src/lib.rs index a6114941..2b25929d 100644 --- a/contracts/payment-stream/src/lib.rs +++ b/contracts/payment-stream/src/lib.rs @@ -102,6 +102,22 @@ pub struct StreamResumedEvent { pub paused_duration: u64, } +/// Emergency paused event data +#[contracttype] +#[derive(Clone)] +pub struct EmergencyPausedEvent { + pub paused_by: Address, + pub paused_at: u64, +} + +/// Emergency unpaused event data +#[contracttype] +#[derive(Clone)] +pub struct EmergencyUnpausedEvent { + pub unpaused_by: Address, + pub unpaused_at: u64, +} + /// Custom errors for the contract #[contracterror] #[derive(Copy, Clone, Debug, Eq, PartialEq, PartialOrd, Ord)] @@ -123,6 +139,12 @@ pub enum Error { DepositExceedsTotal = 14, ArithmeticOverflow = 15, InvalidDelegate = 16, + /// Protocol is globally paused by the emergency circuit breaker + ContractPaused = 17, + /// Emergency pause is already active + AlreadyPaused = 18, + /// Contract is not currently paused + NotPaused = 19, } // Constants @@ -162,6 +184,124 @@ impl PaymentStreamContract { env.storage().instance().extend_ttl(LEDGER_THRESHOLD, LEDGER_BUMP); } + // ----------------------------------------------------------------------- + // Emergency pause circuit breaker + // ----------------------------------------------------------------------- + + /// Internal guard: panics with `ContractPaused` when the global emergency + /// pause flag is active. Call this at the top of every state-mutating + /// entry point that should be halted during an incident. + fn assert_not_paused(env: &Env) { + let paused: bool = env + .storage() + .instance() + .get(&Symbol::new(env, "paused")) + .unwrap_or(false); + if paused { + panic_with_error!(env, Error::ContractPaused); + } + } + + /// Activate the global emergency pause switch. + /// + /// When active, all calls to `create_stream`, `deposit`, `withdraw`, and + /// `withdraw_max` will be rejected with `Error::ContractPaused`. + /// Admin-only operations (fee management, pause/unpause) remain available. + /// + /// # Authorization + /// Requires the stored admin address to sign this transaction. + /// + /// # Errors + /// - `Error::Unauthorized` – caller is not admin. + /// - `Error::AlreadyPaused` – the circuit breaker is already active. + pub fn emergency_pause(env: Env) { + let admin: Address = env + .storage() + .instance() + .get(&Symbol::new(&env, "admin")) + .unwrap_or_else(|| panic_with_error!(&env, Error::NotInitialized)); + admin.require_auth(); + + let already_paused: bool = env + .storage() + .instance() + .get(&Symbol::new(&env, "paused")) + .unwrap_or(false); + if already_paused { + panic_with_error!(&env, Error::AlreadyPaused); + } + + env.storage() + .instance() + .set(&Symbol::new(&env, "paused"), &true); + env.storage() + .instance() + .extend_ttl(LEDGER_THRESHOLD, LEDGER_BUMP); + + let now = env.ledger().timestamp(); + env.events().publish( + ("EmergencyPaused",), + EmergencyPausedEvent { + paused_by: admin, + paused_at: now, + }, + ); + } + + /// Deactivate the global emergency pause switch, resuming normal operation. + /// + /// # Authorization + /// Requires the stored admin address to sign this transaction. + /// + /// # Errors + /// - `Error::Unauthorized` – caller is not admin. + /// - `Error::NotPaused` – the circuit breaker is not currently active. + pub fn emergency_unpause(env: Env) { + let admin: Address = env + .storage() + .instance() + .get(&Symbol::new(&env, "admin")) + .unwrap_or_else(|| panic_with_error!(&env, Error::NotInitialized)); + admin.require_auth(); + + let paused: bool = env + .storage() + .instance() + .get(&Symbol::new(&env, "paused")) + .unwrap_or(false); + if !paused { + panic_with_error!(&env, Error::NotPaused); + } + + env.storage() + .instance() + .set(&Symbol::new(&env, "paused"), &false); + env.storage() + .instance() + .extend_ttl(LEDGER_THRESHOLD, LEDGER_BUMP); + + let now = env.ledger().timestamp(); + env.events().publish( + ("EmergencyUnpaused",), + EmergencyUnpausedEvent { + unpaused_by: admin, + unpaused_at: now, + }, + ); + } + + /// Returns `true` when the global emergency pause is active. + pub fn is_paused(env: Env) -> bool { + env.storage() + .instance() + .get(&Symbol::new(&env, "paused")) + .unwrap_or(false) + } + + // ----------------------------------------------------------------------- + // Core stream operations + // ----------------------------------------------------------------------- + /// Create a new payment stream pub fn create_stream( env: Env, @@ -173,6 +313,7 @@ impl PaymentStreamContract { start_time: u64, end_time: u64, ) -> u64 { + Self::assert_not_paused(&env); sender.require_auth(); // Validate inputs @@ -255,6 +396,7 @@ impl PaymentStreamContract { /// Deposit tokens to an existing stream pub fn deposit(env: Env, stream_id: u64, amount: i128) { + Self::assert_not_paused(&env); let mut stream: Stream = Self::get_stream(env.clone(), stream_id); if matches!(stream.status, StreamStatus::Canceled | StreamStatus::Completed) { @@ -491,6 +633,7 @@ impl PaymentStreamContract { /// Withdraw from a stream pub fn withdraw(env: Env, stream_id: u64, amount: i128) { + Self::assert_not_paused(&env); let mut stream: Stream = Self::get_stream(env.clone(), stream_id); Self::assert_is_recipient_or_delegate(&env, stream_id); @@ -547,6 +690,7 @@ impl PaymentStreamContract { /// Withdraw the maximum available amount from a stream pub fn withdraw_max(env: Env, stream_id: u64) { + Self::assert_not_paused(&env); let available = Self::withdrawable_amount(env.clone(), stream_id); if available <= 0 { panic_with_error!(&env, Error::InsufficientWithdrawable); diff --git a/contracts/payment-stream/src/test.rs b/contracts/payment-stream/src/test.rs index 0e5acef9..6877d938 100644 --- a/contracts/payment-stream/src/test.rs +++ b/contracts/payment-stream/src/test.rs @@ -1829,5 +1829,371 @@ fn test_withdraw_after_pause_and_resume() { assert!(recipient_balance > 0); assert_eq!(recipient_balance, 600); // 100 + 500 } + + // ----------------------------------------------------------------------- + // Emergency pause circuit breaker tests + // ----------------------------------------------------------------------- + + /// Helper: initialise a contract and return (client, contract_id, admin, + /// fee_collector, sender, recipient, token). + fn setup_paused_contract( + env: &Env, + ) -> ( + PaymentStreamContractClient, + Address, // contract_id + Address, // admin + Address, // fee_collector + Address, // sender + Address, // recipient + Address, // token + ) { + let admin = Address::generate(env); + let fee_collector = Address::generate(env); + let sender = Address::generate(env); + let recipient = Address::generate(env); + + let sac = env.register_stellar_asset_contract_v2(admin.clone()); + let token = sac.address(); + + let contract_id = env.register(PaymentStreamContract, ()); + let client = PaymentStreamContractClient::new(env, &contract_id); + + client.initialize(&admin, &fee_collector, &0); + + // Mint tokens to sender + let token_admin = token::StellarAssetClient::new(env, &token); + token_admin.mint(&sender, &2000); + + (client, contract_id, admin, fee_collector, sender, recipient, token) + } + + /// Contract starts unpaused. + #[test] + fn test_is_paused_default_false() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + + assert!(!client.is_paused()); + } + + /// Admin can activate the emergency pause. + #[test] + fn test_emergency_pause_sets_flag() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + + assert!(!client.is_paused()); + client.emergency_pause(); + assert!(client.is_paused()); + } + + /// Admin can deactivate the emergency pause. + #[test] + fn test_emergency_unpause_clears_flag() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + + client.emergency_pause(); + assert!(client.is_paused()); + + client.emergency_unpause(); + assert!(!client.is_paused()); + } + + /// `emergency_pause` emits the correct event. + #[test] + fn test_emergency_pause_emits_event() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + client.emergency_pause(); + + let events = env.events().all(); + assert!(events.len() > 0); + } + + /// `emergency_unpause` emits the correct event. + #[test] + fn test_emergency_unpause_emits_event() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + client.emergency_pause(); + client.emergency_unpause(); + + let events = env.events().all(); + assert!(events.len() > 0); + } + + /// Double-pause is rejected with `AlreadyPaused` (error code 18). + #[test] + #[should_panic(expected = "Error(Contract, #18)")] + fn test_emergency_pause_already_paused() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + + client.emergency_pause(); + client.emergency_pause(); // should panic + } + + /// Unpausing when not paused is rejected with `NotPaused` (error code 19). + #[test] + #[should_panic(expected = "Error(Contract, #19)")] + fn test_emergency_unpause_when_not_paused() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + + client.emergency_unpause(); // should panic + } + + /// `create_stream` is blocked while the circuit breaker is active. + #[test] + #[should_panic(expected = "Error(Contract, #17)")] + fn test_create_stream_blocked_when_paused() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, sender, recipient, token) = setup_paused_contract(&env); + + client.emergency_pause(); + + // This call should panic with ContractPaused (17) + client.create_stream(&sender, &recipient, &token, &1000, &1000, &0, &100); + } + + /// `withdraw` is blocked while the circuit breaker is active. + #[test] + #[should_panic(expected = "Error(Contract, #17)")] + fn test_withdraw_blocked_when_paused() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, sender, recipient, token) = setup_paused_contract(&env); + + // Create a stream before pausing + let stream_id = client.create_stream(&sender, &recipient, &token, &1000, &1000, &0, &100); + env.ledger().set_timestamp(50); + + client.emergency_pause(); + + // Should be blocked + client.withdraw(&stream_id, &500); + } + + /// `withdraw_max` is blocked while the circuit breaker is active. + #[test] + #[should_panic(expected = "Error(Contract, #17)")] + fn test_withdraw_max_blocked_when_paused() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, sender, recipient, token) = setup_paused_contract(&env); + + let stream_id = client.create_stream(&sender, &recipient, &token, &1000, &1000, &0, &100); + env.ledger().set_timestamp(50); + + client.emergency_pause(); + + // Should be blocked + client.withdraw_max(&stream_id); + } + + /// `deposit` is blocked while the circuit breaker is active. + #[test] + #[should_panic(expected = "Error(Contract, #17)")] + fn test_deposit_blocked_when_paused() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, sender, recipient, token) = setup_paused_contract(&env); + + let stream_id = client.create_stream(&sender, &recipient, &token, &1000, &0, &0, &100); + + client.emergency_pause(); + + // Should be blocked + client.deposit(&stream_id, &500); + } + + /// All operations resume normally after emergency_unpause. + #[test] + fn test_operations_resume_after_unpause() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, sender, recipient, token) = setup_paused_contract(&env); + + // Create stream, then pause + let stream_id = client.create_stream(&sender, &recipient, &token, &1000, &1000, &0, &100); + client.emergency_pause(); + assert!(client.is_paused()); + + // Unpause and verify operations work again + client.emergency_unpause(); + assert!(!client.is_paused()); + + env.ledger().set_timestamp(50); + let available = client.withdrawable_amount(&stream_id); + assert_eq!(available, 500); + + // Withdraw should succeed + client.withdraw(&stream_id, &200); + let stream = client.get_stream(&stream_id); + assert_eq!(stream.withdrawn_amount, 200); + } + + /// Pause/unpause can be cycled multiple times. + #[test] + fn test_pause_unpause_cycle() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, _, _, _) = setup_paused_contract(&env); + + for _ in 0..3 { + assert!(!client.is_paused()); + client.emergency_pause(); + assert!(client.is_paused()); + client.emergency_unpause(); + assert!(!client.is_paused()); + } + } + + /// Non-admin callers cannot activate emergency pause. + #[test] + #[should_panic(expected = "Unauthorized")] + fn test_non_admin_cannot_emergency_pause() { + let env = Env::default(); + + let admin = Address::generate(&env); + let fee_collector = Address::generate(&env); + let attacker = Address::generate(&env); + + let sac = env.register_stellar_asset_contract_v2(admin.clone()); + let _token = sac.address(); + + let contract_id = env.register(PaymentStreamContract, ()); + let client = PaymentStreamContractClient::new(&env, &contract_id); + + env.mock_auths(&[MockAuth { + address: &admin, + invoke: &MockAuthInvoke { + contract: &contract_id, + fn_name: "initialize", + args: (&admin, &fee_collector, &0u32).into_val(&env), + sub_invokes: &[], + }, + }]); + client.initialize(&admin, &fee_collector, &0); + + // Now mock only the attacker's auth — admin auth is NOT provided for emergency_pause + env.mock_auths(&[MockAuth { + address: &attacker, + invoke: &MockAuthInvoke { + contract: &contract_id, + fn_name: "emergency_pause", + args: ().into_val(&env), + sub_invokes: &[], + }, + }]); + + // Should panic because admin.require_auth() won't be satisfied + client.emergency_pause(); + } + + /// Non-admin callers cannot deactivate emergency pause. + #[test] + #[should_panic(expected = "Unauthorized")] + fn test_non_admin_cannot_emergency_unpause() { + let env = Env::default(); + + let admin = Address::generate(&env); + let fee_collector = Address::generate(&env); + let attacker = Address::generate(&env); + + let sac = env.register_stellar_asset_contract_v2(admin.clone()); + let _token = sac.address(); + + let contract_id = env.register(PaymentStreamContract, ()); + let client = PaymentStreamContractClient::new(&env, &contract_id); + + // Initialize and pause using real admin auth + env.mock_auths(&[ + MockAuth { + address: &admin, + invoke: &MockAuthInvoke { + contract: &contract_id, + fn_name: "initialize", + args: (&admin, &fee_collector, &0u32).into_val(&env), + sub_invokes: &[], + }, + }, + MockAuth { + address: &admin, + invoke: &MockAuthInvoke { + contract: &contract_id, + fn_name: "emergency_pause", + args: ().into_val(&env), + sub_invokes: &[], + }, + }, + ]); + client.initialize(&admin, &fee_collector, &0); + client.emergency_pause(); + + // Try to unpause as attacker + env.mock_auths(&[MockAuth { + address: &attacker, + invoke: &MockAuthInvoke { + contract: &contract_id, + fn_name: "emergency_unpause", + args: ().into_val(&env), + sub_invokes: &[], + }, + }]); + + client.emergency_unpause(); // should panic + } + + /// Read-only functions remain available while paused. + #[test] + fn test_read_operations_work_while_paused() { + let env = Env::default(); + env.mock_all_auths(); + + let (client, _, _, _, sender, recipient, token) = setup_paused_contract(&env); + + let stream_id = client.create_stream(&sender, &recipient, &token, &1000, &1000, &0, &100); + env.ledger().set_timestamp(50); + + client.emergency_pause(); + + // Read-only calls must not be blocked + let stream = client.get_stream(&stream_id); + assert_eq!(stream.id, stream_id); + + let metrics = client.get_stream_metrics(&stream_id); + assert_eq!(metrics.withdrawal_count, 0); + + let proto = client.get_protocol_metrics(); + assert_eq!(proto.total_streams_created, 1); + + let withdrawable = client.withdrawable_amount(&stream_id); + assert_eq!(withdrawable, 500); + + assert!(client.is_paused()); + } } diff --git a/packages/sdk/src/DistributorClient.ts b/packages/sdk/src/DistributorClient.ts index 6842fb95..dbb24bde 100644 --- a/packages/sdk/src/DistributorClient.ts +++ b/packages/sdk/src/DistributorClient.ts @@ -10,6 +10,12 @@ import { DistributionHistory, } from "./generated/distributor/src/index.js"; import { executeWithErrorHandling } from "./utils/errors.js"; +import { + prepareBatchEqualDistribution, + prepareBatchWeightedDistribution, + BatchDistributionConfig, + BatchDistributionResult, +} from "./utils/batchDistribution.js"; /** * Type alias for address parameters that accept both string and Address objects @@ -248,4 +254,112 @@ export class DistributorClient { "Set protocol fee", ); } + + // --------------------------------------------------------------------------- + // Batch distribution + // --------------------------------------------------------------------------- + + /** + * Distribute tokens equally to a large list of recipients, automatically + * splitting into multiple transactions to stay within Soroban's gas limits. + * + * @param params.sender Sender address (must hold sufficient token balance). + * @param params.token Token contract ID to distribute. + * @param params.total_amount Total amount to distribute (in token base units). + * @param params.recipients Full recipient list — will be chunked automatically. + * @param params.config Optional batch settings (size, progress callbacks). + * @returns A {@link BatchDistributionResult} with one assembled transaction per + * batch, ready to sign and submit sequentially. + * + * @example + * ```ts + * const { transactions } = await client.batchDistribute({ + * sender: 'GAAAA...', + * token: 'CXXXX...', + * total_amount: 1_000_000n, + * recipients: thousandsOfAddresses, + * config: { maxRecipientsPerBatch: 100 }, + * }); + * for (const tx of transactions) await tx.signAndSend({ signTransaction }); + * ``` + */ + public async batchDistribute(params: { + sender: AddressParam; + token: AddressParam; + total_amount: bigint; + recipients: AddressParam[]; + config?: BatchDistributionConfig; + }): Promise; + + /** + * Distribute tokens with per-recipient amounts to a large list of recipients, + * automatically splitting into multiple transactions to stay within Soroban's + * gas limits. + * + * @param params.sender Sender address (must hold sufficient token balance). + * @param params.token Token contract ID to distribute. + * @param params.recipients Full recipient list — will be chunked in parallel with amounts. + * @param params.amounts Per-recipient amounts, parallel to `recipients`. + * @param params.config Optional batch settings (size, progress callbacks). + * @returns A {@link BatchDistributionResult} with one assembled transaction per + * batch, ready to sign and submit sequentially. + * + * @example + * ```ts + * const { transactions } = await client.batchDistribute({ + * sender: 'GAAAA...', + * token: 'CXXXX...', + * recipients: thousandsOfAddresses, + * amounts: correspondingAmounts, + * config: { maxRecipientsPerBatch: 75 }, + * }); + * for (const tx of transactions) await tx.signAndSend({ signTransaction }); + * ``` + */ + public async batchDistribute(params: { + sender: AddressParam; + token: AddressParam; + recipients: AddressParam[]; + amounts: bigint[]; + config?: BatchDistributionConfig; + }): Promise; + + // Implementation signature — handles both overloads + public async batchDistribute( + params: + | { + sender: AddressParam; + token: AddressParam; + total_amount: bigint; + recipients: AddressParam[]; + config?: BatchDistributionConfig; + } + | { + sender: AddressParam; + token: AddressParam; + recipients: AddressParam[]; + amounts: bigint[]; + config?: BatchDistributionConfig; + } + ): Promise { + if ("amounts" in params) { + // Weighted distribution + return prepareBatchWeightedDistribution(this, { + sender: params.sender, + token: params.token, + recipients: params.recipients, + amounts: params.amounts, + config: params.config, + }); + } + + // Equal distribution + return prepareBatchEqualDistribution(this, { + sender: params.sender, + token: params.token, + total_amount: params.total_amount, + recipients: params.recipients, + config: params.config, + }); + } } diff --git a/packages/sdk/src/__tests__/DistributorClient.test.ts b/packages/sdk/src/__tests__/DistributorClient.test.ts index a6f654c1..94434b8f 100644 --- a/packages/sdk/src/__tests__/DistributorClient.test.ts +++ b/packages/sdk/src/__tests__/DistributorClient.test.ts @@ -453,4 +453,314 @@ describe("DistributorClient", () => { }); }); }); + + // ── batchDistribute ──────────────────────────────────────────────────────── + describe("batchDistribute", () => { + /** Build a list of N fake recipient addresses */ + const makeRecipients = (n: number) => + Array.from({ length: n }, (_, i) => + `G${"A".repeat(54)}`.slice(0, 55) + i.toString().padStart(1, "0") + ); + + beforeEach(() => { + mockContractClient.distribute_equal.mockResolvedValue(mockTx(null)); + mockContractClient.distribute_weighted.mockResolvedValue(mockTx(null)); + }); + + // ── equal distribution ──────────────────────────────────────────────────── + describe("equal distribution (total_amount present, no amounts)", () => { + it("returns a single batch when recipients fit in one batch", async () => { + const recipients = makeRecipients(3); + const result = await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 1000n, + recipients, + config: { maxRecipientsPerBatch: 10 }, + }); + + expect(result.batchCount).toBe(1); + expect(result.transactions).toHaveLength(1); + expect(result.recipientBatches).toHaveLength(1); + expect(result.recipientBatches[0]).toHaveLength(3); + expect(mockContractClient.distribute_equal).toHaveBeenCalledTimes(1); + expect(mockContractClient.distribute_weighted).not.toHaveBeenCalled(); + }); + + it("splits into multiple batches when recipients exceed maxRecipientsPerBatch", async () => { + const recipients = makeRecipients(25); + const result = await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 25000n, + recipients, + config: { maxRecipientsPerBatch: 10 }, + }); + + expect(result.batchCount).toBe(3); // ceil(25/10) + expect(result.transactions).toHaveLength(3); + expect(result.recipientBatches[0]).toHaveLength(10); + expect(result.recipientBatches[1]).toHaveLength(10); + expect(result.recipientBatches[2]).toHaveLength(5); + expect(mockContractClient.distribute_equal).toHaveBeenCalledTimes(3); + }); + + it("uses default batch size (100) when config is omitted", async () => { + const recipients = makeRecipients(50); + const result = await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 50000n, + recipients, + }); + + expect(result.batchCount).toBe(1); + expect(mockContractClient.distribute_equal).toHaveBeenCalledTimes(1); + }); + + it("passes the full total_amount to every batch unchanged", async () => { + const recipients = makeRecipients(15); + await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 9999n, + recipients, + config: { maxRecipientsPerBatch: 10 }, + }); + + const calls = mockContractClient.distribute_equal.mock.calls; + expect(calls).toHaveLength(2); + for (const [callParams] of calls) { + expect(callParams.total_amount).toBe(9999n); + expect(callParams.sender).toBe(SENDER); + expect(callParams.token).toBe(TOKEN); + } + }); + + it("rejects an empty recipients list", async () => { + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 1000n, + recipients: [], + }) + ).rejects.toThrow(/empty/i); + expect(mockContractClient.distribute_equal).not.toHaveBeenCalled(); + }); + + it("rejects invalid maxRecipientsPerBatch (0)", async () => { + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 1000n, + recipients: makeRecipients(5), + config: { maxRecipientsPerBatch: 0 }, + }) + ).rejects.toThrow(/positive integer/i); + expect(mockContractClient.distribute_equal).not.toHaveBeenCalled(); + }); + + it("rejects invalid maxRecipientsPerBatch (-1)", async () => { + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 1000n, + recipients: makeRecipients(5), + config: { maxRecipientsPerBatch: -1 }, + }) + ).rejects.toThrow(/positive integer/i); + }); + + it("calls onBatchStart and onBatchComplete for each batch", async () => { + const onBatchStart = vi.fn(); + const onBatchComplete = vi.fn(); + const recipients = makeRecipients(15); + + await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 1000n, + recipients, + config: { maxRecipientsPerBatch: 10, onBatchStart, onBatchComplete }, + }); + + expect(onBatchStart).toHaveBeenCalledTimes(2); + expect(onBatchComplete).toHaveBeenCalledTimes(2); + expect(onBatchStart).toHaveBeenNthCalledWith(1, 1, 2, 10); + expect(onBatchStart).toHaveBeenNthCalledWith(2, 2, 2, 5); + }); + + it("propagates contract errors from distributeEqual", async () => { + mockContractClient.distribute_equal.mockRejectedValue( + new Error("InvalidAmount") + ); + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 0n, + recipients: makeRecipients(3), + }) + ).rejects.toThrow("InvalidAmount"); + }); + + it("returns no amountBatches for equal distribution", async () => { + const result = await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + total_amount: 1000n, + recipients: makeRecipients(3), + }); + + expect(result.amountBatches).toBeUndefined(); + }); + }); + + // ── weighted distribution ───────────────────────────────────────────────── + describe("weighted distribution (amounts present)", () => { + it("returns a single batch when recipients fit in one batch", async () => { + const recipients = makeRecipients(3); + const amounts = [300n, 400n, 300n]; + + const result = await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients, + amounts, + config: { maxRecipientsPerBatch: 10 }, + }); + + expect(result.batchCount).toBe(1); + expect(result.transactions).toHaveLength(1); + expect(result.amountBatches).toHaveLength(1); + expect(result.amountBatches![0]).toEqual(amounts); + expect(mockContractClient.distribute_weighted).toHaveBeenCalledTimes(1); + expect(mockContractClient.distribute_equal).not.toHaveBeenCalled(); + }); + + it("splits recipients and amounts in parallel into multiple batches", async () => { + const recipients = makeRecipients(25); + const amounts = recipients.map((_, i) => BigInt(i + 1)); + + const result = await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients, + amounts, + config: { maxRecipientsPerBatch: 10 }, + }); + + expect(result.batchCount).toBe(3); + expect(result.amountBatches).toHaveLength(3); + expect(result.amountBatches![0]).toHaveLength(10); + expect(result.amountBatches![2]).toHaveLength(5); + expect(result.recipientBatches[0]).toHaveLength(10); + }); + + it("passes correct slices to distributeWeighted", async () => { + const recipients = makeRecipients(15); + const amounts = recipients.map((_, i) => BigInt(i * 100)); + + await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients, + amounts, + config: { maxRecipientsPerBatch: 10 }, + }); + + const calls = mockContractClient.distribute_weighted.mock.calls; + expect(calls).toHaveLength(2); + expect(calls[0][0].recipients).toHaveLength(10); + expect(calls[0][0].amounts).toHaveLength(10); + expect(calls[1][0].recipients).toHaveLength(5); + expect(calls[1][0].amounts).toHaveLength(5); + }); + + it("rejects when recipients and amounts have different lengths", async () => { + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients: makeRecipients(3), + amounts: [100n, 200n], // length mismatch + }) + ).rejects.toThrow(/mismatch/i); + expect(mockContractClient.distribute_weighted).not.toHaveBeenCalled(); + }); + + it("rejects an empty recipients list", async () => { + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients: [], + amounts: [], + }) + ).rejects.toThrow(/empty/i); + }); + + it("rejects invalid maxRecipientsPerBatch", async () => { + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients: makeRecipients(5), + amounts: [1n, 2n, 3n, 4n, 5n], + config: { maxRecipientsPerBatch: 1.5 }, + }) + ).rejects.toThrow(/positive integer/i); + }); + + it("propagates contract errors from distributeWeighted", async () => { + mockContractClient.distribute_weighted.mockRejectedValue( + new Error("Unauthorized") + ); + await expect( + client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients: makeRecipients(3), + amounts: [1n, 2n, 3n], + }) + ).rejects.toThrow("Unauthorized"); + }); + + it("calls onBatchStart and onBatchComplete for each batch", async () => { + const onBatchStart = vi.fn(); + const onBatchComplete = vi.fn(); + const recipients = makeRecipients(15); + const amounts = recipients.map(() => 100n); + + await client.batchDistribute({ + sender: SENDER, + token: TOKEN, + recipients, + amounts, + config: { maxRecipientsPerBatch: 10, onBatchStart, onBatchComplete }, + }); + + expect(onBatchStart).toHaveBeenCalledTimes(2); + expect(onBatchComplete).toHaveBeenCalledTimes(2); + }); + + it("accepts Address objects for sender, token and recipients", async () => { + const recipients = [new Address(RECIPIENT_A), new Address(RECIPIENT_B)]; + const amounts = [600n, 400n]; + + const result = await client.batchDistribute({ + sender: new Address(SENDER), + token: new Address(TOKEN), + recipients, + amounts, + }); + + expect(result.batchCount).toBe(1); + expect(mockContractClient.distribute_weighted).toHaveBeenCalledTimes(1); + }); + }); + }); });