feat: remaining events orchestrator impl

packages/data-flow/src/abis/index.ts

@@ -0,0 +1 @@
+export * from "./strategy.abi.js";

packages/data-flow/src/abis/strategy.abi.ts

@@ -0,0 +1,225 @@
+export const iStrategyAbi = [
+    {
+        type: "function",
+        name: "allocate",
+        inputs: [
+            { name: "_data", type: "bytes", internalType: "bytes" },
+            { name: "_sender", type: "address", internalType: "address" },
+        ],
+        outputs: [],
+        stateMutability: "payable",
+    },
+    {
+        type: "function",
+        name: "distribute",
+        inputs: [
+            { name: "_recipientIds", type: "address[]", internalType: "address[]" },
+            { name: "_data", type: "bytes", internalType: "bytes" },
+            { name: "_sender", type: "address", internalType: "address" },
+        ],
+        outputs: [],
+        stateMutability: "nonpayable",
+    },
+    {
+        type: "function",
+        name: "getAllo",
+        inputs: [],
+        outputs: [{ name: "", type: "address", internalType: "contract IAllo" }],
+        stateMutability: "view",
+    },
+    {
+        type: "function",
+        name: "getPayouts",
+        inputs: [
+            { name: "_recipientIds", type: "address[]", internalType: "address[]" },
+            { name: "_data", type: "bytes[]", internalType: "bytes[]" },
+        ],
+        outputs: [
+            {
+                name: "",
+                type: "tuple[]",
+                internalType: "struct IStrategy.PayoutSummary[]",
+                components: [
+                    {
+                        name: "recipientAddress",
+                        type: "address",
+                        internalType: "address",
+                    },
+                    { name: "amount", type: "uint256", internalType: "uint256" },
+                ],
+            },
+        ],
+        stateMutability: "view",
+    },
+    {
+        type: "function",
+        name: "getPoolAmount",
+        inputs: [],
+        outputs: [{ name: "", type: "uint256", internalType: "uint256" }],
+        stateMutability: "view",
+    },
+    {
+        type: "function",
+        name: "getPoolId",
+        inputs: [],
+        outputs: [{ name: "", type: "uint256", internalType: "uint256" }],
+        stateMutability: "view",
+    },
+    {
+        type: "function",
+        name: "getRecipientStatus",
+        inputs: [{ name: "_recipientId", type: "address", internalType: "address" }],
+        outputs: [{ name: "", type: "uint8", internalType: "enum IStrategy.Status" }],
+        stateMutability: "view",
+    },
+    {
+        type: "function",
+        name: "getStrategyId",
+        inputs: [],
+        outputs: [{ name: "", type: "bytes32", internalType: "bytes32" }],
+        stateMutability: "view",
+    },
+    {
+        type: "function",
+        name: "increasePoolAmount",
+        inputs: [{ name: "_amount", type: "uint256", internalType: "uint256" }],
+        outputs: [],
+        stateMutability: "nonpayable",
+    },
+    {
+        type: "function",
+        name: "initialize",
+        inputs: [
+            { name: "_poolId", type: "uint256", internalType: "uint256" },
+            { name: "_data", type: "bytes", internalType: "bytes" },
+        ],
+        outputs: [],
+        stateMutability: "nonpayable",
+    },
+    {
+        type: "function",
+        name: "isPoolActive",
+        inputs: [],
+        outputs: [{ name: "", type: "bool", internalType: "bool" }],
+        stateMutability: "nonpayable",
+    },
+    {
+        type: "function",
+        name: "isValidAllocator",
+        inputs: [{ name: "_allocator", type: "address", internalType: "address" }],
+        outputs: [{ name: "", type: "bool", internalType: "bool" }],
+        stateMutability: "view",
+    },
+    {
+        type: "function",
+        name: "registerRecipient",
+        inputs: [
+            { name: "_data", type: "bytes", internalType: "bytes" },
+            { name: "_sender", type: "address", internalType: "address" },
+        ],
+        outputs: [{ name: "", type: "address", internalType: "address" }],
+        stateMutability: "payable",
+    },
+    {
+        type: "event",
+        name: "Allocated",
+        inputs: [
+            {
+                name: "recipientId",
+                type: "address",
+                indexed: true,
+                internalType: "address",
+            },
+            {
+                name: "amount",
+                type: "uint256",
+                indexed: false,
+                internalType: "uint256",
+            },
+            {
+                name: "token",
+                type: "address",
+                indexed: false,
+                internalType: "address",
+            },
+            {
+                name: "sender",
+                type: "address",
+                indexed: false,
+                internalType: "address",
+            },
+        ],
+        anonymous: false,
+    },
+    {
+        type: "event",
+        name: "Distributed",
+        inputs: [
+            {
+                name: "recipientId",
+                type: "address",
+                indexed: true,
+                internalType: "address",
+            },
+            {
+                name: "recipientAddress",
+                type: "address",
+                indexed: false,
+                internalType: "address",
+            },
+            {
+                name: "amount",
+                type: "uint256",
+                indexed: false,
+                internalType: "uint256",
+            },
+            {
+                name: "sender",
+                type: "address",
+                indexed: false,
+                internalType: "address",
+            },
+        ],
+        anonymous: false,
+    },
+    {
+        type: "event",
+        name: "Initialized",
+        inputs: [
+            {
+                name: "poolId",
+                type: "uint256",
+                indexed: false,
+                internalType: "uint256",
+            },
+            { name: "data", type: "bytes", indexed: false, internalType: "bytes" },
+        ],
+        anonymous: false,
+    },
+    {
+        type: "event",
+        name: "PoolActive",
+        inputs: [{ name: "active", type: "bool", indexed: false, internalType: "bool" }],
+        anonymous: false,
+    },
+    {
+        type: "event",
+        name: "Registered",
+        inputs: [
+            {
+                name: "recipientId",
+                type: "address",
+                indexed: true,
+                internalType: "address",
+            },
+            { name: "data", type: "bytes", indexed: false, internalType: "bytes" },
+            {
+                name: "sender",
+                type: "address",
+                indexed: false,
+                internalType: "address",
+            },
+        ],
+        anonymous: false,
+    },
+] as const;

