Added cache ttl refresh on heartbeat message for Finnhub

.changeset/sixty-parents-march.md

@@ -0,0 +1,6 @@
+---
+'@chainlink/finnhub-secondary-adapter': patch
+'@chainlink/finnhub-adapter': patch
+---
+
+Added support for cache TTL refresh on heartbeat messages for 'forex' endpoint

packages/sources/finnhub-secondary/README.md

@@ -4,6 +4,12 @@
 
 This document was generated automatically. Please see [README Generator](../../scripts#readme-generator) for more info.
 
+## Known Issues
+
+### CACHE_MAX_AGE interaction with Heartbeat messages
+
+If `CACHE_MAX_AGE` is set below a current heartbeat interval (60000ms), the extended cache TTL feature for out-of-market-hours that relies on heartbeats will not work.
+
 ## Environment Variables
 
 | Required? |      Name       |                      Description                      |  Type   | Options |           Default           |
@@ -36,11 +42,12 @@ Supported names for this endpoint are: `commodities`, `commodity-quote`, `common
 
 ### Input Params
 
-| Required? |   Name   |    Aliases     |                  Description                   |  Type  | Options | Default | Depends On | Not Valid With |
-| :-------: | :------: | :------------: | :--------------------------------------------: | :----: | :-----: | :-----: | :--------: | :------------: |
-|    ✅     |   base   | `coin`, `from` | The symbol of symbols of the currency to query | string |         |         |            |                |
-|           |  quote   | `market`, `to` |    The symbol of the currency to convert to    | string |         |         |            |                |
-|           | exchange |                |         The exchange to fetch data for         | string |         |         |            |                |
+| Required? |     Name     |    Aliases     |                                          Description                                          |  Type  | Options | Default | Depends On | Not Valid With |
+| :-------: | :----------: | :------------: | :-------------------------------------------------------------------------------------------: | :----: | :-----: | :-----: | :--------: | :------------: |
+|    ✅     |     base     | `coin`, `from` |                        The symbol of symbols of the currency to query                         | string |         |         |            |                |
+|           |    quote     | `market`, `to` |                           The symbol of the currency to convert to                            | string |         |         |            |                |
+|           |   exchange   |                |                                The exchange to fetch data for                                 | string |         |         |            |                |
+|           | endpointName |                | Is set automatically based on request endpoint alias/name. Modifying this value has no effect | string |         |         |            |                |
 
 ### Example
 

packages/sources/finnhub-secondary/docs/known-issues.md

@@ -0,0 +1,5 @@
+## Known Issues
+
+### CACHE_MAX_AGE interaction with Heartbeat messages
+
+If `CACHE_MAX_AGE` is set below a current heartbeat interval (60000ms), the extended cache TTL feature for out-of-market-hours that relies on heartbeats will not work.

packages/sources/finnhub-secondary/test/integration/__snapshots__/adapter-ws.test.ts.snap

@@ -8,7 +8,7 @@ exports[`websocket should return success for full symbols 1`] = `
   "result": 1.098455,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 1018,
+    "providerDataReceivedUnixMs": 2028,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },
@@ -23,7 +23,7 @@ exports[`websocket should return success for inverted pairs 1`] = `
   "result": 0.007010066455429998,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 4048,
+    "providerDataReceivedUnixMs": 5058,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },
@@ -38,7 +38,7 @@ exports[`websocket should return success for requests with base and quote 1`] =
   "result": 1.098455,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 2028,
+    "providerDataReceivedUnixMs": 3038,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },
@@ -53,7 +53,22 @@ exports[`websocket should return success for standard pairs, when pair has inver
   "result": 142.652,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 3038,
+    "providerDataReceivedUnixMs": 4048,
+    "providerDataStreamEstablishedUnixMs": 1010,
+    "providerIndicatedTimeUnixMs": 1641035471111,
+  },
+}
+`;
+
+exports[`websocket should update the ttl of forex params after heartbeat is received 1`] = `
+{
+  "data": {
+    "result": 1.098455,
+  },
+  "result": 1.098455,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1018,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },

packages/sources/finnhub-secondary/test/integration/adapter-ws.test.ts

@@ -4,6 +4,7 @@ import {
   TestAdapter,
   mockWebSocketProvider,
   setEnvVariables,
+  runAllUntilTime,
 } from '@chainlink/external-adapter-framework/util/testing-utils'
 import FakeTimers from '@sinonjs/fake-timers'
 
@@ -32,6 +33,9 @@ const mockWebSocketServer = (url: string) => {
           type: 'trade',
         }),
       )
+      setTimeout(() => {
+        socket.send(JSON.stringify({ type: 'ping' }))
+      }, 10000)
     })
   })
   return mockWsServer
