Data Engine: Adds result selection and scaling (#4736)

mohamed-mehany · web-flow · commit 6849d0b0d6f6 · 2026-03-16T14:04:11.000+01:00
diff --git a/.changeset/polite-frogs-pull.md b/.changeset/polite-frogs-pull.md
@@ -0,0 +1,5 @@
+---
+'@chainlink/data-engine-adapter': minor
+---
+
+Add result selection via resultPath and decimal scaling via decimals params to data-engine WS transports.
diff --git a/.pnp.cjs b/.pnp.cjs
diff --git a/packages/sources/data-engine/package.json b/packages/sources/data-engine/package.json
@@ -38,6 +38,7 @@
   "dependencies": {
     "@chainlink/data-streams-sdk": "1.2.0",
     "@chainlink/external-adapter-framework": "2.11.6",
+    "decimal.js": "^10.5.0",
     "tslib": "2.4.1"
   }
 }
diff --git a/packages/sources/data-engine/src/endpoint/common.ts b/packages/sources/data-engine/src/endpoint/common.ts
@@ -0,0 +1,20 @@
+/**
+ * Shared input parameter definitions used across all data-engine endpoints.
+ */
+export const commonInputParams = {
+  feedId: {
+    required: true,
+    type: 'string',
+    description: 'The feed ID to subscribe to',
+  },
+  resultPath: {
+    required: false,
+    type: 'string',
+    description: 'The data field to populate the top-level result',
+  },
+  decimals: {
+    required: false,
+    type: 'number',
+    description: 'Number of decimals to scale the resultPath value to (from native 18)',
+  },
+} as const
diff --git a/packages/sources/data-engine/src/endpoint/cryptoV3.ts b/packages/sources/data-engine/src/endpoint/cryptoV3.ts
@@ -2,14 +2,11 @@ import { AdapterEndpoint } from '@chainlink/external-adapter-framework/adapter'
 import { InputParameters } from '@chainlink/external-adapter-framework/validation'
 import { config } from '../config'
 import { cryptoV3Transport } from '../transport/cryptoV3'
