Add oracle rates

src/utils/rates/rates-oracles.ts

@@ -0,0 +1,170 @@
+import {
+  OracleClient,
+  OracleMetadataMessage,
+  OraclePriceMessage,
+  type OracleMetadataMap,
+} from '@generalprotocols/oracle-client';
+
+import { type RatesEventMap, BaseRates } from './base-rates.js';
+
+// Add the Oracle Price Message to our Events for this Adapter.
+export type RatesOracleEventMap = RatesEventMap & {
+  rateUpdated: {
+    oraclePriceMessage: OraclePriceMessage;
+  };
+};
+
+// TODO: Add RatesHistorical trait since Oracles can provide historical rates.
+export class RatesOracle extends BaseRates<RatesOracleEventMap> {
+  /**
+   * Create a new rates oracle.
+   *
+   * @param client The underlying oracle client. If not provided, a new client will be created.
+   * @returns The rates oracle.
+   */
+  static async from(client?: OracleClient) {
+    const ratesOracle = new RatesOracle(client ?? (await OracleClient.from()));
+
+    return ratesOracle;
+  }
+
+  private client: OracleClient;
+  private oracles: OracleMetadataMap;
+
+  private started: boolean = false;
+
+  private constructor(client: OracleClient) {
+    super();
+
+    this.client = client;
+    this.oracles = {};
+  }
+
+  /**
+   * Start the rates oracle and the underlying client.
+   */
+  async start() {
+    if (this.started) {
+      return;
+    }
+    this.started = true;
+
+    // Create event listeners for the client.
+    this.client.setOnMetadataMessage(this.handleMetadataMessage.bind(this));
+    this.client.setOnPriceMessage(this.handlePriceMessage.bind(this));
+
+    // Get the metadata for the client.
+    this.oracles = await this.client.getMetadataMap();
+
+    // Start the client.
+    await this.client.start();
+
+    // Refresh the prices.
+    await this.refreshPrices();
+  }
+
+  /**
+   * Stop the rates oracle and the underlying client.
+   */
+  async stop() {
+    if (!this.started) {
+      return;
+    }
+    this.started = false;
+
+    // Remove event listeners by setting them to empty functions.
+    this.client.setOnMetadataMessage(() => {});
+    this.client.setOnPriceMessage(() => {});
+
+    await this.client.stop();
+  }
+
+  /**
+   * List the pairs that we are tracking.
+   *
+   * @returns A set of pairs.
+   */
+  async listPairs() {
+    return new Set(
+      Object.values(this.oracles).map((oracle) => {
+        return `${oracle.SOURCE_NUMERATOR_UNIT_CODE}/${oracle.SOURCE_DENOMINATOR_UNIT_CODE}`;
+      }),
+    );
+  }
+
+  /**
+   * Get the latest prices for all the pairs and emit a rate updated event for each.
+   */
+  public async refreshPrices() {
+    const oracles = await this.client.getOracles();
+
+    // For each oracle, get the lastest dataSequence (price) message and emit a rate updated event.
+    await Promise.allSettled(
+      oracles.map(async (oracle) => {
+        try {
+          const messages = await this.client.getOracleMessages({
+            publicKey: oracle.publicKey,
+            minDataSequence: 1,
+            count: 1,
+          });
+
+          // We are only expecting a single message back. Just in case, we take the latest one.
+          const message = messages.reduce((latest, msg) => {
+            if (
+              msg instanceof OraclePriceMessage &&
+              msg.messageSequence > (latest?.messageSequence ?? 0)
+            ) {
+              return msg;
+            }
+            return latest;
+          }, messages[0]);
+
+          // If the message is a price message, handle it.
+          if (message instanceof OraclePriceMessage) {
+            this.handlePriceMessage(message);
+          }
+        } catch (error) {
+          console.error('Error refreshing prices for oracle:', oracle.publicKey, error);
+        }
+      }),
+    );
+  }
+
+  /**
+   * Update the metadata map that we use to track the pairs.
+   *
+   * @param message The metadata message.
+   */
+  private handleMetadataMessage(message: OracleMetadataMessage) {
+    this.oracles = OracleClient.updateMetadataMap(this.oracles, message);
+  }
+
+  /**
+   * Emit a rate updated event for the given pair.
+   *
+   * @param message The price message.
+   */
+  private handlePriceMessage(message: OraclePriceMessage) {
+    const oracle = this.oracles[message.toHexObject().publicKey];
+
+    // If the oracle doesn't have the required metadata, we can't use it.
+    if (
+      !oracle ||
+      !oracle.SOURCE_NUMERATOR_UNIT_CODE ||
+      !oracle.SOURCE_DENOMINATOR_UNIT_CODE ||
+      !oracle.ATTESTATION_SCALING
+    ) {
+      return;
+    }
+
+    // Scale the price
+    const priceValue = message.priceValue / oracle.ATTESTATION_SCALING;
+
+    this.emit('rateUpdated', {
+      numeratorUnitCode: oracle.SOURCE_NUMERATOR_UNIT_CODE,
+      denominatorUnitCode: oracle.SOURCE_DENOMINATOR_UNIT_CODE,
+      price: priceValue,
+      oraclePriceMessage: message,
+    });
+  }
+}