add carp funnel with stake delegation pool tracking (#246)

* add carp funnel with stake delegation pool tracking

* handle presync in carp funnel

* expose cde utils get cardano delegation function

* use local (tmp) carp client lib in the funnel

just to test until the up to date version is published on npm

* add delegation check in batcher's address validator

* add stopSlot setting to stop fetching

* improve logging and make presync concurrent

* address review comments

* fix markCardanoCdeSlotProcessed query

* presync checkpointing for cardano cde

* drop cde_tracking_cardano in down.sql

* address new review comments

* replace carp local client with 2.3.0 from npm

* cache the startingSlot and the previous range upper end

just to reduce the number of rpc calls

* remove old TODO

the cast is probably clearer than the alternatives

* replace map-filter with reduce for getEarlistSlot/Blockheight

* add CARDANO_CONFIRMATION_DEPTH config

* refactor getEarliestStart* functions

* default case of min should be Infinity

* document timestampToAbsoluteSlot and fix missing slot offset

* reinitialize the cache entry in case of error

packages/batcher/address-validator/package.json

@@ -12,7 +12,8 @@
   "dependencies": {
     "pg": "^8.7.3",
     "web3": "1.10.0",
-    "assert-never": "^1.2.1"
+    "assert-never": "^1.2.1",
+    "@dcspark/carp-client": "^2.3.0"
   },
   "devDependencies": {
     "@types/pg": "^8.6.5"

packages/batcher/address-validator/src/index.ts

@@ -17,6 +17,8 @@ import { CryptoManager } from '@paima/crypto';
 import { createMessageForBatcher } from '@paima/concise';
 import { initWeb3, AddressType, getReadNamespaces } from '@paima/utils';
 import assertNever from 'assert-never';
+import { query, getErrorResponse } from '@dcspark/carp-client/client/src/index';
+import { Routes } from '@dcspark/carp-client/shared/routes';
 
 class PaimaAddressValidator {
   private web3: Web3 | undefined;
@@ -59,6 +61,13 @@ class PaimaAddressValidator {
       return GenericRejectionCode.ADDRESS_NOT_ALLOWED;
     }
 
+    // If a list of allowed cardano pools is provided, check that this user is delegating to that address
+    const isDelegatingToAllowedPools = await this.isDelegatingToAllowedPools(input.userAddress);
+    if (!isDelegatingToAllowedPools) {
+      console.log('[address-validator] Address not allowed!');
+      return GenericRejectionCode.ADDRESS_NOT_ALLOWED;
+    }
+
     // All tests passed:
     return 0;
   };
@@ -208,6 +217,31 @@ class PaimaAddressValidator {
   ): boolean {
     return inputsMinute < ENV.MAX_USER_INPUTS_PER_MINUTE && inputsDay < ENV.MAX_USER_INPUTS_PER_DAY;
   }
+
+  private isDelegatingToAllowedPools = async (userAddress: string): Promise<boolean> => {
+    const pools = ENV.BATCHER_CARDANO_ENABLED_POOLS;
+
+    if (!pools) {
+      return true;
+    }
+
+    if (!ENV.CARP_URL) {
+      throw new Error(`[address-validator] missing CARP_URL setting`);
+    }
+
+    const latest = await query(ENV.CARP_URL, Routes.blockLatest, {
+      offset: 0,
+    });
+
+    const delegatingTo = await query(ENV.CARP_URL, Routes.delegationForAddress, {
+      address: userAddress,
+      until: {
+        absoluteSlot: latest.block.slot,
+      },
+    });
+
+    return !!pools.find(pool => delegatingTo.pool === pool);
+  };
 }
 
 export default PaimaAddressValidator;

packages/batcher/utils/src/config.ts

@@ -84,6 +84,14 @@ export class ENV {
     return parseInt(process.env.MAX_USER_INPUTS_PER_DAY || '0', 10);
   }
 
+  static get CARP_URL(): string | undefined {
+    return process.env.BATCHER_CARP_URL;
+  }
+
+  static get BATCHER_CARDANO_ENABLED_POOLS(): string[] | undefined {
+    return process.env.BATCHER_CARDANO_ENABLED_POOLS?.split(',');
+  }
+
   // NOTE: this variable is not currently used, with DEFAULT_VALIDATION_ACTIVE determining the type.
   static get GAME_INPUT_VALIDATION_TYPE_NAME(): string {
     return process.env.GAME_INPUT_VALIDATION_TYPE_NAME || '';

packages/engine/paima-funnel/package.json

@@ -17,6 +17,7 @@
     "typescript": "^5.3.2"
   },
   "dependencies": {
-    "assert-never": "^1.2.1"
+    "assert-never": "^1.2.1",
+    "@dcspark/carp-client": "^2.3.0"
   }
 }

