Add Liquidity Pools methods

src/Centrifuge.ts

@@ -4,20 +4,21 @@ import {
   defaultIfEmpty,
   defer,
   filter,
-  firstValueFrom,
   identity,
   isObservable,
   map,
   mergeMap,
   of,
   Subject,
+  switchMap,
   using,
 } from 'rxjs'
 import { fromFetch } from 'rxjs/fetch'
 import {
   createPublicClient,
   createWalletClient,
   custom,
+  getContract,
   http,
   parseEventLogs,
   type Abi,
@@ -27,14 +28,18 @@ import {
   type WalletClient,
   type WatchEventOnLogsParameter,
 } from 'viem'
+import { ABI } from './abi/index.js'
 import { Account } from './Account.js'
 import { chains } from './config/chains.js'
+import type { CurrencyMetadata } from './config/lp.js'
+import { PERMIT_TYPEHASH } from './constants.js'
 import { Pool } from './Pool.js'
 import type { HexString } from './types/index.js'
 import type { CentrifugeQueryOptions, Query } from './types/query.js'
 import type { OperationStatus, Signer, Transaction, TransactionCallbackParams } from './types/transaction.js'
-import { hashKey, serializeForCache } from './utils/query.js'
-import { makeThenable, shareReplayWithDelayedReset } from './utils/rx.js'
+import { Currency } from './utils/BigInt.js'
+import { hashKey } from './utils/query.js'
+import { makeThenable, repeatOnEvents, shareReplayWithDelayedReset } from './utils/rx.js'
 import { doTransaction, isLocalAccount } from './utils/transaction.js'
 
 export type Config = {
@@ -132,6 +137,83 @@ export class Centrifuge {
     return this._query(null, () => of(new Account(this, address as any, chainId ?? this.config.defaultChain)))
   }
 
+  /**
+   * Get the metadata for an ERC20 token
+   * @param address - The token address
+   * @param chainId - The chain ID
+   */
+  currency(address: string, chainId?: number): Query<CurrencyMetadata> {
+    const curAddress = address.toLowerCase()
+    const cid = chainId ?? this.config.defaultChain
+    return this._query(['currency', curAddress, cid], () =>
+      defer(async () => {
+        const contract = getContract({
+          address: curAddress as any,
+          abi: ABI.Currency,
+          client: this.getClient(cid)!,
+        })
+        const [decimals, name, symbol, supportsPermit] = await Promise.all([
+          contract.read.decimals!() as Promise<number>,
+          contract.read.name!() as Promise<string>,
+          contract.read.symbol!() as Promise<string>,
+          contract.read.PERMIT_TYPEHASH!()
+            .then((hash) => hash === PERMIT_TYPEHASH)
+            .catch(() => false),
+        ])
+        return {
+          address: curAddress as any,
+          decimals,
+          name,
+          symbol,
+          chainId: cid,
+          supportsPermit,
+        }
+      })
+    )
+  }
+
+  /**
+   * Get the balance of an ERC20 token for a given owner.
+   * @param currency - The token address
+   * @param owner - The owner address
+   * @param chainId - The chain ID
+   */
+  balance(currency: string, owner: string, chainId?: number) {
+    const address = owner.toLowerCase()
+    const cid = chainId ?? this.config.defaultChain
+    return this._query(['balance', currency, owner, cid], () => {
+      return this.currency(currency, cid).pipe(
+        switchMap((currencyMeta) =>
+          defer(() =>
+            this.getClient(cid)!
+              .readContract({
+                address: currency as any,
+                abi: ABI.Currency,
+                functionName: 'balanceOf',
+                args: [address],
+              })
+              .then((val: any) => new Currency(val, currencyMeta.decimals))
+          ).pipe(
+            repeatOnEvents(
+              this,
+              {
+                address: currency,
+                abi: ABI.Currency,
+                eventName: 'Transfer',
+                filter: (events) => {
+                  return events.some((event) => {
+                    return event.args.from?.toLowerCase() === address || event.args.to?.toLowerCase() === address
+                  })
+                },
+              },
+              cid
+            )
+          )
+        )
+      )
+    })
+  }
+
   /**
    * Returns an observable of all events on a given chain.
    * @internal
@@ -226,7 +308,7 @@ export class Centrifuge {
 
   #memoized = new Map<string, any>()
   #memoizeWith<T = any>(keys: any[], callback: () => T): T {
-    const cacheKey = hashKey(serializeForCache(keys))
+    const cacheKey = hashKey(keys)
     if (this.#memoized.has(cacheKey)) {
       return this.#memoized.get(cacheKey)
     }
@@ -332,13 +414,16 @@ export class Centrifuge {
       }
 
       const $query = createShared().pipe(
-        // For new subscribers, recreate the shared observable if the previously shared observable has completed
-        // and no longer has a cached value, which can happen with a finite `valueCacheTime`.
+        // When `valueCacheTime` is finite, and the cached value is expired,
+        // the shared observable will immediately complete upon the next subscription.
+        // This will cause the shared observable to be recreated.
         defaultIfEmpty(defer(createShared)),
         // For existing subscribers, merge any newly created shared observable.
         concatWith(sharedSubject),
+        // Flatten observables emitted from the `sharedSubject`
         mergeMap((d) => (isObservable(d) ? d : of(d)))
       )
+
       return makeThenable($query)
     }
     return keys ? this.#memoizeWith(keys, get) : get()
@@ -381,14 +466,12 @@ export class Centrifuge {
    */
   _transact(
     title: string,
-    transactionCallback: (params: TransactionCallbackParams) => Promise<HexString> | Observable<HexString>,
+    transactionCallback: (params: TransactionCallbackParams) => Promise<HexString>,
     chainId?: number
   ) {
     return this._transactSequence(async function* (params) {
       const transaction = transactionCallback(params)
-      yield* doTransaction(title, params.publicClient, () =>
-        'then' in transaction ? transaction : firstValueFrom(transaction)
-      )
+      yield* doTransaction(title, params.publicClient, () => transaction)
     }, chainId)
   }
 
@@ -430,7 +513,7 @@ export class Centrifuge {
       params: TransactionCallbackParams
     ) => AsyncGenerator<OperationStatus> | Observable<OperationStatus>,
     chainId?: number
-  ) {
+  ): Transaction {
     const targetChainId = chainId ?? this.config.defaultChain
     const self = this
     async function* transact() {

src/Pool.test.ts

@@ -0,0 +1,20 @@
+import { expect } from 'chai'
+import { Pool } from './Pool.js'
+import { context } from './tests/setup.js'
+
+const poolId = '2779829532'
+
+describe('Pool', () => {
+  let pool: Pool
+  beforeEach(() => {
+    const { centrifuge } = context
+    pool = new Pool(centrifuge, poolId)
+  })
+
+  it('get active networks of a pool', async () => {
+    const networks = await pool.activeNetworks()
+    expect(networks).to.have.length(2)
+    expect(networks[0]!.chainId).to.equal(11155111)
+    expect(networks[1]!.chainId).to.equal(84532)
+  })
+})

src/Pool.ts

@@ -1,8 +1,8 @@
 import { catchError, combineLatest, map, of, switchMap, timeout } from 'rxjs'
 import type { Centrifuge } from './Centrifuge.js'
 import { Entity } from './Entity.js'
-import { Reports } from './Reports/index.js'
 import { PoolNetwork } from './PoolNetwork.js'
+import { Reports } from './Reports/index.js'
 
 export class Pool extends Entity {
   constructor(
@@ -49,6 +49,21 @@ export class Pool extends Entity {
     })
   }
 
+  /**
+   * Get a specific network where a pool can potentially be deployed.
+   */
+  network(chainId: number) {
+    return this._query(null, () => {
+      return this.networks().pipe(
+        map((networks) => {
+          const network = networks.find((network) => network.chainId === chainId)
+          if (!network) throw new Error(`Network ${chainId} not found`)
+          return network
+        })
+      )
+    })
+  }
+
   /**
    * Get the networks where a pool is active. It doesn't mean that any vaults are deployed there necessarily.
    */
@@ -70,4 +85,8 @@ export class Pool extends Entity {
       )
     })
   }
