[OPDATA-3889] FTSE SFTP EA Downloading and Parsing

.changeset/warm-hornets-compete.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/ftse-sftp-adapter': major
+---
+
+Adding Downloading and parsing logic for russell and ftse csv files from ftse sftp server

packages/sources/ftse-sftp/src/index.ts

@@ -1 +1,12 @@
-export * from './parsing'
+import { expose, ServerInstance } from '@chainlink/external-adapter-framework'
+import { Adapter } from '@chainlink/external-adapter-framework/adapter'
+import { config } from './config'
+import * as endpoints from './endpoint'
+
+export const adapter = new Adapter({
+  name: 'FTSE_SFTP',
+  config,
+  endpoints: [endpoints.sftp.endpoint],
+})
+
+export const server = (): Promise<ServerInstance | undefined> => expose(adapter)

packages/sources/ftse-sftp/src/transport/constants.ts

@@ -0,0 +1,29 @@
+export const instrumentToFilePathMap: Record<string, string> = {
+  FTSE100INDEX: '/data/valuation/uk_all_share/',
+  Russell1000INDEX: '/data/Returns_and_Values/Russell_US_Indexes_Daily_Index_Values_Real_Time_TXT/',
+  Russell2000INDEX: '/data/Returns_and_Values/Russell_US_Indexes_Daily_Index_Values_Real_Time_TXT/',
+  Russell3000INDEX: '/data/Returns_and_Values/Russell_US_Indexes_Daily_Index_Values_Real_Time_TXT/',
+}
+
+export const instrumentToFileTemplateMap: Record<string, string> = {
+  FTSE100INDEX: 'ukallv*.csv',
+  Russell1000INDEX: 'daily_values_russell_*.CSV',
+  Russell2000INDEX: 'daily_values_russell_*.CSV',
+  Russell3000INDEX: 'daily_values_russell_*.CSV',
+}
+
+export const instrumentToFileRegexMap: Record<string, RegExp> = {
+  FTSE100INDEX: /^ukallv\d{4}\.csv$/,
+  Russell1000INDEX: /^daily_values_russell_\d{6}\.CSV$/,
+  Russell2000INDEX: /^daily_values_russell_\d{6}\.CSV$/,
+  Russell3000INDEX: /^daily_values_russell_\d{6}\.CSV$/,
+}
+
+/**
+ * Validates if an instrument is supported by checking if it has all required mappings
+ * @param instrument The instrument identifier to validate
+ * @returns true if the instrument is supported, false otherwise
+ */
+export function isInstrumentSupported(instrument: string): boolean {
+  return !!(instrumentToFilePathMap[instrument] && instrumentToFileRegexMap[instrument])
+}

packages/sources/ftse-sftp/src/transport/sftp.ts

@@ -1,13 +1,35 @@
 import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
+import { ResponseCache } from '@chainlink/external-adapter-framework/cache/response'
 import { TransportDependencies } from '@chainlink/external-adapter-framework/transports'
 import { SubscriptionTransport } from '@chainlink/external-adapter-framework/transports/abstract/subscription'
