feat: dataLoader class

packages/data-flow/package.json

@@ -29,6 +29,7 @@
     },
     "dependencies": {
         "@grants-stack-indexer/indexer-client": "workspace:*",
+        "@grants-stack-indexer/repository": "workspace:*",
         "@grants-stack-indexer/shared": "workspace:*",
         "viem": "2.21.19"
     }

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

@@ -0,0 +1,82 @@
+import {
+    Changeset,
+    IApplicationRepository,
+    IProjectRepository,
+    IRoundRepository,
+} from "@grants-stack-indexer/repository";
+
+import { ExecutionResult, IDataLoader, InvalidChangeset } from "../internal.js";
+import {
+    createApplicationHandlers,
+    createProjectHandlers,
+    createRoundHandlers,
+} from "./handlers/index.js";
+import { ChangesetHandlers } from "./types/index.js";
+
+/**
+ * DataLoader is responsible for applying changesets to the database.
+ * It works by:
+ * 1. Taking an array of changesets representing data modifications
+ * 2. Validating that handlers exist for all changeset types
+ * 3. Sequentially executing each changeset using the appropriate handler
+ * 4. Tracking execution results including successes and failures
+ * 5. Breaking execution if any changeset fails
+ *
+ * The handlers are initialized for different entity types (projects, rounds, applications)
+ * and stored in a map for lookup during execution.
+ */
+
+export class DataLoader implements IDataLoader {
+    private readonly handlers: ChangesetHandlers;
+
+    constructor(
+        private readonly repositories: {
+            project: IProjectRepository;
+            round: IRoundRepository;
+            application: IApplicationRepository;
+        },
+    ) {
+        this.handlers = {
+            ...createProjectHandlers(repositories.project),
+            ...createRoundHandlers(repositories.round),
+            ...createApplicationHandlers(repositories.application),
+        };
+    }
+
+    /** @inheritdoc */
+    public async applyChanges(changesets: Changeset[]): Promise<ExecutionResult> {
+        const result: ExecutionResult = {
+            changesets: [],
+            numExecuted: 0,
+            numSuccessful: 0,
+            numFailed: 0,
+            errors: [],
+        };
+
+        const invalidTypes = changesets.filter((changeset) => !this.handlers[changeset.type]);
+        if (invalidTypes.length > 0) {
+            throw new InvalidChangeset(invalidTypes.map((changeset) => changeset.type));
+        }
+
+        //TODO: research how to manage transactions so we can rollback on error
+        for (const changeset of changesets) {
+            result.numExecuted++;
+            try {
+                await this.handlers[changeset.type](changeset as never);
+                result.changesets.push(changeset.type);
+                result.numSuccessful++;
+            } catch (error) {
+                result.numFailed++;
+                result.errors.push(
+                    `Failed to apply changeset ${changeset.type}: ${
+                        error instanceof Error ? error.message : String(error)
+                    }`,
+                );
+                console.error(error);
+                break;
+            }
+        }
+
+        return result;
+    }
+}

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

