feat: persisting strategy registry into db

apps/processing/src/index.ts

@@ -6,7 +6,7 @@ import { ProcessingService } from "./services/processing.service.js";
 let processor: ProcessingService;
 
 const main = async (): Promise<void> => {
-    processor = new ProcessingService(environment);
+    processor = await ProcessingService.initialize(environment);
     await processor.start();
 };
 

apps/processing/src/services/processing.service.ts

@@ -23,10 +23,9 @@ export class ProcessingService {
     private readonly logger = new Logger({ className: "ProcessingService" });
     private readonly kyselyDatabase: SharedDependencies["kyselyDatabase"];
 
-    constructor(private readonly env: Environment) {
+    private constructor(env: Environment, sharedDependencies: SharedDependencies) {
         const { CHAINS: chains } = env;
-        const { core, registries, indexerClient, kyselyDatabase } =
-            SharedDependenciesService.initialize(env);
+        const { core, registries, indexerClient, kyselyDatabase } = sharedDependencies;
         this.kyselyDatabase = kyselyDatabase;
 
         for (const chain of chains) {
@@ -49,6 +48,11 @@ export class ProcessingService {
         }
     }
 
+    static async initialize(env: Environment): Promise<ProcessingService> {
+        const sharedDependencies = await SharedDependenciesService.initialize(env);
+        return new ProcessingService(env, sharedDependencies);
+    }
+
     /**
      * Start the processor service
      *

apps/processing/src/services/sharedDependencies.service.ts

@@ -1,7 +1,10 @@
 import {
     CoreDependencies,
+    DatabaseStrategyRegistry,
+    IEventsRegistry,
+    InMemoryCachedStrategyRegistry,
     InMemoryEventsRegistry,
-    InMemoryStrategyRegistry,
+    IStrategyRegistry,
 } from "@grants-stack-indexer/data-flow";
 import { EnvioIndexerClient } from "@grants-stack-indexer/indexer-client";
 import { IpfsProvider } from "@grants-stack-indexer/metadata";
@@ -13,6 +16,7 @@ import {
     KyselyDonationRepository,
     KyselyProjectRepository,
     KyselyRoundRepository,
+    KyselyStrategyRegistryRepository,
 } from "@grants-stack-indexer/repository";
 import { Logger } from "@grants-stack-indexer/shared";
 
@@ -21,8 +25,8 @@ import { Environment } from "../config/index.js";
 export type SharedDependencies = {
     core: Omit<CoreDependencies, "evmProvider">;
     registries: {
-        eventsRegistry: InMemoryEventsRegistry;
-        strategyRegistry: InMemoryStrategyRegistry;
+        eventsRegistry: IEventsRegistry;
+        strategyRegistry: IStrategyRegistry;
     };
     indexerClient: EnvioIndexerClient;
     kyselyDatabase: ReturnType<typeof createKyselyDatabase>;
@@ -35,7 +39,7 @@ export type SharedDependencies = {
  * - Initializes indexer client
  */
 export class SharedDependenciesService {
-    static initialize(env: Environment): SharedDependencies {
+    static async initialize(env: Environment): Promise<SharedDependencies> {
         // Initialize repositories
         const kyselyDatabase = createKyselyDatabase({
             connectionString: env.DATABASE_URL,
@@ -68,8 +72,16 @@ export class SharedDependenciesService {
         const eventsRegistry = new InMemoryEventsRegistry(
             new Logger({ className: "InMemoryEventsRegistry" }),
         );
-        const strategyRegistry = new InMemoryStrategyRegistry(
-            new Logger({ className: "InMemoryStrategyRegistry" }),
+        const strategyRepository = new KyselyStrategyRegistryRepository(
+            kyselyDatabase,
+            env.DATABASE_SCHEMA,
+        );
+        const strategyRegistry = await InMemoryCachedStrategyRegistry.initialize(
+            new Logger({ className: "InMemoryCachedStrategyRegistry" }),
+            new DatabaseStrategyRegistry(
+                new Logger({ className: "DatabaseStrategyRegistry" }),
+                strategyRepository,
+            ),
         );
 
         // Initialize indexer client

apps/processing/test/unit/processing.service.spec.ts

@@ -52,9 +52,9 @@ describe("ProcessingService", () => {
         DATABASE_SCHEMA: "public",
     };
 
-    beforeEach(() => {
+    beforeEach(async () => {
         vi.clearAllMocks();
-        processingService = new ProcessingService(mockEnv as Environment);
+        processingService = await ProcessingService.initialize(mockEnv as Environment);
     });
 
     afterEach(() => {

apps/processing/test/unit/sharedDependencies.service.spec.ts

@@ -17,6 +17,11 @@ vi.mock("@grants-stack-indexer/repository", () => ({
     KyselyApplicationRepository: vi.fn(),
     KyselyDonationRepository: vi.fn(),
     KyselyApplicationPayoutRepository: vi.fn(),
+    KyselyStrategyRegistryRepository: vi.fn().mockImplementation(() => ({
+        getStrategies: vi.fn().mockResolvedValue([]),
+        getStrategyId: vi.fn(),
+        saveStrategyId: vi.fn(),
+    })),
 }));
 
 vi.mock("@grants-stack-indexer/pricing", () => ({
@@ -33,6 +38,26 @@ vi.mock("@grants-stack-indexer/indexer-client", () => ({
     EnvioIndexerClient: vi.fn(),
 }));
 
+vi.mock("@grants-stack-indexer/data-flow", () => {
+    const mockStrategyRegistry = {
+        getStrategies: vi.fn(),
+        getStrategyId: vi.fn(),
+        saveStrategyId: vi.fn(),
+    };
+
+    return {
+        InMemoryEventsRegistry: vi.fn(),
+        InMemoryCachedStrategyRegistry: {
+            initialize: vi.fn().mockResolvedValue(mockStrategyRegistry),
+        },
+        DatabaseStrategyRegistry: vi.fn().mockImplementation(() => ({
+            getStrategies: vi.fn(),
+            getStrategyId: vi.fn(),
+            saveStrategyId: vi.fn(),
+        })),
+    };
+});
+
 describe("SharedDependenciesService", () => {
     const mockEnv: Pick<
         Environment,
@@ -51,8 +76,8 @@ describe("SharedDependenciesService", () => {
         PRICING_SOURCE: "dummy",
     };
 
-    it("initializes all dependencies correctly", () => {
-        const dependencies = SharedDependenciesService.initialize(mockEnv as Environment);
+    it("initializes all dependencies correctly", async () => {
+        const dependencies = await SharedDependenciesService.initialize(mockEnv as Environment);
 
         // Verify database initialization
         expect(createKyselyDatabase).toHaveBeenCalledWith({
@@ -89,5 +114,8 @@ describe("SharedDependenciesService", () => {
         // Verify registries
         expect(dependencies.registries).toHaveProperty("eventsRegistry");
         expect(dependencies.registries).toHaveProperty("strategyRegistry");
+
+        // Verify InMemoryCachedStrategyRegistry initialization
+        expect(dependencies.registries.strategyRegistry).toBeDefined();
     });
 });

package.json

@@ -10,15 +10,15 @@
         "build": "turbo run build",
         "check-types": "turbo run check-types",
         "clean": "turbo run clean",
+        "db:migrate": "pnpm run --filter @grants-stack-indexer/migrations db:migrate",
+        "db:reset": "pnpm run --filter @grants-stack-indexer/migrations db:reset",
         "dev": "turbo run dev",
         "format": "turbo run format",
         "format:fix": "turbo run format:fix",
         "preinstall": "npx only-allow pnpm",
         "lint": "turbo run lint",
         "lint:fix": "turbo run lint:fix",
         "prepare": "husky",
-        "script:db:migrate": "pnpm run --filter @grants-stack-indexer/migrations script:db:migrate",
-        "script:db:reset": "pnpm run --filter @grants-stack-indexer/migrations script:db:reset",
         "start": "turbo run start",
         "test": "turbo run test",
         "test:cov": "turbo run test:cov",

packages/data-flow/README.md

@@ -90,9 +90,13 @@ The `EventsProcessor` class is responsible for processing events in the processi
 
 The `EventsFetcher` class is responsible for fetching events from the blockchain.
 
-### [StrategyRegistry](./src/strategyRegistry.ts)
+### [StrategyRegistry](./src/registries/)
 
 The `StrategyRegistry` stores strategy IDs to populate strategy events with them given the Strategy address.
+There are 3 implementations:
+
+-   `DatabaseStrategyRegistry`: persists data to database using IStrategyRepository
+-   `InMemoryCachedStrategyRegistry`: stores map in-memory as cache and persists to database
 
 ### [DataLoader](./src/data-loader/dataLoader.ts)
 

packages/data-flow/src/external.ts

@@ -1,7 +1,8 @@
 export {
     DataLoader,
     InMemoryEventsRegistry,
-    InMemoryStrategyRegistry,
+    InMemoryCachedStrategyRegistry,
+    DatabaseStrategyRegistry,
     Orchestrator,
 } from "./internal.js";
 

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

@@ -1,22 +1,38 @@
 import { Address, Hex } from "viem";
 
+import { Strategy } from "@grants-stack-indexer/repository";
+import { ChainId } from "@grants-stack-indexer/shared";
+
 /**
  * The strategy registry saves the mapping between the strategy address and the strategy id. Serves as a Cache
  * to avoid having to read from the chain to get the strategy id every time.
  */
-//TODO: implement a mechanism to record Strategy that we still don't have a corresponding handler
-// we need to store and mark that this strategy is not handled yet, so when it's supported we can process all of the pending events for it
 export interface IStrategyRegistry {
     /**
-     * Get the strategy id by the strategy address
+     * Get the strategy id by the strategy address and chain id
+     *
+     * @param chainId - The chain id
      * @param strategyAddress - The strategy address
-     * @returns The strategy id or undefined if the strategy address is not registered
+     * @returns The strategy or undefined if the strategy address is not registered
      */
-    getStrategyId(strategyAddress: Address): Promise<Hex | undefined>;
+    getStrategyId(chainId: ChainId, strategyAddress: Address): Promise<Strategy | undefined>;
     /**
-     * Save the strategy id by the strategy address
+     * Save the strategy id by the strategy address and chain id
+     * @param chainId - The chain id
      * @param strategyAddress - The strategy address
      * @param strategyId - The strategy id
+     * @param handled - Whether the strategy is handled
+     */
+    saveStrategyId(
+        chainId: ChainId,
+        strategyAddress: Address,
+        strategyId: Hex,
+        handled: boolean,
+    ): Promise<void>;
+
+    /**
+     * Get all the strategies
+     * @returns The strategies
      */
-    saveStrategyId(strategyAddress: Address, strategyId: Hex): Promise<void>;
+    getStrategies(filters?: { handled?: boolean; chainId?: ChainId }): Promise<Strategy[]>;
 }

packages/data-flow/src/internal.ts

@@ -5,7 +5,7 @@ export * from "./abis/index.js";
 export * from "./utils/index.js";
 export * from "./data-loader/index.js";
 export * from "./eventsFetcher.js";
-export * from "./strategyRegistry.js";
+export * from "./registries/index.js";
 export * from "./eventsRegistry.js";
 export * from "./eventsProcessor.js";
 export * from "./orchestrator.js";

packages/data-flow/src/orchestrator.ts

@@ -113,15 +113,17 @@ export class Orchestrator {
                 await this.eventsRegistry.saveLastProcessedEvent(this.chainId, event);
 
                 event = await this.enhanceStrategyId(event);
-                if (event.contractName === "Strategy" && "strategyId" in event) {
+                if (this.isPoolCreated(event)) {
+                    const handleable = existsHandler(event.strategyId);
+                    await this.strategyRegistry.saveStrategyId(
+                        this.chainId,
+                        event.srcAddress,
+                        event.strategyId,
+                        handleable,
+                    );
+                } else if (event.contractName === "Strategy" && "strategyId" in event) {
                     if (!existsHandler(event.strategyId)) {
-                        //TODO: save to registry as unsupported strategy, so when the strategy is handled it will be backwards compatible and process all of the events
-                        //TODO: decide if we want to log this
-                        // this.logger.info(
-                        //     `No handler found for strategyId: ${event.strategyId}. Event: ${stringify(
-                        //         event,
-                        //     )}`,
-                        // );
+                        // we skip the event if the strategy id is not handled yet
                         continue;
                     }
                 }
@@ -216,9 +218,12 @@ export class Orchestrator {
      * @returns The strategy id
      */
     private async getOrFetchStrategyId(strategyAddress: Address): Promise<Hex> {
-        const existingId = await this.strategyRegistry.getStrategyId(strategyAddress);
-        if (existingId) {
-            return existingId;
+        const cachedStrategy = await this.strategyRegistry.getStrategyId(
+            this.chainId,
+            strategyAddress,
+        );
+        if (cachedStrategy) {
+            return cachedStrategy.id;
         }
 
         const strategyId = await this.dependencies.evmProvider.readContract(
@@ -227,11 +232,20 @@ export class Orchestrator {
             "getStrategyId",
         );
 
-        await this.strategyRegistry.saveStrategyId(strategyAddress, strategyId);
-
         return strategyId;
     }
 
+    /**
+     * Check if the event is a PoolCreated event from Allo contract
+     * @param event - The event
+     * @returns True if the event is a PoolCreated event from Allo contract, false otherwise
+     */
+    private isPoolCreated(
+        event: ProcessorEvent<ContractName, AnyEvent>,
+    ): event is ProcessorEvent<"Allo", "PoolCreated"> {
+        return isAlloEvent(event) && event.eventName === "PoolCreated";
+    }
+
     /**
      * Check if the event requires a strategy id
      * @param event - The event
@@ -240,6 +254,6 @@ export class Orchestrator {
     private requiresStrategyId(
         event: ProcessorEvent<ContractName, AnyEvent>,
     ): event is ProcessorEvent<"Allo", "PoolCreated"> | ProcessorEvent<"Strategy", StrategyEvent> {
-        return (isAlloEvent(event) && event.eventName === "PoolCreated") || isStrategyEvent(event);
+        return this.isPoolCreated(event) || isStrategyEvent(event);
     }
 }

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

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