fix(subscriber): reduce number of liquidity proxy state generations

Author: Ivan011001

src/dex/fluid-dex/fluid-dex-events.test.ts

@@ -40,7 +40,7 @@ describe('FluidDex EventPool Mainnet', function () {
     const eventsToTest: Record<Address, EventMappings> = {
       '0x52aa899454998be5b000ad077a46bbe360f4e497': {
         LogOperate: [
-          21190399, 21190405, 21190420, 21190452, 21190454, 21190465, 21190506,
+          24619596, 24619595, 24619593, 24619592, 24619591, 24619590, 24619589,
         ],
       },
     };
@@ -86,7 +86,7 @@ describe('FluidDex EventPool Mainnet', function () {
 
     const eventsToTest: Record<Address, EventMappings> = {
       '0x91716C4EDA1Fb55e84Bf8b4c7085f84285c19085': {
-        LogDexDeployed: [21199929],
+        LogDexDeployed: [24611892],
       },
     };
 

src/dex/fluid-dex/fluid-dex-liquidity-proxy.ts

@@ -15,6 +15,8 @@ import {
 import { Address } from '../../types';
 import { Contract } from 'ethers';
 
+const NON_POOL_THROTTLE_MS = 60 * 1000; // 1 minute
+
 export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiquidityProxyState> {
   handlers: {
     [event: string]: (
@@ -34,6 +36,10 @@ export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiqu
 
   resolverContract: Contract;
 
+  private poolAddresses: Set<string> = new Set();
+  private lastPoolEventBlockNumber: number = 0;
+  private lastNonPoolEventTimestamp: number = 0;
+
   constructor(
     readonly parentName: string,
     readonly commonAddresses: CommonAddresses,
@@ -57,9 +63,10 @@ export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiqu
     this.handlers['LogOperate'] = this.handleOperate.bind(this);
   }
 
-  /**
-   * Handle a trade rate change on the pool.
-   */
+  setPoolAddresses(addresses: string[]) {
+    this.poolAddresses = new Set(addresses.map(a => a.toLowerCase()));
+  }
+
   async handleOperate(
     event: any,
     state: DeepReadonly<FluidDexLiquidityProxyState>,
@@ -82,6 +89,28 @@ export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiqu
     log: Readonly<Log>,
   ): Promise<DeepReadonly<FluidDexLiquidityProxyState> | null> {
     try {
+      if (log.topics.length >= 2) {
+        const userAddress = ('0x' + log.topics[1].slice(26)).toLowerCase();
+
+        if (this.poolAddresses.has(userAddress)) {
+          // Check if state was already updated for this block
+          if (log.blockNumber === this.lastPoolEventBlockNumber) {
+            return null;
+          }
+
+          this.lastPoolEventBlockNumber = log.blockNumber;
+        } else {
+          const now = Date.now();
+
+          // Throttle non-pool events to avoid generating state too often
+          if (now - this.lastNonPoolEventTimestamp < NON_POOL_THROTTLE_MS) {
+            return null;
+          }
+
+          this.lastNonPoolEventTimestamp = now;
+        }
+      }
+
       const event = this.logDecoder(log);
       if (event.name in this.handlers) {
         return await this.handlers[event.name](event, state, log);

src/dex/fluid-dex/fluid-dex.ts

@@ -124,6 +124,7 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
       }),
     );
 
+    this.liquidityProxy.setPoolAddresses(this.pools.map(p => p.address));
     await this.liquidityProxy.initialize(blockNumber);
   }
 
@@ -135,6 +136,7 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
     poolsFromFactory: readonly PoolWithDecimals[],
   ) {
     this.pools = this.generateFluidDexPoolsFromPoolsFactory(poolsFromFactory);
+    this.liquidityProxy.setPoolAddresses(this.pools.map(p => p.address));
     this.logger.info(`${this.dexKey}: pools list was updated ...`);
   }