feat: make saveLastProcessedEvent part of transaction (#60)

# 🤖 Linear

Closes GIT-232

## Description
We want to make last processed event part of the transaction on the
happy path, so we increase reliability and error handling of the system.
For this we are:
- deprecating the InMemoryCache of EventRegistry
- work directly with IEventRegistryRepository implementations
- add optional txConnection to `saveLastProcessedEvent` method


## Checklist before requesting a review

-   [x] I have conducted a self-review of my code.
-   [x] I have conducted a QA.
-   [ ] 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

## Release Notes

- **New Features**
  - Enhanced event registry handling with improved transaction support.
- Added new processed event changeset type for more flexible event
management.
- Introduced processed event handlers for managing application-related
operations.

- **Improvements**
  - Simplified event processing logic in orchestrators.
  - Updated data loader to support more comprehensive event handling.
  - Improved error handling for event processing workflows.

- **Technical Updates**
  - Refactored event registry interfaces and implementations.
  - Updated type definitions for better type safety and flexibility.
- Expanded test coverage for processed event handlers and orchestrator
functionalities.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: 0xnigir1

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

@@ -4,7 +4,6 @@ import { EvmProvider } from "@grants-stack-indexer/chain-providers";
 import {
     DatabaseEventRegistry,
     DatabaseStrategyRegistry,
-    InMemoryCachedEventRegistry,
     InMemoryCachedStrategyRegistry,
     Orchestrator,
     RetroactiveProcessor,
@@ -67,12 +66,6 @@ export class ProcessingService {
             // Initialize EVM provider
             const evmProvider = new EvmProvider(chain.rpcUrls, optimism, logger);
 
-            // Initialize events registry for the chain
-            const cachedEventsRegistry = await InMemoryCachedEventRegistry.initialize(
-                logger,
-                eventsRegistry,
-                [chain.id as ChainId],
-            );
             const cachedStrategyRegistry = await InMemoryCachedStrategyRegistry.initialize(
                 logger,
                 strategyRegistry,
@@ -84,7 +77,7 @@ export class ProcessingService {
                 { ...core, evmProvider },
                 indexerClient,
                 {
-                    eventsRegistry: cachedEventsRegistry,
+                    eventsRegistry,
                     strategyRegistry: cachedStrategyRegistry,
                 },
                 chain.fetchLimit,
@@ -97,7 +90,7 @@ export class ProcessingService {
                 { ...core, evmProvider },
                 indexerClient,
                 {
-                    eventsRegistry: cachedEventsRegistry,
+                    eventsRegistry,
                     strategyRegistry: cachedStrategyRegistry,
                     checkpointRepository: strategyProcessingCheckpointRepository,
                 },

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

@@ -4,7 +4,6 @@ import { EvmProvider } from "@grants-stack-indexer/chain-providers";
 import {
     DatabaseEventRegistry,
     DatabaseStrategyRegistry,
-    InMemoryCachedEventRegistry,
     InMemoryCachedStrategyRegistry,
     Orchestrator,
     RetroactiveProcessor,
@@ -118,7 +117,6 @@ describe("ProcessingService", () => {
             expect(DatabaseEventRegistry).toHaveBeenCalledTimes(1);
             expect(EvmProvider).toHaveBeenCalledTimes(2);
             expect(InMemoryCachedStrategyRegistry.initialize).toHaveBeenCalledTimes(2);
-            expect(InMemoryCachedEventRegistry.initialize).toHaveBeenCalledTimes(2);
 
             // Verify orchestrators were created with correct parameters
             expect(processingService["orchestrators"].size).toBe(2);

packages/data-flow/src/data-loader/dataLoader.ts

@@ -3,6 +3,7 @@ import {
     IApplicationPayoutRepository,
     IApplicationRepository,
     IDonationRepository,
+    IEventRegistryRepository,
     IProjectRepository,
     IRoundRepository,
     ITransactionManager,
@@ -17,6 +18,7 @@ import {
     createProjectHandlers,
     createRoundHandlers,
 } from "./handlers/index.js";
+import { createProcessedEventHandlers } from "./handlers/processedEvent.handlers.js";
 import { ChangesetHandlers } from "./types/index.js";
 
 /**
@@ -42,6 +44,7 @@ export class DataLoader implements IDataLoader {
             application: IApplicationRepository;
             donation: IDonationRepository;
             applicationPayout: IApplicationPayoutRepository;
+            eventRegistry: IEventRegistryRepository;
         },
         private readonly transactionManager: ITransactionManager,
         private readonly logger: ILogger,
@@ -52,6 +55,7 @@ export class DataLoader implements IDataLoader {
             ...createApplicationHandlers(repositories.application),
             ...createDonationHandlers(repositories.donation),
             ...createApplicationPayoutHandlers(repositories.applicationPayout),
+            ...createProcessedEventHandlers(repositories.eventRegistry),
         };
     }
 

packages/data-flow/src/data-loader/handlers/index.ts

@@ -3,3 +3,4 @@ export * from "./project.handlers.js";
 export * from "./round.handlers.js";
 export * from "./donation.handlers.js";
 export * from "./applicationPayout.handlers.js";
+export * from "./processedEvent.handlers.js";

packages/data-flow/src/data-loader/handlers/processedEvent.handlers.ts

@@ -0,0 +1,32 @@
+import {
+    IEventRegistryRepository,
+    ProcessedEventChangeset,
+} from "@grants-stack-indexer/repository";
+
+import { ChangesetHandler } from "../types/index.js";
+
+/**
+ * Collection of handlers for application-related operations.
+ * Each handler corresponds to a specific Application changeset type.
+ */
+export type ProcessedEventHandlers = {
+    [K in ProcessedEventChangeset["type"]]: ChangesetHandler<K>;
+};
+
+/**
+ * Creates handlers for managing application-related operations.
+ *
+ * @param repository - The application repository instance used for database operations
+ * @returns An object containing all application-related handlers
+ */
+export const createProcessedEventHandlers = (
+    repository: IEventRegistryRepository,
+): ProcessedEventHandlers => ({
+    InsertProcessedEvent: (async (changeset, txConnection): Promise<void> => {
+        await repository.saveLastProcessedEvent(
+            changeset.args.chainId,
+            changeset.args.processedEvent,
+            txConnection,
+        );
+    }) satisfies ChangesetHandler<"InsertProcessedEvent">,
+});

packages/data-flow/src/orchestrator.ts

@@ -7,7 +7,12 @@ import {
     UnsupportedEventException,
     UnsupportedStrategy,
 } from "@grants-stack-indexer/processors";
-import { RoundNotFound, RoundNotFoundForId } from "@grants-stack-indexer/repository";
+import {
+    Changeset,
+    IEventRegistryRepository,
+    RoundNotFound,
+    RoundNotFoundForId,
+} from "@grants-stack-indexer/repository";
 import {
     Address,
     AnyEvent,
@@ -29,7 +34,7 @@ import {
     Token,
 } from "@grants-stack-indexer/shared";
 
-import type { IEventsFetcher, IEventsRegistry, IStrategyRegistry } from "./interfaces/index.js";
+import type { IEventsFetcher, IStrategyRegistry } from "./interfaces/index.js";
 import { EventsFetcher } from "./eventsFetcher.js";
 import { EventsProcessor } from "./eventsProcessor.js";
 import { InvalidEvent } from "./exceptions/index.js";
@@ -72,7 +77,7 @@ export class Orchestrator {
     private readonly eventsByBlockContext: Map<number, AnyIndexerFetchedEvent[]>;
     private readonly eventsFetcher: IEventsFetcher;
     private readonly eventsProcessor: EventsProcessor;
-    private readonly eventsRegistry: IEventsRegistry;
+    private readonly eventsRegistry: IEventRegistryRepository;
     private readonly strategyRegistry: IStrategyRegistry;
     private readonly dataLoader: DataLoader;
     private readonly retryHandler: RetryHandler;
@@ -91,7 +96,7 @@ export class Orchestrator {
         private dependencies: Readonly<CoreDependencies>,
         private indexerClient: IIndexerClient,
         private registries: {
-            eventsRegistry: IEventsRegistry;
+            eventsRegistry: IEventRegistryRepository;
             strategyRegistry: IStrategyRegistry;
         },
         private fetchLimit: number = 1000,
@@ -113,6 +118,7 @@ export class Orchestrator {
                 application: this.dependencies.applicationRepository,
                 donation: this.dependencies.donationRepository,
                 applicationPayout: this.dependencies.applicationPayoutRepository,
+                eventRegistry: this.eventsRegistry,
             },
             this.dependencies.transactionManager,
             this.logger,
@@ -146,18 +152,34 @@ export class Orchestrator {
                     continue;
                 }
 
-                await this.eventsRegistry.saveLastProcessedEvent(this.chainId, {
-                    ...event,
-                    rawEvent: event,
-                });
-
                 await this.retryHandler.execute(
                     async () => {
-                        await this.handleEvent(event!);
+                        const changesets = await this.handleEvent(event!);
+                        if (changesets) {
+                            await this.dataLoader.applyChanges([
+                                ...changesets,
+                                {
+                                    type: "InsertProcessedEvent",
+                                    args: {
+                                        chainId: this.chainId,
+                                        processedEvent: {
+                                            ...event!,
+                                            rawEvent: event,
+                                        },
+                                    },
+                                },
+                            ]);
+                        }
                     },
                     { abortSignal: signal },
                 );
             } catch (error: unknown) {
+                if (event) {
+                    await this.eventsRegistry.saveLastProcessedEvent(this.chainId, {
+                        ...event,
+                        rawEvent: event,
+                    });
+                }
                 // TODO: notify
                 if (
                     error instanceof UnsupportedStrategy ||
@@ -403,7 +425,9 @@ export class Orchestrator {
         return tokenPrices;
     }
 
-    private async handleEvent(event: ProcessorEvent<ContractName, AnyEvent>): Promise<void> {
+    private async handleEvent(
+        event: ProcessorEvent<ContractName, AnyEvent>,
+    ): Promise<Changeset[] | undefined> {
         event = await this.enhanceStrategyId(event);
         if (this.isPoolCreated(event)) {
             const handleable = existsHandler(event.strategyId);
@@ -421,12 +445,11 @@ export class Orchestrator {
                     chainId: this.chainId,
                 });
                 // we skip the event if the strategy id is not handled yet
-                return;
+                return undefined;
             }
         }
 
-        const changesets = await this.eventsProcessor.processEvent(event);
-        await this.dataLoader.applyChanges(changesets);
+        return this.eventsProcessor.processEvent(event);
     }
 
     /**

packages/data-flow/src/retroactiveProcessor.ts

@@ -1,6 +1,9 @@
 import { IIndexerClient } from "@grants-stack-indexer/indexer-client";
 import { existsHandler, UnsupportedEventException } from "@grants-stack-indexer/processors";
-import { IStrategyProcessingCheckpointRepository } from "@grants-stack-indexer/repository";
+import {
+    IEventRegistryRepository,
+    IStrategyProcessingCheckpointRepository,
+} from "@grants-stack-indexer/repository";
 import {
     Address,
     AnyEvent,
@@ -20,7 +23,6 @@ import {
     EventsFetcher,
     EventsProcessor,
     IEventsFetcher,
-    IEventsRegistry,
     InvalidEvent,
     IStrategyRegistry,
     Queue,
@@ -60,7 +62,7 @@ type EventPointer = {
 export class RetroactiveProcessor {
     private readonly eventsFetcher: IEventsFetcher;
     private readonly eventsProcessor: EventsProcessor;
-    private readonly eventsRegistry: IEventsRegistry;
+    private readonly eventsRegistry: IEventRegistryRepository;
     private readonly strategyRegistry: IStrategyRegistry;
     private readonly dataLoader: DataLoader;
     private readonly checkpointRepository: IStrategyProcessingCheckpointRepository;
@@ -81,7 +83,7 @@ export class RetroactiveProcessor {
         private dependencies: Readonly<CoreDependencies>,
         private indexerClient: IIndexerClient,
         private registries: {
-            eventsRegistry: IEventsRegistry;
+            eventsRegistry: IEventRegistryRepository;
             strategyRegistry: IStrategyRegistry;
             checkpointRepository: IStrategyProcessingCheckpointRepository;
         },
@@ -104,6 +106,7 @@ export class RetroactiveProcessor {
                 application: this.dependencies.applicationRepository,
                 donation: this.dependencies.donationRepository,
                 applicationPayout: this.dependencies.applicationPayoutRepository,
+                eventRegistry: this.eventsRegistry,
             },
             this.dependencies.transactionManager,
             this.logger,

packages/data-flow/test/data-loader/dataLoader.spec.ts

@@ -5,6 +5,7 @@ import {
     IApplicationPayoutRepository,
     IApplicationRepository,
     IDonationRepository,
+    IEventRegistryRepository,
     IProjectRepository,
     IRoundRepository,
     ITransactionManager,
@@ -41,6 +42,10 @@ describe("DataLoader", () => {
         insertApplicationPayout: vi.fn(),
     } as IApplicationPayoutRepository;
 
+    const mockEventRegistryRepository = {
+        saveLastProcessedEvent: vi.fn(),
+    } as unknown as IEventRegistryRepository;
+
     const logger: ILogger = {
         debug: vi.fn(),
         error: vi.fn(),
@@ -61,6 +66,7 @@ describe("DataLoader", () => {
                 application: mockApplicationRepository,
                 donation: mockDonationRepository,
                 applicationPayout: mockApplicationPayoutRepository,
+                eventRegistry: mockEventRegistryRepository,
             },
             mockTransactionManager,
             logger,

packages/data-flow/test/data-loader/handlers/processedEvent.handlers.spec.ts

@@ -0,0 +1,69 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { IEventRegistryRepository, TransactionConnection } from "@grants-stack-indexer/repository";
+import { ChainId } from "@grants-stack-indexer/shared";
+
+import { createProcessedEventHandlers } from "../../../src/data-loader/handlers/processedEvent.handlers.js";
+
+describe("ProcessedEvent Handlers", () => {
+    const chainId = 1 as ChainId;
+    const mockEvent = {
+        blockNumber: 1,
+        blockTimestamp: 1234567890,
+        logIndex: 0,
+        rawEvent: {},
+    };
+
+    describe("InsertProcessedEvent", () => {
+        it("saves event to repository within transaction", async () => {
+            const saveLastProcessedEvent = vi.fn();
+            const mockRepository = {
+                saveLastProcessedEvent,
+                getLastProcessedEvent: vi.fn(),
+            } as unknown as IEventRegistryRepository<TransactionConnection>;
+
+            const handlers = createProcessedEventHandlers(mockRepository);
+
+            const mockTx = {} as TransactionConnection;
+
+            await handlers.InsertProcessedEvent(
+                {
+                    type: "InsertProcessedEvent",
+                    args: {
+                        chainId,
+                        processedEvent: mockEvent,
+                    },
+                },
+                mockTx,
+            );
+
+            expect(saveLastProcessedEvent).toHaveBeenCalledWith(chainId, mockEvent, mockTx);
+        });
+
+        it("propagates repository errors", async () => {
+            const error = new Error("Database error");
+            const saveLastProcessedEvent = vi.fn().mockRejectedValue(error);
+            const mockRepository = {
+                saveLastProcessedEvent,
+                getLastProcessedEvent: vi.fn(),
+            } as unknown as IEventRegistryRepository<TransactionConnection>;
+
+            const handlers = createProcessedEventHandlers(mockRepository);
+
+            const mockTx = {} as TransactionConnection;
+
+            await expect(
+                handlers.InsertProcessedEvent(
+                    {
+                        type: "InsertProcessedEvent",
+                        args: {
+                            chainId,
+                            processedEvent: mockEvent,
+                        },
+                    },
+                    mockTx,
+                ),
+            ).rejects.toThrow(error);
+        });
+    });
+});