+import { commonInputParams } from './common'
 
 export const inputParameters = new InputParameters(
   {
-    feedId: {
-      required: true,
-      type: 'string',
-      description: 'The feedId for crypto feed with v3 schema',
-    },
+    ...commonInputParams,
   },
   [
     {
@@ -21,7 +18,7 @@ export const inputParameters = new InputParameters(
 export type BaseEndpointTypes = {
   Parameters: typeof inputParameters.definition
   Response: {
-    Result: null
+    Result: string | null
     Data: {
       bid: string
       ask: string
diff --git a/packages/sources/data-engine/src/endpoint/deutscheBoerseV11.ts b/packages/sources/data-engine/src/endpoint/deutscheBoerseV11.ts
@@ -2,14 +2,11 @@ import { AdapterEndpoint } from '@chainlink/external-adapter-framework/adapter'
 import { InputParameters } from '@chainlink/external-adapter-framework/validation'
 import { config } from '../config'
 import { deutscheBoerseV11Transport } from '../transport/deutscheBoerseV11'
+import { commonInputParams } from './common'
 
 export const inputParameters = new InputParameters(
   {
-    feedId: {
-      required: true,
-      type: 'string',
-      description: 'The feedId for Deutsche Boerse feed with v11 schema',
-    },
+    ...commonInputParams,
   },
   [
     {
@@ -21,7 +18,7 @@ export const inputParameters = new InputParameters(
 export type BaseEndpointTypes = {
   Parameters: typeof inputParameters.definition
   Response: {
-    Result: null
+    Result: string | null
     Data: {
       mid: string
       lastSeenTimestampNs: string
diff --git a/packages/sources/data-engine/src/endpoint/rwaV8.ts b/packages/sources/data-engine/src/endpoint/rwaV8.ts
@@ -2,14 +2,11 @@ import { AdapterEndpoint } from '@chainlink/external-adapter-framework/adapter'
 import { InputParameters } from '@chainlink/external-adapter-framework/validation'
 import { config } from '../config'
 import { rwaV8Transport } from '../transport/rwaV8'
+import { commonInputParams } from './common'
 
 export const inputParameters = new InputParameters(
   {
-    feedId: {
-      required: true,
-      type: 'string',
-      description: 'The feedId for RWA feed with v8 schema',
-    },
+    ...commonInputParams,
   },
   [
     {
@@ -21,7 +18,7 @@ export const inputParameters = new InputParameters(
 export type BaseEndpointTypes = {
   Parameters: typeof inputParameters.definition
   Response: {
-    Result: null
+    Result: string | null
     Data: {
       midPrice: string
       marketStatus: number
diff --git a/packages/sources/data-engine/src/transport/cryptoV3.ts b/packages/sources/data-engine/src/transport/cryptoV3.ts
@@ -1,6 +1,7 @@
 import { DecodedV3Report } from '@chainlink/data-streams-sdk'
 import { BaseEndpointTypes } from '../endpoint/cryptoV3'
-import { createDataEngineTransport, DECIMALS } from './wsTransportBase'
+import { DECIMALS } from './utils'
+import { createDataEngineTransport } from './wsTransportBase'
 
 export const cryptoV3Transport = createDataEngineTransport<BaseEndpointTypes, DecodedV3Report>({
   schemaVersion: 'V3',
diff --git a/packages/sources/data-engine/src/transport/deutscheBoerseV11.ts b/packages/sources/data-engine/src/transport/deutscheBoerseV11.ts
@@ -1,6 +1,7 @@
 import { DecodedV11Report } from '@chainlink/data-streams-sdk'
 import { BaseEndpointTypes } from '../endpoint/deutscheBoerseV11'
-import { createDataEngineTransport, DECIMALS } from './wsTransportBase'
+import { DECIMALS } from './utils'
+import { createDataEngineTransport } from './wsTransportBase'
 
 export const deutscheBoerseV11Transport = createDataEngineTransport<
   BaseEndpointTypes,
diff --git a/packages/sources/data-engine/src/transport/rwaV8.ts b/packages/sources/data-engine/src/transport/rwaV8.ts
@@ -1,6 +1,7 @@
 import { DecodedV8Report } from '@chainlink/data-streams-sdk'
 import { BaseEndpointTypes } from '../endpoint/rwaV8'
-import { createDataEngineTransport, DECIMALS } from './wsTransportBase'
+import { DECIMALS } from './utils'
+import { createDataEngineTransport } from './wsTransportBase'
 
 export const rwaV8Transport = createDataEngineTransport<BaseEndpointTypes, DecodedV8Report>({
   schemaVersion: 'V8',
diff --git a/packages/sources/data-engine/src/transport/utils.ts b/packages/sources/data-engine/src/transport/utils.ts
@@ -0,0 +1,42 @@
+import Decimal from 'decimal.js'
+
+export const DECIMALS = 18
+
+/**
+ * Scales a raw integer string from `fromDecimals` to `toDecimals`.
+ * Uses truncation (floor toward zero) when scaling down.
+ */
+export function scaleDecimals(value: string, fromDecimals: number, toDecimals: number): string {
+  if (fromDecimals === toDecimals) {
+    return value
+  }
+  const diff = fromDecimals - toDecimals
+  const raw = new Decimal(value)
+  const scaled = raw.div(new Decimal(10).pow(diff))
+  return scaled.toFixed(0, Decimal.ROUND_DOWN)
+}
+
+/**
+ * Extracts a value from `data` using `resultPath`, optionally scaling decimals.
+ *
+ * Returns `null` when no `resultPath` is provided (no extraction requested).
+ * Throws if `resultPath` is provided but the key does not exist in `data`.
+ */
+export function resolveResult(
+  data: Record<string, unknown>,
+  resultPath?: string,
+  decimals?: number,
+): string | null {
+  if (!resultPath) return null
+
+  if (!(resultPath in data)) {
+    throw new Error(
+      `resultPath '${resultPath}' not found in data. Available keys: ${Object.keys(data).join(
+        ', ',
+      )}`,
+    )
+  }
+
+  const raw = String(data[resultPath])
+  return decimals !== undefined ? scaleDecimals(raw, DECIMALS, decimals) : raw
+}
diff --git a/packages/sources/data-engine/src/transport/wsTransportBase.ts b/packages/sources/data-engine/src/transport/wsTransportBase.ts
@@ -4,9 +4,13 @@ import {
   WebSocketTransport,
 } from '@chainlink/external-adapter-framework/transports'
 import { makeLogger, ProviderResult } from '@chainlink/external-adapter-framework/util'
+import { TypeFromDefinition } from '@chainlink/external-adapter-framework/validation/input-params'
 import { config } from '../config'
+import { commonInputParams } from '../endpoint/common'
+import { resolveResult } from './utils'
 
-export const DECIMALS = 18
+// Re-export for transport files that reference DECIMALS in extractData
+export { DECIMALS } from './utils'
 
 type ProviderTypes = {
   Provider: {
@@ -25,18 +29,16 @@ type BaseTransportTypes = {
   Settings: TransportGenerics['Settings'] & typeof config.settings
 }
 
-const buildWsUrl = (baseUrl: string, desiredSubs: { feedId?: string }[]) => {
+type DesiredSub = TypeFromDefinition<typeof commonInputParams>
+
+const buildWsUrl = (baseUrl: string, desiredSubs: DesiredSub[]) => {
   const url = new URL(`${baseUrl}/api/v1/ws`)
 
   if (desiredSubs) {
-    url.searchParams.set(
-      'feedIDs',
-      desiredSubs
-        .filter((s) => s.feedId)
-        .map((s) => s.feedId?.toLowerCase())
-        .sort()
-        .join(','),
-    )
+    const feedIds = [
+      ...new Set(desiredSubs.filter((s) => s.feedId).map((s) => s.feedId.toLowerCase())),
+    ].sort()
+    url.searchParams.set('feedIDs', feedIds.join(','))
   }
   return url.toString()
 }
@@ -51,11 +53,20 @@ export function createDataEngineTransport<
 }) {
   const logger = makeLogger(config.loggerName)
 
+  // Tracks the current set of subscriptions so the message handler can build
+  // one cache entry per unique subscription. Updated by the url callback on
+  // each background cycle.
+  let currentDesiredSubs: DesiredSub[] = []
+
   return new WebSocketTransport<BaseEndpointTypes & ProviderTypes>({
-    url: (context, desiredSubs) => buildWsUrl(context.adapterSettings.WS_API_ENDPOINT, desiredSubs),
+    url: (context, desiredSubs) => {
+      const subs = desiredSubs as DesiredSub[]
+      currentDesiredSubs = subs
+      return buildWsUrl(context.adapterSettings.WS_API_ENDPOINT, subs)
+    },
 
     options: (context, desiredSubs) => {
-      const url = buildWsUrl(context.adapterSettings.WS_API_ENDPOINT, desiredSubs)
+      const url = buildWsUrl(context.adapterSettings.WS_API_ENDPOINT, desiredSubs as DesiredSub[])
       return {
         headers: generateAuthHeaders(
           context.adapterSettings.API_USERNAME,
@@ -82,30 +93,46 @@ export function createDataEngineTransport<
         if (decoded?.version.toUpperCase() !== config.schemaVersion.toUpperCase()) {
           return [
             {
-              params: {
-                feedId: msg.report.feedID,
-              } as any,
+              params: { feedId: msg.report.feedID } as DesiredSub,
               response: {
                 statusCode: 400,
                 errorMessage: `${decoded?.version.toUpperCase()} schema from ${
                   msg.report.feedID
                 } is not supported`,
               },
             },
-          ]
+          ] as ProviderResult<BaseEndpointTypes & ProviderTypes>[]
         }
 
-        return [
-          {
-            params: {
-              feedId: msg.report.feedID,
-            } as any,
-            response: {
-              result: null,
-              data: config.extractData(decoded as DecodedReport),
-            },
-          },
-        ]
+        const data = config.extractData(decoded as DecodedReport)
+
+        // Build one result per subscription that matches this message's feedId
+        return currentDesiredSubs
+          .filter((sub) => sub.feedId === msg.report!.feedID)
+          .map((sub) => {
+            const params = sub as ProviderResult<BaseEndpointTypes & ProviderTypes>['params']
+            try {
+              return {
+                params,
+                response: {
+                  result: resolveResult(
+                    data as Record<string, unknown>,
+                    sub.resultPath,
+                    sub.decimals,
+                  ),
+                  data,
+                },
+              }
+            } catch (e) {
+              return {
+                params,
+                response: {
+                  statusCode: 400,
+                  errorMessage: (e as Error).message,
+                },
+              }
+            }
+          })
       },
 
       close: (closeEvent) => {
diff --git a/packages/sources/data-engine/test/integration/__snapshots__/adapter.test.ts.snap b/packages/sources/data-engine/test/integration/__snapshots__/adapter.test.ts.snap
@@ -11,7 +11,7 @@ exports[`execute crypto-v3 endpoint should return success 1`] = `
   "result": null,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 1114,
+    "providerDataReceivedUnixMs": 7114,
     "providerDataStreamEstablishedUnixMs": 1010,
   },
 }
@@ -33,7 +33,7 @@ exports[`execute deutscheBoerse-v11 endpoint should return success 1`] = `
   "result": null,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 3134,
+    "providerDataReceivedUnixMs": 7134,
     "providerDataStreamEstablishedUnixMs": 3030,
   },
 }
@@ -49,7 +49,7 @@ exports[`execute rwa-v8 endpoint should return success 1`] = `
   "result": null,
   "statusCode": 200,
   "timestamps": {
-    "providerDataReceivedUnixMs": 2124,
+    "providerDataReceivedUnixMs": 7124,
     "providerDataStreamEstablishedUnixMs": 2020,
   },
 }
diff --git a/packages/sources/data-engine/test/integration/adapter.test.ts b/packages/sources/data-engine/test/integration/adapter.test.ts
diff --git a/packages/sources/data-engine/test/integration/fixtures.ts b/packages/sources/data-engine/test/integration/fixtures.ts
diff --git a/packages/sources/data-engine/test/unit/utils.test.ts b/packages/sources/data-engine/test/unit/utils.test.ts
diff --git a/packages/sources/data-engine/test/unit/wsTransportBase.test.ts b/packages/sources/data-engine/test/unit/wsTransportBase.test.ts
diff --git a/yarn.lock b/yarn.lock

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +---
 +'@chainlink/data-engine-adapter': minor
 +---
++
 +Add result selection via resultPath and decimal scaling via decimals params to data-engine WS transports.
Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,7 @@`
`38`	`38`	`"dependencies": {`
`39`	`39`	`"@chainlink/data-streams-sdk": "1.2.0",`
`40`	`40`	`"@chainlink/external-adapter-framework": "2.11.6",`
	`41`	`+ "decimal.js": "^10.5.0",`
`41`	`42`	`"tslib": "2.4.1"`
`42`	`43`	`}`
`43`	`44`	`}`