Feat: Add cumulative dex volumes

.github/workflows/build.yaml

@@ -13,7 +13,7 @@ jobs:
             - name: setup node
               uses: actions/setup-node@v3
               with:
-                  node-version: "16"
+                  node-version-file: '.nvmrc'
             - run: yarn install
             - name: Check versions
               run: |

combined.typegen.json

@@ -4,6 +4,8 @@
     "typesBundle": "indexer/typesBundle.json",
     "events": [
         "BTCRelay.StoreMainChainHeader",
+        "DexGeneral.AssetSwap",
+        "DexStable.CurrencyExchange",
         "Escrow.Deposit",
         "Escrow.Withdraw",
         "Issue.CancelIssue",
@@ -30,16 +32,14 @@
         "Tokens.Transfer",
         "VaultRegistry.DecreaseLockedCollateral",
         "VaultRegistry.IncreaseLockedCollateral",
-        "VaultRegistry.RegisterVault",
-        "DexGeneral.AssetSwap",
-        "DexGeneral.LiquidityAdded",
-        "DexGeneral.LiquidityRemoved"
+        "VaultRegistry.RegisterVault"
     ],
     "calls": [
         "BTCRelay.store_block_header",
         "System.set_storage"
     ],
     "storage": [
+        "DexStable.Pools",
         "Issue.IssuePeriod",
         "Redeem.RedeemPeriod"
     ]

db/migrations/1675434206425-Data.js

@@ -0,0 +1,15 @@
+module.exports = class Data1675434206425 {
+    name = 'Data1675434206425'
+
+    async up(db) {
+        await db.query(`CREATE TABLE "cumulative_dex_trading_volume_per_pool" ("id" character varying NOT NULL, "pool_id" text NOT NULL, "pool_type" character varying(8) NOT NULL, "till_timestamp" TIMESTAMP WITH TIME ZONE NOT NULL, "amounts" jsonb NOT NULL, CONSTRAINT "PK_c9bb1ee57bff1390d948e3e6f12" PRIMARY KEY ("id"))`)
+        await db.query(`CREATE INDEX "IDX_bd6bc9a6ce9e1fcb81b0650c0f" ON "cumulative_dex_trading_volume_per_pool" ("pool_id") `)
+        await db.query(`CREATE INDEX "IDX_a903319c2555960f188406a839" ON "cumulative_dex_trading_volume_per_pool" ("till_timestamp") `)
+    }
+
+    async down(db) {
+        await db.query(`DROP TABLE "cumulative_dex_trading_volume_per_pool"`)
+        await db.query(`DROP INDEX "public"."IDX_bd6bc9a6ce9e1fcb81b0650c0f"`)
+        await db.query(`DROP INDEX "public"."IDX_a903319c2555960f188406a839"`)
+    }
+}

distributable/schema.graphql

@@ -243,6 +243,26 @@ type CumulativeVolumePerCurrencyPair @entity {
     collateralCurrency: Currency
 }
 
