[Blockchain Watcher] (APTOS) Integrate aptos source and redeemed events

blockchain-watcher/config/default.json

@@ -3,7 +3,7 @@
  "port": 9090,
  "logLevel": "debug",
  "dryRun": true,
- "enabledPlatforms": ["solana", "evm", "sui"],
+ "enabledPlatforms": ["solana", "evm", "sui", "aptos"],
  "sns": {
  "topicArn": "arn:aws:sns:us-east-1:000000000000:localstack-topic.fifo",
  "region": "us-east-1",
@@ -131,6 +131,13 @@
  "rpcs": ["https://fullnode.testnet.sui.io:443"],
  "timeout": 10000
  },
+ "aptos": {
+ "name": "aptos",
+ "network": "testnet",
+ "chainId": 22,
+ "rpcs": ["https://fullnode.devnet.aptoslabs.com/v1"],
+ "timeout": 10000
+ },
  "arbitrum": {
  "name": "arbitrum",
  "network": "goerli",

blockchain-watcher/config/mainnet.json

@@ -118,6 +118,10 @@
  "sui": {
  "network": "mainnet",
  "rpcs": ["https://fullnode.mainnet.sui.io:443", "http://m-sui-01.nodes.stable.io:8546"]
+ },
+ "aptos": {
+ "network": "mainnet",
+ "rpcs": ["https://fullnode.mainnet.aptoslabs.com"]
  }
  }
 }

blockchain-watcher/src/domain/actions/aptos/GetAptosSequences.ts

@@ -0,0 +1,119 @@
+import { Block, Range, TransactionFilter } from "./PollAptos";
+import { AptosTransaction } from "../../entities/aptos";
+import { AptosRepository } from "../../repositories";
+import { createBatches } from "../../../infrastructure/repositories/common/utils";
+import winston from "winston";
+
+export class GetAptosSequences {
+ protected readonly logger: winston.Logger;
+ private readonly repo: AptosRepository;
+
+ private previousBlock?: bigint;
+ private lastBlock?: bigint;
+
+ constructor(repo: AptosRepository) {
+ this.logger = winston.child({ module: "GetAptosSequences" });
+ this.repo = repo;
+ }
+
+ async execute(range: Block | undefined, opts: GetAptosOpts): Promise<AptosTransaction[]> {
+ let populatedTransactions: AptosTransaction[] = [];
+
+ this.logger.info(
+ `[aptos][exec] Processing blocks [previousBlock: ${opts.previousBlock} - lastBlock: ${opts.lastBlock}]`
+ );
+
+ const batches = createBatches(range);
+
+ for (const toBatch of batches) {
+ const fromBatch = this.lastBlock ? Number(this.lastBlock) : range?.fromBlock;
+
+ const events = await this.repo.getSequenceNumber(
+ {
+ fromBlock: fromBatch,
+ toBlock: toBatch,
+ },
+ opts.filter
+ );
+
+ // Update lastBlock with the new lastBlock
+ this.lastBlock = BigInt(events[events.length - 1].sequence_number);
+
+ if (opts.previousBlock == this.lastBlock) {
+ return [];
+ }
+
+ // Update previousBlock with opts lastBlock
+ this.previousBlock = opts.lastBlock;
+
+ const transactions = await this.repo.getTransactionsByVersionForSourceEvent(
+ events,
+ opts.filter
+ );
+
+ transactions.forEach((tx) => {
+ populatedTransactions.push(tx);
+ });
+ }
+
+ this.logger.info(
+ `[aptos][exec] Got ${populatedTransactions?.length} transactions to process for [addresses:${opts.addresses}][block: ${range?.fromBlock}]`
+ );
+ return populatedTransactions;
+ }
+
+ getBlockRange(
+ cfgBlockBarchSize: number,
+ cfgFromBlock: bigint | undefined,
+ savedPreviousSequence: bigint | undefined,
+ savedLastBlock: bigint | undefined
+ ): Block | undefined {
+ // If [set up a from block for cfg], return the fromBlock and toBlock equal the block batch size
+ if (cfgFromBlock) {
+ return {
+ fromBlock: Number(cfgFromBlock),
+ toBlock: cfgBlockBarchSize,
+ };
+ }
+
+ if (savedPreviousSequence && savedLastBlock) {
+ // If process the [same block], return the same lastBlock and toBlock equal the block batch size
+ if (savedPreviousSequence === savedLastBlock) {
+ return {
+ fromBlock: Number(savedLastBlock),
+ toBlock: cfgBlockBarchSize,
+ };
+ } else {
+ // If process [different sequences], return the difference between the lastBlock and the previousBlock plus 1
+ return {
+ fromBlock: Number(savedLastBlock),
+ toBlock: Number(savedLastBlock) - Number(savedPreviousSequence) + 1,
+ };
+ }
+ }
+
+ if (savedLastBlock) {
+ // If there is [no previous block], return the lastBlock and toBlock equal the block batch size
+ if (!cfgFromBlock || BigInt(cfgFromBlock) < savedLastBlock) {
+ return {
+ fromBlock: Number(savedLastBlock),
+ toBlock: cfgBlockBarchSize,
+ };
+ }
+ }
+ }
+
+ getUpdatedRange(): Range {
+ return {
+ previousBlock: this.previousBlock,
+ lastBlock: this.lastBlock,
+ };
+ }
+}
+
+export type GetAptosOpts = {
+ addresses: string[];
+ filter: TransactionFilter;
+ previousBlock?: bigint | undefined;
+ lastBlock?: bigint | undefined;
+};

