feat: index MerkleDistributor Claimed events

.env.sample

@@ -12,6 +12,7 @@ WEB3_NODE_URL_10=https://optimism-mainnet.infura.io/v3/
 WEB3_NODE_URL_288=https://boba-mainnet.gateway.pokt.network/v1/lb/
 WEB3_NODE_URL_42161=https://arbitrum-mainnet.infura.io/v3/
 WEB3_NODE_URL_137=https://polygon-mainnet.infura.io/v3/
+WEB3_NODE_URL_5=https://goerli.infura.io/v3/
 REFERRAL_DELIMITER_START_TIMESTAMP=1657290720
 ENABLE_SPOKE_POOLS_EVENTS_PROCESSING=false
 ENABLE_REFERRALS_MATERIALIZED_VIEW_REFRESH=false
@@ -27,3 +28,9 @@ DISCORD_CLIENT_ID=clientId
 DISCORD_CLIENT_SECRET=clientSecret
 # the url accessed after the Discord authorization processed is fulfilled
 DISCORD_REDIRECT_URI=http://localhost
+
+# MerkleDistributor overrides
+MERKLE_DISTRIBUTOR_CHAIN_ID=
+MERKLE_DISTRIBUTOR_ADDRESS=
+REFERRALS_START_WINDOW_INDEX=
+ENABLE_MERKLE_DISTRIBUTOR_EVENTS_PROCESSING=false

migrations/1667312452969-MerkleDistributorRecipient.ts

@@ -0,0 +1,29 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class MerkleDistributorRecipient1667312452969 implements MigrationInterface {
+  name = "MerkleDistributorRecipient1667312452969";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE "merkle_distributor_recipient"
+        DROP CONSTRAINT "UK_merkle_distributor_recipient_merkleDistributorWindowId_addre"
+    `);
+    await queryRunner.query(`
+      ALTER TABLE "merkle_distributor_recipient"
+        ADD CONSTRAINT "UK_merkle_distributor_recipient_windowId_address"
+          UNIQUE ("merkleDistributorWindowId", "address")
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE "merkle_distributor_recipient"
+        DROP CONSTRAINT "UK_merkle_distributor_recipient_windowId_address"
+    `);
+    await queryRunner.query(`
+      ALTER TABLE "merkle_distributor_recipient"
+        ADD CONSTRAINT "UK_merkle_distributor_recipient_merkleDistributorWindowId_addre"
+          UNIQUE ("merkleDistributorWindowId", "address")
+    `);
+  }
+}

migrations/1667813310964-MerkleDistributorProcessedBlock.ts

@@ -0,0 +1,20 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class MerkleDistributorProcessedBlock1667813310964 implements MigrationInterface {
+  name = "MerkleDistributorProcessedBlock1667813310964";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      CREATE TABLE "merkle_distributor_processed_block" (
+        "id" SERIAL NOT NULL,
+        "chainId" integer NOT NULL,
+        "latestBlock" integer NOT NULL,
+        "createdAt" TIMESTAMP NOT NULL DEFAULT now(),
+        CONSTRAINT "PK_fb2eb512abaadb453e1cfef109e" PRIMARY KEY ("id"))
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`DROP TABLE "merkle_distributor_processed_block"`);
+  }
+}

migrations/1667813310965-Claim.ts

@@ -0,0 +1,44 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class Claim1667813310965 implements MigrationInterface {
+  name = "Claim1667813310965";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      CREATE TABLE "claim" (
+        "id" SERIAL NOT NULL,
+        "caller" character varying NOT NULL,
+        "accountIndex" integer NOT NULL,
+        "windowIndex" integer NOT NULL,
+        "account" character varying NOT NULL,
+        "rewardToken" character varying NOT NULL,
+        "blockNumber" integer NOT NULL,
+        "claimedAt" TIMESTAMP NOT NULL,
+        "createdAt" TIMESTAMP NOT NULL DEFAULT now(),
+        "updatedAt" TIMESTAMP NOT NULL DEFAULT now(),
+        "merkleDistributorWindowId" integer,
+        CONSTRAINT "UK_claim_windowIndex_accountIndex" UNIQUE (
+          "windowIndex",
+          "accountIndex"
+        ),
+        CONSTRAINT "PK_466b305cc2e591047fa1ce58f81" PRIMARY KEY ("id"))
+    `);
+    await queryRunner.query(`CREATE INDEX "IX_claim_account" ON "claim" ("account")`);
+    await queryRunner.query(`
+      ALTER TABLE "claim"
+        ADD CONSTRAINT "FK_169ca2a2e031f01f62d81dbf1a0"
+          FOREIGN KEY ("merkleDistributorWindowId")
+          REFERENCES "merkle_distributor_window"("id")
+          ON DELETE NO ACTION ON UPDATE NO ACTION
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE "claim" DROP CONSTRAINT "FK_169ca2a2e031f01f62d81dbf1a0"
+    `);
+    await queryRunner.query(`DROP TABLE "merkle_distributor_processed_block"`);
+    await queryRunner.query(`DROP INDEX "public"."IX_claim_account"`);
+    await queryRunner.query(`DROP TABLE "claim"`);
+  }
+}

