feat: retroactively handle strategy

apps/processing/README.md

@@ -55,6 +55,7 @@ Available scripts that can be run using `pnpm`:
 | `start`       | Run the compiled app from dist folder                   |
 | `test`        | Run tests using vitest                                  |
 | `test:cov`    | Run tests with coverage report                          |
+| `retroactive` | Run retroactive processing for all chains               |
 
 TODO: e2e tests
 TODO: Docker image

apps/processing/package.json

@@ -13,6 +13,7 @@
         "format:fix": "prettier --write \"{src,test}/**/*.{js,ts,json}\"",
         "lint": "eslint \"{src,test}/**/*.{js,ts,json}\"",
         "lint:fix": "pnpm lint --fix",
+        "retroactive": "tsx src/retroactiveHandleStrategies.ts",
         "start": "node dist/index.js",
         "test": "vitest run --config vitest.config.ts --passWithNoTests",
         "test:cov": "vitest run --config vitest.config.ts --coverage --passWithNoTests"

apps/processing/src/retroactiveHandleStrategies.ts

@@ -0,0 +1,33 @@
+import { inspect } from "util";
+
+import { environment } from "./config/index.js";
+import { ProcessingService } from "./services/processing.service.js";
+
+let processor: ProcessingService;
+
+const main = async (): Promise<void> => {
+    processor = await ProcessingService.initialize(environment);
+    await processor.processRetroactiveEvents();
+};
+
+process.on("unhandledRejection", (reason, p) => {
+    console.error(`Unhandled Rejection at: \n${inspect(p, undefined, 100)}, \nreason: ${reason}`);
+    process.exit(1);
+});
+
+process.on("uncaughtException", (error: Error) => {
+    console.error(
+        `An uncaught exception occurred: ${error}\n` + `Exception origin: ${error.stack}`,
+    );
+    process.exit(1);
+});
+
+main()
+    .catch((err) => {
+        console.error(`Caught error in main handler: ${err}`);
+        process.exit(1);
+    })
+    // eslint-disable-next-line @typescript-eslint/no-misused-promises
+    .finally(async () => {
+        await processor?.releaseResources();
+    });

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

@@ -7,6 +7,7 @@ import {
     InMemoryCachedEventRegistry,
     InMemoryCachedStrategyRegistry,
     Orchestrator,
+    RetroactiveProcessor,
 } from "@grants-stack-indexer/data-flow";
 import { ChainId, Logger } from "@grants-stack-indexer/shared";
 
