Feat/dvmd dt event handlers (#17)

# 🤖 Linear

Closes GIT-63

## Description

- `Registered` and `Distributed` event handlers for Direct Voting Merkle
Distribution Direct Transfer strategy
- write an abstract class for Distributed since the logic is not
attached to any strategy in particular, so it's reusable

## Checklist before requesting a review

-   [x] I have conducted a self-review of my code.
-   [x] I have conducted a QA.
-   [x] If it is a core feature, I have included comprehensive tests.

packages/processors/src/exceptions/index.ts

@@ -2,3 +2,5 @@ export * from "./tokenPriceNotFound.exception.js";
 export * from "./unsupportedEvent.exception.js";
 export * from "./invalidArgument.exception.js";
 export * from "./unsupportedStrategy.exception.js";
+export * from "./projectNotFound.exception.js";
+export * from "./roundNotFound.exception.js";

packages/processors/src/exceptions/projectNotFound.exception.ts

@@ -0,0 +1,7 @@
+import { ChainId } from "@grants-stack-indexer/shared";
+
+export class ProjectNotFound extends Error {
+    constructor(chainId: ChainId, anchorAddress: string) {
+        super(`Project not found for chainId: ${chainId} and anchorAddress: ${anchorAddress}`);
+    }
+}

packages/processors/src/exceptions/roundNotFound.exception.ts

@@ -0,0 +1,7 @@
+import { ChainId } from "@grants-stack-indexer/shared";
+
+export class RoundNotFound extends Error {
+    constructor(chainId: ChainId, strategyAddress: string) {
+        super(`Round not found for chainId: ${chainId} and strategyAddress: ${strategyAddress}`);
+    }
+}

packages/processors/src/internal.ts

@@ -3,4 +3,5 @@ export * from "./types/index.js";
 export * from "./interfaces/index.js";
 export * from "./exceptions/index.js";
 export * from "./allo/index.js";
+export * from "./strategy/common/index.js";
 export * from "./strategy/index.js";

packages/processors/src/strategy/common/baseDistributed.handler.ts

@@ -0,0 +1,53 @@
+import { getAddress } from "viem";
+
+import { Changeset } from "@grants-stack-indexer/repository";
+import { ChainId, ProtocolEvent } from "@grants-stack-indexer/shared";
+
+import { IEventHandler, ProcessorDependencies } from "../../internal.js";
+
+type Dependencies = Pick<ProcessorDependencies, "roundRepository">;
+
+/**
+ * BaseDistributedHandler: Processes 'Distributed' events
+ *
+ * - Handles distribution events across all strategies.
+ * - Creates a changeset to increment the total distributed amount for a round.
+ * - Serves as a base class as all strategies share the same logic for this event.
+ *
+ * @dev:
+ * - Strategy handlers that want to handle the Distributed event should create an instance of this class corresponding to the event.
+ *
+ */
+
+export class BaseDistributedHandler implements IEventHandler<"Strategy", "Distributed"> {
+    constructor(
+        readonly event: ProtocolEvent<"Strategy", "Distributed">,
+        private readonly chainId: ChainId,
+        private readonly dependencies: Dependencies,
+    ) {}
+
+    async handle(): Promise<Changeset[]> {
+        const { roundRepository } = this.dependencies;
+        const strategyAddress = getAddress(this.event.srcAddress);
+        const round = await roundRepository.getRoundByStrategyAddress(
+            this.chainId,
+            strategyAddress,
+        );
+
+        if (!round) {
+            //TODO: add logging that round was not found
+            return [];
+        }
+
+        return [
+            {
+                type: "IncrementRoundTotalDistributed",
+                args: {
+                    chainId: this.chainId,
+                    roundId: round.id,
+                    amount: BigInt(this.event.params.amount),
+                },
+            },
+        ];
+    }
+}

packages/processors/src/strategy/common/index.ts

@@ -0,0 +1 @@
+export * from "./baseDistributed.handler.js";

packages/processors/src/strategy/donationVotingMerkleDistributionDirectTransfer/dvmdDirectTransfer.handler.ts

@@ -2,8 +2,8 @@ import { Changeset } from "@grants-stack-indexer/repository";
 import { ChainId, ProtocolEvent, StrategyEvent } from "@grants-stack-indexer/shared";
 
 import type { IStrategyHandler, ProcessorDependencies } from "../../internal.js";
-import { UnsupportedEventException } from "../../internal.js";
-import { DVMDDistributedHandler, DVMDRegisteredHandler } from "./handlers/index.js";
+import { BaseDistributedHandler, UnsupportedEventException } from "../../internal.js";
+import { DVMDRegisteredHandler } from "./handlers/index.js";
 
 type Dependencies = Pick<
     ProcessorDependencies,
@@ -33,7 +33,7 @@ export class DVMDDirectTransferHandler implements IStrategyHandler<StrategyEvent
                     this.dependencies,
                 ).handle();
             case "Distributed":