package.json

@@ -26,7 +26,7 @@
     "db:migration:run": "yarn typeorm -- migration:run  -d ormconfig.ts"
   },
   "dependencies": {
-    "@across-protocol/contracts-v2": "^1.0.0",
+    "@across-protocol/contracts-v2": "^1.0.7",
     "@nestjs/axios": "^0.0.8",
     "@nestjs/bull": "^0.5.5",
     "@nestjs/cli": "^8.2.4",

src/modules/airdrop/model/merkle-distributor-recipient.entity.ts

@@ -13,7 +13,7 @@ export type MerkleDistributorRecipientPayload = {
 
 @Entity()
 // A recipient address can't appear twice for the same window
-@Unique("UK_merkle_distributor_recipient_merkleDistributorWindowId_address", ["merkleDistributorWindowId", "address"])
+@Unique("UK_merkle_distributor_recipient_windowId_address", ["merkleDistributorWindowId", "address"])
 export class MerkleDistributorRecipient {
   @PrimaryGeneratedColumn()
   id: number;

src/modules/airdrop/model/merkle-distributor-window.entity.ts

@@ -1,5 +1,6 @@
 import { Column, CreateDateColumn, Entity, OneToMany, PrimaryGeneratedColumn, Unique } from "typeorm";
 import { MerkleDistributorRecipient } from "./merkle-distributor-recipient.entity";
+import { Claim } from "../../scraper/model/claim.entity";
 
 @Entity()
 // Don't allow duplicates of the window index
@@ -29,6 +30,9 @@ export class MerkleDistributorWindow {
   @OneToMany(() => MerkleDistributorRecipient, (recipient) => recipient.merkleDistributorWindow)
   recipients: MerkleDistributorRecipient;
 
+  @OneToMany(() => Claim, (claim) => claim.merkleDistributorWindow)
+  claims: Claim;
+
   @CreateDateColumn()
   createdAt: Date;
 }

src/modules/configuration/index.ts

@@ -39,7 +39,7 @@ export const configValues = () => ({
       // 69: process.env.WEB3_NODE_URL_69,
       // 4: process.env.WEB3_NODE_URL_4,
       // 80001: process.env.WEB3_NODE_URL_80001,
-      // 5: process.env.WEB3_NODE_URL_5,
+      5: process.env.WEB3_NODE_URL_5,
     },
     spokePoolContracts: {
       [ChainIds.mainnet]: {
@@ -63,9 +63,16 @@ export const configValues = () => ({
         startBlockNumber: 28604263,
       },
     },
+    merkleDistributor: {
+      address: process.env.MERKLE_DISTRIBUTOR_ADDRESS || "0xF633b72A4C2Fb73b77A379bf72864A825aD35b6D", // TODO: replace with mainnet
+      chainId: Number(process.env.MERKLE_DISTRIBUTOR_CHAIN_ID || "5"),
+      referralsStartWindowIndex: Number(process.env.REFERRALS_START_WINDOW_INDEX || "1"),
+      startBlockNumber: Number(process.env.MERKLE_DISTRIBUTOR_START_BLOCK || 7884371),
+    },
   },
   acxUsdPrice: 0.1,
   enableSpokePoolsEventsProcessing: process.env.ENABLE_SPOKE_POOLS_EVENTS_PROCESSING === "true",
+  enableMerkleDistributorEventsProcessing: process.env.ENABLE_MERKLE_DISTRIBUTOR_EVENTS_PROCESSING === "true",
   enableReferralsMaterializedViewRefresh: process.env.ENABLE_REFERRALS_MATERIALIZED_VIEW_REFRESH === "true",
   allowWalletRewardsEdit: process.env.ALLOW_WALLET_REWARDS_EDIT === "true",
   stickyReferralAddressesMechanism: process.env.STICKY_REFERRAL_ADDRESSES_MECHANISM

src/modules/database/database.providers.ts

@@ -2,6 +2,8 @@ import { TypeOrmOptionsFactory, TypeOrmModuleOptions } from "@nestjs/typeorm";
 import { Injectable } from "@nestjs/common";
 import { AppConfig } from "../configuration/configuration.service";
 import { ProcessedBlock } from "../scraper/model/ProcessedBlock.entity";
+import { MerkleDistributorProcessedBlock } from "../scraper/model/MerkleDistributorProcessedBlock.entity";
+import { Claim } from "../scraper/model/claim.entity";
 import { Block } from "../web3/model/block.entity";
 import { Deposit } from "../scraper/model/deposit.entity";
 import { Token } from "../web3/model/token.entity";
@@ -17,6 +19,8 @@ import { MerkleDistributorWindow } from "../airdrop/model/merkle-distributor-win
 // TODO: Add db entities here
 const entities = [
   ProcessedBlock,
+  MerkleDistributorProcessedBlock,
+  Claim,
   Block,
   Deposit,
   Token,

src/modules/scraper/adapter/messaging/MerkleDistributorBlocksEventsConsumer.ts

@@ -0,0 +1,63 @@
+import { OnQueueFailed, Process, Processor } from "@nestjs/bull";
+import { Logger } from "@nestjs/common";
+import { Job } from "bull";
+import { InjectRepository } from "@nestjs/typeorm";
+import { Repository, QueryFailedError } from "typeorm";
+
+import { EthProvidersService } from "../../../web3/services/EthProvidersService";
+import { MerkleDistributorBlocksEventsQueueMessage, ScraperQueue } from ".";
+import { ClaimedEvent } from "@across-protocol/contracts-v2/dist/typechain/MerkleDistributor";
+import { Claim } from "../../model/claim.entity";
+import { utils } from "ethers";
+
+@Processor(ScraperQueue.MerkleDistributorBlocksEvents)
+export class MerkleDistributorBlocksEventsConsumer {
+  private logger = new Logger(MerkleDistributorBlocksEventsConsumer.name);
+
+  constructor(
+    private providers: EthProvidersService,
+    @InjectRepository(Claim) private claimRepository: Repository<Claim>,
+  ) {}
+
+  @Process({ concurrency: 1 })
+  private async process(job: Job<MerkleDistributorBlocksEventsQueueMessage>) {
+    const { chainId, from, to } = job.data;
+    const claimedEvents: ClaimedEvent[] = await this.providers.getMerkleDistributorQuerier().getClaimedEvents(from, to);
+    this.logger.log(`(${from}, ${to}) - chainId ${chainId} - found ${claimedEvents.length} ClaimedEvent`);
+
+    for (const event of claimedEvents) {
+      try {
+        const claim = await this.fromClaimedEventToClaim(event, chainId);
+        await this.claimRepository.insert(claim);
+      } catch (error) {
+        if (error instanceof QueryFailedError && error.driverError?.code === "23505") {
+          // Ignore duplicate key value violates unique constraint error.
+          this.logger.warn(error);
+        } else {
+          throw error;
+        }
+      }
+    }
+  }
+
+  private async fromClaimedEventToClaim(event: ClaimedEvent, chainId: number) {
+    const { blockNumber } = event;
+    const { caller, accountIndex, windowIndex, account, rewardToken } = event.args;
+    const blockTimestamp = (await this.providers.getCachedBlock(chainId, blockNumber)).date;
+
+    return this.claimRepository.create({
+      caller,
+      accountIndex: accountIndex.toNumber(),
+      windowIndex: windowIndex.toNumber(),
+      account: utils.getAddress(account),
+      rewardToken: utils.getAddress(rewardToken),
+      blockNumber: blockNumber,
+      claimedAt: blockTimestamp,
+    });
+  }
+
+  @OnQueueFailed()
+  private onQueueFailed(job: Job, error: Error) {
+    this.logger.error(`${ScraperQueue.MerkleDistributorBlocksEvents} ${JSON.stringify(job.data)} failed: ${error}`);
+  }
+}

src/modules/scraper/adapter/messaging/index.ts

@@ -6,6 +6,7 @@ export enum ScraperQueue {
   DepositReferral = "DepositReferral",
   TokenPrice = "TokenPrice",
   DepositFilledDate = "DepositFilledDate",
+  MerkleDistributorBlocksEvents = "MerkleDistributorBlocksEvents",
 }
 
 export type BlocksEventsQueueMessage = {
@@ -14,6 +15,12 @@ export type BlocksEventsQueueMessage = {
   to: number;
 };
 
+export type MerkleDistributorBlocksEventsQueueMessage = {
+  chainId: number;
+  from: number;
+  to: number;
+};
+
 export type FillEventsQueueMessage = {
   realizedLpFeePct: string;
   originChainId: number;

src/modules/scraper/entry-point/http/controller.ts

@@ -5,6 +5,7 @@ import { JwtAuthGuard } from "../../../auth/entry-points/http/jwt.guard";
 import { Role, Roles, RolesGuard } from "../../../auth/entry-points/http/roles";
 import {
   BlocksEventsQueueMessage,
+  MerkleDistributorBlocksEventsQueueMessage,
   DepositFilledDateQueueMessage,
   DepositReferralQueueMessage,
   ScraperQueue,
@@ -19,6 +20,9 @@ export class ScraperController {
 
   @Post("scraper/blocks")
   @ApiTags("scraper")
+  @Roles(Role.Admin)
+  @UseGuards(JwtAuthGuard, RolesGuard)
+  @ApiBearerAuth()
   async processBlocks(@Req() req: Request, @Body() body: ProcessBlocksBody) {
     const { chainId, from, to } = body;
     await this.scraperQueuesService.publishMessage<BlocksEventsQueueMessage>(ScraperQueue.BlocksEvents, {
@@ -28,6 +32,23 @@ export class ScraperController {
     });
   }
 
+  @Post("scraper/blocks/merkle-distributor")
+  @ApiTags("scraper")
+  @Roles(Role.Admin)
+  @UseGuards(JwtAuthGuard, RolesGuard)
+  @ApiBearerAuth()
+  async processMerkleDistributorBlocks(@Req() req: Request, @Body() body: ProcessBlocksBody) {
+    const { chainId, from, to } = body;
+    await this.scraperQueuesService.publishMessage<MerkleDistributorBlocksEventsQueueMessage>(
+      ScraperQueue.MerkleDistributorBlocksEvents,
+      {
+        chainId,
+        from,
+        to,
+      },
+    );
+  }
+
   @Post("scraper/prices")
   @ApiTags("scraper")
   async submitPricesJobs(@Body() body: ProcessPricesBody) {

src/modules/scraper/model/MerkleDistributorProcessedBlock.entity.ts

@@ -0,0 +1,16 @@
+import { Column, CreateDateColumn, Entity, PrimaryGeneratedColumn } from "typeorm";
+
+@Entity()
+export class MerkleDistributorProcessedBlock {
+  @PrimaryGeneratedColumn()
+  id: number;
+
+  @Column()
+  chainId: number;
+
+  @Column()
+  latestBlock: number;
+
+  @CreateDateColumn()
+  createdAt: Date;
+}

src/modules/scraper/model/claim.entity.ts

@@ -0,0 +1,49 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  Index,
+  PrimaryGeneratedColumn,
+  Unique,
+  UpdateDateColumn,
+  ManyToOne,
+} from "typeorm";
+import { MerkleDistributorWindow } from "../../airdrop/model/merkle-distributor-window.entity";
+
+@Entity()
+@Unique("UK_claim_windowIndex_accountIndex", ["windowIndex", "accountIndex"])
+@Index("IX_claim_account", ["account"])
+export class Claim {
+  @PrimaryGeneratedColumn()
+  id: number;
+
+  @Column()
+  caller: string;
+
+  @Column()
+  accountIndex: number;
+
+  @Column()
+  windowIndex: number;
+
+  @Column()
+  account: string;
+
+  @Column()
+  rewardToken: string;
+
+  @Column()
+  blockNumber: number;
+
+  @Column()
+  claimedAt: Date;
+
+  @ManyToOne(() => MerkleDistributorWindow, (window) => window.claims)
+  merkleDistributorWindow: MerkleDistributorWindow;
+
+  @CreateDateColumn()
+  createdAt: Date;
+
+  @UpdateDateColumn()
+  updatedAt: Date;
+}