@@ -27,12 +28,12 @@ import { SharedDependencies, SharedDependenciesService } from "./index.js";
  * - Manages graceful shutdown on termination signals
  */
 export class ProcessingService {
-    private readonly orchestrators: Map<ChainId, Orchestrator> = new Map();
+    private readonly orchestrators: Map<ChainId, [Orchestrator, RetroactiveProcessor]> = new Map();
     private readonly logger = new Logger({ className: "ProcessingService" });
     private readonly kyselyDatabase: SharedDependencies["kyselyDatabase"];
 
     private constructor(
-        orchestrators: Map<ChainId, Orchestrator>,
+        orchestrators: Map<ChainId, [Orchestrator, RetroactiveProcessor]>,
         kyselyDatabase: SharedDependencies["kyselyDatabase"],
     ) {
         this.orchestrators = orchestrators;
@@ -43,8 +44,12 @@ export class ProcessingService {
         const sharedDependencies = await SharedDependenciesService.initialize(env);
         const { CHAINS: chains } = env;
         const { core, registriesRepositories, indexerClient, kyselyDatabase } = sharedDependencies;
-        const { eventRegistryRepository, strategyRegistryRepository } = registriesRepositories;
-        const orchestrators: Map<ChainId, Orchestrator> = new Map();
+        const {
+            eventRegistryRepository,
+            strategyRegistryRepository,
+            strategyProcessingCheckpointRepository,
+        } = registriesRepositories;
+        const orchestrators: Map<ChainId, [Orchestrator, RetroactiveProcessor]> = new Map();
 
         const strategyRegistry = new DatabaseStrategyRegistry(
             new Logger({ className: "DatabaseStrategyRegistry" }),
@@ -72,21 +77,32 @@ export class ProcessingService {
                 chain.id as ChainId,
             );
 
-            orchestrators.set(
+            const orchestrator = new Orchestrator(
                 chain.id as ChainId,
-                new Orchestrator(
-                    chain.id as ChainId,
-                    { ...core, evmProvider },
-                    indexerClient,
-                    {
-                        eventsRegistry: cachedEventsRegistry,
-                        strategyRegistry: cachedStrategyRegistry,
-                    },
-                    chain.fetchLimit,
-                    chain.fetchDelayMs,
-                    chainLogger,
-                ),
+                { ...core, evmProvider },
+                indexerClient,
+                {
+                    eventsRegistry: cachedEventsRegistry,
+                    strategyRegistry: cachedStrategyRegistry,
+                },
+                chain.fetchLimit,
+                chain.fetchDelayMs,
+                chainLogger,
             );
+            const retroactiveProcessor = new RetroactiveProcessor(
+                chain.id as ChainId,
+                { ...core, evmProvider },
+                indexerClient,
+                {
+                    eventsRegistry: cachedEventsRegistry,
+                    strategyRegistry: cachedStrategyRegistry,
+                    checkpointRepository: strategyProcessingCheckpointRepository,
+                },
+                chain.fetchLimit,
+                chainLogger,
+            );
+
+            orchestrators.set(chain.id as ChainId, [orchestrator, retroactiveProcessor]);
         }
 
         return new ProcessingService(orchestrators, kyselyDatabase);
@@ -116,7 +132,7 @@ export class ProcessingService {
         });
 
         try {
-            for (const orchestrator of this.orchestrators.values()) {
+            for (const [orchestrator, _] of this.orchestrators.values()) {
                 this.logger.info(`Starting orchestrator for chain ${orchestrator.chainId}...`);
                 orchestratorProcesses.push(orchestrator.run(abortController.signal));
             }
@@ -128,6 +144,17 @@ export class ProcessingService {
         }
     }
 
+    /**
+     * Process retroactive events for all chains
+     * - This is a blocking operation that will run until all retroactive events are processed
+     */
+    async processRetroactiveEvents(): Promise<void> {
+        this.logger.info("Processing retroactive events...");
+        for (const [_, retroactiveProcessor] of this.orchestrators.values()) {
+            await retroactiveProcessor.processRetroactiveStrategies();
+        }
+    }
+
     /**
      * Call this function when the processor service is terminated
      * - Releases database resources

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

@@ -5,13 +5,15 @@ import { PricingProviderFactory } from "@grants-stack-indexer/pricing";
 import {
     createKyselyDatabase,
     IEventRegistryRepository,
+    IStrategyProcessingCheckpointRepository,
     IStrategyRegistryRepository,
     KyselyApplicationPayoutRepository,
     KyselyApplicationRepository,
     KyselyDonationRepository,
     KyselyEventRegistryRepository,
     KyselyProjectRepository,
     KyselyRoundRepository,
+    KyselyStrategyProcessingCheckpointRepository,
     KyselyStrategyRegistryRepository,
 } from "@grants-stack-indexer/repository";
 import { Logger } from "@grants-stack-indexer/shared";
@@ -23,6 +25,7 @@ export type SharedDependencies = {
     registriesRepositories: {
         eventRegistryRepository: IEventRegistryRepository;
         strategyRegistryRepository: IStrategyRegistryRepository;
+        strategyProcessingCheckpointRepository: IStrategyProcessingCheckpointRepository;
     };
     indexerClient: EnvioIndexerClient;
     kyselyDatabase: ReturnType<typeof createKyselyDatabase>;
@@ -73,6 +76,9 @@ export class SharedDependenciesService {
             env.DATABASE_SCHEMA,
         );
 
+        const strategyProcessingCheckpointRepository =
+            new KyselyStrategyProcessingCheckpointRepository(kyselyDatabase, env.DATABASE_SCHEMA);
+
         // Initialize indexer client
         const indexerClient = new EnvioIndexerClient(
             env.INDEXER_GRAPHQL_URL,
@@ -92,6 +98,7 @@ export class SharedDependenciesService {
             registriesRepositories: {
                 eventRegistryRepository,
                 strategyRegistryRepository,
+                strategyProcessingCheckpointRepository,
             },
             indexerClient,
             kyselyDatabase,