Feat/distributor client batch distribute

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;

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<typeof distribute>[0];
 

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';

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<string>;
@@ -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;

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
  */

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()

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);