feat(price-server): support parsing accumul data

package-lock.json

@@ -57447,7 +57447,7 @@
     },
     "price_service/server": {
       "name": "@pythnetwork/price-service-server",
-      "version": "3.0.8",
+      "version": "3.1.0",
       "license": "Apache-2.0",
       "dependencies": {
         "@certusone/wormhole-sdk": "^0.9.9",

price_service/server/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pythnetwork/price-service-server",
-  "version": "3.0.8",
+  "version": "3.1.0",
   "description": "Webservice for retrieving prices from the Pyth oracle.",
   "private": "true",
   "main": "index.js",

price_service/server/src/__tests__/rest.test.ts

@@ -508,4 +508,59 @@ describe("Get VAA endpoint and Get VAA CCIP", () => {
       expect(ccipResp.status).toBe(StatusCodes.INTERNAL_SERVER_ERROR);
     }
   );
+
+  test("vaaToPriceInfo works with accumulator update data", () => {
+    // An update data taken from Hermes with the following price feed:
+    // {
+    //   "id":"e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43",
+    //   "price":{"price":"2836040669135","conf":"3282830965","expo":-8,"publish_time":1692280808},
+    //   "ema_price":{"price":"2845324900000","conf":"3211773100","expo":-8,"publish_time":1692280808},
+    //   "metadata":{"slot":89783664,"emitter_chain":26,"price_service_receive_time":1692280809}
+    // }
+    const updateData = Buffer.from(
+      "UE5BVQEAAAADuAEAAAADDQAsKPsmb7Vz7io3taJQKgoi1m/z0kqKgtpmlkv+ZuunX2Iegsf+8fuUtpHPLKgCWPU8PN2x9NyAZz5" +
+        "BY9M3SWwJAALYlM0U7f2GFWfEjKwSJlHZ5sf+n6KXCocVC66ImS2o0TD0SBhTWcp0KdcuzR1rY1jfIHaFpVneroRLbTjNrk/WAA" +
+        "MuAYxPVPf1DR30wYQo12Dbf+in3akTjhKERNQ+nPwRjxAyIQD+52LU3Rh2VL7nOIStMNTiBMaiWHywaPoXowWAAQbillhhX4MR+" +
+        "7h81PfxHIbiXBmER4c5M7spilWKkROb+VXhrqnVJL162t9TdhYk56PDIhvXO1Tm/ldjVJw130y0AAk6qpccfsxDZEmVN8LI4z87" +
+        "39Ni/kb+CB3yW2l2dWhKTjBeNanhK6TCCoNH/jRzWfrjrEk5zjNrUr82JwL4fR1OAQrYZescxbH26m8QHiH+RHzwlXpUKJgbHD5" +
+        "NnWtB7oFb9AFM15jbjd4yIEBEtAlXPE0Q4j+X+DLnCtZbLSQiYNh5AQvz70LTbYry1lEExuUcO+IRJiysw5AFyqZ9Y1E//WKIqg" +
+        "EysfcnHwoOxtDtAc5Z9sTUEYfPqQ1d27k3Yk0X7dvCAQ10cdG0qYHb+bQrYRIKKnb0aeCjkCs0HZQY2fXYmimyfTNfECclmPW9k" +
+        "+CfOvW0JKuFxC1l11zJ3zjsgN/peA8BAQ5oIFQGjq9qmf5gegE1DjuzXsGksKao6nsjTXYIspCczCe2h5KNQ9l5hws11hauUKS2" +
+        "0JoOYjHwxPD2x0adJKvkAQ+4UjVcZgVEQP8y3caqUDH81Ikcadz2bESpYg93dpnzZTH6A7Ue+RL34PTNx6cCRzukwQuhiStuyL1" +
+        "WYEIrLI4nABAjGv3EBXjWaPLUj59OzVnGkzxkr6C4KDjMmpsYNzx7I2lp2iQV46TM78El8i9h7twiEDUOSdC5CmfQjRpkP72yAB" +
+        "GVAQELUm2/SjkpF0O+/rVDgA/Y2/wMacD1ZDahdyvSNSFThn5NyRYA1JXGgIDxoYeAZgkr1gL1cjCLWiO+Bs9QARIiCvHfIkn2a" +
+        "YhYHQq/u6cHB/2DxE3OgbCZyTv8OVO55hQDkJ1gDwAec+IJ4M5Od4OxWEu+OywhJT7zUmwZko9MAGTeJ+kAAAAAABrhAfrtrFhR" +
+        "4yubI7X5QRqMK6xKrj7U3XuBHdGnLqSqcQAAAAAAWllxAUFVV1YAAAAAAAVZ/XAAACcQ8Xfx5wQ+nj1rn6IeTUAy+VER1nUBAFU" +
+        "A5i32yLSoX+GmfbRNwS3l2zMPesZrctxliv7fD0pBW0MAAAKUUTJXzwAAAADDrAZ1////+AAAAABk3ifoAAAAAGTeJ+cAAAKWep" +
+        "R2oAAAAAC/b8SsCasjFzENKvXWwOycuzCVaDWfm0IuuuesmamDKl2lNXss15orlNN+xHVNEEIIq7Xg8GRZGVLt43fkg7xli6EPQ" +
+        "/Nyxl6SixiYteNt1uTTh4M1lQTUjPxKnkE5JEea4RnhOWgmSAWMf8ft4KgE7hvRifV1JP0rOsNgsOYFRbs6iDKW1qLpxgZLMAiO" +
+        "clwS3Tjw2hj8sPfq1NHeVttsBEK5SIM14GjAuD/p2V0+NqHqMHxU/kfftg==",
+      "base64"
+    );
+
+    const priceInfo = RestAPI.vaaToPriceInfo(
+      "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43",
+      updateData
+    );
+
+    expect(priceInfo).toBeDefined();
+    expect(priceInfo?.priceFeed).toEqual(
+      new PriceFeed({
+        id: "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43",
+        price: new Price({
+          price: "2836040669135",
+          conf: "3282830965",
+          publishTime: 1692280808,
+          expo: -8,
+        }),
+        emaPrice: new Price({
+          price: "2845324900000",
+          conf: "3211773100",
+          publishTime: 1692280808,
+          expo: -8,
+        }),
+      })
+    );
+    expect(priceInfo?.emitterChainId).toEqual(26);
+  });
 });

