refactor: use cross-chain stream api

Author: piavgh

apps/kyberswap-interface/src/pages/CrossChainSwap/hooks/useCrossChainSwap.tsx

@@ -34,11 +34,74 @@ import { CrossChainSwapFactory } from '../factory'
 import { CrossChainSwapAdapterRegistry, Quote } from '../registry'
 import { NEAR_STABLE_COINS, SOLANA_STABLE_COINS, isCanonicalPair } from '../utils'
 
+// Mapping from ChainId/NonEvmChain to aggregator API chain names
+const CHAIN_TO_API_NAME: Partial<Record<Chain, string>> = {
+  [NonEvmChain.Bitcoin]: 'bitcoin',
+  [NonEvmChain.Near]: 'near',
+  [NonEvmChain.Solana]: 'solana',
+  [ChainId.MAINNET]: 'ethereum',
+  [ChainId.BASE]: 'base',
+  [ChainId.ARBITRUM]: 'arbitrum',
+  [ChainId.OPTIMISM]: 'optimism',
+  [ChainId.MATIC]: 'polygon',
+  [ChainId.BSCMAINNET]: 'bsc',
+  [ChainId.AVAXMAINNET]: 'avalanche',
+  [ChainId.LINEA]: 'linea',
+  [ChainId.SCROLL]: 'scroll',
+  [ChainId.BLAST]: 'blast',
+  [ChainId.ZKSYNC]: 'zksync',
+  [ChainId.SONIC]: 'sonic',
+  [ChainId.FANTOM]: 'fantom',
+  [ChainId.BERA]: 'bera',
+  [ChainId.MANTLE]: 'mantle',
+  [ChainId.RONIN]: 'ronin',
+  [ChainId.UNICHAIN]: 'unichain',
+  [ChainId.HYPEREVM]: 'hyperevm',
+}
+
+// Reverse mapping from API chain names to ChainId/NonEvmChain
+export const API_NAME_TO_CHAIN: Record<string, Chain> = {
+  bitcoin: NonEvmChain.Bitcoin,
+  near: NonEvmChain.Near,
+  solana: NonEvmChain.Solana,
+  ethereum: ChainId.MAINNET,
+  base: ChainId.BASE,
+  arbitrum: ChainId.ARBITRUM,
+  optimism: ChainId.OPTIMISM,
+  polygon: ChainId.MATIC,
+  bsc: ChainId.BSCMAINNET,
+  avalanche: ChainId.AVAXMAINNET,
+  linea: ChainId.LINEA,
+  scroll: ChainId.SCROLL,
+  blast: ChainId.BLAST,
+  zksync: ChainId.ZKSYNC,
+  sonic: ChainId.SONIC,
+  fantom: ChainId.FANTOM,
+  bera: ChainId.BERA,
+  mantle: ChainId.MANTLE,
+  ronin: ChainId.RONIN,
+  unichain: ChainId.UNICHAIN,
+  hyperevm: ChainId.HYPEREVM,
+}
+
+const getChainName = (chain: Chain): string => {
+  return CHAIN_TO_API_NAME[chain] || chain.toString()
+}
+
+export const getChainIdFromName = (chainName: string): Chain => {
+  return API_NAME_TO_CHAIN[chainName.toLowerCase()] || chainName
+}
+
 export const registry = new CrossChainSwapAdapterRegistry()
 CrossChainSwapFactory.getAllAdapters().forEach(adapter => {
   registry.registerAdapter(adapter)
 })
 