-import { sleep } from '@chainlink/external-adapter-framework/util'
-import SftpClient from 'ssh2-sftp-client'
-import { BaseEndpointTypes } from '../endpoint/sftp'
+import { AdapterResponse, makeLogger, sleep } from '@chainlink/external-adapter-framework/util'
+import { AdapterInputError } from '@chainlink/external-adapter-framework/validation/error'
+import SftpClient, { FileInfo } from 'ssh2-sftp-client'
+import { BaseEndpointTypes, IndexResponseData, inputParameters } from '../endpoint/sftp'
+import { CSVParserFactory } from '../parsing/factory'
+import {
+  instrumentToFilePathMap,
+  instrumentToFileRegexMap,
+  isInstrumentSupported,
+} from './constants'
+
+const logger = makeLogger('FTSE SFTP Adapter')
+
+type RequestParams = typeof inputParameters.validated
+
+interface SftpConnectionConfig {
+  host: string
+  port: number
+  username: string
+  password: string
+  readyTimeout: number
+}
 
 export class SftpTransport extends SubscriptionTransport<BaseEndpointTypes> {
   config!: BaseEndpointTypes['Settings']
   endpointName!: string
+  name!: string
+  responseCache!: ResponseCache<BaseEndpointTypes>
   sftpClient: SftpClient
 
   constructor() {
@@ -24,12 +46,156 @@ export class SftpTransport extends SubscriptionTransport<BaseEndpointTypes> {
     await super.initialize(dependencies, adapterSettings, endpointName, transportName)
     this.config = adapterSettings
     this.endpointName = endpointName
+    this.name = transportName
+    this.responseCache = dependencies.responseCache
   }
 
-  async backgroundHandler(context: EndpointContext<BaseEndpointTypes>): Promise<void> {
+  async backgroundHandler(
+    context: EndpointContext<BaseEndpointTypes>,
+    entries: RequestParams[],
+  ): Promise<void> {
+    await Promise.all(entries.map(async (param) => this.handleRequest(param)))
     await sleep(context.adapterSettings.BACKGROUND_EXECUTE_MS)
   }
 
+  async handleRequest(param: RequestParams) {
+    let response: AdapterResponse<BaseEndpointTypes['Response']>
+    try {
+      response = await this._handleRequest(param)
+    } catch (e) {
+      const errorMessage = e instanceof Error ? e.message : 'Unknown error occurred'
+      response = {
+        statusCode: 502,
+        errorMessage,
+        timestamps: {
+          providerDataRequestedUnixMs: 0,
+          providerDataReceivedUnixMs: 0,
+          providerIndicatedTimeUnixMs: undefined,
+        },
+      }
+    } finally {
+      try {
+        await this.sftpClient.end()
+        logger.info('SFTP connection closed')
+      } catch (error) {
+        logger.error('Error closing SFTP connection:', error)
+      }
+    }
+
+    await this.responseCache.write(this.name, [{ params: param, response }])
+  }
+
+  async _handleRequest(
+    param: RequestParams,
+  ): Promise<AdapterResponse<BaseEndpointTypes['Response']>> {
+    const providerDataRequestedUnixMs = Date.now()
+
+    await this.connectToSftp()
+
+    const parsedData = await this.tryDownloadAndParseFile(param.instrument)
+
+    // Extract the numeric result based on the data type
+    let result: number
+    if ('gbpIndex' in parsedData) {
+      // FTSE data
+      result = (parsedData.gbpIndex as number) ?? 0
+    } else if ('close' in parsedData) {
+      // Russell data
+      result = parsedData.close as number
+    } else {
+      throw new Error('Unknown data format received from parser')
+    }
+
+    logger.info(`Successfully processed data for instrument: ${param.instrument}`)
+    return {
+      data: {
+        result: parsedData,
+      },
+      statusCode: 200,
+      result,
+      timestamps: {
+        providerDataRequestedUnixMs,
+        providerDataReceivedUnixMs: Date.now(),
+        providerIndicatedTimeUnixMs: undefined,
+      },
+    }
+  }
+
+  private async connectToSftp(): Promise<void> {
+    const connectConfig: SftpConnectionConfig = {
+      host: this.config.SFTP_HOST,
+      port: this.config.SFTP_PORT || 22,
+      username: this.config.SFTP_USERNAME,
+      password: this.config.SFTP_PASSWORD,
+      readyTimeout: 30000,
+    }
+
+    try {
+      // Create a new client instance to avoid connection state issues
+      this.sftpClient = new SftpClient()
+      await this.sftpClient.connect(connectConfig)
+      logger.info('Successfully connected to SFTP server')
+    } catch (error) {
+      logger.error(error, 'Failed to connect to SFTP server')
+      throw new AdapterInputError({
+        statusCode: 500,
+        message: `Failed to connect to SFTP server: ${
+          error instanceof Error ? error.message : 'Unknown error'
+        }`,
+      })
+    }
+  }
+
+  private async tryDownloadAndParseFile(instrument: string): Promise<IndexResponseData> {
+    // Validate that the instrument is supported
+    if (!isInstrumentSupported(instrument)) {
+      throw new AdapterInputError({
+        statusCode: 400,
+        message: `Unsupported instrument: ${instrument}`,
+      })
+    }
+
+    const filePath = instrumentToFilePathMap[instrument]
+    const fileRegex = instrumentToFileRegexMap[instrument]
+
+    const fileList = await this.sftpClient.list(filePath)
+    // Filter files based on the regex pattern
+    const matchingFiles = fileList
+      .map((file: FileInfo) => file.name)
+      .filter((fileName: string) => fileRegex.test(fileName))
+
+    if (matchingFiles.length === 0) {
+      throw new AdapterInputError({
+        statusCode: 500,
+        message: `No files matching pattern ${fileRegex} found in directory: ${filePath}`,
+      })
+    } else if (matchingFiles.length > 1) {
+      throw new AdapterInputError({
+        statusCode: 500,
+        message: `Multiple files matching pattern ${fileRegex} found in directory: ${filePath}.`,
+      })
+    }
+    const fullPath = `${filePath}${matchingFiles[0]}`
+
+    // Log the download attempt
+    logger.info(`Downloading file: ${fullPath}`)
+
+    const fileContent = await this.sftpClient.get(fullPath)
+    // we need latin1 here because the file contains special characters like "®"
+    const csvContent = fileContent.toString('latin1')
+
+    const parser = CSVParserFactory.detectParserByInstrument(instrument)
+
+    if (!parser) {
+      throw new AdapterInputError({
+        statusCode: 500,
+        message: `Parser initialization failed for instrument: ${instrument}`,
+      })
+    }
+
+    return (await parser.parse(csvContent)) as IndexResponseData
+  }
+
   getSubscriptionTtlFromConfig(adapterSettings: BaseEndpointTypes['Settings']): number {
     return adapterSettings.BACKGROUND_EXECUTE_MS || 60000
   }

packages/sources/ftse-sftp/test/integration/__snapshots__/adapter.test.ts.snap

@@ -0,0 +1,72 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`execute ftse_sftp endpoint should return success for FTSE100INDEX 1`] = `
+{
+  "data": {
+    "result": {
+      "gbpIndex": 8045.12345678,
+      "indexBaseCurrency": "GBP",
+      "indexCode": "UKX",
+      "indexSectorName": "FTSE 100 Index",
+      "numberOfConstituents": 100,
+    },
+  },
+  "result": 8045.12345678,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1641035471111,
+    "providerDataRequestedUnixMs": 1641035471111,
+  },
+}
+`;
+
+exports[`execute ftse_sftp endpoint should return success for Russell1000INDEX 1`] = `
+{
+  "data": {
+    "result": {
+      "close": 2654.123456,
+      "indexName": "Russell 1000® Index",
+    },
+  },
+  "result": 2654.123456,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1641035471111,
+    "providerDataRequestedUnixMs": 1641035471111,
+  },
+}
+`;
+
+exports[`execute ftse_sftp endpoint should return success for Russell2000INDEX 1`] = `
+{
+  "data": {
+    "result": {
+      "close": 1987.654321,
+      "indexName": "Russell 2000® Index",
+    },
+  },
+  "result": 1987.654321,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1641035471111,
+    "providerDataRequestedUnixMs": 1641035471111,
+  },
+}
+`;
+
+exports[`execute ftse_sftp endpoint should return success for Russell3000INDEX 1`] = `
+{
+  "data": {
+    "result": {
+      "close": 3456.789012,
+      "indexName": "Russell 3000® Index",
+    },
+  },
+  "result": 3456.789012,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1641035471111,
+    "providerDataRequestedUnixMs": 1641035471111,
+  },
+}
+`;