blockchain-watcher/src/domain/actions/aptos/GetAptosTransactions.ts

@@ -0,0 +1,114 @@
+import { Block, Range, TransactionFilter } from "./PollAptos";
+import { AptosTransaction } from "../../entities/aptos";
+import { AptosRepository } from "../../repositories";
+import { createBatches } from "../../../infrastructure/repositories/common/utils";
+import winston from "winston";
+
+export class GetAptosTransactions {
+ protected readonly logger: winston.Logger;
+ private readonly repo: AptosRepository;
+
+ private previousBlock?: bigint;
+ private lastBlock?: bigint;
+
+ constructor(repo: AptosRepository) {
+ this.logger = winston.child({ module: "GetAptosTransactions" });
+ this.repo = repo;
+ }
+
+ async execute(range: Block | undefined, opts: GetAptosOpts): Promise<AptosTransaction[]> {
+ let populatedTransactions: AptosTransaction[] = [];
+
+ this.logger.info(
+ `[aptos][exec] Processing blocks [previousBlock: ${opts.previousBlock} - latestBlock: ${opts.lastBlock}]`
+ );
+
+ const batches = createBatches(range);
+
+ for (const toBatch of batches) {
+ const fromBatch = this.lastBlock ? Number(this.lastBlock) : range?.fromBlock;
+
+ const transaction = await this.repo.getTransactions({
+ fromBlock: fromBatch,
+ toBlock: toBatch,
+ });
+
+ // Only process transactions to the contract address configured
+ const transactionsByAddressConfigured = transaction.filter((transaction) =>
+ opts.filter?.type?.includes(String(transaction.payload?.function).toLowerCase())
+ );
+
+ // Update lastBlock with the new lastBlock
+ this.lastBlock = BigInt(transaction[transaction.length - 1].version);
+
+ if (opts.previousBlock == this.lastBlock) {
+ return [];
+ }
+
+ // Update previousBlock with opts lastBlock
+ this.previousBlock = opts.lastBlock;
+
+ if (transactionsByAddressConfigured.length > 0) {
+ const transactions = await this.repo.getTransactionsByVersionForRedeemedEvent(
+ transactionsByAddressConfigured,
+ opts.filter
+ );
+
+ transactions.forEach((tx) => {
+ populatedTransactions.push(tx);
+ });
+ }
+ }
+
+ return populatedTransactions;
+ }
+
+ getBlockRange(
+ cfgBlockBarchSize: number,
+ cfgFromBlock: bigint | undefined,
+ savedPreviousBlock: bigint | undefined,
+ savedLastBlock: bigint | undefined
+ ): Block | undefined {
+ // If [set up a from block for cfg], return the fromBlock and toBlock equal the block batch size
+ if (cfgFromBlock) {
+ return {
+ fromBlock: Number(cfgFromBlock),
+ toBlock: cfgBlockBarchSize,
+ };
+ }
+
+ if (savedPreviousBlock && savedLastBlock) {
+ // If process [equal or different blocks], return the same lastBlock and toBlock equal the block batch size
+ if (savedPreviousBlock === savedLastBlock || savedPreviousBlock !== savedLastBlock) {
+ return {
+ fromBlock: Number(savedLastBlock),
+ toBlock: cfgBlockBarchSize,
+ };
+ }
+ }
+
+ if (savedLastBlock) {
+ // If there is [no previous block], return the lastBlock and toBlock equal the block batch size
+ if (!cfgFromBlock || BigInt(cfgFromBlock) < savedLastBlock) {
+ return {
+ fromBlock: Number(savedLastBlock),
+ toBlock: cfgBlockBarchSize,
+ };
+ }
+ }
+ }
+
+ getUpdatedRange(): Range {
+ return {
+ previousBlock: this.previousBlock,
+ lastBlock: this.lastBlock,
+ };
+ }
+}
+
+export type GetAptosOpts = {
+ addresses: string[];
+ filter: TransactionFilter;
+ previousBlock?: bigint | undefined;
+ lastBlock?: bigint | undefined;
+};

blockchain-watcher/src/domain/actions/aptos/HandleAptosTransactions.ts

@@ -0,0 +1,46 @@
+import { TransactionFoundEvent } from "../../entities";
+import { AptosTransaction } from "../../entities/aptos";
+import { StatRepository } from "../../repositories";
+
+export class HandleAptosTransactions {
+ constructor(
+ private readonly cfg: HandleAptosTransactionsOptions,
+ private readonly mapper: (tx: AptosTransaction) => TransactionFoundEvent,
+ private readonly target: (parsed: TransactionFoundEvent[]) => Promise<void>,
+ private readonly statsRepo: StatRepository
+ ) {}
+
+ public async handle(txs: AptosTransaction[]): Promise<TransactionFoundEvent[]> {
+ const items: TransactionFoundEvent[] = [];
+
+ for (const tx of txs) {
+ const txMapped = this.mapper(tx);
+ this.report(txMapped.attributes.protocol);
+ items.push(txMapped);
+ }
+
+ await this.target(items);
+
+ return items;
+ }
+
+ private report(protocol: string) {
+ if (!this.cfg.metricName) return;
+
+ const labels = this.cfg.metricLabels ?? {
+ job: this.cfg.id,
+ chain: "aptos",
+ commitment: "finalized",
+ protocol: protocol,
+ };
+
+ this.statsRepo.count(this.cfg.metricName, labels);
+ }
+}
+
+export interface HandleAptosTransactionsOptions {
+ metricLabels?: { job: string; chain: string; commitment: string };
+ eventTypes?: string[];
+ metricName?: string;
+ id: string;
+}