@@ -42,6 +46,7 @@ describe('websocket', () => {
 
   const data = {
     base: 'OANDA:EUR_USD',
+    endpoint: 'forex',
   }
 
   let spy: jest.SpyInstance
@@ -118,4 +123,16 @@ describe('websocket', () => {
     expect(response.statusCode).toBe(200)
     expect(response.json()).toMatchSnapshot()
   })
+
+  it('should update the ttl of forex params after heartbeat is received', async () => {
+    await runAllUntilTime(testAdapter.clock, 93000)
+    const expiredCacheResponse = await testAdapter.request({ base: 'JPY', quote: 'USD' })
+    expect(expiredCacheResponse.statusCode).toBe(504)
+
+    // The cache ttl is 90 seconds. Mocked heartbeat message is sent after 10s after connection which should
+    // update the ttl and therefore after 93 seconds (from the initial message) we can access the asset
+    const response = await testAdapter.request(data)
+    expect(response.statusCode).toBe(200)
+    expect(response.json()).toMatchSnapshot()
+  })
 })

packages/sources/finnhub/README.md

@@ -4,6 +4,12 @@
 
 This document was generated automatically. Please see [README Generator](../../scripts#readme-generator) for more info.
 
+## Known Issues
+
+### CACHE_MAX_AGE interaction with Heartbeat messages
+
+If `CACHE_MAX_AGE` is set below a current heartbeat interval (60000ms), the extended cache TTL feature for out-of-market-hours that relies on heartbeats will not work.
+
 ## Environment Variables
 
 | Required? |      Name       |                      Description                      |  Type   | Options |           Default           |
@@ -36,11 +42,12 @@ Supported names for this endpoint are: `commodities`, `commodity-quote`, `common
 
 ### Input Params
 
-| Required? |   Name   |    Aliases     |                  Description                   |  Type  | Options | Default | Depends On | Not Valid With |
-| :-------: | :------: | :------------: | :--------------------------------------------: | :----: | :-----: | :-----: | :--------: | :------------: |
-|    ✅     |   base   | `coin`, `from` | The symbol of symbols of the currency to query | string |         |         |            |                |
-|           |  quote   | `market`, `to` |    The symbol of the currency to convert to    | string |         |         |            |                |
-|           | exchange |                |         The exchange to fetch data for         | string |         |         |            |                |
+| Required? |     Name     |    Aliases     |                                          Description                                          |  Type  | Options | Default | Depends On | Not Valid With |
+| :-------: | :----------: | :------------: | :-------------------------------------------------------------------------------------------: | :----: | :-----: | :-----: | :--------: | :------------: |
+|    ✅     |     base     | `coin`, `from` |                        The symbol of symbols of the currency to query                         | string |         |         |            |                |
+|           |    quote     | `market`, `to` |                           The symbol of the currency to convert to                            | string |         |         |            |                |
+|           |   exchange   |                |                                The exchange to fetch data for                                 | string |         |         |            |                |
+|           | endpointName |                | Is set automatically based on request endpoint alias/name. Modifying this value has no effect | string |         |         |            |                |
 
 ### Example
 

packages/sources/finnhub/docs/known-issues.md

@@ -0,0 +1,5 @@
+## Known Issues
+
+### CACHE_MAX_AGE interaction with Heartbeat messages
+
+If `CACHE_MAX_AGE` is set below a current heartbeat interval (60000ms), the extended cache TTL feature for out-of-market-hours that relies on heartbeats will not work.

packages/sources/finnhub/src/endpoint/quote.ts

@@ -32,6 +32,12 @@ export const inputParameters = new InputParameters(
       description: 'The exchange to fetch data for',
       required: false,
     },
+    endpointName: {
+      type: 'string',
+      description:
+        'Is set automatically based on request endpoint alias/name. Providing a custom value has no effect',
+      required: false,
+    },
   },
   [
     {
@@ -135,6 +141,13 @@ const validateExchange = (req: AdapterRequest<typeof inputParameters.validated>)
   }
 }
 
+// Explicitly sets the requested endpoint alias into requestContext.data.endpointName, so it is available in the subscription set
+const setEndpointName = (req: AdapterRequest<typeof inputParameters.validated>) => {
+  req.requestContext.data.endpointName = (
+    req.body.data as unknown as typeof inputParameters.validated & { endpoint?: string }
+  ).endpoint
+}
+
 export const buildQuoteEndpoint = (overrides?: Record<string, string>) =>
   new PriceEndpoint<BaseEndpointTypes>({
     name: 'quote',
@@ -152,7 +165,7 @@ export const buildQuoteEndpoint = (overrides?: Record<string, string>) =>
       .register('rest', httpTransport),
     defaultTransport: 'rest',
     customRouter: (_req, adapterConfig) => (adapterConfig.WS_ENABLED ? 'ws' : 'rest'),
-    requestTransforms: [requestTransform, validateExchange],
+    requestTransforms: [requestTransform, validateExchange, setEndpointName],
     inputParameters: inputParameters,
     overrides,
   })

packages/sources/finnhub/src/transport/quote-ws.ts

@@ -1,5 +1,8 @@
 import { BaseEndpointTypes, buildSymbol } from '../endpoint/quote'
-import { WebsocketReverseMappingTransport } from '@chainlink/external-adapter-framework/transports'
+import {
+  WebsocketReverseMappingTransport,
+  WebSocketTransport,
+} from '@chainlink/external-adapter-framework/transports'
 import { ProviderResult, makeLogger } from '@chainlink/external-adapter-framework/util'
 import { parseResult } from './utils'
 
@@ -10,6 +13,10 @@ type WsMessageError = {
   msg: string
 }
 
+type WsMessageHeartbeat = {
+  type: 'ping'
+}
+
 type WsMessageTrade = {
   type: 'trade'
   data: {
@@ -21,24 +28,40 @@ type WsMessageTrade = {
   }[]
 }
 
-type WsMessage = WsMessageError | WsMessageTrade
+type WsMessage = WsMessageError | WsMessageTrade | WsMessageHeartbeat
 
 type WsEndpointTypes = BaseEndpointTypes & {
   Provider: {
     WsMessage: WsMessage
   }
 }
+/*
+Finnhub EA currently does not receive asset prices during off-market hours. When a heartbeat message is received during these hours,
+we update the TTL of **forex** cache entries that EA is requested to provide a price during off-market hours.
+ */
+const updateTTL = async (transport: WebSocketTransport<WsEndpointTypes>, ttl: number) => {
+  const allParams = await transport.subscriptionSet.getAll()
+  const forexParams = allParams.filter((p) => p.endpointName === 'forex')
+  transport.responseCache.writeTTL(transport.name, forexParams, ttl)
+}
 
 export const wsTransport = new WebsocketReverseMappingTransport<WsEndpointTypes, string>({
   url: ({ adapterSettings }) =>
     `${adapterSettings.WS_API_ENDPOINT}?token=${adapterSettings.API_KEY}`,
   handlers: {
-    message: (message) => {
+    message: (message, context) => {
       if (message.type === 'error') {
         logger.error(message.msg)
         return
       }
 
+      // Check for a heartbeat message, refresh the TTLs of all requested entries in the cache
+      if (message.type === 'ping') {
+        wsTransport.lastMessageReceivedAt = Date.now()
+        updateTTL(wsTransport, context.adapterSettings.CACHE_MAX_AGE)
+        return []
+      }
+
       if (message.type === 'trade') {
         const results: ProviderResult<WsEndpointTypes>[] = []
 

packages/sources/finnhub/test/integration/__snapshots__/adapter-ws.test.ts.snap

@@ -8,7 +8,7 @@ exports[`websocket should return success for full symbols 1`] = `
   "result": 1.098455,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 1018,
+    "providerDataReceivedUnixMs": 2028,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },
@@ -23,7 +23,7 @@ exports[`websocket should return success for inverted pairs 1`] = `
   "result": 0.007010066455429998,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 4048,
+    "providerDataReceivedUnixMs": 5058,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },
@@ -38,7 +38,7 @@ exports[`websocket should return success for requests with base and quote 1`] =
   "result": 1.098455,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 2028,
+    "providerDataReceivedUnixMs": 3038,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },
@@ -53,7 +53,22 @@ exports[`websocket should return success for standard pairs, when pair has inver
   "result": 142.652,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 3038,
