Monitor for polymesh polkadot assets

Polymesh events.ts

import { ArgumentsCamelCase, Argv } from "yargs";
import { logger } from "../../logger";
import { red, green, blue, yellow } from "picocolors";
import { Polymesh } from "@polymeshassociation/polymesh-sdk";
import { UnsubCallback } from "@polymeshassociation/polymesh-sdk/types";
import {
  balanceToBigNumber,
  tickerToString,
  instructionMemoToString,
  bytesToString,
} from "@polymeshassociation/polymesh-sdk/utils/conversion";
import {
  PolymeshPrimitivesIdentityIdPortfolioId,
  PolymeshPrimitivesNftNfTs,
  PolymeshPrimitivesPortfolioPortfolioUpdateReason,
} from "@polymeshassociation/polymesh-sdk/polkadot/types-lookup";
import { polymesh } from "../../helpers/polymesh";

interface GetTransactionEventsArgv {
  type: string;
  assets: string;
}

let sdk: Polymesh | null = null;
let api: Polymesh["_polkadotApi"] | null = null;

interface DecodedPortfolioId {
  did: string;
  portfolioType: "Default" | "User";
  portfolioNumber?: string; // only present if portfolioType is 'User'
}

interface PortfolioUpdateReason {
  type: "Issued" | "Redeemed" | "Transferred" | "ControllerTransfer";
  fundingRoundName?: string | null;
  instructionId?: string | null;
  instructionMemo?: string | null;
}

/**
 * Decodes the portfolio ID into a human-readable object.
 * @param portfolioId The raw portfolio ID from the blockchain.
 * @returns Decoded portfolio information.
 */
const decodePortfolioId = (
  portfolioId: PolymeshPrimitivesIdentityIdPortfolioId,
): DecodedPortfolioId => {
  const { kind } = portfolioId;

  if (kind.isDefault) {
    return {
      did: portfolioId.did.toString(),
      portfolioType: "Default",
    };
  }

  if (kind.isUser) {
    return {
      did: portfolioId.did.toString(),
      portfolioType: "User",
      portfolioNumber: kind.asUser.toString(),
    };
  }

  throw new Error("Unknown portfolio kind");
};

/**
 * Decodes the portfolio update reason into an object.
 * @param reason The raw portfolio update reason from the blockchain.
 * @returns Decoded portfolio update reason.
 */
const decodePortfolioUpdateReason = (
  reason: PolymeshPrimitivesPortfolioPortfolioUpdateReason,
): PortfolioUpdateReason => {
  if (reason.isIssued) {
    return {
      type: "Issued",
      fundingRoundName: reason.asIssued.fundingRoundName.isSome
        ? bytesToString(reason.asIssued.fundingRoundName.unwrap())
        : null,
    };
  }

  if (reason.isTransferred) {
    return {
      type: "Transferred",
      instructionId: reason.asTransferred.instructionId.isSome
        ? reason.asTransferred.instructionId.unwrap().toString()
        : null,
      instructionMemo: reason.asTransferred.instructionMemo.isSome
        ? instructionMemoToString(reason.asTransferred.instructionMemo.unwrap())
        : null,
    };
  }

  if (reason.isRedeemed) {
    return {
      type: "Redeemed",
    };
  }

  if (reason.isControllerTransfer) {
    return {
      type: "ControllerTransfer",
    };
  }

  throw new Error("Unknown portfolio update reason");
};

/**
 * Decodes NFT details such as ticker and IDs.
 * @param nfts The raw NFT data from the blockchain.
 * @returns Decoded NFT information.
 */
const decodeNFTs = (nfts: PolymeshPrimitivesNftNfTs) => {
  const ticker = tickerToString(nfts.ticker);
  const ids = nfts.ids.map((id) => id.toString());
  return { ticker, ids };
};

// CLI Command to get transaction events
export const getTransactionEventsCommand = {
  command: "get-transaction-events",
  describe: "Fetch the transaction events by type",
  aliases: ["gte"],
  builder: (yargs: Argv): Argv<GetTransactionEventsArgv> => {
    return yargs
      .option("type", {
        type: "string",
        alias: "t",
        describe:
          "Type of event: 'issued', 'redeemed', 'transferred' (Defaults to 'any')",
        demandOption: false,
        default: "any",
      })
      .option("assets", {
        type: "string",
        alias: "at",
        describe: "The ticker(s) of the asset(s), comma-separated",
        demandOption: false,
        default: "any",
      });
  },
  handler: async (argv: GetTransactionEventsArgv) => {
    try {
      // Parse arguments
      const { type, assets } = argv;
      const types =
        type !== "any"
          ? type.split(",").map((type) => type.toLocaleLowerCase())
          : [];
      const tickers = assets !== "any" ? assets.split(",") : [];

      logger.info(
        `Listening to events of types: ${types.length === 0 ? "*" : types.toString()}`,
      );
      logger.info(`Listening to events for tickers: ${tickers}`);

      // Initialize Polymesh SDK
      const { polyClient } = await polymesh("", true);

      sdk = polyClient;
      api = polyClient._polkadotApi;

      // Subscribe to finalized heads (finalized blocks only)
      const unsubscribeFinalizedHeads =
        await api.rpc.chain.subscribeFinalizedHeads(async (header) => {
          const blockHash = header.hash.toString();
          await processBlockEvents(blockHash, types, tickers);
        });

      // Catch errors and unsubscribe if necessary
    } catch (error) {
      logger.error(
        red("An error occurred while fetching transaction events:"),
        error,
      );
      process.exit(1);
    }
  },
};

