add cache ttl refresh feature

.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

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           |

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

@@ -59,3 +59,18 @@ exports[`websocket should return success for standard pairs, when pair has inver
   },
 }
 `;
+
+exports[`websocket should update the ttl 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
@@ -118,4 +122,13 @@ describe('websocket', () => {
     expect(response.statusCode).toBe(200)
     expect(response.json()).toMatchSnapshot()
   })
+
+  it('should update the ttl after heartbeat is received', async () => {
+    // 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
+    await runAllUntilTime(testAdapter.clock, 93000)
+    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           |

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/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,39 @@ 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 cache entries that EA is requested to provide a price during off-market hours.
+ */
+const updateTTL = async (transport: WebSocketTransport<WsEndpointTypes>, ttl: number) => {
+  const params = await transport.subscriptionSet.getAll()
+  transport.responseCache.writeTTL(transport.name, params, 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

@@ -59,3 +59,18 @@ exports[`websocket should return success for standard pairs, when pair has inver
   },
 }
 `;
+
+exports[`websocket should update the ttl 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
@@ -118,4 +122,13 @@ describe('websocket', () => {
     expect(response.statusCode).toBe(200)
     expect(response.json()).toMatchSnapshot()
   })
+
+  it('should update the ttl after heartbeat is received', async () => {
+    // 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
+    await runAllUntilTime(testAdapter.clock, 93000)
+    const response = await testAdapter.request(data)
+    expect(response.statusCode).toBe(200)
+    expect(response.json()).toMatchSnapshot()
+  })
 })