fix(coinmetrics): filter invalid assets in lwba ws transport

.changeset/fix-coinmetrics-lwba-invalid-assets.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/coinmetrics-adapter': patch
+---
+
+fix(coinmetrics): filter invalid assets in lwba ws transport

packages/sources/coinmetrics/src/transport/error-handling.ts

@@ -2,6 +2,7 @@ import { makeLogger } from '@chainlink/external-adapter-framework/util'
 import { ResponseError } from './types'
 
 const logger = makeLogger('CoinMetrics Crypto error handling')
+const unsupportedAssetRegex = /\bValue\s+'([^']+)'/i
 
 export const logPossibleSolutionForKnownErrors = (error: ResponseError) => {
   if (error['type'] === 'wrong_credentials') {
@@ -19,3 +20,13 @@ export const logPossibleSolutionForKnownErrors = (error: ResponseError) => {
       2. There maybe an issue with the job spec or the Data Provider may have delisted the asset. Reach out to Chainlink Labs.`)
   }
 }
+
+export const getUnsupportedAssetFromBadParameterError = (
+  error: ResponseError,
+): string | undefined => {
+  if (error.type !== 'bad_parameter') {
+    return undefined
+  }
+
+  return error.message.match(unsupportedAssetRegex)?.[1]?.toLowerCase()
+}

packages/sources/coinmetrics/src/transport/lwba.ts

@@ -8,10 +8,15 @@ import {
 import { TypeFromDefinition } from '@chainlink/external-adapter-framework/validation/input-params'
 import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
 import { WebSocketTransport } from '@chainlink/external-adapter-framework/transports/websocket'
-import { logPossibleSolutionForKnownErrors } from './error-handling'
+import {
+  getUnsupportedAssetFromBadParameterError,
+  logPossibleSolutionForKnownErrors,
+} from './error-handling'
 import { ResponseError } from './types'
 
 const logger = makeLogger('CoinMetrics Crypto LWBA WS')
+// base currencies that break the ws connection
+export const invalidBaseAssets: string[] = []
 
 export type MultiVarResult<T extends ProviderResultGenerics> = {
   params: TypeFromDefinition<T['Parameters']>
@@ -65,8 +70,18 @@ export const calculateUrl = (
 ): string => {
   const { API_KEY, WS_API_ENDPOINT } = context.adapterSettings
 
+  // Remove assets from desiredSubs that are invalid.
+  const validDesiredSubs = desiredSubs.filter(
+    ({ base }) => !invalidBaseAssets.includes(base.toLowerCase()),
+  )
+
   const generated = new URL('/v4/timeseries-stream/asset-quotes', WS_API_ENDPOINT)
-  const assets = [...new Set(desiredSubs.map((pair) => pair.base.toLowerCase()))].sort().join(',')
+  const assets = [...new Set(validDesiredSubs.map((pair) => pair.base.toLowerCase()))]
+    .sort()
+    .join(',')
+  if (!assets) {
+    logger.warn('No valid assets remain after filtering unsupported assets')
+  }
   generated.searchParams.append('assets', assets)
 
   generated.searchParams.append('api_key', API_KEY)
@@ -80,6 +95,12 @@ export const handleCryptoLwbaMessage = (
   if ('error' in message) {
     logger.error(message, `Error response from websocket`)
     logPossibleSolutionForKnownErrors(message.error)
+
+    const unsupportedAsset = getUnsupportedAssetFromBadParameterError(message.error)
+    if (unsupportedAsset && !invalidBaseAssets.includes(unsupportedAsset)) {
+      invalidBaseAssets.push(unsupportedAsset)
+      logger.warn({ asset: unsupportedAsset }, 'Added unsupported asset to invalid asset list')
+    }
   } else if ('warning' in message) {
     logger.warn(message, `Warning response from websocket`)
   } else if ('type' in message && message.type === 'reorg') {

packages/sources/coinmetrics/src/transport/price-ws.ts

@@ -3,7 +3,10 @@ import { WebSocketTransport } from '@chainlink/external-adapter-framework/transp
 import { makeLogger, ProviderResult } from '@chainlink/external-adapter-framework/util'
 import { VALID_QUOTES } from '../config'
 import { assetMetricsInputParameters, BaseEndpointTypes } from '../endpoint/price'
-import { logPossibleSolutionForKnownErrors } from './error-handling'
+import {
+  getUnsupportedAssetFromBadParameterError,
+  logPossibleSolutionForKnownErrors,
+} from './error-handling'
 import { ResponseError } from './types'
 
 const logger = makeLogger('CoinMetrics WS')
@@ -63,11 +66,21 @@ export const calculateAssetMetricsUrl = (
 ): string => {
   const { API_KEY, WS_API_ENDPOINT } = context.adapterSettings
 
-  //remove assets from desiredSubs that are invalid
-  desiredSubs = desiredSubs.filter(({ base }) => !invalidBaseAssets.includes(base.toLowerCase()))
+  // Remove assets from desiredSubs that are invalid.
+  const validDesiredSubs = desiredSubs.filter(
+    ({ base }) => !invalidBaseAssets.includes(base.toLowerCase()),
+  )
 
-  const assets = [...new Set(desiredSubs.map((sub) => sub.base.toLowerCase()))].sort().join(',')
-  const metrics = [...new Set(desiredSubs.map((sub) => `ReferenceRate${sub.quote.toUpperCase()}`))]
+  const assets = [...new Set(validDesiredSubs.map((sub) => sub.base.toLowerCase()))]
+    .sort()
+    .join(',')
+  if (!assets) {
+    logger.warn('No valid assets remain after filtering unsupported assets')
+  }
+
+  const metrics = [
+    ...new Set(validDesiredSubs.map((sub) => `ReferenceRate${sub.quote.toUpperCase()}`)),
+  ]
     .sort()
     .join(',')
   const generated = new URL('/v4/timeseries-stream/asset-metrics', WS_API_ENDPOINT)
@@ -90,17 +103,10 @@ export const handleAssetMetricsMessage = (
     logger.error(message, `Error response from websocket`)
     logPossibleSolutionForKnownErrors(message.error)
 
-    const findBaseCurrenciesRegex = new RegExp(/'([^']+)'/g)
-    if (message['error']['type'] === 'bad_parameter') {
-      //Bad Parameter Error Message
-      // type: 'bad_paramter'
-      // message: 'Bad parameter 'assets'. Value 'ohmv2' is not supported.'
-
-      const matches = [...message.error.message.matchAll(findBaseCurrenciesRegex)]
-
-      if (matches && !invalidBaseAssets.includes(matches[1][1])) {
-        invalidBaseAssets.push(matches[1][1])
-      }
+    const unsupportedAsset = getUnsupportedAssetFromBadParameterError(message.error)
+    if (unsupportedAsset && !invalidBaseAssets.includes(unsupportedAsset)) {
+      invalidBaseAssets.push(unsupportedAsset)
+      logger.warn({ asset: unsupportedAsset }, 'Added unsupported asset to invalid asset list')
     }
   } else if ('warning' in message) {
     // Is WsAssetMetricsWarningResponse

packages/sources/coinmetrics/test/unit/lwba-ws.test.ts

@@ -0,0 +1,194 @@
+import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
+import { LoggerFactoryProvider } from '@chainlink/external-adapter-framework/util'
+import * as queryString from 'querystring'
+import { config } from '../../src/config'
+import { BaseEndpointTypes, inputParameters } from '../../src/endpoint/lwba'
+import {
+  calculateUrl,
+  handleCryptoLwbaMessage,
+  invalidBaseAssets,
+  WsCryptoLwbaErrorResponse,
+  WsCryptoLwbaSuccessResponse,
+  WsCryptoLwbaWarningResponse,
+} from '../../src/transport/lwba'
+
+// Since the test is directly using endpoint functions, we need to initialize the logger here
+LoggerFactoryProvider.set()
+
+const EXAMPLE_SUCCESS_MESSAGE: WsCryptoLwbaSuccessResponse = {
+  pair: 'eth-usd',
+  time: Date.now().toString(),
+  ask_price: '1500.5',
+  ask_size: '20',
+  bid_price: '1499.5',
+  bid_size: '15',
+  mid_price: '1500',
+  spread: '1',
+  cm_sequence_id: '9',
+}
+
+const EXAMPLE_WARNING_MESSAGE: WsCryptoLwbaWarningResponse = {
+  warning: {
+    type: 'warning',
+    message: 'This is a warning message',
+  },
+}
+
+const EXAMPLE_ERROR_MESSAGE: WsCryptoLwbaErrorResponse = {
+  error: {
+    type: 'error',
+    message: 'This is an error message',
+  },
+}
+
+const EXAMPLE_BAD_PARAMETER_ERROR_MESSAGE: WsCryptoLwbaErrorResponse = {
+  error: {
+    type: 'bad_parameter',
+    message: "Bad parameter 'assets'. Value 'ohmv2' is not supported.",
+  },
+}
+
+const EXAMPLE_BAD_PARAMETER_UPPERCASE_ERROR_MESSAGE: WsCryptoLwbaErrorResponse = {
+  error: {
+    type: 'bad_parameter',
+    message: "Value 'OHMV2' is not supported for parameter 'assets'.",
+  },
+}
+
+const EXAMPLE_MALFORMED_BAD_PARAMETER_ERROR_MESSAGE: WsCryptoLwbaErrorResponse = {
+  error: {
+    type: 'bad_parameter',
+    message: "Bad parameter 'assets'.",
+  },
+}
+
+const EXAMPLE_REORG_MESSAGE = {
+  time: Date.now().toString(),
+  asset: 'eth',
+  height: 99999,
+  hash: 'nwiiwefepnfpnwiwiwfi',
+  parent_hash: 'iriwwfnpfpuffp',
+  type: 'reorg' as const,
+  cm_sequence_id: 9,
+}
+
+config.initialize()
+const EXAMPLE_CONTEXT: EndpointContext<BaseEndpointTypes> = {
+  endpointName: 'crypto-lwba',
+  inputParameters,
+  adapterSettings: config.settings,
+}
+
+describe('lwba-ws url generator', () => {
+  let oldEnv: NodeJS.ProcessEnv
+  const invalidCurrencies = invalidBaseAssets
+
+  beforeAll(() => {
+    oldEnv = JSON.parse(JSON.stringify(process.env))
+    process.env['API_KEY'] = 'someKey'
+  })
+
+  beforeEach(() => {
+    invalidBaseAssets.length = 0
+  })
+
+  afterAll(() => {
+    process.env = oldEnv
+  })
+
+  it('should correct casing in url', async () => {
+    const url = await calculateUrl(EXAMPLE_CONTEXT, [
+      {
+        base: 'ETH',
+        quote: 'usd',
+      },
+    ])
+    expect(url).toContain(queryString.stringify({ assets: 'eth' }))
+  })
+
+  it('should compose the url using all desired subs', async () => {
+    const url = await calculateUrl(EXAMPLE_CONTEXT, [
+      {
+        base: 'btc',
+        quote: 'usd',
+      },
+      {
+        base: 'eth',
+        quote: 'EUR',
+      },
+    ])
+    expect(url).toContain(new URLSearchParams({ assets: 'btc,eth' }).toString())
+  })
+
+  it('invalid request, should compose url without invalid asset', async () => {
+    invalidCurrencies.push('usd')
+    const url = await calculateUrl(EXAMPLE_CONTEXT, [
+      {
+        base: 'BTC',
+        quote: 'USD',
+      },
+      {
+        base: 'USD',
+        quote: 'BTC',
+      },
+    ])
+    expect(url).toContain(new URLSearchParams({ assets: 'btc' }).toString())
+  })
+
+  it('should compose url with empty assets when all desired subs are invalid', async () => {
+    invalidCurrencies.push('btc')
+    const url = await calculateUrl(EXAMPLE_CONTEXT, [
+      {
+        base: 'BTC',
+        quote: 'USD',
+      },
+    ])
+    expect(new URL(url).searchParams.get('assets')).toEqual('')
+  })
+})
+
+describe('lwba-ws message handler', () => {
+  beforeEach(() => {
+    invalidBaseAssets.length = 0
+  })
+
+  it('success message results in value', () => {
+    const res = handleCryptoLwbaMessage({ ...EXAMPLE_SUCCESS_MESSAGE })
+    expect(res).toBeDefined()
+    expect(res?.length).toEqual(1)
+    expect(res?.[0].response.data?.mid).toEqual(1500)
+  })
+
+  it('warning message results in undefined', () => {
+    const res = handleCryptoLwbaMessage(EXAMPLE_WARNING_MESSAGE)
+    expect(res).toBeUndefined()
+  })
+
+  it('error message results in undefined', () => {
+    const res = handleCryptoLwbaMessage(EXAMPLE_ERROR_MESSAGE)
+    expect(res).toBeUndefined()
+  })
+
+  it('bad parameter error stores the unsupported asset', () => {
+    const res = handleCryptoLwbaMessage(EXAMPLE_BAD_PARAMETER_ERROR_MESSAGE)
+    expect(res).toBeUndefined()
+    expect(invalidBaseAssets).toContain('ohmv2')
+  })
+
+  it('bad parameter error stores the unsupported asset in lowercase', () => {
+    const res = handleCryptoLwbaMessage(EXAMPLE_BAD_PARAMETER_UPPERCASE_ERROR_MESSAGE)
+    expect(res).toBeUndefined()
+    expect(invalidBaseAssets).toContain('ohmv2')
+  })
+
+  it('malformed bad parameter error does not throw', () => {
+    const res = handleCryptoLwbaMessage(EXAMPLE_MALFORMED_BAD_PARAMETER_ERROR_MESSAGE)
+    expect(res).toBeUndefined()
+    expect(invalidBaseAssets).toEqual([])
+  })
+
+  it('reorg message results in undefined', () => {
+    const res = handleCryptoLwbaMessage(EXAMPLE_REORG_MESSAGE)
+    expect(res).toBeUndefined()
+  })
+})