556 add prices table to indexeddb (#764)

* PRICES table added

* wip

* save prices

* impl numerairePerUnit calculation

* add amount tests

* review fixes

* add test for price-indexer

* use .env for numeraireAssetId

* review fix

apps/extension/.env

@@ -1,6 +1,6 @@
 
 PRAX=lkpmkhpnhknhmibgnmmhdhgdilepfghe
-IDB_VERSION=28
+IDB_VERSION=29
 USDC_ASSET_ID="reum7wQmk/owgvGMWMZn/6RFPV24zIKq3W6In/WwZgg="
 MINIFRONT_URL=https://app.testnet.penumbra.zone/
 PENUMBRA_NODE_PD_URL=https://grpc.testnet.penumbra.zone/

apps/extension/src/service-worker.ts

@@ -48,6 +48,7 @@ const startServices = async () => {
     grpcEndpoint,
     walletId: wallet0.id,
     fullViewingKey: wallet0.fullViewingKey,
+    numeraireAssetId: USDC_ASSET_ID,
   });
   await services.initialize();
   return services;

packages/getters/src/batch-swap-output-data.ts

@@ -0,0 +1,14 @@
+import { createGetter } from './utils/create-getter';
+import { getAsset1, getAsset2 } from './trading-pair';
+import { BatchSwapOutputData } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/component/dex/v1/dex_pb';
+
+export const getTradingPair = createGetter((b?: BatchSwapOutputData) => b?.tradingPair);
+export const getSwapAsset1 = getTradingPair.pipe(getAsset1);
+export const getSwapAsset2 = getTradingPair.pipe(getAsset2);
+
+export const getDelta1Amount = createGetter((b?: BatchSwapOutputData) => b?.delta1);
+export const getDelta2Amount = createGetter((b?: BatchSwapOutputData) => b?.delta2);
+export const getLambda1Amount = createGetter((b?: BatchSwapOutputData) => b?.lambda1);
+export const getLambda2Amount = createGetter((b?: BatchSwapOutputData) => b?.lambda2);
+export const getUnfilled1Amount = createGetter((b?: BatchSwapOutputData) => b?.unfilled1);
+export const getUnfilled2Amount = createGetter((b?: BatchSwapOutputData) => b?.unfilled2);

packages/query/package.json

