feat: dvmd direct transfer allocated handler

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

@@ -9,6 +9,7 @@ import { CoingeckoProvider } from "@grants-stack-indexer/pricing";
 import {
     createKyselyDatabase,
     KyselyApplicationRepository,
+    KyselyDonationRepository,
     KyselyProjectRepository,
     KyselyRoundRepository,
 } from "@grants-stack-indexer/repository";
@@ -45,6 +46,10 @@ export class SharedDependenciesService {
             kyselyDatabase,
             env.DATABASE_SCHEMA,
         );
+        const donationRepository = new KyselyDonationRepository(
+            kyselyDatabase,
+            env.DATABASE_SCHEMA,
+        );
         const pricingProvider = new CoingeckoProvider(
             {
                 apiKey: env.COINGECKO_API_KEY,
@@ -71,6 +76,7 @@ export class SharedDependenciesService {
                 roundRepository,
                 applicationRepository,
                 pricingProvider,
+                donationRepository,
                 metadataProvider,
             },
             registries: {

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

@@ -1,6 +1,7 @@
 import {
     Changeset,
     IApplicationRepository,
+    IDonationRepository,
     IProjectRepository,
     IRoundRepository,
 } from "@grants-stack-indexer/repository";
@@ -9,6 +10,7 @@ import { ILogger, stringify } from "@grants-stack-indexer/shared";
 import { ExecutionResult, IDataLoader, InvalidChangeset } from "../internal.js";
 import {
     createApplicationHandlers,
+    createDonationHandlers,
     createProjectHandlers,
     createRoundHandlers,
 } from "./handlers/index.js";
@@ -35,13 +37,15 @@ export class DataLoader implements IDataLoader {
             project: IProjectRepository;
             round: IRoundRepository;
             application: IApplicationRepository;
+            donation: IDonationRepository;
         },
         private readonly logger: ILogger,
     ) {
         this.handlers = {
             ...createProjectHandlers(repositories.project),
             ...createRoundHandlers(repositories.round),
             ...createApplicationHandlers(repositories.application),
+            ...createDonationHandlers(repositories.donation),
         };
     }
 

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

@@ -0,0 +1,27 @@
+import { DonationChangeset, IDonationRepository } 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 DonationHandlers = {
+    [K in DonationChangeset["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 createDonationHandlers = (repository: IDonationRepository): DonationHandlers => ({
+    InsertDonation: (async (changeset): Promise<void> => {
+        await repository.insertDonation(changeset.args.donation);
+    }) satisfies ChangesetHandler<"InsertDonation">,
+
+    InsertManyDonations: (async (changeset): Promise<void> => {
+        await repository.insertManyDonations(changeset.args.donations);
+    }) satisfies ChangesetHandler<"InsertManyDonations">,
+});

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

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

packages/data-flow/src/orchestrator.ts

@@ -90,6 +90,7 @@ export class Orchestrator {
                 project: this.dependencies.projectRepository,
                 round: this.dependencies.roundRepository,
                 application: this.dependencies.applicationRepository,
+                donation: this.dependencies.donationRepository,
             },
             this.logger,
         );

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

@@ -2,6 +2,7 @@ import { ProcessorDependencies } from "@grants-stack-indexer/processors";
 import {
     Changeset,
     IApplicationRepository,
+    IDonationRepository,
     IProjectRepository,
     IRoundRepository,
 } from "@grants-stack-indexer/repository";
@@ -31,4 +32,5 @@ export type CoreDependencies = Pick<
     roundRepository: IRoundRepository;
     projectRepository: IProjectRepository;
     applicationRepository: IApplicationRepository;
+    donationRepository: IDonationRepository;
 };

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

@@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 import {
     Changeset,
     IApplicationRepository,
+    IDonationRepository,
     IProjectRepository,
     IRoundRepository,
 } from "@grants-stack-indexer/repository";
@@ -27,6 +28,11 @@ describe("DataLoader", () => {
         updateApplication: vi.fn(),
     } as unknown as IApplicationRepository;
 
+    const mockDonationRepository = {
+        insertDonation: vi.fn(),
+        insertManyDonations: vi.fn(),
+    } as IDonationRepository;
+
     const logger: ILogger = {
         debug: vi.fn(),
         error: vi.fn(),
@@ -39,6 +45,7 @@ describe("DataLoader", () => {
                 project: mockProjectRepository,
                 round: mockRoundRepository,
                 application: mockApplicationRepository,
+                donation: mockDonationRepository,
             },
             logger,
         );

packages/data-flow/test/unit/eventsFetcher.spec.ts

@@ -1,7 +1,7 @@
 import { beforeEach, describe, expect, it, Mocked, vi } from "vitest";
 
 import { IIndexerClient } from "@grants-stack-indexer/indexer-client";
-import { AnyIndexerFetchedEvent, ChainId } from "@grants-stack-indexer/shared";
+import { AnyIndexerFetchedEvent, ChainId, PoolCreatedParams } from "@grants-stack-indexer/shared";
 
 import { EventsFetcher } from "../../src/eventsFetcher.js";
 
@@ -27,7 +27,11 @@ describe("EventsFetcher", () => {
                 eventName: "PoolCreated",
                 srcAddress: "0x1234567890123456789012345678901234567890",
                 logIndex: 0,
-                params: { contractAddress: "0x1234", tokenAddress: "0x1234", amount: 1000 },
+                params: {
+                    contractAddress: "0x1234",
+                    tokenAddress: "0x1234",
+                    amount: 1000n,
+                } as unknown as PoolCreatedParams,
                 transactionFields: { hash: "0x1234", transactionIndex: 0 },
             },
             {
@@ -41,8 +45,8 @@ describe("EventsFetcher", () => {
                 params: {
                     contractAddress: "0x1234",
                     tokenAddress: "0x1234",
-                    amount: 1000,
-                },
+                    amount: 1000n,
+                } as unknown as PoolCreatedParams,
                 transactionFields: { hash: "0x1234", transactionIndex: 1 },
             },
         ];

packages/data-flow/test/unit/orchestrator.spec.ts

@@ -7,6 +7,7 @@ import { UnsupportedStrategy } from "@grants-stack-indexer/processors";
 import {
     Changeset,
     IApplicationRepository,
+    IDonationRepository,
     IProjectRepository,
     IRoundRepository,
 } from "@grants-stack-indexer/repository";
@@ -92,6 +93,7 @@ describe("Orchestrator", { sequential: true }, () => {
             projectRepository: {} as unknown as IProjectRepository,
             roundRepository: {} as unknown as IRoundRepository,
             applicationRepository: {} as unknown as IApplicationRepository,
+            donationRepository: {} as unknown as IDonationRepository,
             pricingProvider: {
                 getTokenPrice: vi.fn(),
             },
@@ -269,6 +271,10 @@ describe("Orchestrator", { sequential: true }, () => {
             RegisteredWithData: "",
             DistributedWithData: "",
             DistributedWithFlowRate: "",
+            AllocatedWithOrigin: "",
+            AllocatedWithData: "",
+            AllocatedWithVotes: "",
+            AllocatedWithStatus: "",
         };
 
         for (const event of Object.keys(strategyEvents) as StrategyEvent[]) {

packages/indexer-client/test/unit/envioIndexerClient.spec.ts

@@ -1,7 +1,7 @@
 import { GraphQLClient, RequestDocument, RequestOptions } from "graphql-request";
 import { afterEach, beforeEach, describe, expect, it, Mocked, vi } from "vitest";
 
-import { AnyIndexerFetchedEvent, ChainId } from "@grants-stack-indexer/shared";
+import { AnyIndexerFetchedEvent, ChainId, PoolCreatedParams } from "@grants-stack-indexer/shared";
 
 import { IndexerClientError, InvalidIndexerResponse } from "../../src/exceptions/index.js";
 import { EnvioIndexerClient } from "../../src/providers/envioIndexerClient.js";
@@ -32,7 +32,11 @@ describe("EnvioIndexerClient", () => {
             eventName: "PoolCreated",
             srcAddress: "0x1234",
             logIndex: 1,
-            params: { contractAddress: "0x1234", tokenAddress: "0x1234", amount: 1000 },
+            params: {
+                contractAddress: "0x1234",
+                tokenAddress: "0x1234",
+                amount: 1000n,
+            } as unknown as PoolCreatedParams,
             transactionFields: { hash: "0x123", transactionIndex: 1 },
         },
         {
@@ -43,7 +47,11 @@ describe("EnvioIndexerClient", () => {
             eventName: "PoolCreated",
             srcAddress: "0x1234",
             logIndex: 3,
-            params: { contractAddress: "0x1234", tokenAddress: "0x1234", amount: 1000 },
+            params: {
+                contractAddress: "0x1234",
+                tokenAddress: "0x1234",
+                amount: 1000n,
+            } as unknown as PoolCreatedParams,
             transactionFields: { hash: "0x123", transactionIndex: 1 },
         },
         {
@@ -54,7 +62,11 @@ describe("EnvioIndexerClient", () => {
             eventName: "PoolCreated",
             srcAddress: "0x1234",
             logIndex: 1,
-            params: { contractAddress: "0x1234", tokenAddress: "0x1234", amount: 1000 },
+            params: {
+                contractAddress: "0x1234",
+                tokenAddress: "0x1234",
+                amount: 1000n,
+            } as unknown as PoolCreatedParams,
             transactionFields: { hash: "0x123", transactionIndex: 1 },
         },
         {
@@ -65,7 +77,11 @@ describe("EnvioIndexerClient", () => {
             eventName: "PoolCreated",
             srcAddress: "0x1234",
             logIndex: 1,
-            params: { contractAddress: "0x1234", tokenAddress: "0x1234", amount: 1000 },
+            params: {
+                contractAddress: "0x1234",
+                tokenAddress: "0x1234",
+                amount: 1000n,
+            } as unknown as PoolCreatedParams,
             transactionFields: { hash: "0x123", transactionIndex: 1 },
         },
     ];

packages/processors/src/exceptions/applicationNotFound.exception.ts

@@ -0,0 +1,9 @@
+import { ChainId } from "@grants-stack-indexer/shared";
+
+export class ApplicationNotFound extends Error {
+    constructor(chainId: ChainId, roundId: string, recipientId: string) {
+        super(
+            `Application not found on chain ${chainId} for round ${roundId} and recipient ${recipientId}`,
+        );
+    }
+}

packages/processors/src/exceptions/index.ts

@@ -4,3 +4,6 @@ export * from "./invalidArgument.exception.js";
 export * from "./unsupportedStrategy.exception.js";
 export * from "./projectNotFound.exception.js";
 export * from "./roundNotFound.exception.js";
+export * from "./applicationNotFound.exception.js";
+export * from "./unknownToken.exception.js";
+export * from "./metadataParsingFailed.exception.js";

packages/processors/src/exceptions/metadataParsingFailed.exception.ts

@@ -0,0 +1,5 @@
+export class MetadataParsingFailed extends Error {
+    constructor(additionalInfo?: string) {
+        super(`Failed to parse application metadata: ${additionalInfo}`);
+    }
+}

packages/processors/src/exceptions/unknownToken.exception.ts

@@ -0,0 +1,7 @@
+import { ChainId } from "@grants-stack-indexer/shared";
+
+export class UnknownToken extends Error {
+    constructor(tokenAddress: string, chainId?: ChainId) {
+        super(`Unknown token: ${tokenAddress} ${chainId ? `on chain ${chainId}` : ""}`);
+    }
+}

packages/processors/src/helpers/index.ts

@@ -1,3 +1,4 @@
 export * from "./roles.js";
 export * from "./utils.js";
 export * from "./tokenMath.js";
+export * from "./pricing.js";

packages/processors/src/helpers/pricing.ts

@@ -0,0 +1,71 @@
+import { IPricingProvider } from "@grants-stack-indexer/pricing";
+import { Token } from "@grants-stack-indexer/shared";
+
+import { TokenPriceNotFoundError } from "../internal.js";
+import { calculateAmountInToken, calculateAmountInUsd } from "./index.js";
+
+// sometimes coingecko returns no prices for 1 hour range, 2 hours works better
+const TIMESTAMP_DELTA_RANGE = 2 * 60 * 60 * 1000;
+
+/**
+ * Get the amount in USD for a given amount in the token
+ * @param pricingProvider - The pricing provider to use
+ * @param token - The token to get the amount in
+ * @param amount - The amount in the token
+ * @param timestamp - The timestamp to get the price at
+ * @returns The amount in USD
+ * @throws TokenPriceNotFoundError if the price is not found
+ */
+export const getTokenAmountInUsd = async (
+    pricingProvider: IPricingProvider,
+    token: Token,
+    amount: bigint,
+    timestamp: number,
+): Promise<{ amountInUsd: string; timestamp: number }> => {
+    const tokenPrice = await pricingProvider.getTokenPrice(
+        token.priceSourceCode,
+        timestamp,
+        timestamp + TIMESTAMP_DELTA_RANGE,
+    );
+
+    if (!tokenPrice) {
+        throw new TokenPriceNotFoundError(token.address, timestamp);
+    }
+
+    return {
+        amountInUsd: calculateAmountInUsd(amount, tokenPrice.priceUsd, token.decimals),
+        timestamp: tokenPrice.timestampMs,
+    };
+};
+
+/**
+ * Get the amount in the token for a given amount in USD
+ * @param pricingProvider - The pricing provider to use
+ * @param token - The token to get the amount in
+ * @param amountInUSD - The amount in USD
+ * @param timestamp - The timestamp to get the price at
+ * @returns The amount in the token
+ * @throws TokenPriceNotFoundError if the price is not found
+ */
+export const getUsdInTokenAmount = async (
+    pricingProvider: IPricingProvider,
+    token: Token,
+    amountInUSD: string,
+    timestamp: number,
+): Promise<{ amount: bigint; price: number; timestamp: Date }> => {
+    const closestPrice = await pricingProvider.getTokenPrice(
+        token.priceSourceCode,
+        timestamp,
+        timestamp + TIMESTAMP_DELTA_RANGE,
+    );
+
+    if (!closestPrice) {
+        throw new TokenPriceNotFoundError(token.address, timestamp);
+    }
+
+    return {
+        amount: calculateAmountInToken(amountInUSD, closestPrice.priceUsd, token.decimals),
+        timestamp: new Date(closestPrice.timestampMs),
+        price: 1 / closestPrice.priceUsd, // price is the token price in USD, we return the inverse
+    };
+};

packages/processors/src/helpers/tokenMath.ts

@@ -32,3 +32,28 @@ export const calculateAmountInUsd = (
 
     return amountInUsd.toString();
 };
+
+/**
+ * Calculates the amount in token
+ * @param amountInUSD - The amount in USD
+ * @param tokenPriceInUsd - The price of the token in USD
+ * @param tokenDecimals - The number of decimals the token has
+ * @returns The amount in token
+ * @throws Error if tokenPriceInUsd is 0 (division by zero)
+ */
+export const calculateAmountInToken = (
+    amountInUSD: string,
+    tokenPriceInUsd: string | number,
+    tokenDecimals: number,
+): bigint => {
+    const amountInUsdBN = new BigNumber(amountInUSD);
+    const tokenPriceInUsdBN = new BigNumber(tokenPriceInUsd);
+    const scaleFactor = new BigNumber(10).pow(tokenDecimals);
+
+    return BigInt(
+        amountInUsdBN
+            .multipliedBy(scaleFactor)
+            .dividedBy(tokenPriceInUsdBN)
+            .toFixed(0, BigNumber.ROUND_FLOOR),
+    );
+};