Get price feed endpoint (#764)

* Add get price feed endpoint

* fix stuff

* lint

---------

Co-authored-by: Jayant Krishnamurthy <jkrishnamurthy@jumptrading.com>

price_service/server/src/listen.ts

@@ -14,6 +14,7 @@ import {
   getBatchSummary,
   parseBatchPriceAttestation,
   priceAttestationToPriceFeed,
+  PriceAttestation,
 } from "@pythnetwork/wormhole-attester-sdk";
 import { HexString, PriceFeed } from "@pythnetwork/price-service-sdk";
 import LRUCache from "lru-cache";
@@ -31,6 +32,24 @@ export type PriceInfo = {
   priceServiceReceiveTime: number;
 };
 
+export function createPriceInfo(
+  priceAttestation: PriceAttestation,
+  vaa: Buffer,
+  sequence: bigint,
+  emitterChain: number
+): PriceInfo {
+  const priceFeed = priceAttestationToPriceFeed(priceAttestation);
+  return {
+    seqNum: Number(sequence),
+    vaa,
+    publishTime: priceAttestation.publishTime,
+    attestationTime: priceAttestation.attestationTime,
+    priceFeed,
+    emitterChainId: emitterChain,
+    priceServiceReceiveTime: Math.floor(new Date().getTime() / 1000),
+  };
+}
+
 export interface PriceStore {
   getPriceIds(): Set<HexString>;
   getLatestPriceInfo(priceFeedId: HexString): PriceInfo | undefined;
@@ -324,17 +343,12 @@ export class Listener implements PriceStore {
     for (const priceAttestation of batchAttestation.priceAttestations) {
       const key = priceAttestation.priceId;
 
-      const priceFeed = priceAttestationToPriceFeed(priceAttestation);
+      const priceInfo = createPriceInfo(
-      const priceInfo = {
+        priceAttestation,
-        seqNum: Number(parsedVaa.sequence),
         vaa,
-        publishTime: priceAttestation.publishTime,
+        parsedVaa.sequence,
-        attestationTime: priceAttestation.attestationTime,
+        parsedVaa.emitterChain
-        priceFeed,
+      );
-        emitterChainId: parsedVaa.emitterChain,
-        priceServiceReceiveTime: Math.floor(new Date().getTime() / 1000),
-      };
-
       const cachedPriceInfo = this.priceFeedVaaMap.get(key);
 
       if (this.isNewPriceInfo(cachedPriceInfo, priceInfo)) {

price_service/server/src/rest.ts

@@ -6,11 +6,16 @@ import { Server } from "http";
 import { StatusCodes } from "http-status-codes";
 import morgan from "morgan";
 import fetch from "node-fetch";
+import {
+  parseBatchPriceAttestation,
+  priceAttestationToPriceFeed,
+} from "@pythnetwork/wormhole-attester-sdk";
 import { removeLeading0x, TimestampInSec } from "./helpers";
-import { PriceStore, VaaConfig } from "./listen";
+import { createPriceInfo, PriceInfo, PriceStore, VaaConfig } from "./listen";
 import { logger } from "./logging";
 import { PromClient } from "./promClient";
 import { retry } from "ts-retry-promise";
+import { parseVaa } from "@certusone/wormhole-sdk";
 
 const MORGAN_LOG_FORMAT =
   ':remote-addr - :remote-user ":method :url HTTP/:http-version"' +
@@ -71,7 +76,10 @@ export class RestAPI {
     this.promClient = promClient;
   }
 
-  async getVaaWithDbLookup(priceFeedId: string, publishTime: TimestampInSec) {
+  async getVaaWithDbLookup(
+    priceFeedId: string,
+    publishTime: TimestampInSec
+  ): Promise<VaaConfig | undefined> {
     // Try to fetch the vaa from the local cache
     let vaa = this.priceFeedVaaInfo.getVaa(priceFeedId, publishTime);
 
@@ -104,6 +112,56 @@ export class RestAPI {
     return vaa;
   }
 
+  vaaToPriceInfo(priceFeedId: string, vaa: Buffer): PriceInfo | undefined {
+    const parsedVaa = parseVaa(vaa);
+
+    let batchAttestation;
+
+    try {
+      batchAttestation = parseBatchPriceAttestation(
+        Buffer.from(parsedVaa.payload)
+      );
+    } catch (e: any) {
+      logger.error(e, e.stack);
+      logger.error("Parsing historical VAA failed: %o", parsedVaa);
+      return undefined;
+    }
+
+    for (const priceAttestation of batchAttestation.priceAttestations) {
+      if (priceAttestation.priceId === priceFeedId) {
+        return createPriceInfo(
+          priceAttestation,
+          vaa,
+          parsedVaa.sequence,
+          parsedVaa.emitterChain
+        );
+      }
+    }
+
+    return undefined;
+  }
+
+  priceInfoToJson(
+    priceInfo: PriceInfo,
+    verbose: boolean,
+    binary: boolean
+  ): object {
+    return {
+      ...priceInfo.priceFeed.toJson(),
+      ...(verbose && {
+        metadata: {
+          emitter_chain: priceInfo.emitterChainId,
+          attestation_time: priceInfo.attestationTime,
+          sequence_number: priceInfo.seqNum,
+          price_service_receive_time: priceInfo.priceServiceReceiveTime,
+        },
+      }),
+      ...(binary && {
+        vaa: priceInfo.vaa.toString("base64"),
+      }),
+    };
+  }
+
   // Run this function without blocking (`await`) if you want to run it async.
   async createApp() {
     const app = express();
@@ -283,21 +341,9 @@ export class RestAPI {
             continue;
           }
 
-          responseJson.push({
+          responseJson.push(
-            ...latestPriceInfo.priceFeed.toJson(),
+            this.priceInfoToJson(latestPriceInfo, verbose, binary)
-            ...(verbose && {
+          );
-              metadata: {
-                emitter_chain: latestPriceInfo.emitterChainId,
-                attestation_time: latestPriceInfo.attestationTime,
-                sequence_number: latestPriceInfo.seqNum,
-                price_service_receive_time:
-                  latestPriceInfo.priceServiceReceiveTime,
-              },
-            }),
-            ...(binary && {
-              vaa: latestPriceInfo.vaa.toString("base64"),
-            }),
-          });
         }
 
         if (notFoundIds.length > 0) {
@@ -317,6 +363,62 @@ export class RestAPI {
       "api/latest_price_feeds?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..&verbose=true&binary=true"
     );
 
+    const getPriceFeedInputSchema: schema = {
+      query: Joi.object({
+        id: Joi.string()
+          .regex(/^(0x)?[a-f0-9]{64}$/)
+          .required(),
+        publish_time: Joi.number().required(),
+        verbose: Joi.boolean(),
+        binary: Joi.boolean(),
+      }).required(),
+    };
+
+    app.get(
+      "/api/get_price_feed",
+      validate(getPriceFeedInputSchema),
+      asyncWrapper(async (req: Request, res: Response) => {
+        const priceFeedId = removeLeading0x(req.query.id as string);
+        const publishTime = Number(req.query.publish_time as string);
+        // verbose is optional, default to false
+        const verbose = req.query.verbose === "true";
+        // binary is optional, default to false
+        const binary = req.query.binary === "true";
+
+        if (
+          this.priceFeedVaaInfo.getLatestPriceInfo(priceFeedId) === undefined
+        ) {
+          throw RestException.PriceFeedIdNotFound([priceFeedId]);
+        }
+
+        const vaa = await this.getVaaWithDbLookup(priceFeedId, publishTime);
+        if (vaa === undefined) {
+          throw RestException.VaaNotFound();
+        }
+
+        const priceInfo = this.vaaToPriceInfo(
+          priceFeedId,
+          Buffer.from(vaa.vaa, "base64")
+        );
+
+        if (priceInfo === undefined) {
+          throw RestException.VaaNotFound();
+        } else {
+          res.json(this.priceInfoToJson(priceInfo, verbose, binary));
+        }
+      })
+    );
+
+    endpoints.push(
+      "api/get_price_feed?id=<price_feed_id>&publish_time=<publish_time_in_unix_timestamp>"
+    );
+    endpoints.push(
+      "api/get_price_feed?id=<price_feed_id>&publish_time=<publish_time_in_unix_timestamp>&verbose=true"
+    );
+    endpoints.push(
+      "api/get_price_feed?id=<price_feed_id>&publish_time=<publish_time_in_unix_timestamp>&binary=true"
+    );
+
     app.get("/api/price_feed_ids", (req: Request, res: Response) => {
       const availableIds = this.priceFeedVaaInfo.getPriceIds();
       res.json([...availableIds]);

tilt_devnet/docker_images/Dockerfile.lerna

@@ -18,6 +18,8 @@ COPY ./tsconfig.base.json ./
 
 FROM node:18.13.0@sha256:d9061fd0205c20cd47f70bdc879a7a84fb472b822d3ad3158aeef40698d2ce36 as lerna
 
+RUN apt-get update && apt-get install -y libusb-dev
+
 # 1000 is the uid and gid of the node user
 USER 1000
 RUN mkdir -p /home/node/.npm