price_service/server/src/rest.ts

@@ -1,4 +1,4 @@
-import { HexString } from "@pythnetwork/price-service-sdk";
+import { HexString, Price, PriceFeed } from "@pythnetwork/price-service-sdk";
 import cors from "cors";
 import express, { NextFunction, Request, Response } from "express";
 import { Joi, schema, validate, ValidationError } from "express-validation";
@@ -6,12 +6,9 @@ import { Server } from "http";
 import { StatusCodes } from "http-status-codes";
 import morgan from "morgan";
 import fetch from "node-fetch";
-import {
+import { parseBatchPriceAttestation } from "@pythnetwork/wormhole-attester-sdk";
-  parseBatchPriceAttestation,
-  priceAttestationToPriceFeed,
-} from "@pythnetwork/wormhole-attester-sdk";
 import { removeLeading0x, TimestampInSec } from "./helpers";
-import { createPriceInfo, PriceInfo, PriceStore, VaaConfig } from "./listen";
+import { createPriceInfo, PriceInfo, PriceStore } from "./listen";
 import { logger } from "./logging";
 import { PromClient } from "./promClient";
 import { retry } from "ts-retry-promise";
@@ -21,7 +18,6 @@ import {
   TargetChain,
   validTargetChains,
   defaultTargetChain,
-  VaaEncoding,
   encodeVaaForChain,
 } from "./encoding";
 
@@ -136,7 +132,128 @@ export class RestAPI {
     return vaa;
   }
 