+
+  vault(chainId: number, trancheId: string, asset: string) {
+    return this._query(null, () => this.network(chainId).pipe(switchMap((network) => network.vault(trancheId, asset))))
+  }
 }

src/PoolNetwork.test.ts

@@ -0,0 +1,37 @@
+import { expect } from 'chai'
+import sinon from 'sinon'
+import { Pool } from './Pool.js'
+import { PoolNetwork } from './PoolNetwork.js'
+import { context } from './tests/setup.js'
+
+const poolId = '2779829532'
+const trancheId = '0xac6bffc5fd68f7772ceddec7b0a316ca'
+const vaultAddress = '0x05eb35c2e4fa21fb06d3fab92916191b254b3504'
+
+describe('PoolNetwork', () => {
+  let poolNetwork: PoolNetwork
+  beforeEach(() => {
+    const { centrifuge } = context
+    const pool = new Pool(centrifuge, poolId)
+    poolNetwork = new PoolNetwork(centrifuge, pool, 11155111)
+  })
+
+  it('should get whether a pool is deployed to a network', async () => {
+    const isActive = await poolNetwork.isActive()
+    expect(isActive).to.equal(true)
+
+    // non-active pool/network
+    const poolNetwork2 = new PoolNetwork(context.centrifuge, new Pool(context.centrifuge, '123'), 11155111)
+    const isActive2 = await poolNetwork2.isActive()
+    expect(isActive2).to.equal(false)
+  })
+
+  it('get vaults for a tranche', async () => {
+    const fetchSpy = sinon.spy(globalThis, 'fetch')
+    const vaults = await poolNetwork.vaults(trancheId)
+    expect(vaults).to.have.length(1)
+    expect(vaults[0]!.address.toLowerCase()).to.equal(vaultAddress)
+    // Calls should get batched
+    expect(fetchSpy.getCalls().length).to.equal(1)
+  })
+})