-                return new DVMDDistributedHandler(
+                return new BaseDistributedHandler(
                     event as ProtocolEvent<"Strategy", "Distributed">,
                     this.chainId,
                     this.dependencies,

packages/processors/src/strategy/donationVotingMerkleDistributionDirectTransfer/handlers/index.ts

@@ -1,2 +1 @@
-export * from "./distributed.handler.js";
 export * from "./registered.handler.js";

packages/processors/src/strategy/donationVotingMerkleDistributionDirectTransfer/handlers/registered.handler.ts

@@ -1,7 +1,15 @@
-import { Changeset } from "@grants-stack-indexer/repository";
+import { getAddress } from "viem";
+
+import { Changeset, NewApplication } from "@grants-stack-indexer/repository";
 import { ChainId, ProtocolEvent } from "@grants-stack-indexer/shared";
 
-import { IEventHandler, ProcessorDependencies } from "../../../internal.js";
+import {
+    IEventHandler,
+    ProcessorDependencies,
+    ProjectNotFound,
+    RoundNotFound,
+} from "../../../internal.js";
+import { decodeDVMDApplicationData } from "../helpers/decoder.js";
 
 type Dependencies = Pick<
     ProcessorDependencies,
@@ -16,7 +24,64 @@ export class DVMDRegisteredHandler implements IEventHandler<"Strategy", "Registe
     ) {}
 
     async handle(): Promise<Changeset[]> {
-        //TODO: Implement
-        throw new Error("Not implemented");
+        const { projectRepository, roundRepository, metadataProvider } = this.dependencies;
+        const { data: encodedData, recipientId, sender } = this.event.params;
+        const { blockNumber, blockTimestamp } = this.event;
+
+        const anchorAddress = getAddress(recipientId);
+        const project = await projectRepository.getProjectByAnchor(this.chainId, anchorAddress);
+
+        if (!project) {
+            throw new ProjectNotFound(this.chainId, anchorAddress);
+        }
+
+        const strategyAddress = getAddress(this.event.srcAddress);
+        const round = await roundRepository.getRoundByStrategyAddress(
+            this.chainId,
+            strategyAddress,
+        );
+
+        if (!round) {
+            throw new RoundNotFound(this.chainId, strategyAddress);
+        }
+
+        const values = decodeDVMDApplicationData(encodedData);
+        // ID is defined as recipientsCounter - 1, which is a value emitted by the strategy
+        const id = (Number(values.recipientsCounter) - 1).toString();
+
+        const metadata = await metadataProvider.getMetadata(values.metadata.pointer);
+
+        const application: NewApplication = {
+            chainId: this.chainId,
+            id: id,
+            projectId: project.id,
+            anchorAddress,
+            roundId: round.id,
+            status: "PENDING",
+            metadataCid: values.metadata.pointer,
+            metadata: metadata ?? null,
+            createdAtBlock: BigInt(blockNumber),
+            createdByAddress: getAddress(sender),
+            statusUpdatedAtBlock: BigInt(blockNumber),
+            statusSnapshots: [
+                {
+                    status: "PENDING",
+                    updatedAtBlock: blockNumber.toString(),
+                    updatedAt: new Date(blockTimestamp * 1000), // timestamp is in seconds, convert to ms
+                },
+            ],
+            distributionTransaction: null,
+            totalAmountDonatedInUsd: 0,
+            totalDonationsCount: 0,
+            uniqueDonorsCount: 0,
+            tags: ["allo-v2"],
+        };
+
+        return [
+            {
+                type: "InsertApplication",
+                args: application,
+            },
+        ];
     }
 }

packages/processors/src/strategy/donationVotingMerkleDistributionDirectTransfer/helpers/decoder.ts