/**
 * Processes block events and handles specific events such as Transfers, AssetBalanceUpdated, and NFTPortfolioUpdated.
 * Filters events based on types and asset tickers.
 * @param hash The block hash for the finalized block.
 * @param types The types of events to filter.
 * @param tickers The asset tickers to filter.
 */
async function processBlockEvents(
  hash: string,
  types: string[],
  tickers: string[],
): Promise<void> {
  if (!api) throw new Error("API not initialized");

  const finalizedBlock = await api.rpc.chain.getBlock(hash);
  const blockNumber = finalizedBlock.block.header.number.unwrap().toString();
  const { extrinsics } = finalizedBlock.block;
  const apiAtBlock = await api.at(hash);

  logger.info(`Processing Block number ${blockNumber}`);

  const events = await apiAtBlock.query.system.events();

  events.forEach((record) => {
    if (!api) throw new Error("API not initialized");

    const { event, phase } = record;
    let extrinsicHash: string | undefined;
    let extrinsicId: number | undefined;
    let method: string | undefined;
    let section: string | undefined;

    if (phase.isApplyExtrinsic) {
      extrinsicId = phase.asApplyExtrinsic.toNumber();
      const extrinsic = extrinsics[extrinsicId];
      extrinsicHash = extrinsic?.hash.toString();
      method = extrinsic?.method.method;
      section = extrinsic?.method.section;
    }

    // Handle Balances.Transfer event
    if (api.events.balances.Transfer.is(event)) {
      const balancesEvent = event.data;
      const fromDid = balancesEvent[0].unwrapOrDefault().toString();
      const fromAddress = balancesEvent[1].toString();
      const toDid = balancesEvent[2].unwrapOrDefault().toString();
      const toAddress = balancesEvent[3].toString();
      const amount = balanceToBigNumber(balancesEvent[4]).toString();
      const memo = balancesEvent[5].isSome
        ? instructionMemoToString(balancesEvent[5].unwrap())
        : undefined;

      logger.info(
        "Transfer event details:",
        {
          fromDid,
          fromAddress,
          toDid,
          toAddress,
          amount,
          memo,
        },
        {
          blockNumber,
          extrinsicId,
          extrinsicHash,
          section,
          method,
        },
      );
    }

    // Handle Asset.AssetBalanceUpdated event
    if (api.events.asset.AssetBalanceUpdated.is(event)) {
      const assetBalanceUpdatedEvent = event.data;
      const did = assetBalanceUpdatedEvent[0].toString();
      const ticker = tickerToString(assetBalanceUpdatedEvent[1]);
      const assetAmount = balanceToBigNumber(
        assetBalanceUpdatedEvent[2],
      ).toString();
      const fromPortfolio = assetBalanceUpdatedEvent[3].isSome
        ? decodePortfolioId(assetBalanceUpdatedEvent[3].unwrap())
        : null;
      const toPortfolio = assetBalanceUpdatedEvent[4].isSome
        ? decodePortfolioId(assetBalanceUpdatedEvent[4].unwrap())
        : null;
      const updateReason = decodePortfolioUpdateReason(
        assetBalanceUpdatedEvent[5],
      );

      if (
        tickers.length === 0 ||
        (tickers.includes(ticker) &&
          types.includes(updateReason.type.toLocaleLowerCase()))
      ) {
        logger.info(
          "AssetBalanceUpdated event details:",
          {
            did,
            ticker,
            assetAmount,
            fromPortfolio,
            toPortfolio,
            updateReason,
          },
          {
            blockNumber,
            extrinsicId,
            extrinsicHash,
            section,
            method,
          },
        );
      }
    }

    // Handle NFT.NFTPortfolioUpdated event
    if (api.events.nft.NFTPortfolioUpdated.is(event)) {
      const nftPortfolioUpdatedEvent = event.data;
      const did = nftPortfolioUpdatedEvent[0].toString();
      const nfts = decodeNFTs(nftPortfolioUpdatedEvent[1]);
      const fromPortfolio = nftPortfolioUpdatedEvent[2].isSome
        ? decodePortfolioId(nftPortfolioUpdatedEvent[2].unwrap())
        : null;
      const toPortfolio = nftPortfolioUpdatedEvent[3].isSome
        ? decodePortfolioId(nftPortfolioUpdatedEvent[3].unwrap())
        : null;
      const updateReason = decodePortfolioUpdateReason(
        nftPortfolioUpdatedEvent[4],
      );

      logger.info(
        "NFTPortfolioUpdated event details:",
        {
          did,
          nfts,
          fromPortfolio,
          toPortfolio,
          updateReason,
        },
        {
          blockNumber,
          extrinsicId,
          extrinsicHash,
          section,
          method,
        },
      );
    }
  });
}