@@ -0,0 +1,31 @@
+import { IApplicationRepository } 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 ApplicationHandlers = {
+    InsertApplication: ChangesetHandler<"InsertApplication">;
+    UpdateApplication: ChangesetHandler<"UpdateApplication">;
+};
+
+/**
+ * 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 createApplicationHandlers = (
+    repository: IApplicationRepository,
+): ApplicationHandlers => ({
+    InsertApplication: (async (changeset): Promise<void> => {
+        await repository.insertApplication(changeset.args);
+    }) satisfies ChangesetHandler<"InsertApplication">,
+
+    UpdateApplication: (async (changeset): Promise<void> => {
+        const { chainId, roundId, applicationId, application } = changeset.args;
+        await repository.updateApplication({ chainId, roundId, id: applicationId }, application);
+    }) satisfies ChangesetHandler<"UpdateApplication">,
+});

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

@@ -0,0 +1,3 @@
+export * from "./application.handlers.js";
+export * from "./project.handlers.js";
+export * from "./round.handlers.js";

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

@@ -0,0 +1,60 @@
+import { IProjectRepository } from "@grants-stack-indexer/repository";
+
+import { ChangesetHandler } from "../types/index.js";
+
+/**
+ * Collection of handlers for project-related operations.
+ * Each handler corresponds to a specific Project changeset type.
+ */
+export type ProjectHandlers = {
+    InsertProject: ChangesetHandler<"InsertProject">;
+    UpdateProject: ChangesetHandler<"UpdateProject">;
+    InsertPendingProjectRole: ChangesetHandler<"InsertPendingProjectRole">;
+    DeletePendingProjectRoles: ChangesetHandler<"DeletePendingProjectRoles">;
+    InsertProjectRole: ChangesetHandler<"InsertProjectRole">;
+    DeleteAllProjectRolesByRole: ChangesetHandler<"DeleteAllProjectRolesByRole">;
+    DeleteAllProjectRolesByRoleAndAddress: ChangesetHandler<"DeleteAllProjectRolesByRoleAndAddress">;
+};
+
+/**
+ * Creates handlers for managing project-related operations.
+ *
+ * @param repository - The project repository instance used for database operations
+ * @returns An object containing all project-related handlers
+ */
+export const createProjectHandlers = (repository: IProjectRepository): ProjectHandlers => ({
+    InsertProject: (async (changeset): Promise<void> => {
+        const { project } = changeset.args;
+        await repository.insertProject(project);
+    }) satisfies ChangesetHandler<"InsertProject">,
+
+    UpdateProject: (async (changeset): Promise<void> => {
+        const { chainId, projectId, project } = changeset.args;
+        await repository.updateProject({ id: projectId, chainId }, project);
+    }) satisfies ChangesetHandler<"UpdateProject">,
+
+    InsertPendingProjectRole: (async (changeset): Promise<void> => {
+        const { pendingProjectRole } = changeset.args;
+        await repository.insertPendingProjectRole(pendingProjectRole);
+    }) satisfies ChangesetHandler<"InsertPendingProjectRole">,
+
+    DeletePendingProjectRoles: (async (changeset): Promise<void> => {
+        const { ids } = changeset.args;
+        await repository.deleteManyPendingProjectRoles(ids);
+    }) satisfies ChangesetHandler<"DeletePendingProjectRoles">,
+
+    InsertProjectRole: (async (changeset): Promise<void> => {
+        const { projectRole } = changeset.args;
+        await repository.insertProjectRole(projectRole);
+    }) satisfies ChangesetHandler<"InsertProjectRole">,
+
+    DeleteAllProjectRolesByRole: (async (changeset): Promise<void> => {
+        const { chainId, projectId, role } = changeset.args.projectRole;
+        await repository.deleteManyProjectRoles(chainId, projectId, role);
+    }) satisfies ChangesetHandler<"DeleteAllProjectRolesByRole">,
+
+    DeleteAllProjectRolesByRoleAndAddress: (async (changeset): Promise<void> => {
+        const { chainId, projectId, role, address } = changeset.args.projectRole;
+        await repository.deleteManyProjectRoles(chainId, projectId, role, address);
+    }) satisfies ChangesetHandler<"DeleteAllProjectRolesByRoleAndAddress">,
+});

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

@@ -0,0 +1,87 @@
+import { IRoundRepository } from "@grants-stack-indexer/repository";
+
+import { ChangesetHandler } from "../types/index.js";
+
+/**
+ * Collection of handlers for round-related operations.
+ * Each handler corresponds to a specific Round changeset type.
+ */
+export type RoundHandlers = {
+    InsertRound: ChangesetHandler<"InsertRound">;
+    UpdateRound: ChangesetHandler<"UpdateRound">;
+    UpdateRoundByStrategyAddress: ChangesetHandler<"UpdateRoundByStrategyAddress">;
+    IncrementRoundFundedAmount: ChangesetHandler<"IncrementRoundFundedAmount">;
+    IncrementRoundTotalDistributed: ChangesetHandler<"IncrementRoundTotalDistributed">;
+    InsertPendingRoundRole: ChangesetHandler<"InsertPendingRoundRole">;
+    DeletePendingRoundRoles: ChangesetHandler<"DeletePendingRoundRoles">;
+    InsertRoundRole: ChangesetHandler<"InsertRoundRole">;
+    DeleteAllRoundRolesByRoleAndAddress: ChangesetHandler<"DeleteAllRoundRolesByRoleAndAddress">;
+};
+
+/**
+ * Creates handlers for managing round-related operations.
+ *
+ * @param repository - The round repository instance used for database operations
+ * @returns An object containing all round-related handlers
+ */
+export const createRoundHandlers = (repository: IRoundRepository): RoundHandlers => ({
+    InsertRound: (async (changeset): Promise<void> => {
+        const { round } = changeset.args;
+        await repository.insertRound(round);
+    }) satisfies ChangesetHandler<"InsertRound">,
+
+    UpdateRound: (async (changeset): Promise<void> => {
+        const { chainId, roundId, round } = changeset.args;
+        await repository.updateRound({ id: roundId, chainId }, round);
+    }) satisfies ChangesetHandler<"UpdateRound">,
+
+    UpdateRoundByStrategyAddress: (async (changeset): Promise<void> => {
+        const { chainId, strategyAddress, round } = changeset.args;
+        if (round) {
+            await repository.updateRound({ strategyAddress, chainId: chainId }, round);
+        }
+    }) satisfies ChangesetHandler<"UpdateRoundByStrategyAddress">,
+
+    IncrementRoundFundedAmount: (async (changeset): Promise<void> => {
+        const { chainId, roundId, fundedAmount, fundedAmountInUsd } = changeset.args;
+        await repository.incrementRoundFunds(
+            {
+                chainId,
+                roundId,
+            },
+            fundedAmount,
+            fundedAmountInUsd,
+        );
+    }) satisfies ChangesetHandler<"IncrementRoundFundedAmount">,
+
+    IncrementRoundTotalDistributed: (async (changeset): Promise<void> => {
+        const { chainId, roundId, amount } = changeset.args;
+        await repository.incrementRoundTotalDistributed(
+            {
+                chainId,
+                roundId,
+            },
+            amount,
+        );
+    }) satisfies ChangesetHandler<"IncrementRoundTotalDistributed">,
+
+    InsertPendingRoundRole: (async (changeset): Promise<void> => {
+        const { pendingRoundRole } = changeset.args;
+        await repository.insertPendingRoundRole(pendingRoundRole);
+    }) satisfies ChangesetHandler<"InsertPendingRoundRole">,
+
+    DeletePendingRoundRoles: (async (changeset): Promise<void> => {
+        const { ids } = changeset.args;
+        await repository.deleteManyPendingRoundRoles(ids);
+    }) satisfies ChangesetHandler<"DeletePendingRoundRoles">,
+
+    InsertRoundRole: (async (changeset): Promise<void> => {
+        const { roundRole } = changeset.args;
+        await repository.insertRoundRole(roundRole);
+    }) satisfies ChangesetHandler<"InsertRoundRole">,
+
+    DeleteAllRoundRolesByRoleAndAddress: (async (changeset): Promise<void> => {
+        const { chainId, roundId, role, address } = changeset.args.roundRole;
+        await repository.deleteManyRoundRolesByRoleAndAddress(chainId, roundId, role, address);
+    }) satisfies ChangesetHandler<"DeleteAllRoundRolesByRoleAndAddress">,
+});

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

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

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