+    "providerDataReceivedUnixMs": 4048,
+    "providerDataStreamEstablishedUnixMs": 1010,
+    "providerIndicatedTimeUnixMs": 1641035471111,
+  },
+}
+`;
+
+exports[`websocket should update the ttl of forex params after heartbeat is received 1`] = `
+{
+  "data": {
+    "result": 1.098455,
+  },
+  "result": 1.098455,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1018,
     "providerDataStreamEstablishedUnixMs": 1010,
     "providerIndicatedTimeUnixMs": 1641035471111,
   },

packages/sources/finnhub/test/integration/adapter-ws.test.ts

@@ -4,6 +4,7 @@ import {
   TestAdapter,
   mockWebSocketProvider,
   setEnvVariables,
+  runAllUntilTime,
 } from '@chainlink/external-adapter-framework/util/testing-utils'
 import FakeTimers from '@sinonjs/fake-timers'
 
@@ -32,6 +33,9 @@ const mockWebSocketServer = (url: string) => {
           type: 'trade',
         }),
       )
+      setTimeout(() => {
+        socket.send(JSON.stringify({ type: 'ping' }))
+      }, 10000)
     })
   })
   return mockWsServer
@@ -42,6 +46,7 @@ describe('websocket', () => {
 
   const data = {
     base: 'FHFX:EUR-USD',
+    endpoint: 'forex',
   }
 
   let spy: jest.SpyInstance
@@ -118,4 +123,16 @@ describe('websocket', () => {
     expect(response.statusCode).toBe(200)
     expect(response.json()).toMatchSnapshot()
   })
+
+  it('should update the ttl of forex params after heartbeat is received', async () => {
+    await runAllUntilTime(testAdapter.clock, 93000)
+    const expiredCacheResponse = await testAdapter.request({ base: 'EUR', quote: 'USD' })
+    expect(expiredCacheResponse.statusCode).toBe(504)
+
+    // The cache ttl is 90 seconds. Mocked heartbeat message is sent after 10s after connection which should
+    // update the ttl and therefore after 93 seconds (from the initial message) we can access the asset
+    const response = await testAdapter.request(data)
+    expect(response.statusCode).toBe(200)
+    expect(response.json()).toMatchSnapshot()
+  })
 })