+type PooledAmount {
+    amount: BigInt!
+    amountHuman: BigDecimal
+    token: PooledToken!
+    # TODO: check if this should be Currency?
+}
+
+enum PoolType {
+    Standard
+    Stable
+}
+
+type CumulativeDexTradingVolumePerPool @entity {
+    id: ID!
+    poolId: String! @index
+    poolType: PoolType!
+    tillTimestamp: DateTime! @index
+    amounts: [PooledAmount!]!
+}
+
 type IssuePeriod @entity {
     height: Height!
     timestamp: DateTime!

package.json

@@ -1,7 +1,7 @@
 {
     "name": "interbtc-indexer",
     "private": "true",
-    "version": "0.14.1",
+    "version": "0.14.2",
     "description": "GraphQL server and Substrate indexer for the interBTC parachain",
     "author": "",
     "license": "ISC",

schema.graphql

@@ -243,6 +243,26 @@ type CumulativeVolumePerCurrencyPair @entity {
     collateralCurrency: Currency
 }
 
+type PooledAmount {
+    amount: BigInt!
+    amountHuman: BigDecimal
+    token: PooledToken!
+    # TODO: check if this should be Currency?
+}
+
+enum PoolType {
+    Standard
+    Stable
+}
+
+type CumulativeDexTradingVolumePerPool @entity {
+    id: ID!
+    poolId: String! @index
+    poolType: PoolType!
+    tillTimestamp: DateTime! @index
+    amounts: [PooledAmount!]!
+}
+
 type IssuePeriod @entity {
     height: Height!
     timestamp: DateTime!

src/mappings/_utils.ts

@@ -80,7 +80,7 @@ export async function getForeignAsset(id: number): Promise<AssetMetadata> {
     }
     try {
         const wsProvider = new WsProvider(process.env.CHAIN_ENDPOINT);
-        const api = await ApiPromise.create({ provider: wsProvider });
+        const api = await ApiPromise.create({ provider: wsProvider, noInitWarn: true });
         const assets = await api.query.assetRegistry.metadata(id);
         const assetsJSON = assets.toHuman();
         const metadata = assetsJSON as AssetMetadata;
@@ -170,4 +170,9 @@ export async function convertAmountToHuman(currency: Currency, amount: bigint )
     const currencyInfo: CurrencyExt = await currencyToLibCurrencyExt(currency);
     const monetaryAmount = newMonetaryAmount(amount.toString(), currencyInfo);
     return BigDecimal(monetaryAmount.toString());
+}
+
+// helper method to switch around key/value pairs for a given map
+export function invertMap<K extends Object, V extends Object>(map: Map<K, V>): Map<V, K> {
+    return new Map(Array.from(map, ([key, value]) => [value, key]));
 }

src/mappings/event/dex.ts

@@ -0,0 +1,177 @@
+import { SubstrateBlock } from "@subsquid/substrate-processor";
+import { CumulativeDexTradingVolumePerPool, Currency, fromJsonPooledToken, PooledToken } from "../../model";
+import { Ctx, EventItem } from "../../processor";
+import { DexGeneralAssetSwapEvent, DexStableCurrencyExchangeEvent } from "../../types/events";
+import { currencyId } from "../encoding";
+import { SwapDetails, updateCumulativeDexVolumesForStablePool, updateCumulativeDexVolumesForStandardPool } from "../utils/cumulativeVolumes";
+import EntityBuffer from "../utils/entityBuffer";
+import { getStablePoolCurrencyByIndex } from "../utils/pools";
+
+function isPooledToken(currency: Currency): currency is PooledToken {
+    try {
+        fromJsonPooledToken(currency);
+        return true;
+    } catch (e) {
+        return false;
+    }
+}
+
+/**
+ * Combines the given arrays into in/out pairs as an array of {@link SwapDetails}.
+ * @param currencies The currencies in the swap path
+ * @param atomicBalances The swapped balances, in atomic units and same order as currencies
+ * @returns An array of pair-wise combined {@link SwapDetails}.
+ * @throws {@link Error}
+ * Throws an error if currencies length does not match balances length, or if a passed in currency is not a {@link PooledToken}
+ */
+function createPairWiseSwapDetails(currencies: Currency[], atomicBalances: bigint[]): SwapDetails[] {
+    if (currencies.length !== atomicBalances.length) {
+        throw new Error(`Cannot combine pair wise swap details; currency count [${
+            currencies.length
+        }] does not match balance count [${
+            atomicBalances.length
+        }]`);
+    }
+
+    const swapDetailsList: SwapDetails[] = [];
+    for(let idx = 0; (idx + 1) < currencies.length; idx++) {
+        const inIdx = idx;
+        const outIdx = idx + 1;
+        const currencyIn = currencies[inIdx];
+        const currencyOut = currencies[outIdx];
+
+        if (!isPooledToken(currencyIn)) {
+            throw new Error(`Cannot combine pair wise swap details; unexpected currency type ${
+                currencyIn.isTypeOf
+            } in pool, skip processing of DexGeneralAssetSwapEvent`);
+        } else if (!isPooledToken(currencyOut)) {
+            throw new Error(`Unexpected currency type ${
+                currencyOut.isTypeOf
+            } in pool, skip processing of DexGeneralAssetSwapEvent`);
+        }
+
+        swapDetailsList.push({
+            from: {
+                currency: currencyIn,
+                atomicAmount: atomicBalances[inIdx]
+            },
+            to: {
+                currency: currencyOut,
+                atomicAmount: atomicBalances[outIdx]
+            }
+        });
+    }
+
+    return swapDetailsList;
+}
+
+export async function dexGeneralAssetSwap(
+    ctx: Ctx,
+    block: SubstrateBlock,
+    item: EventItem,
+    entityBuffer: EntityBuffer
+): Promise<void> {
+    const rawEvent = new DexGeneralAssetSwapEvent(ctx, item.event);
+    let currencies: Currency[] = [];
+    let atomicBalances: bigint[] = [];
+
+    if (rawEvent.isV1021000) {
+        const [, , swapPath, balances] = rawEvent.asV1021000;
+        currencies = swapPath.map(currencyId.encode);
+        atomicBalances = balances;
+    } else {
+        ctx.log.warn("UNKOWN EVENT VERSION: DexGeneral.AssetSwap");
+        return;
+    }
+
+    // we can only use pooled tokens, check we have not other ones
+    for (const currency of currencies) {
+        if (!isPooledToken(currency)) {
+            ctx.log.error(`Unexpected currency type ${currency.isTypeOf} in pool, skip processing of DexGeneralAssetSwapEvent`);
+            return;
+        }
+    }
+
+    let swapDetailsList: SwapDetails[];
+    try {
+        swapDetailsList = createPairWiseSwapDetails(currencies, atomicBalances);
+    } catch (e) {
+        ctx.log.error((e as Error).message);
+        return;
+    }
+
+    // construct and await sequentially, otherwise some operations may try to read values from 
+    // the entity buffer before it has been updated
+    for (const swapDetails of swapDetailsList) {
+        const entity = await updateCumulativeDexVolumesForStandardPool(
+            ctx.store,
+            new Date(block.timestamp),
+            swapDetails,
+            entityBuffer
+        );
+
+        entityBuffer.pushEntity(CumulativeDexTradingVolumePerPool.name, entity);
+    }
+}
+
+export async function dexStableCurrencyExchange(
+    ctx: Ctx,
+    block: SubstrateBlock,
+    item: EventItem,
+    entityBuffer: EntityBuffer
+): Promise<void> {
+    const rawEvent = new DexStableCurrencyExchangeEvent(ctx, item.event);
+    let poolId: number;
+    let inIndex: number;
+    let outIndex: number;
+    let inAmount: bigint;
+    let outAmount: bigint;
+
+    if (rawEvent.isV1021000) {
+        const event = rawEvent.asV1021000;
+        poolId = event.poolId;
+        inIndex = event.inIndex;
+        outIndex = event.outIndex;
+        inAmount = event.inAmount;
+        outAmount = event.outAmount;
+    } else {
+        ctx.log.warn("UNKOWN EVENT VERSION: DexStable.CurrencyExchange");
+        return;
+    }
+
+    const outCurrency = await getStablePoolCurrencyByIndex(ctx, block, poolId, outIndex);
+    const inCurrency = await getStablePoolCurrencyByIndex(ctx, block, poolId, inIndex);
+
+    if (!isPooledToken(inCurrency)) {
+        ctx.log.error(`Unexpected currencyIn type ${inCurrency.isTypeOf}, skip processing of DexGeneralAssetSwapEvent`);
+        return;
+    }
+    if (!isPooledToken(outCurrency)) {
+        ctx.log.error(`Unexpected currencyOut type ${outCurrency.isTypeOf}, skip processing of DexGeneralAssetSwapEvent`);
+        return;
+    }
+
+    const swapDetails: SwapDetails = {
+        from: {
+            currency: inCurrency,
+            atomicAmount: inAmount
+        },
+        to: {
+            currency: outCurrency,
+            atomicAmount: outAmount
+        }
+    };
+
+    const entityPromise = updateCumulativeDexVolumesForStablePool(
+        ctx.store,
+        new Date(block.timestamp),
+        poolId,
+        swapDetails,
+        entityBuffer
+    );
+
+    entityBuffer.pushEntity(
+        CumulativeDexTradingVolumePerPool.name,
+        await entityPromise
+    );
+}

src/mappings/event/index.ts

@@ -1,3 +1,4 @@
+export * from "./dex";
 export * from "./issue";
 export * from "./redeem";
 export * from "./vault";

src/mappings/event/loans.ts

@@ -176,8 +176,6 @@ export async function activatedMarket(
     marketDb.activation = activation;
     await entityBuffer.pushEntity(LoanMarketActivation.name, activation);
     await entityBuffer.pushEntity(LoanMarket.name, marketDb);
-
-    console.log(`Activated ${marketDb.id}`);
 }
 
 export async function borrow(