packages/data-flow/src/eventsProcessor.ts

@@ -1,13 +1,10 @@
+import type { Changeset } from "@grants-stack-indexer/repository";
 import {
     AlloProcessor,
     ProcessorDependencies,
     RegistryProcessor,
     StrategyProcessor,
 } from "@grants-stack-indexer/processors";
-import { Changeset } from "@grants-stack-indexer/repository";
-
-import "@grants-stack-indexer/processors/dist/src/internal.js";
-
 import {
     AnyEvent,
     ChainId,
@@ -20,6 +17,11 @@ import {
 
 import { InvalidEvent } from "./exceptions/index.js";
 
+/**
+ * EventsProcessor handles the processing of Allo V2 events by delegating them to the appropriate processor
+ * (Allo, Registry, or Strategy) based on the event type. Each processor generates changesets that represent
+ * the required database updates.
+ */
 export class EventsProcessor {
     alloProcessor: AlloProcessor;
     registryProcessor: RegistryProcessor;
@@ -31,6 +33,12 @@ export class EventsProcessor {
         this.strategyProcessor = new StrategyProcessor(chainId, dependencies);
     }
 
+    /**
+     * Process an Allo V2 event and return the changesets
+     * @param event - The event to process
+     * @returns The changesets
+     * @throws InvalidEvent if the event is not a valid Allo V2 event (Allo, Registry or Strategy)
+     */
     public async processEvent(event: ProcessorEvent<ContractName, AnyEvent>): Promise<Changeset[]> {
         if (isAlloEvent(event)) {
             return await this.alloProcessor.process(event);

packages/data-flow/src/eventsRegistry.ts

@@ -1,17 +1,18 @@
-import { AnyEvent, ContractName, ProcessorEvent } from "@grants-stack-indexer/shared";
+import type { AnyEvent, ContractName, ProcessorEvent } from "@grants-stack-indexer/shared";
+
+import type { IEventsRegistry } from "./internal.js";
 
 /**
  * Class to store the last processed event
  */
-export class EventsRegistry {
-    //TODO: Implement storage to persist the last processed event
-    private lastProcessedEvent: ProcessorEvent<ContractName, AnyEvent> | null = null;
-
-    constructor() {}
+export class InMemoryEventsRegistry implements IEventsRegistry {
+    //TODO: Implement storage to persist the last processed event. we need to store it by chainId
+    private lastProcessedEvent: ProcessorEvent<ContractName, AnyEvent> | undefined;
 
-    async getLastProcessedEvent(): Promise<ProcessorEvent<ContractName, AnyEvent> | null> {
+    async getLastProcessedEvent(): Promise<ProcessorEvent<ContractName, AnyEvent> | undefined> {
         return this.lastProcessedEvent;
     }
+
     async saveLastProcessedEvent(event: ProcessorEvent<ContractName, AnyEvent>): Promise<void> {
         this.lastProcessedEvent = event;
     }

packages/data-flow/src/interfaces/eventsRegistry.interface.ts

@@ -0,0 +1,6 @@
+import { AnyEvent, ContractName, ProcessorEvent } from "@grants-stack-indexer/shared";
+
+export interface IEventsRegistry {
+    getLastProcessedEvent(): Promise<ProcessorEvent<ContractName, AnyEvent> | undefined>;
+    saveLastProcessedEvent(event: ProcessorEvent<ContractName, AnyEvent>): Promise<void>;
+}

packages/data-flow/src/interfaces/index.ts

@@ -1,2 +1,4 @@
 export * from "./eventsFetcher.interface.js";
 export * from "./dataLoader.interface.js";
+export * from "./eventsRegistry.interface.js";
+export * from "./strategyRegistry.interface.js";

packages/data-flow/src/interfaces/strategyRegistry.interface.ts

@@ -0,0 +1,6 @@
+import { Address, Hex } from "viem";
+
+export interface IStrategyRegistry {
+    getStrategyId(strategyAddress: Address): Promise<Hex | undefined>;
+    saveStrategyId(strategyAddress: Address, strategyId: Hex): Promise<void>;
+}

packages/data-flow/src/internal.ts

@@ -1,5 +1,7 @@
 export * from "./types/index.js";
 export * from "./interfaces/index.js";
 export * from "./exceptions/index.js";
+export * from "./abis/index.js";
+export * from "./utils/index.js";
 export * from "./data-loader/index.js";
 export * from "./eventsFetcher.js";