Properly Index Relayer Stats and Collator Reward Indexing

packages/indexer/scripts/templates/substrate-chain.yaml.hbs

@@ -32,6 +32,11 @@ dataSources:
           filter:
            module: relayer
            method: AccumulateFees
+        - handler: handleCollatorRewardedEvent
+          kind: substrate/EventHandler
+          filter:
+            module: collatorManager
+            method: CollatorRewarded
         - handler: handleSubstrateAssetTeleportedEvent
           kind: substrate/EventHandler
           filter:

packages/indexer/src/configs/schema.graphql

@@ -1873,3 +1873,38 @@ type DustSwept @entity {
 	"""
 	lastUpdated: Date! @index
 }
+
+"""
+Collator rewards indexed from treasury transfers to block authors
+"""
+type HyperbridgeCollatorReward @entity {
+	"""
+	Unique identifier for the collator reward entry (collator address)
+	"""
+	id: ID!
+
+	"""
+	The total amount of rewards this collator has ever received
+	"""
+	totalRewardAmount: BigInt
+
+	"""
+	Total number of blocks authored by this collator
+	"""
+	totalBlocksAuthored: BigInt
+
+	"""
+	Timestamp of the last reward received
+	"""
+	lastRewardedAt: Date
+}
+
+"""
+Record of daily collator rewards paid out by the treasury
+"""
+type DailyTreasuryCollatorReward @entity {
+	id: ID! # ID is: {YYYY-MM-DD}
+	date: Date!
+	dailyRewardAmount: BigInt!
+}
+

packages/indexer/src/constants.ts

@@ -12,6 +12,7 @@ enum SupportedAssets {
 export const HYPERBRIDGE = {
 	testnet: "KUSAMA-4009",
 	mainnet: "POLKADOT-3367",
+	local: "KUSAMA-2000",
 } as const
 
 export const BIFROST = {

packages/indexer/src/handlers/events/collators/collatorRewarded.event.handler.ts

@@ -0,0 +1,70 @@
+import { SubstrateEvent } from "@subql/types"
+import { HyperbridgeCollatorReward, DailyTreasuryCollatorReward } from "@/configs/src/types"
+import { wrap } from "@/utils/event.utils"
+import { Balance } from "@polkadot/types/interfaces"
+import { getBlockTimestamp } from "@/utils/rpc.helpers"
+import { getHostStateMachine } from "@/utils/substrate.helpers"
+import { timestampToDate } from "@/utils/date.helpers"
+
+/**
+ * Handles CollatorRewarded events from pallet-collator-manager
+ * Emitted when a collator is rewarded for authoring a block
+ */
+export const handleCollatorRewardedEvent = wrap(async (event: SubstrateEvent): Promise<void> => {
+	try {
+		const {
+			event: { data, method },
+			block,
+		} = event
+
+		const [collator, amount] = data
+		const collatorAddress = collator.toString()
+		const rewardAmount = (amount as unknown as Balance).toBigInt()
+
+		let record = await HyperbridgeCollatorReward.get(collatorAddress)
+		if (!record) {
+			record = HyperbridgeCollatorReward.create({
+				id: collatorAddress,
+				totalRewardAmount: BigInt(0),
+				totalBlocksAuthored: BigInt(0),
+			})
+		}
+
+		record.totalRewardAmount = (record.totalRewardAmount ?? BigInt(0)) + rewardAmount
+		record.totalBlocksAuthored = (record.totalBlocksAuthored ?? BigInt(0)) + BigInt(1)
+		record.lastRewardedAt = new Date(block.timestamp!)
+
+		await record.save()
+
+		const hyperbridgeChain = getHostStateMachine(chainId)
+		const blockTimestamp = await getBlockTimestamp(block.block.header.hash.toString(), hyperbridgeChain)
+		await updateDailyCollatorReward(blockTimestamp, rewardAmount)
+
+		logger.info(`Collator reward indexed: ${collatorAddress} received ${rewardAmount.toString()}`)
+	} catch (e) {
+		const errorMessage = e instanceof Error ? e.message : String(e)
+		logger.error(`Failed to handle collator reward event: ${errorMessage}`)
+	}
+})
+
+/**
+ * Updates the daily collator reward tracking
+ */
+async function updateDailyCollatorReward(timestamp: bigint, amount: bigint): Promise<void> {
+	const day = timestampToDate(timestamp)
+	day.setUTCHours(0, 0, 0, 0)
+	const id = day.toISOString().slice(0, 10)
+
+	let record = await DailyTreasuryCollatorReward.get(id)
+
+	if (!record) {
+		record = DailyTreasuryCollatorReward.create({
+			id: id,
+			date: day,
+			dailyRewardAmount: BigInt(0),
+		})
+	}
+
+	record.dailyRewardAmount += amount
+	await record.save()
+}

packages/indexer/src/handlers/events/evmHost/getRequestHandled.event.handler.ts

@@ -35,7 +35,7 @@ export const handleGetRequestHandledEvent = wrap(async (event: GetRequestHandled
 	const blockTimestamp = await getBlockTimestamp(blockHash, chain)
 
 	try {
-		await HyperBridgeService.handlePostRequestOrResponseHandledEvent(relayer_id, chain, blockTimestamp)
+		await HyperBridgeService.handlePostRequestOrResponseHandledEvent(relayer_id, chain, blockTimestamp, transaction)
 
 		await GetRequestService.updateStatus({
 			commitment,

packages/indexer/src/handlers/events/evmHost/postRequestHandled.event.handler.ts

@@ -34,7 +34,7 @@ export const handlePostRequestHandledEvent = wrap(async (event: PostRequestHandl
 	const blockTimestamp = await getBlockTimestamp(blockHash, chain)
 
 	try {
-		await HyperBridgeService.handlePostRequestOrResponseHandledEvent(relayer_id, chain, blockTimestamp)
+		await HyperBridgeService.handlePostRequestOrResponseHandledEvent(relayer_id, chain, blockTimestamp, transaction)
 
 		await RequestService.updateStatus({
 			commitment,

packages/indexer/src/handlers/events/evmHost/postResponseHandled.event.handler.ts

@@ -35,7 +35,7 @@ export const handlePostResponseHandledEvent = wrap(async (event: PostResponseHan
 	const blockTimestamp = await getBlockTimestamp(blockHash, chain)
 
 	try {
-		await HyperBridgeService.handlePostRequestOrResponseHandledEvent(relayer_id, chain, blockTimestamp)
+		await HyperBridgeService.handlePostRequestOrResponseHandledEvent(relayer_id, chain, blockTimestamp, transaction)
 
 		await ResponseService.updateStatus({
 			commitment,

packages/indexer/src/mappings/mappingHandlers.ts

@@ -61,3 +61,4 @@ export { handleFeeRewardedEvent } from "@/handlers/events/incentives/feeRewarded
 
 export { handleTreasuryTransferEvent } from "@/handlers/events/treasury/treasuryTransfer.event.handler"
 export { handleAccumulateFeesEvent } from "@/handlers/events/fees/accumulatedFees.event.handler"
+export { handleCollatorRewardedEvent } from "@/handlers/events/collators/collatorRewarded.event.handler"

packages/indexer/src/services/hyperbridge.service.ts

@@ -1,4 +1,5 @@
 import { PostRequestEventLog, PostResponseEventLog } from "@/configs/src/types/abi-interfaces/EthereumHostAbi"
+import { EthereumTransaction } from "@subql/types-ethereum"
 import { DailyProtocolFeesStats, Relayer, Transfer } from "@/configs/src/types/models"
 import { HyperBridgeChainStatsService } from "@/services/hyperbridgeChainStats.service"
 import { isHexString } from "ethers/lib/utils"
@@ -8,10 +9,6 @@ import { getBlockTimestamp } from "@/utils/rpc.helpers"
 import stringify from "safe-stable-stringify"
 import { getDateFormatFromTimestamp, isWithin24Hours, timestampToDate } from "@/utils/date.helpers"
 import { RelayerService } from "./relayer.service"
-// import {
-// 	HandlePostRequestsTransaction,
-// 	HandlePostResponsesTransaction,
-// } from "@/configs/src/types/abi-interfaces/HandlerV1Abi"
 
 export class HyperBridgeService {
 	/**
@@ -47,35 +44,18 @@ export class HyperBridgeService {
 
 	/**
 	 * Perform the necessary actions related to Hyperbridge stats when a PostRequestHandled/PostResponseHandled event is indexed
+	 * @param transaction Optional Ethereum transaction for EVM chains.
 	 */
 	static async handlePostRequestOrResponseHandledEvent(
 		relayer_id: string,
 		chain: string,
 		timestamp: bigint,
+		transaction?: EthereumTransaction,
 	): Promise<void> {
 		await this.incrementNumberOfDeliveredMessages(chain)
-		await RelayerService.updateMessageDelivered(relayer_id, chain, timestamp)
+		await RelayerService.updateMessageDelivered(relayer_id, chain, timestamp, transaction)
 	}
 
-	/**
-	 * Handle PostRequest or PostResponse transactions
-	 */
-	// static async handlePostRequestOrResponseTransaction(
-	// 	chain: string,
-	// 	transaction: HandlePostRequestsTransaction | HandlePostResponsesTransaction,
-	// ): Promise<void> {
-	// 	logger.info(
-	// 		`Creating PostRequest or PostResponse transaction update: ${JSON.stringify({
-	// 			transaction,
-	// 		})}`,
-	// 	)
-	// 	const { status } = await transaction.receipt()
-
-	// 	if (status === false) {
-	// 		await this.incrementNumberOfFailedDeliveries(chain)
-	// 	}
-	// }
-
 	/**
 	 * Increment the total number of messages sent on hyperbridge
 	 */

packages/indexer/src/services/relayer.service.ts

@@ -1,14 +1,9 @@
 import { Relayer, RelayerActivity, Transfer } from "@/configs/src/types/models"
 import { RelayerChainStatsService } from "@/services/relayerChainStats.service"
-// import {
-// 	HandlePostRequestsTransaction,
-// 	HandlePostResponsesTransaction,
-// } from "@/configs/src/types/abi-interfaces/HandlerV1Abi"
+import { EthereumTransaction } from "@subql/types-ethereum"
 import PriceHelper from "@/utils/price.helpers"
 import { PointsService } from "@/services/points.service"
 import { PointsActivityType, ProtocolParticipantType } from "@/configs/src/types"
-import { getHostStateMachine } from "@/utils/substrate.helpers"
-import { getBlockTimestamp } from "@/utils/rpc.helpers"
 import { GET_ETHEREUM_L2_STATE_MACHINES } from "@/utils/l2-state-machine.helper"
 
 export class RelayerService {
@@ -61,17 +56,69 @@ export class RelayerService {
 	}
 
 	/**
-	 * Update message delivered by the relayer
+	 * Update message delivered by the relayer with gas cost tracking
 	 * @param relayer_id The relayer address
 	 * @param chain The chain identifier
+	 * @param timestamp The block timestamp
+	 * @param transaction Optional Ethereum transaction for EVM chains. When provided, gas costs are tracked.
 	 */
-	static async updateMessageDelivered(relayer_id: string, chain: string, timestamp: bigint): Promise<void> {
+	static async updateMessageDelivered(
+		relayer_id: string,
+		chain: string,
+		timestamp: bigint,
+		transaction?: EthereumTransaction,
+	): Promise<void> {
 		const relayer = await this.findOrCreate(relayer_id, chain, timestamp)
 		const relayer_chain_stats = await RelayerChainStatsService.findOrCreate(relayer.id, chain)
 
-		relayer_chain_stats.numberOfSuccessfulMessagesDelivered += BigInt(1)
-		await this.updateRelayerActivity(relayer.id, timestamp)
+		if (transaction) {
+			const receipt = await transaction.receipt()
+			const { status, gasUsed, effectiveGasPrice } = receipt
+
+			const nativeCurrencyPrice = await PriceHelper.getNativeCurrencyPrice(chain)
+			let gasFee = BigInt(effectiveGasPrice) * BigInt(gasUsed)
+
+			// Add L1 fee for L2 chains
+			if (GET_ETHEREUM_L2_STATE_MACHINES().includes(chain)) {
+				const l1Fee = BigInt((receipt as any).l1Fee ?? 0)
+				gasFee += l1Fee
+			}
+
+			const usdFee = (gasFee * nativeCurrencyPrice) / 10n ** 18n
+
+			let pointsToAward = 50
+			let description = "Points awarded for successful message delivered"
+
+			if (status === true) {
+				relayer_chain_stats.numberOfSuccessfulMessagesDelivered += BigInt(1)
+				relayer_chain_stats.gasUsedForSuccessfulMessages += BigInt(gasUsed)
+				relayer_chain_stats.gasFeeForSuccessfulMessages += gasFee
+				relayer_chain_stats.usdGasFeeForSuccessfulMessages += usdFee
+			} else {
+				relayer_chain_stats.numberOfFailedMessagesDelivered += BigInt(1)
+				relayer_chain_stats.gasUsedForFailedMessages += BigInt(gasUsed)
+				relayer_chain_stats.gasFeeForFailedMessages += gasFee
+				relayer_chain_stats.usdGasFeeForFailedMessages += usdFee
+				pointsToAward = pointsToAward / 2
+				description = "Points awarded for failed message delivery"
+			}
+
+			await PointsService.awardPoints(
+				relayer_id,
+				chain,
+				BigInt(pointsToAward),
+				ProtocolParticipantType.RELAYER,
+				PointsActivityType.REWARD_POINTS_EARNED,
+				transaction.hash,
+				description,
+				timestamp,
+			)
+		} else {
+			// For non evm chains without transaction
+			relayer_chain_stats.numberOfSuccessfulMessagesDelivered += BigInt(1)
+		}
 
+		await this.updateRelayerActivity(relayer.id, timestamp)
 		await relayer.save()
 		await relayer_chain_stats.save()
 	}
@@ -91,82 +138,4 @@ export class RelayerService {
 		await activity.save()
 	}
 
-	/**
-	 * Computes relayer specific stats from the handlePostRequest/handlePostResponse transactions on the handlerV1 contract
-	 */
-	// static async handlePostRequestOrResponseTransaction(
-	// 	chain: string,
-	// 	transaction: HandlePostRequestsTransaction | HandlePostResponsesTransaction,
-	// ): Promise<void> {
-	// 	const { from: relayer_id, hash: transaction_hash, blockHash } = transaction
-	// 	const receipt = await transaction.receipt()
-	// 	const { status, gasUsed, effectiveGasPrice } = receipt
-
-	// 	const nativeCurrencyPrice = await PriceHelper.getNativeCurrencyPrice(chain)
-
-	// 	let gasFee = BigInt(effectiveGasPrice) * BigInt(gasUsed)
-
-	// 	// Add the L1 Gas Used for L2 chains
-	// 	if (GET_ETHEREUM_L2_STATE_MACHINES().includes(chain)) {
-	// 		if ((receipt as any).l1Fee) {
-	// 			const l1Fee = BigInt((receipt as any).l1Fee ?? 0)
-	// 			gasFee += l1Fee
-	// 		} else {
-	// 			logger.error(
-	// 				`Could not find l1Fee in transaction receipt: ${JSON.stringify({
-	// 					chain,
-	// 					transactionHash: transaction.hash,
-	// 				})}`,
-	// 			)
-	// 		}
-	// 	}
-
-	// 	const usdFee = (gasFee * nativeCurrencyPrice) / (10n ** 18n);
-	// 	const gasFeeInEth = Number(gasFee) / 1e18;
-
-	// 	try {
-	// 		const timestamp = await getBlockTimestamp(blockHash, chain)
-
-	// 		let relayer = await RelayerService.findOrCreate(relayer_id, chain, timestamp)
-	// 		let relayer_chain_stats = await RelayerChainStatsService.findOrCreate(relayer_id, chain)
-
-	// 		let pointsToAWard = 50;
-	// 		let description = "`Points awarded for successful message delivered`";
-	// 		if (status === true) {
-	// 			relayer_chain_stats.numberOfSuccessfulMessagesDelivered += BigInt(1)
-	// 			relayer_chain_stats.gasUsedForSuccessfulMessages += BigInt(gasUsed)
-	// 			relayer_chain_stats.gasFeeForSuccessfulMessages += BigInt(gasFee)
-	// 			relayer_chain_stats.usdGasFeeForSuccessfulMessages += usdFee
-	// 		} else {
-	// 			relayer_chain_stats.numberOfFailedMessagesDelivered += BigInt(1)
-	// 			relayer_chain_stats.gasUsedForFailedMessages += BigInt(gasUsed)
-	// 			relayer_chain_stats.gasFeeForFailedMessages += BigInt(gasFee)
-	// 			relayer_chain_stats.usdGasFeeForFailedMessages += usdFee
-
-	// 			pointsToAWard = pointsToAWard / 2;
-	// 			description = "`Points awarded for failed message delivery`"
-	// 		}
-
-	// 		await PointsService.awardPoints(
-	// 			relayer_id,
-	// 			chain,
-	// 			BigInt(pointsToAWard),
-	// 			ProtocolParticipantType.RELAYER,
-	// 			PointsActivityType.REWARD_POINTS_EARNED,
-	// 			transaction_hash,
-	// 			description,
-	// 			timestamp,
-	// 		)
-
-	// 		await relayer.save()
-	// 		await relayer_chain_stats.save()
-
-	// 		logger.info(`Relayer: ${relayer_id} updated successfully for chain: ${chain}`)
-	// 	} catch (e) {
-	// 		const errorMessage = e instanceof Error ? e.message : String(e)
-	// 		logger.error(
-	// 			`Error while handling PostRequest/PostResponse transaction relayer updates: ${errorMessage}`,
-	// 		)
-	// 	}
-	// }
 }

packages/indexer/src/utils/substrate.helpers.ts

@@ -109,7 +109,7 @@ export function getHostStateMachine(chainId: string): StateMachineId {
 }
 
 export function isHyperbridge(host: StateMachineId): boolean {
-	return host === HYPERBRIDGE.mainnet || host === HYPERBRIDGE.testnet
+	return host === HYPERBRIDGE.mainnet || host === HYPERBRIDGE.testnet || host === HYPERBRIDGE.local
 }
 
 /**