Refactored supplier-related handlers and logic (#65)

## Summary

Recreated index supplier logic to handle the messages and events that
updates suppliers in parallel:

- Removed unused supplier handlers and related code (e.g.,
`_handleOldEventSupplierSlashed`).
- Consolidated supplier processing into `indexSupplier`.
- Exported internal utility methods (`parseAttribute`, `getAttributes`,
etc.) for broader use.
- Deprecated event/message handling for suppliers now processed via
`indexSupplier`.
- Enhanced modularity for supplier-related functions.

## Issue

Our previous logic was indexing every supplier update sequentially, one
by one. This caused node runners updating their suppliers' stakes (i.e.,
updating their services) to take an extremely long time, causing the
indexer to get stuck on that block and fall out of sync.

## Type of change

Select one or more:

- [x] New feature, functionality or library
- [ ] Bug fix
- [ ] Code health or cleanup
- [ ] Documentation
- [ ] Other (specify)

## Sanity Checklist

- [x] I have tested my changes using the available tooling
- [x] I have commented my code
- [x] I have performed a self-review of my own code; both comments &
source code
- [ ] I create and reference any new tickets, if applicable
- [ ] I have left TODOs throughout the codebase, if applicable

Author: Alann27

src/mappings/handlers.ts

@@ -31,19 +31,10 @@ import {
   handleEventClaimUpdated,
   handleEventProofUpdated,
   handleEventProofValidityChecked,
-  handleEventSupplierSlashed,
   handleMsgCreateClaim,
   handleMsgSubmitProof,
 } from "./pocket/relays";
 import { handleEventRelayMiningDifficultyUpdated, handleMsgAddService } from "./pocket/services";
