cardano transfer cde and cursor based pagination

packages/engine/paima-funnel/package.json

@@ -18,6 +18,6 @@
   },
   "dependencies": {
     "assert-never": "^1.2.1",
-    "@dcspark/carp-client": "^2.3.1"
+    "@dcspark/carp-client": "file:../../../tmp-carp-client"
   }
 }

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

@@ -1,8 +1,4 @@
-import type {
-  CdeCardanoPoolDatum,
-  ChainDataExtensionCardanoDelegation,
-  ChainDataExtensionDatum,
-} from '@paima/sm';
+import type { CdeCardanoPoolDatum, ChainDataExtensionCardanoDelegation } 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';
@@ -14,7 +10,7 @@ export default async function getCdeData(
   toAbsoluteSlot: number,
   getBlockNumber: (slot: number) => number,
   absoluteSlotToEpoch: (slot: number) => number
-): Promise<ChainDataExtensionDatum[]> {
+): Promise<CdeCardanoPoolDatum[]> {
   const events = await timeout(
     query(url, Routes.delegationForPool, {
       pools: extension.pools,

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

@@ -0,0 +1,88 @@
+import type { CdeCardanoTransferDatum, ChainDataExtensionCardanoTransfer } from '@paima/sm';
+import { ChainDataExtensionDatumType, DEFAULT_FUNNEL_TIMEOUT, timeout } from '@paima/utils';
+import { Routes, query } from '@dcspark/carp-client/client/src';
+import type { TxAndBlockInfo } from '@dcspark/carp-client/shared/models/TransactionHistory';
+import type { BlockTxPair } from 'tmp-carp-client/shared/models/common';
+
+export const PAGINATION_LIMIT = 1000;
+
+export default async function getCdeData(
+  url: string,
+  extension: ChainDataExtensionCardanoTransfer,
+  fromAbsoluteSlot: number,
+  toAbsoluteSlot: number,
+  getBlockNumber: (slot: number) => number,
+  // if we are in the presync phase, we don't care about the slots since we
+  // don't need to deterministically pair this with the evm blocks, so in this
+  // case we only fetch one page and break.
+  isPresync: boolean,
+  untilBlock: string,
+  // only should be used during the presync phase, to be able to resume from the
+  // previous point
+  fromTx: BlockTxPair | undefined
+): Promise<CdeCardanoTransferDatum[]> {
+  let result = [] as CdeCardanoTransferDatum[];
+
+  while (true) {
+    const event = await timeout(
+      query(url, Routes.transactionHistory, {
+        // TODO: maybe it should be Output
+        addresses: [extension.credential],
+        slotLimits: {
+          from: fromAbsoluteSlot,
+          to: toAbsoluteSlot,
+        },
+        limit: PAGINATION_LIMIT,
+        untilBlock,
+        after: fromTx,
+      }),
+      DEFAULT_FUNNEL_TIMEOUT
+    );
+
+    const transactions = event.transactions;
+
+    if (transactions.length > 0) {
+      const last = transactions[transactions.length - 1];
+
+      fromTx = {
+        tx: last.transaction.hash,
+        block: last.block.hash,
+      };
+    }
+
+    transactions
+      .map(e => eventToCdeDatum(e, extension, getBlockNumber(e.block.slot)))
+      .forEach(element => {
+        result.push(element);
+      });
+
+    if (transactions.length === 0 || isPresync) {
+      break;
+    }
+  }
+
+  return result;
+}
+
+function eventToCdeDatum(
+  event: TxAndBlockInfo,
+  extension: ChainDataExtensionCardanoTransfer,
+  blockNumber: number
+): CdeCardanoTransferDatum {
+  const cursor: BlockTxPair = {
+    block: event.block.hash,
+    tx: event.transaction.hash,
+  };
+
+  return {
+    cdeId: extension.cdeId,
+    cdeDatumType: ChainDataExtensionDatumType.CardanoTransfer,
+    blockNumber,
+    payload: {
+      rawTx: event.transaction.payload,
+      txId: event.transaction.hash,
+    },
+    scheduledPrefix: extension.scheduledPrefix,
+    paginationCursor: { cursor: JSON.stringify(cursor), finished: false },
+  };
+}

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

@@ -59,6 +59,10 @@ async function getSpecificCdeData(
       // this is used by the block funnel, which can't get information for this
       // extension
       return [];
+    case ChainDataExtensionType.CardanoTransfer:
+      // 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/FunnelCache.ts

@@ -77,14 +77,26 @@ export type CarpFunnelCacheEntryState = {
   startingSlot: number;
   lastPoint: { blockHeight: number; timestamp: number } | undefined;
   epoch: number | undefined;
+  cursors:
+    | {
+        [cdeId: number]:
+          | { kind: 'paginationCursor'; cursor: string; finished: boolean }
+          | { kind: 'slot'; slot: number; finished: boolean };
+      }
+    | 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, epoch: this.state?.epoch };
+    this.state = {
+      startingSlot,
+      lastPoint: this.state?.lastPoint,
+      epoch: this.state?.epoch,
+      cursors: this.state?.cursors,
+    };
   }
 
   public updateLastPoint(blockHeight: number, timestamp: number): void {
@@ -99,6 +111,21 @@ export class CarpFunnelCacheEntry implements FunnelCacheEntry {
     }
   }
 
+  public updateCursor(
+    cdeId: number,
+    presyncCursor:
+      | { kind: 'paginationCursor'; cursor: string; finished: boolean }
+      | { kind: 'slot'; slot: number; finished: boolean }
+  ): void {
+    if (this.state) {
+      if (!this.state.cursors) {
+        this.state.cursors = {};
+      }
+
+      this.state.cursors[cdeId] = presyncCursor;
+    }
+  }
+
   public initialized(): boolean {
     return !!this.state;
   }

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

@@ -118,7 +118,7 @@ export class BlockFunnel extends BaseFunnel implements ChainFunnel {
         ),
         getUngroupedCdeData(this.sharedData.web3, this.sharedData.extensions, fromBlock, toBlock),
       ]);
-      const cdeData = groupCdeData(Network.CARDANO, fromBlock, toBlock, ungroupedCdeData);
+      const cdeData = groupCdeData(fromBlock, toBlock, ungroupedCdeData);
       return composeChainData(baseChainData, cdeData);
     } catch (err) {
       doLog(`[funnel] at ${fromBlock}-${toBlock} caught ${err}`);
@@ -127,14 +127,8 @@ export class BlockFunnel extends BaseFunnel implements ChainFunnel {
   };
 
   public override async readPresyncData(
-    args: ReadPresyncDataFrom
+    arg: 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;
 
@@ -155,7 +149,7 @@ export class BlockFunnel extends BaseFunnel implements ChainFunnel {
         fromBlock,
         toBlock
       );
-      return { [Network.EVM]: groupCdeData(Network.EVM, fromBlock, toBlock, ungroupedCdeData) };
+      return { [Network.EVM]: groupCdeData(fromBlock, toBlock, ungroupedCdeData) };
     } catch (err) {
       doLog(`[paima-funnel::readPresyncData] Exception occurred while reading blocks: ${err}`);
       throw err;