feat: add balancec-watcher updater

.github/workflows/ci.yml

@@ -18,6 +18,7 @@ env:
   REACT_APP_GOOGLE_ANALYTICS_ID: ${{ secrets.REACT_APP_GOOGLE_ANALYTICS_ID }}
   REACT_APP_BLOCKNATIVE_API_KEY: ${{ secrets.REACT_APP_BLOCKNATIVE_API_KEY }}
   REACT_APP_BFF_BASE_URL: ${{ secrets.BFF_BASE_URL }}
+  REACT_APP_BALANCES_WATCHER_BASE_URL: ${{ secrets.BALANCES_WATCHER_BASE_URL }}
   REACT_APP_CMS_BASE_URL: ${{ secrets.CMS_BASE_URL }}
   NEXT_PUBLIC_CMS_BASE_URL: ${{ secrets.CMS_BASE_URL }}
   PACKAGE_READ_AUTH_TOKEN: ${{ secrets.PACKAGE_READ_AUTH_TOKEN }}

.github/workflows/ipfs.yml

@@ -18,6 +18,7 @@ env:
   IPFS_DEPLOY_PINATA__API_KEY: ${{ secrets.REACT_APP_PINATA_API_KEY }}
   IPFS_DEPLOY_PINATA__SECRET_API_KEY: ${{ secrets.REACT_APP_PINATA_SECRET_API_KEY }}
   REACT_APP_BFF_BASE_URL: ${{ secrets.BFF_BASE_URL }}
+  REACT_APP_BALANCES_WATCHER_BASE_URL: ${{ secrets.BALANCES_WATCHER_BASE_URL }}
   REACT_APP_CMS_BASE_URL: ${{ secrets.CMS_BASE_URL }}
   NEXT_PUBLIC_CMS_BASE_URL: ${{ secrets.CMS_BASE_URL }}
   NODE_VERSION: lts/jod

.github/workflows/vercel.yml

@@ -81,6 +81,7 @@ jobs:
           REACT_APP_SUBGRAPH_URL_BASE: ${{ secrets.REACT_APP_SUBGRAPH_URL_BASE }}
           REACT_APP_SUBGRAPH_URL_GNOSIS_CHAIN: ${{ secrets.REACT_APP_SUBGRAPH_URL_GNOSIS_CHAIN }}
           REACT_APP_BFF_BASE_URL: ${{ secrets.BFF_BASE_URL }}
+          REACT_APP_BALANCES_WATCHER_BASE_URL: ${{ secrets.BALANCES_WATCHER_BASE_URL }}
           REACT_APP_CMS_BASE_URL: ${{ secrets.CMS_BASE_URL }}
           NEXT_PUBLIC_CMS_BASE_URL: ${{ secrets.CMS_BASE_URL }}
           REACT_APP_NEAR_API_KEY: ${{ secrets.REACT_APP_NEAR_API_KEY }}

apps/cowswap-frontend/src/modules/affiliate/api/bffAffiliateApi.ts

@@ -6,6 +6,7 @@ import {
   parseJsonResponse,
   RetryableResponseError,
   STATUS_CODES_TO_RETRY,
+  stripTrailingSlash,
   unwrapOk,
 } from '@cowprotocol/common-utils'
 import type { ApiErrorPayload, FetchJsonResponse } from '@cowprotocol/common-utils'
