[CU-35d6wrw] store total volume changes and handle resolver

subsquid/schema.graphql

@@ -157,6 +157,15 @@ type HistoricalLockedValue @entity {
   source: LockedSource!
 }
 
+type HistoricalVolume @entity {
+  id: ID!
+  event: Event! @index
+  amount: BigInt!
+  currency: Currency!
+  timestamp: BigInt!
+  assetId: String!
+}
+
 type PabloTransaction @entity {
   id: ID!
   event: Event! @unique @index

subsquid/src/dbHelper.ts

@@ -12,6 +12,7 @@ import {
   Event,
   EventType,
   HistoricalLockedValue,
+  HistoricalVolume,
   LockedSource,
   PabloPool,
 } from "./model";
@@ -248,6 +249,60 @@ export async function storeHistoricalLockedValue(
   await ctx.store.save(historicalLockedValueSource);
 }
 
+/**
+ * Stores a new HistoricalVolume for the specified quote asset
+ * @param ctx
+ * @param quoteAssetId
+ * @param amount
+ */
+export async function storeHistoricalVolume(
+  ctx: EventHandlerContext<Store>,
+  quoteAssetId: string,
+  amount: bigint
+): Promise<void> {
+  const wsProvider = new WsProvider(chain());
+  const api = await ApiPromise.create({ provider: wsProvider });
+  let assetPrice = 0n;
+
+  try {
+    const oraclePrice = await api.query.oracle.prices(quoteAssetId);
+    console.log(oraclePrice);
+    if (!oraclePrice?.price) {
+      // TODO: handle missing oracle price
+      // NOTE: should we look at the latest known price for this asset?
+      return;
+    }
+    assetPrice = BigInt(oraclePrice.price.toString());
+  } catch (error) {
+    console.error(error);
+    return;
+  }
+
+  // TODO: get decimals for this asset
+  // NOTE: normalize to 12 decimals for historical values?
+
+  const volume = amount * assetPrice;
+
+  const lastVolume = await getLastVolume(ctx, quoteAssetId);
+
+  const event = await ctx.store.get(Event, { where: { id: ctx.event.id } });
+
+  if (!event) {
+    return Promise.reject(new Error("Event not found"));
+  }
+
+  const historicalVolume = new HistoricalVolume({
+    id: randomUUID(),
+    event,
+    amount: lastVolume + volume,
+    currency: Currency.USD,
+    assetId: quoteAssetId,
+    timestamp: BigInt(new Date(ctx.block.timestamp).valueOf()),
+  });
+
+  await ctx.store.save(historicalVolume);
+}
+
 /**
  * Get latest locked value
  */
@@ -258,7 +313,24 @@ export async function getLastLockedValue(
   const lastLockedValue = await ctx.store.find(HistoricalLockedValue, {
     where: { source },
     order: { timestamp: "DESC" },
+    relations: { event: true },
   });
 
   return BigInt(lastLockedValue.length > 0 ? lastLockedValue[0].amount : 0);
 }
+
+/**
+ * Get latest volume
+ */
+export async function getLastVolume(
+  ctx: EventHandlerContext<Store>,
+  assetId: string
+): Promise<bigint> {
+  const lastVolume = await ctx.store.find(HistoricalVolume, {
+    where: { assetId },
+    order: { timestamp: "DESC" },
+    relations: { event: true },
+  });
+
+  return BigInt(lastVolume.length > 0 ? lastVolume[0].amount : 0);
+}

subsquid/src/model/generated/historicalVolume.model.ts

@@ -0,0 +1,30 @@
+import {Entity as Entity_, Column as Column_, PrimaryColumn as PrimaryColumn_, ManyToOne as ManyToOne_, Index as Index_} from "typeorm"
+import * as marshal from "./marshal"
+import {Event} from "./event.model"
+import {Currency} from "./_currency"
+
+@Entity_()
+export class HistoricalVolume {
+  constructor(props?: Partial<HistoricalVolume>) {
+    Object.assign(this, props)
+  }
+
+  @PrimaryColumn_()
+  id!: string
+
+  @Index_()
+  @ManyToOne_(() => Event, {nullable: true})
+  event!: Event
+
+  @Column_("numeric", {transformer: marshal.bigintTransformer, nullable: false})
+  amount!: bigint
+
+  @Column_("varchar", {length: 3, nullable: false})
+  currency!: Currency
+
+  @Column_("numeric", {transformer: marshal.bigintTransformer, nullable: false})
+  timestamp!: bigint
+
+  @Column_("text", {nullable: false})
+  assetId!: string
+}

subsquid/src/model/generated/index.ts

@@ -13,6 +13,7 @@ export * from "./rewardPool.model"
 export * from "./stakingPosition.model"
 export * from "./_lockedSource"
 export * from "./historicalLockedValue.model"
+export * from "./historicalVolume.model"
 export * from "./pabloTransaction.model"
 export * from "./event.model"
 export * from "./_eventType"

subsquid/src/processors/pablo.ts

@@ -14,6 +14,7 @@ import {
   getLatestPoolByPoolId,
   getOrCreate,
   storeHistoricalLockedValue,
+  storeHistoricalVolume,
 } from "../dbHelper";
 import {
   Event,
@@ -118,7 +119,6 @@ export async function processPoolCreatedEvent(
 
     let tx = await ctx.store.get(Event, ctx.event.id);
     if (tx != undefined) {
-      console.log("qwe");
       console.error("Unexpected event in db", tx);
       throw new Error("Unexpected event in db");
     }
@@ -551,6 +551,12 @@ export async function processSwappedEvent(
     await ctx.store.save(quoteAsset);
     await ctx.store.save(eventEntity);
     await ctx.store.save(pabloTransaction);
+
+    await storeHistoricalVolume(
+      ctx,
+      quoteAsset.assetId,
+      swappedEvt.quoteAmount
+    );
   } else {
     throw new Error("Pool not found");
   }
@@ -571,7 +577,6 @@ export async function processPoolDeletedEvent(
   ctx: EventHandlerContext<Store, { event: true }>,
   event: PabloPoolDeletedEvent
 ): Promise<void> {
-  console.debug("processing LiquidityAddedEvent", ctx.event.id);
   const poolDeletedEvent = getPoolDeletedEvent(event);
   const pool = await getLatestPoolByPoolId(ctx.store, poolDeletedEvent.poolId);
   // only set values if the owner was missing, i.e a new pool

subsquid/src/server-extension/resolvers/index.ts

@@ -3,6 +3,7 @@ import { PicassoOverviewStatsResolver } from "./picassoOverviewStats";
 import { PabloOverviewStatsResolver } from "./pabloOverviewStats";
 import { AssetsResolver } from "./assets";
 import { TotalValueLockedResolver } from "./totalValueLocked";
+import { TotalVolume } from "./totalVolume";
 import { StakingRewardsStatsResolver } from "./stakingRewards";
 
 export {
@@ -11,5 +12,6 @@ export {
   PabloOverviewStatsResolver,
   AssetsResolver,
   TotalValueLockedResolver,
+  TotalVolume,
   StakingRewardsStatsResolver,
 };

subsquid/src/server-extension/resolvers/totalVolume.ts

@@ -0,0 +1,84 @@
+import {
+  Arg,
+  Field,
+  InputType,
+  Int,
+  ObjectType,
+  Query,
+  Resolver,
+} from "type-graphql";
+import type { EntityManager } from "typeorm";
+import { IsDateString, Min } from "class-validator";
+import { HistoricalVolume } from "../../model";
+import { getTimelineParams } from "./common";
+
+@ObjectType()
+export class TotalVolume {
+  @Field(() => String, { nullable: false })
+  date!: string;
+
+  @Field(() => BigInt, { nullable: false })
+  totalVolume!: bigint;
+
+  constructor(props: Partial<TotalVolume>) {
+    Object.assign(this, props);
+  }
+}
+
+@InputType()
+export class TotalVolumeInput {
+  @Field(() => Int, { nullable: false })
+  @Min(1)
+  intervalMinutes!: number;
+
+  @Field(() => String, { nullable: true })
+  @IsDateString()
+  dateFrom?: string;
+
+  @Field(() => String, { nullable: true })
+  @IsDateString()
+  dateTo?: string;
+}
+
+@Resolver()
+export class TotalVolumeResolver {
+  constructor(private tx: () => Promise<EntityManager>) {}
+
+  @Query(() => [TotalVolume])
+  async totalVolume(
+    @Arg("params", { validate: true }) input: TotalVolumeInput
+  ): Promise<TotalVolume[]> {
+    const { intervalMinutes, dateFrom, dateTo } = input;
+    const { where, params } = getTimelineParams(
+      intervalMinutes,
+      dateFrom,
+      dateTo
+    );
+
+    const manager = await this.tx();
+
+    const rows: {
+      period: string;
+      total_volume: string;
+    }[] = await manager.getRepository(HistoricalVolume).query(
+      `
+            SELECT
+              round(timestamp / $1) * $1 as period,
+              max(amount) as total_volume
+            FROM historical_volume
+            WHERE ${where.join(" AND ")}
+            GROUP BY period
+            ORDER BY period DESC
+        `,
+      params
+    );
+
+    return rows.map(
+      (row) =>
+        new TotalVolume({
+          date: new Date(parseInt(row.period, 10)).toISOString(),
+          totalVolume: BigInt(row.total_volume),
+        })
+    );
+  }
+}