updating holderCount stat for v3 pools (#1125)

* updating holderCount stat for v3 pools

* include swap lifetime values

* fixes

* renaming

* aggregate lifetime values from snapshots

* sync lifetime values based on snapshots

.changeset/rich-horses-rush.md

@@ -0,0 +1,5 @@
+---
+'backend': patch
+---
+
+updating holderCount stat for v3 pools

modules/actions/pool/update-liftetime-values.ts

@@ -0,0 +1,113 @@
+import { Chain, PrismaPoolType } from '@prisma/client';
+import { prisma } from '../../../prisma/prisma-client';
+
+// This is a helper for V3 and CowAmm pools only. V2 is already handled and concidered legacy.
+export const updateLifetimeValues = async (chain: Chain, protocolVersion?: number, type?: PrismaPoolType) => {
+    const holders = await getHoldersCount(chain, protocolVersion, type);
+    const lifetime = await getSwapLifetimeValues(chain, protocolVersion, type);
+
+    // Merge all keys into an unique list
+    const allKeys = [...Object.keys(holders), ...Object.keys(lifetime)].reduce((acc, key) => {
+        if (!acc.includes(key)) acc.push(key);
+        return acc;
+    }, [] as string[]);
+
+    const data = allKeys.map((key) => {
+        const [poolId, chain] = key.split('-');
+        const holdersCount = holders[key] || 0;
+        const lifetimeSwapFees = lifetime[key]?.lifetimeSwapFees || 0;
+        const lifetimeVolume = lifetime[key]?.lifetimeVolume || 0;
+
+        return {
+            where: {
+                poolId_chain: {
+                    poolId,
+                    chain: chain as Chain,
+                },
+            },
+            data: {
+                holdersCount,
+                lifetimeSwapFees,
+                lifetimeVolume,
+            },
+        };
+    });
+
+    const updates = data.map((record) => {
+        const { where, data } = record;
+
+        return prisma.prismaPoolDynamicData.update({
+            where,
+            data,
+        });
+    });
+
+    return prisma.$transaction(updates);
+};
+
+const getHoldersCount = async (chain: Chain, protocolVersion?: number, type?: PrismaPoolType) => {
+    const holders = await prisma.prismaUserWalletBalance.groupBy({
+        by: ['poolId', 'chain'],
+        _count: { userAddress: true },
+        where: {
+            chain,
+            pool: {
+                protocolVersion,
+                type,
+            },
+        },
+    });
+    // This is overfetching, because of V2 pools
+    const stakers = await prisma.prismaUserStakedBalance.groupBy({
+        by: ['poolId', 'chain'],
+        _count: { userAddress: true },
+        where: {
+            chain,
+            pool: {
+                protocolVersion,
+                type,
+            },
+        },
+    });
+
+    // Merge the two arrays
+    const pools = [...holders, ...stakers].reduce((acc, item) => {
+        const { poolId, chain } = item;
+        if (!poolId) return acc;
+        acc[`${poolId}-${chain}`] ||= 0;
+        acc[`${poolId}-${chain}`] += item._count.userAddress;
+        return acc;
+    }, {} as Record<string, number>);
+
+    return pools;
+};
+
+const getSwapLifetimeValues = async (chain: Chain, protocolVersion?: number, type?: PrismaPoolType) => {
+    // Get latest snapshots for each pool
+    const swapLifetimeValues = await prisma.prismaPoolSnapshot.groupBy({
+        by: ['poolId', 'chain'],
+        _sum: {
+            fees24h: true,
+            volume24h: true,
+        },
+        where: {
+            chain,
+            protocolVersion,
+            pool: {
+                type,
+            },
+        },
+    });
+
+    const lifetimeValues = swapLifetimeValues.reduce((acc, { poolId, chain, _sum }) => {
+        const key = `${poolId}-${chain}`;
+        if (!acc[key]) {
+            acc[key] = { lifetimeSwapFees: 0, lifetimeVolume: 0 };
+        }
+        acc[key].lifetimeSwapFees += _sum.fees24h || 0;
+        acc[key].lifetimeVolume += _sum.volume24h || 0;
+        return acc;
+    }, {} as Record<string, { lifetimeSwapFees: number; lifetimeVolume: number }>);
+
+    return lifetimeValues;
+};

modules/controllers/cow-amm-controller.ts

@@ -17,6 +17,7 @@ import { updateVolumeAndFees } from '../actions/pool/update-volume-and-fees';
 import moment from 'moment';
 import { upsertBptBalances } from '../actions/cow-amm/upsert-bpt-balances';
 import { getLastSyncedBlock, upsertLastSyncedBlock } from '../actions/pool/last-synced-block';
+import { updateLifetimeValues } from '../actions/pool/update-liftetime-values';
 
 export function CowAmmController(tracer?: any) {
     const getSubgraphClient = (chain: Chain) => {
@@ -125,6 +126,8 @@ export function CowAmmController(tracer?: any) {
         async syncSnapshots(chain: Chain) {
             const subgraphClient = getSubgraphClient(chain);
             const timestamp = await syncSnapshots(subgraphClient, chain);
+            // update lifetime values based on snapshots
+            await updateLifetimeValues(chain, undefined, 'COW_AMM');
             return timestamp;
         },
         async syncAllSnapshots(chain: Chain) {

modules/controllers/snapshots-controller.ts

@@ -6,6 +6,7 @@ import { PoolSnapshotService } from '../actions/snapshots/pool-snapshot-service'
 import { chainIdToChain } from '../network/chain-id-to-chain';
 import { getVaultSubgraphClient } from '../sources/subgraphs';
 import { getV2SubgraphClient } from '../subgraphs/balancer-subgraph';
+import { updateLifetimeValues } from '../actions/pool/update-liftetime-values';
 
 /**
  * Controller responsible for configuring and executing ETL actions.
@@ -78,6 +79,8 @@ export function SnapshotsController(tracer?: any) {
 
             const vaultSubgraphClient = getVaultSubgraphClient(balancerV3);
             const entries = await syncSnapshotsV3(vaultSubgraphClient, chain);
+            // update lifetime values based on snapshots
+            await updateLifetimeValues(chain, 3);
             return entries;
         },
         async fillMissingSnapshotsV2(chain: Chain) {

tasks/index.ts

@@ -45,6 +45,8 @@ async function run(job: string = process.argv[2], chainId: string = process.argv
         return PoolController().reloadPoolsV3(chain);
     } else if (job === 'sync-pools-v3') {
         return PoolController().syncChangedPoolsV3(chain);
+    } else if (job === 'update-liquidity-for-active-pools') {
+        return PoolController().updateLiquidityValuesForActivePools(chain);
     } else if (job === 'sync-staking') {
         return StakingController().syncStaking(chain);
     } else if (job === 'sync-join-exits-v3') {