@@ -42,7 +43,7 @@ class BffAffiliateApi {
    */
 
   constructor(baseUrl: string, timeoutMs: number = AFFILIATE_API_TIMEOUT_MS) {
-    this.baseUrl = baseUrl.replace(/\/$/, '')
+    this.baseUrl = stripTrailingSlash(baseUrl)
     this.timeoutMs = timeoutMs
     this.fetchRateLimited = fetchWithRateLimit({
       rateLimit: {

apps/cowswap-frontend/src/modules/balancesAndAllowances/updaters/CommonPriorityBalancesAndAllowancesUpdater.tsx

@@ -2,9 +2,11 @@ import { ReactNode, useEffect, useMemo, useState } from 'react'
 
 import {
   BalancesAndAllowancesUpdater,
+  BalancesWatcherUpdater,
   PRIORITY_TOKENS_REFRESH_INTERVAL,
   PriorityTokensUpdater,
 } from '@cowprotocol/balances-and-allowances'
+import { useFeatureFlags } from '@cowprotocol/common-hooks'
 import { useWalletInfo } from '@cowprotocol/wallet'
 
 import { useBalancesContext } from 'entities/balancesContext/useBalancesContext'
@@ -20,6 +22,8 @@ export function CommonPriorityBalancesAndAllowancesUpdater(): ReactNode {
   const balancesContext = useBalancesContext()
   const balancesAccount = balancesContext.account || account
 
+  const { useBalancesWatcher: isBalancesWatcherEnabled } = useFeatureFlags()
+
   const priorityTokenAddresses = usePriorityTokenAddresses()
   const priorityTokenAddressesAsArray = useMemo(() => {
     return Array.from(priorityTokenAddresses.values())
@@ -52,6 +56,10 @@ export function CommonPriorityBalancesAndAllowancesUpdater(): ReactNode {
 
   const refreshTrigger = useOrdersFilledEventsTrigger()
 
+  if (isBalancesWatcherEnabled) {
+    return <BalancesWatcherUpdater account={balancesAccount} chainId={sourceChainId} />
+  }
+
   return (
     <>
       <PriorityTokensUpdater

libs/balances-and-allowances/src/balancesWatcher/createSession.test.ts

@@ -0,0 +1,101 @@
+import { SupportedChainId } from '@cowprotocol/cow-sdk'
+
+import { createBalancesWatcherSession } from './createSession'
+import { BalancesWatcherApiError } from './types'
+
+const OWNER = '0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045'
+const BASE_URL = 'https://watcher.example'
+
+function mockFetchResponse(status: number, body: unknown): jest.SpyInstance {
+  const init: ResponseInit = {
+    status,
+    headers: { 'Content-Type': 'application/json' },
+  }
+  const text = typeof body === 'string' ? body : JSON.stringify(body)
+  return jest.spyOn(global, 'fetch').mockResolvedValue(new Response(text, init))
+}
+
+describe('createBalancesWatcherSession', () => {
+  afterEach(() => {
+    jest.restoreAllMocks()
+  })
+
+  it('POSTs to /{chainId}/sessions/{owner} with the request body and resolves on 2xx', async () => {
+    const fetchSpy = mockFetchResponse(200, '')
+
+    await createBalancesWatcherSession({
+      chainId: SupportedChainId.MAINNET,
+      owner: OWNER,
+      body: { tokensListsUrls: ['https://lists.example/uni.json'], customTokens: ['0xabc'] },
+      baseUrl: BASE_URL,
+    })
+
+    expect(fetchSpy).toHaveBeenCalledTimes(1)
+    const [calledUrl, calledInit] = fetchSpy.mock.calls[0] as [string, RequestInit]
+    expect(calledUrl).toBe(`${BASE_URL}/1/sessions/${OWNER}`)
+    expect(calledInit.method).toBe('POST')
+    expect(JSON.parse(calledInit.body as string)).toEqual({
+      tokensListsUrls: ['https://lists.example/uni.json'],
+      customTokens: ['0xabc'],
+    })
+  })
+
+  it('strips a trailing slash from baseUrl', async () => {
+    const fetchSpy = mockFetchResponse(200, '')
+
+    await createBalancesWatcherSession({
+      chainId: SupportedChainId.GNOSIS_CHAIN,
+      owner: OWNER,
+      body: { tokensListsUrls: [], customTokens: ['0xabc'] },
+      baseUrl: `${BASE_URL}/`,
+    })
+
+    const [calledUrl] = fetchSpy.mock.calls[0]
+    expect(calledUrl).toBe(`${BASE_URL}/100/sessions/${OWNER}`)
+  })
+
+  it('throws BalancesWatcherApiError with code+status when the server returns the JSON envelope', async () => {
+    mockFetchResponse(400, { code: 400, message: 'Bad request: tokens_lists_urls && custom_tokens are empty' })
+
+    await expect(
+      createBalancesWatcherSession({
+        chainId: SupportedChainId.MAINNET,
+        owner: OWNER,
+        body: { tokensListsUrls: [], customTokens: [] },
+        baseUrl: BASE_URL,
+      }),
+    ).rejects.toMatchObject({
+      name: 'BalancesWatcherApiError',
+      status: 400,
+      code: 400,
+      message: 'Bad request: tokens_lists_urls && custom_tokens are empty',
+    })
+  })
+
+  it('falls back to {code: status, message: raw body} when the error body is not JSON', async () => {
+    mockFetchResponse(503, 'upstream unreachable')
+
+    const error = await createBalancesWatcherSession({
+      chainId: SupportedChainId.MAINNET,
+      owner: OWNER,
+      body: { tokensListsUrls: ['https://lists.example/x.json'], customTokens: [] },
+      baseUrl: BASE_URL,
+    }).catch((e: unknown) => e)
+
+    expect(error).toBeInstanceOf(BalancesWatcherApiError)
+    expect(error).toMatchObject({ status: 503, code: 503, message: 'upstream unreachable' })
+  })
+
+  it('surfaces 404 chain mismatch responses', async () => {
+    mockFetchResponse(404, { code: 404, message: 'Not found' })
+
+    await expect(
+      createBalancesWatcherSession({
+        chainId: SupportedChainId.MAINNET,
+        owner: OWNER,
+        body: { tokensListsUrls: ['https://lists.example/x.json'], customTokens: [] },
+        baseUrl: BASE_URL,
+      }),
+    ).rejects.toMatchObject({ status: 404, code: 404 })
+  })
+})

libs/balances-and-allowances/src/balancesWatcher/createSession.ts

@@ -0,0 +1,38 @@
+import { BALANCES_WATCHER_BASE_URL } from '@cowprotocol/common-const'
+import { fetchWithTimeout, JSON_HEADERS, parseJsonResponse, stripTrailingSlash } from '@cowprotocol/common-utils'
+import type { SupportedChainId } from '@cowprotocol/cow-sdk'
+
+import { BalancesWatcherApiError, type BalancesWatcherErrorPayload, type CreateSessionRequest } from './types'
+
+const DEFAULT_SESSION_TIMEOUT_MS = 10_000
+
+export interface CreateSessionParams {
+  chainId: SupportedChainId
+  owner: string
+  body: CreateSessionRequest
+  baseUrl?: string
+  timeoutMs?: number
+}
+
+/**
+ * Step 1 of 2 in the balances-watcher handshake: registers the wallet with the
+ * watcher and tells it which token lists and individual token addresses to
+ * track. Step 2 — opening the SSE balance stream — is done by
+ * `subscribeToBalancesEvents` after this call resolves.
+ */
+export async function createBalancesWatcherSession(params: CreateSessionParams): Promise<void> {
+  const baseUrl = stripTrailingSlash(params.baseUrl ?? BALANCES_WATCHER_BASE_URL)
+  const url = `${baseUrl}/${params.chainId}/sessions/${params.owner}`
+
+  const response = await fetchWithTimeout(url, {
+    method: 'POST',
+    headers: JSON_HEADERS,
+    body: JSON.stringify(params.body),
+    timeout: params.timeoutMs ?? DEFAULT_SESSION_TIMEOUT_MS,
+  })
+
+  if (response.ok) return
+
+  const { data, text } = await parseJsonResponse<BalancesWatcherErrorPayload>(response)
+  throw new BalancesWatcherApiError(response.status, data ?? { code: response.status, message: text })
+}

libs/balances-and-allowances/src/balancesWatcher/index.ts

@@ -0,0 +1,8 @@
+export { createBalancesWatcherSession } from './createSession'
+export type { CreateSessionParams } from './createSession'
+
+export { subscribeToBalancesEvents } from './subscribeToBalancesEvents'
+export type { BalancesSubscription, SubscribeToBalancesEventsParams } from './subscribeToBalancesEvents'
+
+export { BalancesWatcherApiError, BalancesWatcherStreamError } from './types'
+export type { BalanceUpdateEvent, BalancesMap, BalancesWatcherErrorPayload, CreateSessionRequest } from './types'