-import {
-  handleEventSupplierServiceConfigActivated,
-  handleMsgClaimMorseSupplier,
-  handleSupplierStakeMsg,
-  handleSupplierUnbondingBeginEvent,
-  handleSupplierUnbondingEndEvent,
-  handleUnstakeSupplierMsg,
-} from "./pocket/suppliers";
 import {
   handleValidatorCommission,
   handleValidatorMsgCreate,
@@ -68,7 +59,6 @@ export const MsgHandlers: Record<string, (messages: Array<CosmosMessage>) => Pro
   // this is currently being handle inside Authz handler
   "/pocket.migration.MsgRecoverMorseAccount": noOp,
   "/pocket.migration.MsgClaimMorseApplication": handleMsgClaimMorseApplication,
-  "/pocket.migration.MsgClaimMorseSupplier": handleMsgClaimMorseSupplier,
   // bank
   "/cosmos.bank.v1beta1.MsgSend": handleNativeTransfer,
   // validator
@@ -83,9 +73,10 @@ export const MsgHandlers: Record<string, (messages: Array<CosmosMessage>) => Pro
   "/pocket.application.MsgTransferApplication": handleTransferApplicationMsg,
   // service
   "/pocket.service.MsgAddService": handleMsgAddService,
-  // supplier
-  "/pocket.supplier.MsgStakeSupplier": handleSupplierStakeMsg,
-  "/pocket.supplier.MsgUnstakeSupplier": handleUnstakeSupplierMsg,
+  // supplier - handled by batch processing in indexSupplier (called from indexStake)
+  "/pocket.supplier.MsgStakeSupplier": noOp,// - now handled in indexSupplier
+  "/pocket.supplier.MsgUnstakeSupplier": noOp,// - now handled in indexSupplier
+  "/pocket.migration.MsgClaimMorseSupplier": noOp,// - now handled in indexSupplier
   // gateway
   "/pocket.gateway.MsgStakeGateway": handleGatewayMsgStake,
   "/pocket.gateway.MsgUnstakeGateway": handleGatewayMsgUnstake,
@@ -117,10 +108,10 @@ export const EventHandlers: Record<string, (events: Array<CosmosEvent>) => Promi
   "pocket.application.EventTransferError": handleTransferApplicationErrorEvent,
   "pocket.application.EventApplicationUnbondingBegin": handleApplicationUnbondingBeginEvent,
   "pocket.application.EventApplicationUnbondingEnd": handleApplicationUnbondingEndEvent,
-  // supplier
-  "pocket.supplier.EventSupplierServiceConfigActivated": handleEventSupplierServiceConfigActivated,
-  "pocket.supplier.EventSupplierUnbondingBegin": handleSupplierUnbondingBeginEvent,
-  "pocket.supplier.EventSupplierUnbondingEnd": handleSupplierUnbondingEndEvent,
+  // supplier - handled by batch processing in indexSupplier (called from indexStake)
+  "pocket.supplier.EventSupplierServiceConfigActivated": noOp, // - now handled in indexSupplier
+  "pocket.supplier.EventSupplierUnbondingBegin": noOp, // - now handled in indexSupplier
+  "pocket.supplier.EventSupplierUnbondingEnd": noOp, // - now handled in indexSupplier
   // service
   "pocket.service.EventRelayMiningDifficultyUpdated": handleEventRelayMiningDifficultyUpdated,
   // gateway
@@ -130,7 +121,7 @@ export const EventHandlers: Record<string, (events: Array<CosmosEvent>) => Promi
   // relay
   "pocket.tokenomics.EventClaimSettled": handleEventClaimSettled,
   "pocket.tokenomics.EventClaimExpired": handleEventClaimExpired,
-  "pocket.tokenomics.EventSupplierSlashed": handleEventSupplierSlashed,
+  "pocket.tokenomics.EventSupplierSlashed": noOp, // - now handled in indexSupplier
   "pocket.tokenomics.EventApplicationOverserviced": handleEventApplicationOverserviced,
   "pocket.tokenomics.EventApplicationReimbursementRequest": handleEventApplicationReimbursementRequest,
   "pocket.proof.EventClaimUpdated": handleEventClaimUpdated,

src/mappings/indexer.manager.ts

@@ -25,6 +25,7 @@ import {
   MsgHandlers,
 } from "./handlers";
 import { handleAddBlockReports } from "./pocket/reports";
+import { indexSupplier } from "./pocket/suppliers";
 import {
   handleBlock,
   handleGenesis,
@@ -501,80 +502,6 @@ async function indexStakeEntity(allData: Array<CosmosEvent | CosmosMessage>, get
   )
 }
 
-// any supplier msg or event
-async function indexSupplier(msgByType: MessageByType, eventByType: EventByType): Promise<void> {
-  const msgTypes = [
-    "/pocket.supplier.MsgUnstakeSupplier",
-    "/pocket.migration.MsgClaimMorseSupplier",
-    "/pocket.supplier.MsgStakeSupplier",
-  ];
-  const eventTypes = [
-    "pocket.supplier.EventSupplierUnbondingBegin",
-    "pocket.supplier.EventSupplierUnbondingEnd",
-    "pocket.supplier.EventSupplierServiceConfigActivated",
-    // this is here because it modifies the staked tokens of the supplier
-    "pocket.tokenomics.EventSupplierSlashed"
-  ];
-
-
-  const eventGetId = (attributes: CosmosEvent["event"]["attributes"]) => {
-    for (const attribute of attributes) {
-      if (attribute.key === "supplier") {
-        return JSON.parse(attribute.value as string).operator_address
-      }
-
-      if (attribute.key === "operator_address") {
-        return attribute.value as string
-      }
-    }
-
-    return null
-  }
-
-  await indexStakeEntity(
-    [
-      ...msgTypes.map(type => msgByType[type]).flat(),
-      ...eventTypes.map(type => eventByType[type]).flat()
-    ],
-  {
-    "/pocket.supplier.MsgUnstakeSupplier": "operatorAddress",
-    "/pocket.supplier.MsgStakeSupplier": "operatorAddress",
-    "/pocket.migration.MsgClaimMorseSupplier": "shannonOperatorAddress",
-    "pocket.supplier.EventSupplierUnbondingBegin": eventGetId,
-    "pocket.supplier.EventSupplierUnbondingEnd": eventGetId,
-    "pocket.supplier.EventSupplierServiceConfigActivated": eventGetId,
-    "pocket.tokenomics.EventSupplierSlashed": (attributes) => {
-      /*
-        [
-          {"key":"application_address","value":"\"pokt16wwc45wjc4ulne7wmaawxhju00vwf900lscfld\""},
-          {"key":"claim_proof_status_int","value":"2"},
-          {"key":"proof_missing_penalty","value":"\"1upokt\""},
-          {"key":"service_id","value":"\"hey\""},
-          {"key":"session_end_block_height","value":"\"363540\""},
-          {"key":"supplier_operator_address","value":"\"pokt1wua234ulad3vkcsqmasu845mn4ugu9aa6jcv23\""},
-          {"key":"mode","value":"EndBlock"}
-        ]
-       */
-      for (const attribute of attributes) {
-        // in the previous version of this event this is the key to get the supplierId
-        if (attribute.key === "supplier_operator_addr" || attribute.key === "supplier_operator_address") {
-          return attribute.value as string
-        }
-
-        if (attribute.key === "claim") {
-          return JSON.parse(attribute.value as string).supplier_operator_address
-        }
-
-        if (attribute.key === "supplier_operator_address") {
-          return attribute.value as string
-        }
-      }
-
-      return null
-    }
-  })
-}
-
 // any message or event related to stake (supplier, gateway, application, service)
 async function indexStake(msgByType: MessageByType, eventByType: EventByType): Promise<void> {
   await Promise.all([

src/mappings/pocket/relays.ts

@@ -57,7 +57,7 @@ import {
 import { getDenomAndAmount } from "../utils/primitives";
 
 // this can return undefined because older events do not have this attribute
-function getClaimProofStatusFromSDK(item: typeof ClaimProofStatusSDKType | string | number): ClaimProofStatus | undefined {
+export function getClaimProofStatusFromSDK(item: typeof ClaimProofStatusSDKType | string | number): ClaimProofStatus | undefined {
   if (!item) return undefined;
 
   switch (item) {
@@ -151,7 +151,7 @@ function parseAttribute(attribute: unknown = ""): string {
 }
 
 // eslint-disable-next-line complexity
-function getAttributes(attributes: CosmosEvent["event"]["attributes"]) {
+export function getAttributes(attributes: CosmosEvent["event"]["attributes"]) {
 
   // eslint-disable-next-line @typescript-eslint/ban-ts-comment
   // @ts-ignore
@@ -890,121 +890,6 @@ function _handleEventApplicationReimbursementRequest(event: CosmosEvent): EventA
   }
 }
 
-async function _handleOldEventSupplierSlashed(event: CosmosEvent) {
-  let slashingCoin: CoinSDKType | null = null, operatorAddress = "";
-
-  for (const attribute of event.event.attributes) {
-    if (attribute.key === "slashing_amount") {
-      slashingCoin = getDenomAndAmount(attribute.value as string);
-      if (!slashingCoin) {
-        throw new Error(`[handleEventSupplierSlashed] event attribute key=${attribute.key} value=${attribute.value} is not a valid coin`);
-      }
-    }
-
-    if (attribute.key === "proof_missing_penalty") {
-      /*
-        [
-        {"key":"application_address","value":"\"pokt16wwc45wjc4ulne7wmaawxhju00vwf900lscfld\""},
-        {"key":"claim_proof_status_int","value":"2"},
-        {"key":"proof_missing_penalty","value":"\"1upokt\""},
-        {"key":"service_id","value":"\"hey\""},
-        {"key":"session_end_block_height","value":"\"363540\""},
-        {"key":"supplier_operator_address","value":"\"pokt1wua234ulad3vkcsqmasu845mn4ugu9aa6jcv23\""},
-        {"key":"mode","value":"EndBlock"}
-        ]
-     */
-      const coins = parseCoins(parseAttribute(attribute.value));
-      if (!coins.length) {
-        throw new Error(`[handleEventSupplierSlashed] event attribute key=${attribute.key} value=${attribute.value} is not a valid coin`);
-      }
-    }
-
-    if (attribute.key === "supplier_operator_addr" || attribute.key === "supplier_operator_address") {
-      operatorAddress = parseAttribute(attribute.value);
-    }
-  }
-
-  if (!slashingCoin) {
-    throw new Error(`[handleEventSupplierSlashed] slashingCoin not found in event`);
-  }
-
-  if (!operatorAddress) {
-    logger.error(`[handleEventSupplierSlashed] operatorAddress not found in event=${event.kind} attributes=${stringify(event.event.attributes)}`);
-    throw new Error(`[handleEventSupplierSlashed] operatorAddress not found in event`);
-  }
-
-  const supplier = await Supplier.get(operatorAddress);
-
-  if (!supplier) {
-    throw new Error(`[handleEventSupplierSlashed] supplier not found for operator address ${operatorAddress}`);
-  }
-
-  supplier.stakeAmount -= BigInt(slashingCoin.amount);
-
-  await Promise.all([
-    supplier.save(),
-    EventSupplierSlashed.create({
-      id: getEventId(event),
-      supplierId: operatorAddress,
-      blockId: getBlockId(event.block),
-      eventId: getEventId(event),
-      proofMissingPenalty: BigInt(slashingCoin.amount),
-      proofMissingPenaltyDenom: slashingCoin.denom,
-      previousStakeAmount: supplier.stakeAmount,
-      afterStakeAmount: supplier.stakeAmount,
-      // in alpha this event does not have the values below, so we are setting them to empty values for now
-      applicationId: "",
-      serviceId: "",
-      sessionId: "",
-      sessionStartHeight: BigInt(0),
-      sessionEndHeight: BigInt(0),
-    }).save(),
-  ]);
-}
-
-async function _handleEventSupplierSlashed(event: CosmosEvent) {
-  const {
-    claim,
-    proofMissingPenalty,
-  } = getAttributes(event.event.attributes);
-
-  if (!claim || !claim.session_header || Object.keys(claim).length === 0) {
-    logger.warn(`[handleEventSupplierSlashed] claim not found in event, trying to handle with previous version`);
-    await _handleOldEventSupplierSlashed(event);
-    return;
-  }
-
-  const supplier = await Supplier.get(claim.supplier_operator_address);
-
-  if (!supplier) {
-    throw new Error(`[handleEventSupplierSlashed] supplier not found for address: ${claim.supplier_operator_address}`);
-  }
-
-  const previousStakeAmount = supplier.stakeAmount.valueOf();
-  supplier.stakeAmount -= BigInt(proofMissingPenalty.amount);
-
-  await Promise.all([
-    supplier.save(),
-    EventSupplierSlashed.create({
-      id: getEventId(event),
-      supplierId: claim.supplier_operator_address,
-      // TODO: Create entity for session header
-      applicationId: claim.session_header.application_address,
-      serviceId: claim.session_header.service_id,
-      sessionId: claim.session_header.session_id || "",
-      sessionEndHeight: BigInt(claim.session_header.session_end_block_height || "0"),
-      sessionStartHeight: BigInt(claim.session_header.session_start_block_height || "0"),
-      blockId: getBlockId(event.block),
-      eventId: getEventId(event),
-      proofMissingPenalty: BigInt(proofMissingPenalty.amount),
-      proofMissingPenaltyDenom: proofMissingPenalty.denom,
-      previousStakeAmount,
-      afterStakeAmount: supplier.stakeAmount,
-      proofValidationStatus: getClaimProofStatusFromSDK(claim.proof_validation_status),
-    }).save(),
-  ]);
-}
-
 // eslint-disable-next-line complexity
 function _handleEventProofValidityChecked(event: CosmosEvent): EventProofValidityCheckedProps {
   let supplierOperatorAddress = "",
@@ -1158,10 +1043,6 @@ export async function handleEventApplicationReimbursementRequest(events: Array<C
   await optimizedBulkCreate("EventApplicationReimbursementRequest", eventsUpdated);
 }
 
-export async function handleEventSupplierSlashed(events: Array<CosmosEvent>): Promise<void> {
-  await Promise.all(events.map(_handleEventSupplierSlashed));
-}
-
 export async function handleEventProofValidityChecked(events: Array<CosmosEvent>): Promise<void> {
   await optimizedBulkCreate("EventProofValidityChecked", events.map(_handleEventProofValidityChecked));
 }