+console.log(
+  'Registered adapters:',
+  registry.getAllAdapters().map(a => a.getName()),
+)
+
 // Helper function to create a timeout promise
 const createTimeoutPromise = (ms: number) => {
   return new Promise<never>((_, reject) => {
@@ -516,6 +579,7 @@ export const CrossChainSwapRegistryProvider = ({ children }: { children: React.R
 
     setLoading(true)
     setAllLoading(true)
+    setQuotes([]) // Clear previous quotes
 
     const getQuotesWithCancellation = async (params: QuoteParams | NearQuoteParams) => {
       let allAdapters = registry.getAllAdapters().filter(a => !excludedSources.includes(a.getName()))
@@ -525,8 +589,10 @@ export const CrossChainSwapRegistryProvider = ({ children }: { children: React.R
         allAdapters = registry.getAllAdapters()
       }
 
-      // Create a modified version of getQuotes that can be cancelled
+      // Build API URL with query parameters
       const quotes: Quote[] = []
+
+      // Determine which adapters should be used (for filtering later)
       const adapters =
         params.fromChain === params.toChain &&
         isEvmChain(params.fromChain) &&
@@ -541,38 +607,169 @@ export const CrossChainSwapRegistryProvider = ({ children }: { children: React.R
                 adapter.getSupportedChains().includes(params.fromChain) &&
                 adapter.getSupportedChains().includes(params.toChain),
             )
-      // Map each adapter to a promise that can be cancelled
-      const quotePromises = adapters.map(async adapter => {
-        try {
-          // Check for cancellation before starting
-          if (signal.aborted) throw new Error('Cancelled')
-
-          // Race between the adapter quote and timeout
-          const quote = await Promise.race([adapter.getQuote(params), createTimeoutPromise(9_000)])
-
-          // Check for cancellation after getting quote
-          if (signal.aborted) throw new Error('Cancelled')
-
-          quotes.push({ adapter, quote })
-          const sortedQuotes = [...quotes].sort((a, b) => (a.quote.outputAmount < b.quote.outputAmount ? 1 : -1))
-          setQuotes(sortedQuotes)
-          setLoading(false)
-        } catch (err) {
-          if (err.message === 'Cancelled' || signal.aborted) {
+
+      // Construct the API URL with parameters
+      const fromToken = (params.fromToken as any).isNative
+        ? ZERO_ADDRESS
+        : (params.fromToken as any).wrapped?.address ||
+          (params.fromToken as any).address ||
+          (params.fromToken as any).id ||
+          (params.fromToken as any).assetId
+      const toToken = (params.toToken as any).isNative
+        ? ZERO_ADDRESS
+        : (params.toToken as any).wrapped?.address ||
+          (params.toToken as any).address ||
+          (params.toToken as any).id ||
+          (params.toToken as any).assetId
+      const fromTokenDecimals = params.fromToken.decimals
+      const toTokenDecimals = params.toToken.decimals
+
+      const queryParams = new URLSearchParams({
+        fromChain: getChainName(params.fromChain),
+        fromToken,
+        fromTokenDecimals: fromTokenDecimals.toString(),
+        fromAddress: params.sender,
+        fromAmount: params.amount,
+        toChain: getChainName(params.toChain),
+        toToken,
+        toTokenDecimals: toTokenDecimals.toString(),
+        toAddress: params.recipient,
+        fee: params.feeBps.toString(),
+        integrator: 'kyberswap',
+        stream: 'true',
+        slippage: params.slippage.toString(),
+        fromTokenUsd: (params as any).tokenInUsd?.toString() || '0',
+        toTokenUsd: (params as any).tokenOutUsd?.toString() || '0',
+      })
+
+      const apiUrl = `https://pre-crosschain-aggregator.kyberengineering.io/api/v1/quotes?${queryParams.toString()}`
+
+      console.log('Fetching quotes from streaming API:', apiUrl)
+
+      try {
+        // Check for cancellation before starting
+        if (signal.aborted) throw new Error('Cancelled')
+
+        // Fetch with streaming
+        const response = await fetch(apiUrl, { signal })
+
+        console.log('Streaming API response status:', response.status)
+
+        if (!response.ok) {
+          throw new Error(`HTTP error! status: ${response.status}`)
+        }
+
+        const reader = response.body?.getReader()
+        const decoder = new TextDecoder()
+
+        if (!reader) {
+          throw new Error('No response body reader available')
+        }
+
+        let buffer = ''
+
+        // Read the stream
+        while (true) {
+          // Check for cancellation
+          if (signal.aborted) {
+            reader.cancel()
             throw new Error('Cancelled')
           }
-          console.error(`Failed to get quote from ${adapter.getName()}:`, err)
+
+          const { done, value } = await reader.read()
+
+          if (done) break
+
+          buffer += decoder.decode(value, { stream: true })
+
+          // Process complete SSE messages
+          const lines = buffer.split('\n')
+          buffer = lines.pop() || '' // Keep incomplete line in buffer
+
+          for (const line of lines) {
+            // Skip empty lines and event lines
+            if (!line || line.startsWith('event:')) {
+              continue
+            }
+
+            if (line.startsWith('data:')) {
+              try {
+                // Handle both "data:{json}" and "data: {json}" formats
+                const jsonStr = line.startsWith('data: ') ? line.slice(6) : line.slice(5)
+                const data = JSON.parse(jsonStr)
+
+                // Skip complete messages
+                if (data.message === 'All quotes received') {
+                  console.log('All quotes received from streaming API')
+                  continue
+                }
+
+                // Find the corresponding adapter
+                console.log('Processing quote from provider:', data.provider)
+                const adapter = registry.getAdapter(data.provider)
+
+                if (adapter) {
+                  console.log('Adapter found:', adapter.getName(), 'Output amount:', data.outputAmount)
+                  // Convert the API response to NormalizedQuote format
+                  // Need to convert chain names back to Chain IDs for the adapters to work correctly
+                  const normalizedQuote = {
+                    quoteParams: {
+                      ...data.quoteParams,
+                      fromChain: getChainIdFromName(data.quoteParams.fromChain),
+                      toChain: getChainIdFromName(data.quoteParams.toChain),
+                    },
+                    outputAmount: BigInt(data.outputAmount),
+                    formattedOutputAmount: data.formattedOutputAmount,
+                    inputUsd: data.inputUsd,
+                    outputUsd: data.outputUsd,
+                    rate: data.rate,
+                    timeEstimate: data.timeEstimate,
+                    priceImpact: data.priceImpact,
+                    gasFeeUsd: data.gasFeeUsd,
+                    contractAddress: data.contractAddress,
+                    rawQuote: data.rawQuote,
+                    protocolFee: 0, // Protocol fee is included in the quote
+                    protocolFeeString: data.protocolFeeString,
+                    platformFeePercent: data.platformFeePercent,
+                  }
+
+                  quotes.push({ adapter, quote: normalizedQuote })
+                  console.log('Total quotes so far:', quotes.length)
+                  const sortedQuotes = [...quotes].sort((a, b) =>
+                    a.quote.outputAmount < b.quote.outputAmount ? 1 : -1,
+                  )
+                  console.log('Setting quotes state with', sortedQuotes.length, 'quotes')
+                  setQuotes(sortedQuotes)
+                  setLoading(false)
+                } else {
+                  console.warn(`Adapter not found in registry: ${data.provider}`)
+                  console.log(
+                    'Available adapters:',
+                    registry.getAllAdapters().map(a => a.getName()),
+                  )
+                }
+              } catch (err) {
+                console.error('Failed to parse SSE data:', err)
+                console.error('Problematic line:', line)
+                console.error('Line length:', line.length)
+              }
+            }
+          }
         }
-      })
 
-      await Promise.all(quotePromises)
+        if (quotes.length === 0) {
+          throw new Error('No valid quotes found for the requested swap')
+        }
 
-      if (quotes.length === 0) {
-        throw new Error('No valid quotes found for the requested swap')
+        quotes.sort((a, b) => (a.quote.outputAmount < b.quote.outputAmount ? 1 : -1))
+        return quotes
+      } catch (err) {
+        if ((err as Error).message === 'Cancelled' || signal.aborted) {
+          throw new Error('Cancelled')
+        }
+        console.error('Failed to get quotes from streaming API:', err)
+        throw err
       }
-
-      quotes.sort((a, b) => (a.quote.outputAmount < b.quote.outputAmount ? 1 : -1))
-      return quotes
     }
 
     const adaptedWallet = adaptSolanaWallet(
@@ -582,28 +779,32 @@ export const CrossChainSwapRegistryProvider = ({ children }: { children: React.R
       connection.sendTransaction as any,
     )
 
-    const q = await getQuotesWithCancellation({
-      feeBps,
-      tokenInUsd: tokenInUsd,
-      tokenOutUsd: tokenOutUsd,
-      fromChain: fromChainId,
-      toChain: toChainId,
-      fromToken: currencyIn,
-      toToken: currencyOut,
-      amount: inputAmount,
-      slippage,
-      walletClient: fromChainId === 'solana' ? adaptedWallet : walletClient?.data,
-      sender,
-      recipient: receiver,
-      nearTokens,
-      publicKey: btcPublicKey || '',
-    }).catch(e => {
-      console.log(e)
-      return []
-    })
-    setQuotes(q)
-    setLoading(false)
-    setAllLoading(false)
+    try {
+      await getQuotesWithCancellation({
+        feeBps,
+        tokenInUsd: tokenInUsd,
+        tokenOutUsd: tokenOutUsd,
+        fromChain: fromChainId,
+        toChain: toChainId,
+        fromToken: currencyIn,
+        toToken: currencyOut,
+        amount: inputAmount,
+        slippage,
+        walletClient: fromChainId === 'solana' ? adaptedWallet : walletClient?.data,
+        sender,
+        recipient: receiver,
+        nearTokens,
+        publicKey: btcPublicKey || '',
+      })
+    } catch (e) {
+      console.error('Error getting quotes:', e)
+      if ((e as Error).message !== 'Cancelled') {
+        setQuotes([])
+      }
+    } finally {
+      setLoading(false)
+      setAllLoading(false)
+    }
   }, [
     sender,
     receiver,

apps/kyberswap-interface/src/pages/CrossChainSwap/registry.ts

@@ -14,11 +14,11 @@ export class CrossChainSwapAdapterRegistry {
   private adapters: Map<string, SwapProvider> = new Map()
 
   registerAdapter(adapter: SwapProvider): void {
-    this.adapters.set(adapter.getName().toLowerCase(), adapter)
+    this.adapters.set(adapter.getName().toLowerCase().replace(/\s+/g, ''), adapter)
   }
 
   getAdapter(name: string): SwapProvider | undefined {
-    return this.adapters.get(name.toLowerCase())
+    return this.adapters.get(name.toLowerCase().replace(/\s+/g, ''))
   }
 
   getAllAdapters(): SwapProvider[] {