@@ -10,6 +10,7 @@
   "dependencies": {
     "@penumbra-zone/crypto-web": "workspace:*",
     "@penumbra-zone/constants": "workspace:*",
+    "@penumbra-zone/getters": "workspace:*",
     "@penumbra-zone/wasm": "workspace:*",
     "@penumbra-zone/types": "workspace:*",
     "exponential-backoff": "^3.1.1"

packages/query/src/block-processor.ts

@@ -22,11 +22,13 @@ import type { BlockProcessorInterface } from '@penumbra-zone/types/src/block-pro
 import type { IndexedDbInterface } from '@penumbra-zone/types/src/indexed-db';
 import type { ViewServerInterface } from '@penumbra-zone/types/src/servers';
 import { customizeSymbol } from '@penumbra-zone/types/src/customize-symbol';
+import { updatePrices } from './price-indexer';
 
 interface QueryClientProps {
   querier: RootQuerier;
   indexedDb: IndexedDbInterface;
   viewServer: ViewServerInterface;
+  numeraireAssetId: string;
 }
 
 const blankTxSource = new CommitmentSource({
@@ -38,12 +40,14 @@ export class BlockProcessor implements BlockProcessorInterface {
   private readonly indexedDb: IndexedDbInterface;
   private readonly viewServer: ViewServerInterface;
   private readonly abortController: AbortController = new AbortController();
+  private readonly numeraireAssetId: string;
   private syncPromise: Promise<void> | undefined;
 
-  constructor({ indexedDb, viewServer, querier }: QueryClientProps) {
+  constructor({ indexedDb, viewServer, querier, numeraireAssetId }: QueryClientProps) {
     this.indexedDb = indexedDb;
     this.viewServer = viewServer;
     this.querier = querier;
+    this.numeraireAssetId = numeraireAssetId;
   }
 
   // If syncBlocks() is called multiple times concurrently, they'll all wait for
@@ -232,6 +236,18 @@ export class BlockProcessor implements BlockProcessorInterface {
         await this.saveTransactions(compactBlock.height, relevantTx);
       }
 
+      // we can't use third-party price oracles for privacy reasons,
+      // so we have to get asset prices from swap results during block scans
+      // and store them locally in indexed-db.
+      if (compactBlock.swapOutputs.length) {
+        await updatePrices(
+          this.indexedDb,
+          this.numeraireAssetId,
+          compactBlock.swapOutputs,
+          compactBlock.height,
+        );
+      }
+
       // We only query Tendermint for the latest known block height once, when
       // the block processor starts running. Once we're caught up, though, the
       // chain will of course continue adding blocks, and we'll keep processing

packages/query/src/price-indexer.test.ts

@@ -0,0 +1,115 @@
+import { updatePrices } from './price-indexer';
+import { BatchSwapOutputData } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/component/dex/v1/dex_pb';
+import { beforeEach, describe, expect, it, Mock, vi } from 'vitest';
+import { IndexedDbInterface } from '@penumbra-zone/types/src/indexed-db';
+import { AssetId } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/asset/v1/asset_pb';
+import { base64ToUint8Array } from '@penumbra-zone/types/src/base64';
+
+describe('update prices', () => {
+  let indexedDbMock: IndexedDbInterface;
+  const updatePriceMock: Mock = vi.fn();
+  const height = 123n;
+  const numeraireAssetId = 'reum7wQmk/owgvGMWMZn/6RFPV24zIKq3W6In/WwZgg=';
+  const numeraireAsset: AssetId = new AssetId({
+    inner: base64ToUint8Array(numeraireAssetId),
+  });
+  beforeEach(() => {
+    vi.clearAllMocks();
+
+    indexedDbMock = {
+      updatePrice: updatePriceMock,
+    } as unknown as IndexedDbInterface;
+  });
+
+  it('should update prices correctly for a swapOutput with NUMERAIRE as swapAsset2', async () => {
+    const asset1 = new AssetId({ inner: new Uint8Array(12) });
+    const swapOutputs: BatchSwapOutputData[] = [
+      new BatchSwapOutputData({
+        tradingPair: {
+          asset1: asset1,
+          asset2: numeraireAsset,
+        },
+        delta1: { lo: 250n },
+        lambda2: { lo: 1200n },
+        unfilled1: { lo: 0n },
+      }),
+    ];
+
+    await updatePrices(indexedDbMock, numeraireAssetId, swapOutputs, height);
+    expect(updatePriceMock).toBeCalledTimes(1);
+    expect(updatePriceMock).toBeCalledWith(asset1, numeraireAsset, 4.8, height);
+  });
+
+  it('should update prices correctly for a swapOutput with NUMERAIRE as swapAsset1', async () => {
+    const asset1 = new AssetId({ inner: new Uint8Array(12) });
+    const swapOutputs: BatchSwapOutputData[] = [
+      new BatchSwapOutputData({
+        tradingPair: {
+          asset1: numeraireAsset,
+          asset2: asset1,
+        },
+        delta2: { lo: 40n },
+        lambda1: { lo: 12740n },
+        unfilled2: { lo: 0n },
+      }),
+    ];
+
+    await updatePrices(indexedDbMock, numeraireAssetId, swapOutputs, height);
+    expect(updatePriceMock).toBeCalledTimes(1);
+    expect(updatePriceMock).toBeCalledWith(asset1, numeraireAsset, 318.5, height);
+  });
+
+  it('should not update prices if delta is zero', async () => {
+    const asset1 = new AssetId({ inner: new Uint8Array(12) });
+    const swapOutputs: BatchSwapOutputData[] = [
+      new BatchSwapOutputData({
+        tradingPair: {
+          asset1: numeraireAsset,
+          asset2: asset1,
+        },
+        delta2: { lo: 0n },
+        lambda1: { lo: 12740n },
+        unfilled2: { lo: 0n },
+      }),
+    ];
+
+    await updatePrices(indexedDbMock, numeraireAssetId, swapOutputs, height);
+    expect(updatePriceMock).toBeCalledTimes(0);
+  });
+
+  it('should update prices correctly for partially filled', async () => {
+    const asset1 = new AssetId({ inner: new Uint8Array(12) });
+    const swapOutputs: BatchSwapOutputData[] = [
+      new BatchSwapOutputData({
+        tradingPair: {
+          asset1: asset1,
+          asset2: numeraireAsset,
+        },
+        delta1: { lo: 250n },
+        lambda2: { lo: 1200n },
+        unfilled1: { lo: 100n },
+      }),
+    ];
+    await updatePrices(indexedDbMock, numeraireAssetId, swapOutputs, height);
+    expect(updatePriceMock).toBeCalledTimes(1);
+    expect(updatePriceMock).toBeCalledWith(asset1, numeraireAsset, 8, height);
+  });
+
+  it('should not update prices if swap is fully unfilled', async () => {
+    const asset1 = new AssetId({ inner: new Uint8Array(12) });
+    const swapOutputs: BatchSwapOutputData[] = [
+      new BatchSwapOutputData({
+        tradingPair: {
+          asset1: numeraireAsset,
+          asset2: asset1,
+        },
+        delta2: { lo: 100n },
+        lambda1: { lo: 12740n },
+        unfilled2: { lo: 100n },
+      }),
+    ];
+
+    await updatePrices(indexedDbMock, numeraireAssetId, swapOutputs, height);
+    expect(updatePriceMock).toBeCalledTimes(0);
+  });
+});

packages/query/src/price-indexer.ts

@@ -0,0 +1,91 @@
+import { BatchSwapOutputData } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/component/dex/v1/dex_pb';
+import { IndexedDbInterface } from '@penumbra-zone/types/src/indexed-db';
+import { divideAmounts, isZero, subtractAmounts } from '@penumbra-zone/types/src/amount';
+import { AssetId } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/asset/v1/asset_pb';
+import { Amount } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/num/v1/num_pb';
+import {
+  getDelta1Amount,
+  getDelta2Amount,
+  getLambda1Amount,
+  getLambda2Amount,
+  getSwapAsset1,
+  getSwapAsset2,
+  getUnfilled1Amount,
+  getUnfilled2Amount,
+} from '@penumbra-zone/getters/src/batch-swap-output-data';
+import { base64ToUint8Array } from '@penumbra-zone/types/src/base64';
+
+/**
+ *
+ * @param delta -  total amount of 'pricedAsset' that was input to the batch swap
+ * @param unfilled - total amount of 'pricedAsset' that was returned unfilled
+ * @param lambda - total amount of 'numeraire' that was output from the batch swap
+ *  Price formula:
+ *  price = (lambda)/(delta - unfilled)
+ *  The price cannot be calculated if
+ *  - lambda is zero
+ *  - delta is zero
+ *  - (delta - unfilled) is zero
+ * @return 0 if the price cannot be calculated and some positive number if the price has been calculated.
+ */
+export const calculatePrice = (delta: Amount, unfilled: Amount, lambda: Amount): number => {
+  const filledAmount = subtractAmounts(delta, unfilled);
+  //
+  return isZero(delta) || isZero(lambda) || isZero(filledAmount)
+    ? 0
+    : divideAmounts(lambda, filledAmount).toNumber();
+};
+
+/**
+ * Each 'BatchSwapOutputData' (BSOD) can generate up to two prices
+ * Each BSOD in block has a unique trading pair
+ * Trading pair has a canonical ordering, there's only one trading pair per pair of assets
+ * Each BSOD can generate up to two prices
+ * 1. pricedAsset -> numeraire (selling price)
+ * 2. numeraire -> pricedAsset (buying price)
+ * This function processes only (1) price and ignores (2) price
+ * We can get a BSOD with zero deltas(inputs), and we shouldn't save the price in that case
+ */
+export const updatePrices = async (
+  indexedDb: IndexedDbInterface,
+  numeraireAssetId: string,
+  swapOutputs: BatchSwapOutputData[],
+  height: bigint,
+) => {
+  const numeraireAsset: AssetId = new AssetId({
+    inner: base64ToUint8Array(numeraireAssetId),
+  });
+
+  for (const swapOutput of swapOutputs) {
+    const swapAsset1 = getSwapAsset1(swapOutput);
+    const swapAsset2 = getSwapAsset2(swapOutput);
+
+    let numerairePerUnit = 0;
+    let pricedAsset: AssetId | undefined = undefined;
+
+    // case for trading pair <pricedAsset,numéraire>
+    if (swapAsset2.equals(numeraireAsset)) {
+      pricedAsset = swapAsset1;
+      // numerairePerUnit = lambda2/(delta1-unfilled1)
+      numerairePerUnit = calculatePrice(
+        getDelta1Amount(swapOutput),
+        getUnfilled1Amount(swapOutput),
+        getLambda2Amount(swapOutput),
+      );
+    }
+    // case for trading pair <numéraire,pricedAsset>
+    else if (swapAsset1.equals(numeraireAsset)) {
+      pricedAsset = swapAsset2;
+      // numerairePerUnit = lambda1/(delta2-unfilled2)
+      numerairePerUnit = calculatePrice(
+        getDelta2Amount(swapOutput),
+        getUnfilled2Amount(swapOutput),
+        getLambda1Amount(swapOutput),
+      );
+    }
+
+    if (pricedAsset === undefined || numerairePerUnit === 0) continue;
+
+    await indexedDb.updatePrice(pricedAsset, numeraireAsset, numerairePerUnit, height);
+  }
+};

packages/services/src/index.ts

@@ -14,10 +14,15 @@ export interface ServicesConfig {
   readonly grpcEndpoint?: string;
   readonly walletId?: string;
   readonly fullViewingKey?: string;
+  readonly numeraireAssetId: string;
 }
 
 const isCompleteServicesConfig = (c: Partial<ServicesConfig>): c is Required<ServicesConfig> =>
-  c.grpcEndpoint != null && c.idbVersion != null && c.walletId != null && c.fullViewingKey != null;
+  c.grpcEndpoint != null &&
+  c.idbVersion != null &&
+  c.walletId != null &&
+  c.fullViewingKey != null &&
+  c.numeraireAssetId != null;
 
 export class Services implements ServicesInterface {
   private walletServicesPromise: Promise<WalletServices> | undefined;
@@ -83,7 +88,7 @@ export class Services implements ServicesInterface {
   }
 
   private async initializeWalletServices(): Promise<WalletServices> {
-    const { walletId, fullViewingKey, idbVersion: dbVersion } = await this.config;
+    const { walletId, fullViewingKey, idbVersion: dbVersion, numeraireAssetId } = await this.config;
     const params = await this.querier.app.appParams();
     if (!params.sctParams?.epochDuration) throw new Error('Epoch duration unknown');
     const {
@@ -110,6 +115,7 @@ export class Services implements ServicesInterface {
       viewServer,
       querier: this.querier,
       indexedDb,
+      numeraireAssetId,
     });
 
     return { viewServer, blockProcessor, indexedDb, querier: this.querier };

packages/storage/src/indexed-db/index.ts

@@ -14,6 +14,7 @@ import {
 } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/view/v1/view_pb';
 import {
   AssetId,
+  EstimatedPrice,
   Metadata,
 } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/asset/v1/asset_pb';
 import { StateCommitment } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/crypto/tct/v1/tct_pb';
@@ -104,6 +105,7 @@ export class IndexedDb implements IndexedDbInterface {
         db.createObjectStore('POSITIONS', { keyPath: 'id.inner' });
         db.createObjectStore('EPOCHS', { autoIncrement: true });
         db.createObjectStore('VALIDATOR_INFOS');
+        db.createObjectStore('PRICES', { keyPath: ['pricedAsset.inner', 'numeraire.inner'] });
       },
     });
     const constants = {
@@ -536,6 +538,25 @@ export class IndexedDb implements IndexedDbInterface {
     );
   }
 
+  async updatePrice(
+    pricedAsset: AssetId,
+    numeraire: AssetId,
+    numerairePerUnit: number,
+    height: bigint,
+  ) {
+    const estimatedPrice = new EstimatedPrice({
+      pricedAsset,
+      numeraire,
+      numerairePerUnit,
+      asOfHeight: height,
+    });
+
+    await this.u.update({
+      table: 'PRICES',
+      value: estimatedPrice.toJson() as Jsonified<EstimatedPrice>,
+    });
+  }
+
   private addSctUpdates(txs: IbdUpdates, sctUpdates: ScanBlockResult['sctUpdates']): void {
     if (sctUpdates.set_position) {
       txs.add({