packages/engine/paima-funnel/src/cde/cardanoPool.ts

@@ -0,0 +1,43 @@
+import type {
+  CdeCardanoPoolDatum,
+  ChainDataExtensionCardanoDelegation,
+  ChainDataExtensionDatum,
+} from '@paima/sm';
+import { ChainDataExtensionDatumType, DEFAULT_FUNNEL_TIMEOUT, timeout } from '@paima/utils';
+import { Routes, query } from '@dcspark/carp-client/client/src';
+import type { DelegationForPoolResponse } from '@dcspark/carp-client/shared/models/DelegationForPool';
+
+export default async function getCdeData(
+  url: string,
+  extension: ChainDataExtensionCardanoDelegation,
+  fromAbsoluteSlot: number,
+  toAbsoluteSlot: number,
+  getBlockNumber: (slot: number) => number
+): Promise<ChainDataExtensionDatum[]> {
+  const events = await timeout(
+    query(url, Routes.delegationForPool, {
+      pools: extension.pools,
+      range: { minSlot: fromAbsoluteSlot, maxSlot: toAbsoluteSlot },
+    }),
+    DEFAULT_FUNNEL_TIMEOUT
+  );
+
+  return events.map(e => eventToCdeDatum(e, extension, getBlockNumber(e.slot)));
+}
+
+function eventToCdeDatum(
+  event: DelegationForPoolResponse[0],
+  extension: ChainDataExtensionCardanoDelegation,
+  blockNumber: number
+): CdeCardanoPoolDatum {
+  return {
+    cdeId: extension.cdeId,
+    cdeDatumType: ChainDataExtensionDatumType.CardanoPool,
+    blockNumber,
+    payload: {
+      address: event.credential,
+      pool: event.pool || undefined,
+    },
+    scheduledPrefix: extension.scheduledPrefix,
+  };
+}

packages/engine/paima-funnel/src/cde/reading.ts

