Sequential rank-ordered balance fetcher (#4063)

- **feat(app2): rank in token schema**
- **feat(app2): sequential token fetcher**
- **refactor(app2): more effective balance fetcher**
- **feat(app2): query tokens by rank**
- **feat(app2): general token blacklist mechanism**
- **fix(app2): cleanup**

Author: cor

app2/src/generated/graphql-env.d.ts

@@ -11,9 +11,10 @@ interface Props {
   chain: Chain
   denom: TokenRawDenom
   amount?: TokenRawAmount
+  showRank?: boolean
 }
 
-const { chain, denom, amount = undefined }: Props = $props()
+const { chain, denom, amount = undefined, showRank = true }: Props = $props()
 
 // Start the query when the component mounts
 $effect(() => {
@@ -89,6 +90,13 @@ const displayDenom = $derived(
           <section class="flex justify-between items-center">
             {#if token.value.representations.length > 0}
               <h2 class="text-white font-bold text-lg">{token.value.representations[0].symbol}</h2>
+              <span class="text-neutral-500">
+                {#if Option.isSome(token.value.rank)}
+                  Rank: #{token.value.rank.value}
+                {:else}
+                  Unranked
+                {/if}
+              </span>
             {/if}
           </section>
           <section>

app2/src/lib/components/model/TokenComponent.svelte

@@ -0,0 +1,16 @@
+import { TokenRawDenom } from "$lib/schema/token"
+
+/**
+ * List of token denoms that should be excluded from display and processing
+ */
+export const TOKEN_BLACKLIST: Array<TokenRawDenom> = [
+  TokenRawDenom.make("0x0000000000000000000000000000000000000000"),
+  TokenRawDenom.make("0xb7fb16053a3e3d4306791045769ec686f6ec4432")
+]
+
+/**
+ * Checks if a token denom is blacklisted
+ */
+export const isTokenBlacklisted = (denom: TokenRawDenom): boolean => {
+  return TOKEN_BLACKLIST.includes(denom)
+}

app2/src/lib/constants/tokens.ts

@@ -1,6 +1,7 @@
 import { Effect, Option, Schema } from "effect"
 import type { UniversalChainId } from "$lib/schema/chain"
-import { Token, TokenRawDenom } from "$lib/schema/token"
+import { Tokens } from "$lib/schema/token"
+import { isTokenBlacklisted } from "$lib/constants/tokens"
 import { createQueryGraphql } from "$lib/utils/queries"
 import { tokensStore } from "$lib/stores/tokens.svelte"
 import { graphql } from "gql.tada"
@@ -9,10 +10,10 @@ export const tokensQuery = (universalChainId: UniversalChainId) =>
   Effect.gen(function* () {
     yield* Effect.log(`zkgm starting token fetcher for ${universalChainId}`)
     const response = yield* createQueryGraphql({
-      schema: Schema.Struct({ v2_tokens: Schema.Array(Token) }),
+      schema: Schema.Struct({ v2_tokens: Tokens }),
       document: graphql(`
         query TokensForChain($universal_chain_id: String!) @cached(ttl: 60) {
-          v2_tokens(args: { p_universal_chain_id: $universal_chain_id }) {
+          v2_tokens(args: { p_universal_chain_id: $universal_chain_id }, order_by: {rank: asc_nulls_last}) {
             rank
             denom
             representations {
@@ -47,15 +48,8 @@ export const tokensQuery = (universalChainId: UniversalChainId) =>
         Effect.runSync(Effect.log(`storing new tokens for ${universalChainId}`))
         tokensStore.setData(
           universalChainId,
-          // Can be removed when this invalid denom is removed from hubble
-          data.pipe(
-            Option.map(d =>
-              d.v2_tokens.filter(
-                token =>
-                  token.denom !== TokenRawDenom.make("0x0000000000000000000000000000000000000000")
-              )
-            )
-          )
+          // Filter out blacklisted tokens
+          data.pipe(Option.map(d => d.v2_tokens.filter(token => !isTokenBlacklisted(token.denom))))
         )
       },
       writeError: error => {

app2/src/lib/queries/tokens.svelte.ts

@@ -51,6 +51,7 @@ export class TokenWrapping extends Schema.Class<TokenWrapping>("TokenWrapping")(
 }) {}
 
 export class Token extends Schema.Class<Token>("Token")({
+  rank: Schema.OptionFromNullOr(Schema.Int),
   denom: TokenRawDenom,
   representations: Schema.Array(TokenRepresentation),
   wrapping: Schema.Array(TokenWrapping)

app2/src/lib/schema/token.ts

@@ -1,6 +1,5 @@
-import { Data, Effect, Option, Schema, Schedule } from "effect"
+import { Data, Effect, Option, Schema } from "effect"
 import { fetchDecode } from "$lib/utils/queries"
-import type { DurationInput } from "effect/Duration"
 import { RawTokenBalance, TokenRawAmount, type TokenRawDenom } from "$lib/schema/token"
 import type { Chain, NoRpcError } from "$lib/schema/chain"
 import { type AddressCosmosCanonical, AddressCosmosDisplay } from "$lib/schema/address"
@@ -74,28 +73,23 @@ const fetchCosmosBankBalance = ({
     `${rpcUrl}/cosmos/bank/v1beta1/balances/${walletAddress}/by_denom?denom=${denom}`
   ).pipe(Effect.map(response => response.balance.amount))
 
-export const createCosmosBalanceQuery = ({
+// Core function to fetch a single Cosmos balance
+export const fetchCosmosBalance = ({
   chain,
   tokenAddress,
-  walletAddress,
-  refetchInterval,
-  writeData,
-  writeError
+  walletAddress
 }: {
   chain: Chain
   tokenAddress: TokenRawDenom
   walletAddress: AddressCosmosCanonical
-  refetchInterval: DurationInput
-  writeData: (data: RawTokenBalance) => void
-  writeError: (error: Option.Option<FetchCosmosBalanceError>) => void
 }) => {
-  const fetcherPipeline = Effect.gen(function* () {
+  return Effect.gen(function* () {
     const rpcUrl = yield* chain.requireRpcUrl("rest")
     const displayAddress = yield* chain.toCosmosDisplay(walletAddress)
     const decodedDenom = yield* fromHexString(tokenAddress)
 
     yield* Effect.log(
-      `starting balance fetcher for ${chain.universal_chain_id}:${displayAddress}:${decodedDenom}`
+      `fetching balance for ${chain.universal_chain_id}:${displayAddress}:${decodedDenom}`
     )
 
     const fetchBalance = decodedDenom.startsWith(`${chain.addr_prefix}1`)
@@ -112,17 +106,8 @@ export const createCosmosBalanceQuery = ({
 
     let balance = yield* Effect.retry(fetchBalance, cosmosBalanceRetrySchedule)
 
-    writeData(RawTokenBalance.make(Option.some(TokenRawAmount.make(balance))))
-    writeError(Option.none())
+    return RawTokenBalance.make(Option.some(TokenRawAmount.make(balance)))
   }).pipe(
-    Effect.tapError(error => Effect.sync(() => writeError(Option.some(error)))),
-    Effect.catchAll(_ => Effect.succeed(null))
-  )
-
-  return Effect.repeat(
-    fetcherPipeline,
-    Schedule.addDelay(Schedule.repeatForever, () => refetchInterval)
-  ).pipe(
     Effect.scoped,
     Effect.provide(FetchHttpClient.layer),
     withTracerDisabledWhen(() => true) // important! this prevents CORS issues: https://github.com/Effect-TS/effect/issues/4568

app2/src/lib/services/cosmos/balances.ts

@@ -1,7 +1,6 @@
-import { Schedule, Data, Effect, Option, Schema } from "effect"
+import { Data, Effect, Option, Schema } from "effect"
 import { erc20Abi, type PublicClient } from "viem"
 import type { TimeoutException } from "effect/Cause"
-import type { DurationInput } from "effect/Duration"
 import type { GetBalanceErrorType, ReadContractErrorType } from "viem"
 import { getPublicClient } from "$lib/services/evm/clients"
 import { RawTokenBalance, TokenRawAmount, type TokenRawDenom } from "$lib/schema/token"
@@ -66,27 +65,22 @@ const fetchEvmErc20Balance = ({
     catch: err => new ReadContractError({ cause: err as ReadContractErrorType })
   })
 
-export const createEvmBalanceQuery = ({
+// Core function to fetch a single EVM balance
+export const fetchEvmBalance = ({
   chain,
   tokenAddress,
-  walletAddress,
-  refetchInterval,
-  writeData,
-  writeError
+  walletAddress
 }: {
   chain: Chain
   tokenAddress: TokenRawDenom
   walletAddress: AddressEvmCanonical
-  refetchInterval: DurationInput
-  writeData: (data: RawTokenBalance) => void
-  writeError: (error: Option.Option<FetchEvmBalanceError>) => void
 }) => {
-  const fetcherPipeline = Effect.gen(function* (_) {
+  return Effect.gen(function* (_) {
     const client = yield* getPublicClient(chain)
     const decodedDenom = yield* fromHexString(tokenAddress)
 
     yield* Effect.log(
-      `starting balances fetcher for ${chain.universal_chain_id}:${walletAddress}:${tokenAddress}`
+      `fetching balance for ${chain.universal_chain_id}:${walletAddress}:${tokenAddress}`
     )
 
     const fetchBalance =
@@ -96,15 +90,6 @@ export const createEvmBalanceQuery = ({
 
     const balance = yield* Effect.retry(fetchBalance, evmBalanceRetrySchedule)
 
-    writeData(RawTokenBalance.make(Option.some(TokenRawAmount.make(balance))))
-    writeError(Option.none())
-  }).pipe(
-    Effect.tapError(error => Effect.sync(() => writeError(Option.some(error)))),
-    Effect.catchAll(_ => Effect.succeed(null))
-  )
-
-  return Effect.repeat(
-    fetcherPipeline,
-    Schedule.addDelay(Schedule.repeatForever, () => refetchInterval)
-  )
+    return RawTokenBalance.make(Option.some(TokenRawAmount.make(balance)))
+  })
 }

app2/src/lib/services/evm/balances.ts

@@ -1,12 +1,9 @@
-import { Effect, type Fiber, Option } from "effect"
+import { Effect, type Fiber, Option, Queue } from "effect"
 import type { TokenRawDenom } from "$lib/schema/token"
 import type { Chain, UniversalChainId } from "$lib/schema/chain"
 import { RawTokenBalance } from "$lib/schema/token"
-import { createEvmBalanceQuery, type FetchEvmBalanceError } from "$lib/services/evm/balances"
-import {
-  createCosmosBalanceQuery,
-  type FetchCosmosBalanceError
-} from "$lib/services/cosmos/balances"
+import { fetchEvmBalance, type FetchEvmBalanceError } from "$lib/services/evm/balances"
+import { fetchCosmosBalance, type FetchCosmosBalanceError } from "$lib/services/cosmos/balances"
 import { SvelteMap } from "svelte/reactivity"
 import {
   AddressEvmCanonical,
@@ -24,12 +21,29 @@ const createKey = (
   denom: TokenRawDenom
 ): BalanceKey => `${universalChainId}:${address}:${denom}`
 
+// Type for a balance fetch request
+type BalanceFetchRequest = {
+  chain: Chain
+  address: AddressCanonicalBytes
+  denom: TokenRawDenom
+}
+
+// Type for chain key
+type ChainKey = `${UniversalChainId}:${AddressCanonicalBytes}`
+
+// Helper to create the chain key
+const createChainKey = (
+  universalChainId: UniversalChainId,
+  address: AddressCanonicalBytes
+): ChainKey => `${universalChainId}:${address}`
+
 export class BalancesStore {
   data = $state(new SvelteMap<BalanceKey, RawTokenBalance>())
   errors = $state(
     new SvelteMap<BalanceKey, Option.Option<FetchEvmBalanceError | FetchCosmosBalanceError>>()
   )
-  fibers = $state(new SvelteMap<BalanceKey, Fiber.RuntimeFiber<number, never>>())
+  chainFibers = $state(new SvelteMap<ChainKey, Fiber.RuntimeFiber<void, never>>())
+  pendingRequests = $state(new SvelteMap<ChainKey, Array<BalanceFetchRequest>>())
 
   setBalance(
     universalChainId: UniversalChainId,
@@ -65,37 +79,102 @@ export class BalancesStore {
     return this.errors.get(createKey(universalChainId, address, denom)) ?? Option.none()
   }
 
-  fetchBalance(chain: Chain, address: AddressCanonicalBytes, denom: TokenRawDenom) {
-    const key = createKey(chain.universal_chain_id, address, denom)
+  // Process balance requests for a specific chain one at a time
+  private processBatchedBalances(
+    chain: Chain,
+    address: AddressCanonicalBytes,
+    denoms: Array<TokenRawDenom>
+  ) {
+    const chainKey = createChainKey(chain.universal_chain_id, address)
+    const self = this
 
-    // If there's already a query running for this combination, don't start another one
-    if (this.fibers.has(key)) {
+    // If there's already a query running for this chain, don't start another one
+    if (this.chainFibers.has(chainKey)) {
+      // Add these requests to pending
+      const existing = this.pendingRequests.get(chainKey) || []
+      const newRequests = denoms.map(denom => ({ chain, address, denom }))
+      this.pendingRequests.set(chainKey, [...existing, ...newRequests])
       return
     }
 
-    let query =
-      chain.rpc_type === "evm"
-        ? createEvmBalanceQuery({
-            chain,
-            tokenAddress: denom,
-            walletAddress: AddressEvmCanonical.make(address),
-            refetchInterval: "15 minutes",
-            writeData: balance =>
-              this.setBalance(chain.universal_chain_id, address, denom, balance),
-            writeError: error => this.setError(chain.universal_chain_id, address, denom, error)
-          })
-        : createCosmosBalanceQuery({
-            chain,
-            tokenAddress: denom,
-            walletAddress: AddressCosmosCanonical.make(address),
-            refetchInterval: "15 minutes",
-            writeData: balance =>
-              this.setBalance(chain.universal_chain_id, address, denom, balance),
-            writeError: error => this.setError(chain.universal_chain_id, address, denom, error)
-          })
-
-    const fiber = Effect.runFork(query)
-    this.fibers.set(key, fiber)
+    // Create a queue for processing balance requests
+    const batchProcessor = Effect.gen(function* (_) {
+      // Create a queue for balance requests
+      const queue = yield* Queue.unbounded<BalanceFetchRequest>()
+
+      // Add all denoms to the queue
+      for (const denom of denoms) {
+        yield* Queue.offer(queue, { chain, address, denom })
+      }
+
+      yield* Effect.forever(
+        Effect.gen(function* (_) {
+          // Take the next request from the queue
+          const request = yield* Queue.take(queue)
+          const { chain, address, denom } = request
+
+          // Process the balance request
+          yield* Effect.gen(function* (_) {
+            let balance: RawTokenBalance
+            if (chain.rpc_type === "evm") {
+              balance = yield* fetchEvmBalance({
+                chain,
+                tokenAddress: denom,
+                walletAddress: AddressEvmCanonical.make(address)
+              })
+            } else {
+              balance = yield* fetchCosmosBalance({
+                chain,
+                tokenAddress: denom,
+                walletAddress: AddressCosmosCanonical.make(address)
+              })
+            }
+
+            // Update the balance
+            self.setBalance(chain.universal_chain_id, address, denom, balance)
+            self.setError(chain.universal_chain_id, address, denom, Option.none())
+          }).pipe(
+            Effect.catchAll(error => {
+              // Update the error
+              self.setError(chain.universal_chain_id, address, denom, Option.some(error))
+              return Effect.succeed(undefined)
+            })
+          )
+        })
+      )
+    }).pipe(
+      Effect.catchAll(error => {
+        Effect.logError("error processing balance batch:", error)
+        return Effect.succeed(undefined)
+      }),
+      Effect.ensuring(
+        Effect.sync(() => {
+          // Check if there are pending requests for this chain
+          const pending = self.pendingRequests.get(chainKey) || []
+          self.pendingRequests.delete(chainKey)
+          self.chainFibers.delete(chainKey)
+
+          // If there are pending requests, process them
+          if (pending.length > 0) {
+            const pendingDenoms = pending.map(req => req.denom)
+            self.processBatchedBalances(chain, address, pendingDenoms)
+          }
+        })
+      )
+    )
+
+    // Run the batch processor
+    const fiber = Effect.runFork(batchProcessor)
+    this.chainFibers.set(chainKey, fiber)
+  }
+
+  fetchBalance(chain: Chain, address: AddressCanonicalBytes, denom: TokenRawDenom) {
+    this.processBatchedBalances(chain, address, [denom])
+  }
+
+  fetchBalances(chain: Chain, address: AddressCanonicalBytes, denoms: Array<TokenRawDenom>) {
+    if (denoms.length === 0) return
+    this.processBatchedBalances(chain, address, denoms)
   }
 }