@@ -0,0 +1,9 @@
+import { Changeset } from "@grants-stack-indexer/repository";
+
+export type ChangesetHandler<T extends Changeset["type"]> = (
+    changeset: Extract<Changeset, { type: T }>,
+) => Promise<void>;
+
+export type ChangesetHandlers = {
+    [K in Changeset["type"]]: ChangesetHandler<K>;
+};

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

@@ -0,0 +1 @@
+export * from "./invalidChangeset.exception.js";

packages/data-flow/src/exceptions/invalidChangeset.exception.ts

@@ -0,0 +1,5 @@
+export class InvalidChangeset extends Error {
+    constructor(invalidTypes: string[]) {
+        super(`Invalid changeset types: ${invalidTypes.join(", ")}`);
+    }
+}

packages/data-flow/src/external.ts

@@ -1 +1,3 @@
 export { EventsFetcher } from "./internal.js";
+
+export { DataLoader } from "./internal.js";

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

@@ -0,0 +1,13 @@
+import type { Changeset } from "@grants-stack-indexer/repository";
+
+import type { ExecutionResult } from "../internal.js";
+
+export interface IDataLoader {
+    /**
+     * Applies the changesets to the database.
+     * @param changesets - The changesets to apply.
+     * @returns The execution result.
+     * @throws {InvalidChangeset} if there are changesets with invalid types.
+     */
+    applyChanges(changesets: Changeset[]): Promise<ExecutionResult>;
+}

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

@@ -0,0 +1,20 @@
+import { AnyProtocolEvent } from "@grants-stack-indexer/shared";
+
+/**
+ * Interface for the events fetcher
+ */
+export interface IEventsFetcher {
+    /**
+     * Fetch the events by block number and log index for a chain
+     * @param chainId id of the chain
+     * @param blockNumber block number to fetch events from
+     * @param logIndex log index in the block to fetch events from
+     * @param limit limit of events to fetch
+     */
+    fetchEventsByBlockNumberAndLogIndex(
+        chainId: bigint,
+        blockNumber: bigint,
+        logIndex: number,
+        limit?: number,
+    ): Promise<AnyProtocolEvent[]>;
+}

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

@@ -1,20 +1,2 @@
-import { AnyProtocolEvent } from "@grants-stack-indexer/shared";
-
-/**
- * Interface for the events fetcher
- */
-export interface IEventsFetcher {
-    /**
-     * Fetch the events by block number and log index for a chain
-     * @param chainId id of the chain
-     * @param blockNumber block number to fetch events from
-     * @param logIndex log index in the block to fetch events from
-     * @param limit limit of events to fetch
-     */
-    fetchEventsByBlockNumberAndLogIndex(
-        chainId: bigint,
-        blockNumber: bigint,
-        logIndex: number,
-        limit?: number,
-    ): Promise<AnyProtocolEvent[]>;
-}
+export * from "./eventsFetcher.interface.js";
+export * from "./dataLoader.interface.js";

packages/data-flow/src/internal.ts

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

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

@@ -0,0 +1,9 @@
+import { Changeset } from "@grants-stack-indexer/repository";
+
+export type ExecutionResult = {
+    changesets: Changeset["type"][];
+    numExecuted: number;
+    numSuccessful: number;
+    numFailed: number;
+    errors: string[];
+};