@@ -21,13 +21,15 @@ export async function getUngroupedCdeData(
     return extensions.map(_ => []);
   }
   const allData = await Promise.all(
-    extensions.map(extension => getSpecificCdeData(extension, fromBlock, toBlock))
+    extensions.map(extension =>
+      'startBlockHeight' in extension ? getSpecificCdeData(extension, fromBlock, toBlock) : []
+    )
   );
   return allData;
 }
 
 async function getSpecificCdeData(
-  extension: ChainDataExtension,
+  extension: ChainDataExtension & { startBlockHeight: number },
   fromBlock: number,
   toBlock: number
 ): Promise<ChainDataExtensionDatum[]> {
@@ -49,6 +51,10 @@ async function getSpecificCdeData(
       return await getCdeGenericData(extension, fromBlock, toBlock);
     case ChainDataExtensionType.ERC6551Registry:
       return await getCdeErc6551RegistryData(extension, fromBlock, toBlock);
+    case ChainDataExtensionType.CardanoPool:
+      // this is used by the block funnel, which can't get information for this
+      // extension
+      return [];
     default:
       assertNever(extension);
   }

packages/engine/paima-funnel/src/funnels/BaseFunnel.ts

@@ -1,8 +1,9 @@
-import type { ChainFunnel } from '@paima/runtime';
+import type { ChainFunnel, ReadPresyncDataFrom } from '@paima/runtime';
 import type { ChainData, ChainDataExtension, PresyncChainData } from '@paima/sm';
 import type { PaimaL2Contract, Web3 } from '@paima/utils';
 import type { FunnelCacheManager } from './FunnelCache.js';
 import type { PoolClient } from 'pg';
+import { FUNNEL_PRESYNC_FINISHED } from '@paima/utils';
 
 export type FunnelSharedData = {
   readonly web3: Web3;
@@ -30,8 +31,10 @@ export class BaseFunnel implements ChainFunnel {
     return [];
   }
 
-  public async readPresyncData(_fromBlock: number, _toBlock: number): Promise<PresyncChainData[]> {
-    return [];
+  public async readPresyncData(
+    _args: ReadPresyncDataFrom
+  ): Promise<{ [network: number]: PresyncChainData[] | typeof FUNNEL_PRESYNC_FINISHED }> {
+    return {};
   }
 
   public getDbTx(): PoolClient {

packages/engine/paima-funnel/src/funnels/FunnelCache.ts

@@ -11,6 +11,7 @@ export interface FunnelCacheEntry {
 export type CacheMapType = {
   [QueuedBlockCacheEntry.SYMBOL]?: QueuedBlockCacheEntry;
   [RpcCacheEntry.SYMBOL]?: RpcCacheEntry;
+  [CarpFunnelCacheEntry.SYMBOL]?: CarpFunnelCacheEntry;
 };
 export class FunnelCacheManager {
   public cacheEntries: CacheMapType = {};
@@ -71,3 +72,38 @@ export class RpcCacheEntry implements FunnelCacheEntry {
     this.rpcResult = {};
   };
 }
+
+export type CarpFunnelCacheEntryState = {
+  startingSlot: number;
+  lastPoint: { blockHeight: number; timestamp: number } | undefined;
+};
+
+export class CarpFunnelCacheEntry implements FunnelCacheEntry {
+  private state: CarpFunnelCacheEntryState | null = null;
+  public static readonly SYMBOL = Symbol('CarpFunnelStartingSlot');
+
+  public updateStartingSlot(startingSlot: number): void {
+    this.state = { startingSlot, lastPoint: this.state?.lastPoint };
+  }
+
+  public updateLastPoint(blockHeight: number, timestamp: number): void {
+    if (this.state) {
+      this.state.lastPoint = { blockHeight, timestamp };
+    }
+  }
+
+  public initialized(): boolean {
+    return !!this.state;
+  }
+
+  public getState(): Readonly<CarpFunnelCacheEntryState> {
+    if (!this.state) {
+      throw new Error('[carp-funnel] Uninitialized cache entry');
+    }
+    return this.state;
+  }
+
+  clear: FunnelCacheEntry['clear'] = () => {
+    this.state = null;
+  };
+}

packages/engine/paima-funnel/src/funnels/block/funnel.ts

@@ -1,5 +1,5 @@
-import { ENV, doLog, timeout } from '@paima/utils';
-import type { ChainFunnel } from '@paima/runtime';
+import { ENV, Network, doLog, timeout } from '@paima/utils';
+import type { ChainFunnel, ReadPresyncDataFrom } from '@paima/runtime';
 import type { ChainData, PresyncChainData } from '@paima/sm';
 import { getBaseChainDataMulti, getBaseChainDataSingle } from '../../reading.js';
 import { getUngroupedCdeData } from '../../cde/reading.js';
@@ -8,6 +8,7 @@ import { BaseFunnel } from '../BaseFunnel.js';
 import type { FunnelSharedData } from '../BaseFunnel.js';
 import { RpcCacheEntry, RpcRequestState } from '../FunnelCache.js';
 import type { PoolClient } from 'pg';
+import { FUNNEL_PRESYNC_FINISHED } from '@paima/utils';
 
 const GET_BLOCK_NUMBER_TIMEOUT = 5000;
 
@@ -111,7 +112,7 @@ export class BlockFunnel extends BaseFunnel implements ChainFunnel {
         ),
         getUngroupedCdeData(this.sharedData.web3, this.sharedData.extensions, fromBlock, toBlock),
       ]);
-      const cdeData = groupCdeData(fromBlock, toBlock, ungroupedCdeData);
+      const cdeData = groupCdeData(Network.CARDANO, fromBlock, toBlock, ungroupedCdeData);
       return composeChainData(baseChainData, cdeData);
     } catch (err) {
       doLog(`[funnel] at ${fromBlock}-${toBlock} caught ${err}`);
@@ -120,9 +121,21 @@ export class BlockFunnel extends BaseFunnel implements ChainFunnel {
   };
 
   public override async readPresyncData(
-    fromBlock: number,
-    toBlock: number
-  ): Promise<PresyncChainData[]> {
+    args: ReadPresyncDataFrom
+  ): Promise<{ [network: number]: PresyncChainData[] | typeof FUNNEL_PRESYNC_FINISHED }> {
+    let arg = args.find(arg => arg.network == Network.EVM);
+
+    if (!arg) {
+      return [];
+    }
+
+    let fromBlock = arg.from;
+    let toBlock = arg.to;
+
+    if (fromBlock >= ENV.START_BLOCKHEIGHT) {
+      return { [Network.EVM]: FUNNEL_PRESYNC_FINISHED };
+    }
+
     try {
       toBlock = Math.min(toBlock, ENV.START_BLOCKHEIGHT);
       fromBlock = Math.max(fromBlock, 0);
@@ -136,7 +149,7 @@ export class BlockFunnel extends BaseFunnel implements ChainFunnel {
         fromBlock,
         toBlock
       );
-      return groupCdeData(fromBlock, toBlock, ungroupedCdeData);
+      return { [Network.EVM]: groupCdeData(Network.EVM, fromBlock, toBlock, ungroupedCdeData) };
     } catch (err) {
       doLog(`[paima-funnel::readPresyncData] Exception occurred while reading blocks: ${err}`);
       throw err;