feat: add handler for RF distribution updated (#116)

# 🤖 Linear

Closes PAR-XXX

## Description

## 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.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Enhanced the distribution update process with an improved event
handling mechanism.
- Introduced a simplified distribution data format for more accurate
round updates.
  - Added a new handler for processing "DistributionUpdated" events.
- Expanded type definitions to accommodate new distribution structures.

- **Refactor**
  - Upgraded the existing event handler for more reliable processing.
  - Streamlined distribution type definitions to ensure consistency.

- **Tests**
- Added comprehensive testing to verify successful outcomes and robust
error handling during distribution updates.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: 0xKurt

packages/data-flow/test/integration/strategy.integration.spec.ts

@@ -209,30 +209,14 @@ describe("Orchestrator Integration - Strategy Events Processing", () => {
             "fetchEventsByBlockNumberAndLogIndex",
         );
 
-        const mockDistribution = {
-            matchingDistribution: [
-                {
-                    contributionsCount: 44,
-                    projectPayoutAddress: "0x7340F1a1e4e38F43d2FCC85cdb2b764de36B40c0",
-                    applicationId: "3",
-                    matchPoolPercentage: 0.099999,
-                    projectId: "0x15c5e4db5530e05216abc9484025e2f1c4fb55b8525d29ef38fde237e767e324",
-                    projectName: "ReFi DAO - A Network Society to Regenerate Earth. ✨ 🌱",
-                    matchAmountInToken: "999999999999999475712",
-                    originalMatchAmountInToken: "999999999999999475712",
-                },
-                {
-                    contributionsCount: 69,
-                    projectPayoutAddress: "0x01d1909cA27E364904934849eab8399532dd5c8b",
-                    applicationId: "11",
-                    matchPoolPercentage: 0.099999,
-                    projectId: "0xca460772f5ba0840a589d2c19fb7e17ac259e4be884313e3712c06c9c885dc93",
-                    projectName: "Giveth",
-                    matchAmountInToken: "999999999999999475712",
-                    originalMatchAmountInToken: "999999999999999475712",
-                },
-            ],
-        };
+        const mockDistribution = [
+            {
+                anchorAddress: "0x7340F1a1e4e38F43d2FCC85cdb2b764de36B40c0",
+                payoutAddress: "0x7340F1a1e4e38F43d2FCC85cdb2b764de36B40c0",
+                amount: "1000000000000000000",
+                index: 0,
+            },
+        ];
 
         vi.spyOn(metadataProvider, "getMetadata").mockResolvedValue(mockDistribution);
 
@@ -254,7 +238,7 @@ describe("Orchestrator Integration - Strategy Events Processing", () => {
             { chainId, strategyAddress: distributionUpdatedEvent.srcAddress },
             {
                 readyForPayoutTransaction: distributionUpdatedEvent.transactionFields.hash,
-                matchingDistribution: mockDistribution.matchingDistribution,
+                matchingDistribution: mockDistribution,
             },
             {},
         );

packages/processors/src/processors/strategy/easyRetroFunding/easyRetroFunding.handler.ts

@@ -23,12 +23,12 @@ import {
     UnsupportedEventException,
 } from "../../../internal.js";
 import {
-    BaseDistributionUpdatedHandler,
     BaseFundsDistributedHandler,
     BaseRecipientStatusUpdatedHandler,
     BaseStrategyHandler,
 } from "../common/index.js";
 import {
+    ERFDistributionUpdatedHandler,
     ERFRegisteredHandler,
     ERFTimestampsUpdatedHandler,
     ERFUpdatedRegistrationHandler,
@@ -86,7 +86,7 @@ export class EasyRetroFundingStrategyHandler extends BaseStrategyHandler {
                     this.dependencies,
                 ).handle();
             case "DistributionUpdated":
-                return new BaseDistributionUpdatedHandler(
+                return new ERFDistributionUpdatedHandler(
                     event as ProcessorEvent<"Strategy", "DistributionUpdated">,
                     this.chainId,
                     this.dependencies,

packages/processors/src/processors/strategy/easyRetroFunding/handlers/distributionUpdated.handler.ts

@@ -0,0 +1,79 @@
+import { getAddress } from "viem";
+
+import { Changeset } from "@grants-stack-indexer/repository";
+import { ChainId, ProcessorEvent } from "@grants-stack-indexer/shared";
+
+import {
+    IEventHandler,
+    MetadataNotFound,
+    MetadataParsingFailed,
+    ProcessorDependencies,
+} from "../../../../internal.js";
+import {
+    SimpleMatchingDistribution,
+    SimpleMatchingDistributionSchema,
+} from "../../../../schemas/index.js";
+
+type Dependencies = Pick<ProcessorDependencies, "metadataProvider" | "logger">;
+
+/**
+ * ERFDistributionUpdatedHandler: Processes 'DistributionUpdated' events
+ *
+ * - Decodes the updated distribution metadata
+ * - Creates a changeset to update the round with the new distribution
+ *
+ * @dev:
+ * - Strategy handlers that want to handle the DistributionUpdated event should create an instance of this class corresponding to the event.
+ *
+ */
+
+export class ERFDistributionUpdatedHandler
+    implements IEventHandler<"Strategy", "DistributionUpdated">
+{
+    constructor(
+        readonly event: ProcessorEvent<"Strategy", "DistributionUpdated">,
+        private readonly chainId: ChainId,
+        private readonly dependencies: Dependencies,
+    ) {}
+
+    /* @inheritdoc */
+    async handle(): Promise<Changeset[]> {
+        const { logger, metadataProvider } = this.dependencies;
+        const [_, pointer] = this.event.params.metadata;
+
+        const strategyAddress = getAddress(this.event.srcAddress);
+        const rawDistribution = await metadataProvider.getMetadata<
+            SimpleMatchingDistribution | undefined
+        >(pointer);
+
+        if (!rawDistribution) {
+            logger.warn(`No matching distribution found for pointer: ${pointer}`);
+
+            throw new MetadataNotFound(`No matching distribution found for pointer: ${pointer}`);
+        }
+
+        const distribution = SimpleMatchingDistributionSchema.safeParse(rawDistribution);
+
+        if (!distribution.success) {
+            logger.warn(`Failed to parse matching distribution: ${distribution.error.message}`);
+
+            throw new MetadataParsingFailed(
+                `Failed to parse matching distribution: ${distribution.error.message}`,
+            );
+        }
+
+        return [
+            {
+                type: "UpdateRoundByStrategyAddress",
+                args: {
+                    chainId: this.chainId,
+                    strategyAddress,
+                    round: {
+                        readyForPayoutTransaction: this.event.transactionFields.hash,
+                        matchingDistribution: distribution.data,
+                    },
+                },
+            },
+        ];
+    }
+}

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

@@ -1,3 +1,4 @@
 export * from "./registered.handler.js";
 export * from "./timestampsUpdated.handler.js";
 export * from "./updatedRegistration.handler.js";
+export * from "./distributionUpdated.handler.js";

packages/processors/src/schemas/matchingDistribution.ts

@@ -1,6 +1,7 @@
 import { z } from "zod";
 
 export type MatchingDistribution = z.infer<typeof MatchingDistributionSchema>;
+export type SimpleMatchingDistribution = z.infer<typeof SimpleMatchingDistributionSchema>;
 
 const BigIntSchema = z.string().or(
     z.object({ type: z.literal("BigNumber"), hex: z.string() }).transform((val) => {
@@ -22,3 +23,12 @@ export const MatchingDistributionSchema = z.object({
         }),
     ),
 });
+
+export const SimpleMatchingDistributionSchema = z.array(
+    z.object({
+        anchorAddress: z.string(),
+        payoutAddress: z.string(),
+        amount: BigIntSchema.default("0"),
+        index: z.coerce.number(),
+    }),
+);

packages/processors/test/strategy/easyRetroFunding/easyRetroFunding.handler.spec.ts

@@ -12,12 +12,12 @@ import { ChainId, ILogger, ProcessorEvent, StrategyEvent } from "@grants-stack-i
 
 import { UnsupportedEventException } from "../../../src/internal.js";
 import {
-    BaseDistributionUpdatedHandler,
     BaseFundsDistributedHandler,
     BaseRecipientStatusUpdatedHandler,
 } from "../../../src/processors/strategy/common/index.js";
 import { EasyRetroFundingStrategyHandler } from "../../../src/processors/strategy/easyRetroFunding/easyRetroFunding.handler.js";
 import {
+    ERFDistributionUpdatedHandler,
     ERFRegisteredHandler,
     ERFTimestampsUpdatedHandler,
     ERFUpdatedRegistrationHandler,
@@ -27,38 +27,37 @@ vi.mock("../../../src/processors/strategy/easyRetroFunding/handlers/index.js", a
     const ERFRegisteredHandler = vi.fn();
     const ERFTimestampsUpdatedHandler = vi.fn();
     const ERFUpdatedRegistrationHandler = vi.fn();
-
+    const ERFDistributionUpdatedHandler = vi.fn();
     // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
     ERFRegisteredHandler.prototype.handle = vi.fn();
     // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
     ERFTimestampsUpdatedHandler.prototype.handle = vi.fn();
     // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
     ERFUpdatedRegistrationHandler.prototype.handle = vi.fn();
     // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
+    ERFDistributionUpdatedHandler.prototype.handle = vi.fn();
+
     return {
         ERFRegisteredHandler,
         ERFTimestampsUpdatedHandler,
         ERFUpdatedRegistrationHandler,
+        ERFDistributionUpdatedHandler,
     };
 });
 
 vi.mock("../../../src/processors/strategy/common/index.js", async (importOriginal) => {
     const original =
         await importOriginal<typeof import("../../../src/processors/strategy/common/index.js")>();
     const BaseFundsDistributedHandler = vi.fn();
-    const BaseDistributionUpdatedHandler = vi.fn();
     const BaseRecipientStatusUpdatedHandler = vi.fn();
 
     // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
     BaseFundsDistributedHandler.prototype.handle = vi.fn();
     // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-    BaseDistributionUpdatedHandler.prototype.handle = vi.fn();
-    // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
     BaseRecipientStatusUpdatedHandler.prototype.handle = vi.fn();
     return {
         ...original,
         BaseFundsDistributedHandler,
-        BaseDistributionUpdatedHandler,
         BaseRecipientStatusUpdatedHandler,
     };
 });
@@ -197,16 +196,16 @@ describe("EasyRetroFundingStrategyHandler", () => {
         expect(ERFTimestampsUpdatedHandler.prototype.handle).toHaveBeenCalled();
     });
 
-    it("calls BaseDistributionUpdatedHandler for DistributionUpdated event", async () => {
+    it("calls ERFDistributionUpdatedHandler for DistributionUpdated event", async () => {
         const mockEvent = {
             eventName: "DistributionUpdated",
         } as ProcessorEvent<"Strategy", "DistributionUpdated">;
 
-        vi.spyOn(BaseDistributionUpdatedHandler.prototype, "handle").mockResolvedValue([]);
+        vi.spyOn(ERFDistributionUpdatedHandler.prototype, "handle").mockResolvedValue([]);
 
         await handler.handle(mockEvent);
 
-        expect(BaseDistributionUpdatedHandler).toHaveBeenCalledWith(mockEvent, chainId, {
+        expect(ERFDistributionUpdatedHandler).toHaveBeenCalledWith(mockEvent, chainId, {
             metadataProvider: mockMetadataProvider,
             roundRepository: mockRoundRepository,
             projectRepository: mockProjectRepository,
@@ -215,7 +214,7 @@ describe("EasyRetroFundingStrategyHandler", () => {
             applicationRepository: mockApplicationRepository,
             logger,
         });
-        expect(BaseDistributionUpdatedHandler.prototype.handle).toHaveBeenCalled();
+        expect(ERFDistributionUpdatedHandler.prototype.handle).toHaveBeenCalled();
     });
 
     it("calls FundsDistributedHandler for FundsDistributed event", async () => {

packages/processors/test/strategy/easyRetroFunding/handlers/distributionUpdated.handler.spec.ts

@@ -0,0 +1,133 @@
+import { getAddress } from "viem";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { IMetadataProvider } from "@grants-stack-indexer/metadata";
+import { PartialRound } from "@grants-stack-indexer/repository";
+import { Bytes32String, ChainId, Logger, ProcessorEvent } from "@grants-stack-indexer/shared";
+
+import { MetadataNotFound, MetadataParsingFailed } from "../../../../src/internal.js";
+import { ERFDistributionUpdatedHandler } from "../../../../src/processors/strategy/easyRetroFunding/index.js";
+import { createMockEvent } from "../../../mocks/index.js";
+
+describe("ERFDistributionUpdatedHandler", () => {
+    let handler: ERFDistributionUpdatedHandler;
+    let mockMetadataProvider: IMetadataProvider;
+    let mockLogger: Logger;
+    let mockEvent: ProcessorEvent<"Strategy", "DistributionUpdated">;
+    const chainId = 10 as ChainId;
+    const eventName = "DistributionUpdated";
+    const defaultParams = {
+        metadata: ["1", "ipfs://QmTestHash"] as [string, string],
+        merkleRoot: "0xroot" as Bytes32String,
+    };
+    const defaultStrategyId = "0x9fa6890423649187b1f0e8bf4265f0305ce99523c3d11aa36b35a54617bb0ec0";
+
+    beforeEach(() => {
+        mockMetadataProvider = {
+            getMetadata: vi.fn(),
+        } as unknown as IMetadataProvider;
+        mockLogger = {
+            warn: vi.fn(),
+            error: vi.fn(),
+            info: vi.fn(),
+            debug: vi.fn(),
+        } as unknown as Logger;
+    });
+
+    it("handles a valid distribution update event", async () => {
+        mockEvent = createMockEvent(eventName, defaultParams, defaultStrategyId);
+        const mockDistribution = [
+            {
+                anchorAddress: "anchorAddress",
+                payoutAddress: "payoutAddress",
+                amount: "10",
+                index: 0,
+            },
+        ];
+
+        vi.spyOn(mockMetadataProvider, "getMetadata").mockResolvedValue(mockDistribution);
+
+        handler = new ERFDistributionUpdatedHandler(mockEvent, chainId, {
+            metadataProvider: mockMetadataProvider,
+            logger: mockLogger,
+        });
+
+        const result = await handler.handle();
+
+        expect(result).toEqual([
+            {
+                type: "UpdateRoundByStrategyAddress",
+                args: {
+                    chainId,
+                    strategyAddress: getAddress(mockEvent.srcAddress),
+                    round: {
+                        readyForPayoutTransaction: mockEvent.transactionFields.hash,
+                        matchingDistribution: mockDistribution,
+                    },
+                },
+            },
+        ]);
+    });
+
+    it("throws MetadataNotFound if distribution metadata is not found", async () => {
+        mockEvent = createMockEvent(eventName, defaultParams, defaultStrategyId);
+        vi.spyOn(mockMetadataProvider, "getMetadata").mockResolvedValue(undefined);
+
+        handler = new ERFDistributionUpdatedHandler(mockEvent, chainId, {
+            metadataProvider: mockMetadataProvider,
+            logger: mockLogger,
+        });
+
+        await expect(handler.handle()).rejects.toThrow(MetadataNotFound);
+        expect(mockLogger.warn).toHaveBeenCalledWith(
+            expect.stringContaining("No matching distribution found for pointer:"),
+        );
+    });
+
+    it("throw MatchingDistributionParsingError if distribution format is invalid", async () => {
+        mockEvent = createMockEvent(eventName, defaultParams, defaultStrategyId);
+        const invalidDistribution = {
+            matchingDistribution: [
+                {
+                    amount: "not_a_number", // Invalid amount format
+                    applicationId: "app1",
+                    recipientAddress: "0x1234567890123456789012345678901234567890",
+                },
+            ],
+        };
+
+        vi.spyOn(mockMetadataProvider, "getMetadata").mockResolvedValue(invalidDistribution);
+
+        handler = new ERFDistributionUpdatedHandler(mockEvent, chainId, {
+            metadataProvider: mockMetadataProvider,
+            logger: mockLogger,
+        });
+
+        await expect(handler.handle()).rejects.toThrow(MetadataParsingFailed);
+        expect(mockLogger.warn).toHaveBeenCalledWith(
+            expect.stringContaining("Failed to parse matching distribution:"),
+        );
+    });
+
+    it("handles empty matching distribution array", async () => {
+        mockEvent = createMockEvent(eventName, defaultParams, defaultStrategyId);
+
+        vi.spyOn(mockMetadataProvider, "getMetadata").mockResolvedValue([]);
+
+        handler = new ERFDistributionUpdatedHandler(mockEvent, chainId, {
+            metadataProvider: mockMetadataProvider,
+            logger: mockLogger,
+        });
+
+        const result = await handler.handle();
+        expect(result).toHaveLength(1);
+
+        const changeset = result[0] as {
+            type: "UpdateRoundByStrategyAddress";
+            args: {
+                round: PartialRound;
+            };
+        };
+        expect(changeset.args.round.matchingDistribution).toEqual([]);
+    });
+});