@@ -0,0 +1,41 @@
+import { decodeAbiParameters, Hex } from "viem";
+
+import { Address } from "@grants-stack-indexer/shared";
+
+import { DVMDApplicationData } from "../types/index.js";
+
+const DVMD_EVENT_DATA_DECODER = [
+    { name: "data", type: "bytes" },
+    { name: "recipientsCounter", type: "uint256" },
+] as const;
+
+const DVMD_DATA_DECODER = [
+    { name: "registryAnchor", type: "address" },
+    { name: "recipientAddress", type: "address" },
+    {
+        name: "metadata",
+        type: "tuple",
+        components: [
+            { name: "protocol", type: "uint256" },
+            { name: "pointer", type: "string" },
+        ],
+    },
+] as const;
+
+export const decodeDVMDApplicationData = (encodedData: Hex): DVMDApplicationData => {
+    const values = decodeAbiParameters(DVMD_EVENT_DATA_DECODER, encodedData);
+
+    const decodedData = decodeAbiParameters(DVMD_DATA_DECODER, values[0]);
+
+    const results: DVMDApplicationData = {
+        recipientsCounter: values[1].toString(),
+        anchorAddress: decodedData[0] as Address,
+        recipientAddress: decodedData[1] as Address,
+        metadata: {
+            protocol: Number(decodedData[2].protocol),
+            pointer: decodedData[2].pointer,
+        },
+    };
+
+    return results;
+};

packages/processors/src/strategy/donationVotingMerkleDistributionDirectTransfer/types/index.ts

@@ -0,0 +1,11 @@
+import { Address } from "@grants-stack-indexer/shared";
+
+export type DVMDApplicationData = {
+    recipientsCounter: string;
+    anchorAddress: Address;
+    recipientAddress: Address;
+    metadata: {
+        protocol: number;
+        pointer: string;
+    };
+};

packages/processors/test/strategy/common/baseDistributed.handler.spec.ts

@@ -0,0 +1,85 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { IRoundReadRepository, Round } from "@grants-stack-indexer/repository";
+import { ChainId, ProtocolEvent } from "@grants-stack-indexer/shared";
+
+import { BaseDistributedHandler } from "../../../src/strategy/common/baseDistributed.handler.js";
+
+function createMockEvent(
+    overrides: Partial<ProtocolEvent<"Strategy", "Distributed">> = {},
+): ProtocolEvent<"Strategy", "Distributed"> {
+    const defaultEvent: ProtocolEvent<"Strategy", "Distributed"> = {
+        params: {
+            amount: 1000,
+            recipientAddress: "0x1234567890123456789012345678901234567890",
+            recipientId: "0x1234567890123456789012345678901234567890",
+            sender: "0x1234567890123456789012345678901234567890",
+        },
+        eventName: "Distributed",
+        srcAddress: "0x1234567890123456789012345678901234567890",
+        blockNumber: 12345,
+        blockTimestamp: 1000000000,
+        chainId: 10 as ChainId,
+        contractName: "Strategy",
+        logIndex: 1,
+        transactionFields: {
+            hash: "0xd2352acdcd59e312370831ea927d51a1917654697a72434cd905a60897a5bb8b",
+            transactionIndex: 6,
+            from: "0xcBf407C33d68a55CB594Ffc8f4fD1416Bba39DA5",
+        },
+        strategyId: "0x9fa6890423649187b1f0e8bf4265f0305ce99523c3d11aa36b35a54617bb0ec0",
+    };
+
+    return { ...defaultEvent, ...overrides };
+}
+
+describe("BaseDistributedHandler", () => {
+    let handler: BaseDistributedHandler;
+    let mockRoundRepository: IRoundReadRepository;
+    let mockEvent: ProtocolEvent<"Strategy", "Distributed">;
+    const chainId = 10 as ChainId;
+
+    beforeEach(() => {
+        mockRoundRepository = {
+            getRoundByStrategyAddress: vi.fn(),
+        } as unknown as IRoundReadRepository;
+    });
+
+    it("increment round total distributed when round is found", async () => {
+        mockEvent = createMockEvent();
+        const mockRound = { id: "round1" } as Round;
+
+        vi.spyOn(mockRoundRepository, "getRoundByStrategyAddress").mockResolvedValue(mockRound);
+
+        handler = new BaseDistributedHandler(mockEvent, chainId, {
+            roundRepository: mockRoundRepository,
+        });
+
+        const result = await handler.handle();
+
+        expect(result).toEqual([
+            {
+                type: "IncrementRoundTotalDistributed",
+                args: {
+                    chainId,
+                    roundId: "round1",
+                    amount: BigInt(mockEvent.params.amount),
+                },
+            },
+        ]);
+    });
+
+    it("returns an empty array when round is not found", async () => {
+        mockEvent = createMockEvent();
+
+        vi.spyOn(mockRoundRepository, "getRoundByStrategyAddress").mockResolvedValue(undefined);
+
+        handler = new BaseDistributedHandler(mockEvent, chainId, {
+            roundRepository: mockRoundRepository,
+        });
+
+        const result = await handler.handle();
+
+        expect(result).toEqual([]);
+    });
+});