-  vaaToPriceInfo(priceFeedId: string, vaa: Buffer): PriceInfo | undefined {
+  // Extract the price info from an Accumulator update. This is a temporary solution until hermes adoption
+  // to maintain backward compatibility when the db migrates to the new update format.
+  static extractPriceInfoFromAccumulatorUpdate(
+    priceFeedId: string,
+    updateData: Buffer
+  ): PriceInfo | undefined {
+    let offset = 0;
+    offset += 4; // magic
+    offset += 1; // major version
+    offset += 1; // minor version
+
+    const trailingHeaderSize = updateData.readUint8(offset);
+    offset += 1 + trailingHeaderSize;
+
+    const updateType = updateData.readUint8(offset);
+    offset += 1;
+
+    // There is a single update type of 0 for now.
+    if (updateType !== 0) {
+      logger.error(`Invalid accumulator update type: ${updateType}`);
+      return undefined;
+    }
+
+    const vaaLength = updateData.readUint16BE(offset);
+    offset += 2;
+
+    const vaaBuffer = updateData.slice(offset, offset + vaaLength);
+    const vaa = parseVaa(vaaBuffer);
+    offset += vaaLength;
+
+    const numUpdates = updateData.readUint8(offset);
+    offset += 1;
+
+    // Iterate through the updates to find the price info with the given id
+    for (let i = 0; i < numUpdates; i++) {
+      const messageLength = updateData.readUint16BE(offset);
+      offset += 2;
+
+      const message = updateData.slice(offset, offset + messageLength);
+      offset += messageLength;
+
+      const proofLength = updateData.readUint8(offset);
+      offset += 1;
+
+      // ignore proofs
+      offset += proofLength;
+
+      // Checket whether the message is a price feed update
+      // from the given price id and if so, extract the price info
+      let messageOffset = 0;
+      const messageType = message.readUint8(messageOffset);
+      messageOffset += 1;
+
+      // MessageType of 0 is a price feed update
+      if (messageType !== 0) {
+        continue;
+      }
+
+      const priceId = message
+        .slice(messageOffset, messageOffset + 32)
+        .toString("hex");
+      messageOffset += 32;
+
+      if (priceId !== priceFeedId) {
+        continue;
+      }
+
+      const price = message.readBigInt64BE(messageOffset);
+      messageOffset += 8;
+      const conf = message.readBigUint64BE(messageOffset);
+      messageOffset += 8;
+      const expo = message.readInt32BE(messageOffset);
+      messageOffset += 4;
+      const publishTime = message.readBigInt64BE(messageOffset);
+      messageOffset += 8;
+      const prevPublishTime = message.readBigInt64BE(messageOffset);
+      messageOffset += 8;
+      const emaPrice = message.readBigInt64BE(messageOffset);
+      messageOffset += 8;
+      const emaConf = message.readBigUint64BE(messageOffset);
+
+      return {
+        priceFeed: new PriceFeed({
+          id: priceFeedId,
+          price: new Price({
+            price: price.toString(),
+            conf: conf.toString(),
+            expo,
+            publishTime: Number(publishTime),
+          }),
+          emaPrice: new Price({
+            price: emaPrice.toString(),
+            conf: emaConf.toString(),
+            expo,
+            publishTime: Number(publishTime),
+          }),
+        }),
+        publishTime: Number(publishTime),
+        vaa: vaaBuffer,
+        seqNum: Number(vaa.sequence),
+        emitterChainId: vaa.emitterChain,
+        // These are not available in the accumulator update format
+        // but are required by the PriceInfo type.
+        attestationTime: Number(publishTime),
+        lastAttestedPublishTime: Number(prevPublishTime),
+        priceServiceReceiveTime: Number(publishTime),
+      };
+    }
+
+    return undefined;
+  }
+
+  static vaaToPriceInfo(
+    priceFeedId: string,
+    vaa: Buffer
+  ): PriceInfo | undefined {
+    // Vaa could be the update data from the db with the Accumulator format.
+    const ACCUMULATOR_MAGIC = "504e4155";
+    if (vaa.slice(0, 4).toString("hex") === ACCUMULATOR_MAGIC) {
+      return RestAPI.extractPriceInfoFromAccumulatorUpdate(priceFeedId, vaa);
+    }
+
     const parsedVaa = parseVaa(vaa);
 
     let batchAttestation;
@@ -454,7 +571,7 @@ export class RestAPI {
           throw RestException.VaaNotFound();
         }
 
-        const priceInfo = this.vaaToPriceInfo(
+        const priceInfo = RestAPI.vaaToPriceInfo(
           priceFeedId,
           Buffer.from(vaa.vaa, "base64")
         );