feat: fetch events by src addresses

packages/data-flow/src/eventsFetcher.ts

@@ -1,4 +1,4 @@
-import { IIndexerClient } from "@grants-stack-indexer/indexer-client";
+import { GetEventsFilters, IIndexerClient } from "@grants-stack-indexer/indexer-client";
 import { AnyIndexerFetchedEvent, ChainId } from "@grants-stack-indexer/shared";
 
 import { IEventsFetcher } from "./interfaces/index.js";
@@ -19,4 +19,9 @@ export class EventsFetcher implements IEventsFetcher {
             limit,
         );
     }
+
+    /** @inheritdoc */
+    async fetchEvents(params: GetEventsFilters): Promise<AnyIndexerFetchedEvent[]> {
+        return this.indexerClient.getEvents(params);
+    }
 }

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

@@ -1,3 +1,4 @@
+import { GetEventsFilters } from "@grants-stack-indexer/indexer-client";
 import { AnyIndexerFetchedEvent, ChainId } from "@grants-stack-indexer/shared";
 
 /**
@@ -17,4 +18,14 @@ export interface IEventsFetcher {
         logIndex: number,
         limit?: number,
     ): Promise<AnyIndexerFetchedEvent[]>;
+
+    /**
+     * Fetch the events by src address, block number and log index for a chain
+     * @param chainId id of the chain
+     * @param srcAddresses src addresses to fetch events from
+     * @param toBlock block number to fetch events from
+     * @param logIndex log index in the block to fetch events from
+     * @param limit limit of events to fetch
+     */
+    fetchEvents(params: GetEventsFilters): Promise<AnyIndexerFetchedEvent[]>;
 }

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

@@ -12,6 +12,7 @@ describe("EventsFetcher", () => {
     beforeEach(() => {
         indexerClientMock = {
             getEventsAfterBlockNumberAndLogIndex: vi.fn(),
+            getEvents: vi.fn(),
         };
 
         eventsFetcher = new EventsFetcher(indexerClientMock);

packages/indexer-client/src/external.ts

@@ -1,3 +1,5 @@
 export type { IIndexerClient } from "./internal.js";
 
 export { EnvioIndexerClient } from "./internal.js";
+
+export type { GetEventsFilters } from "./internal.js";

packages/indexer-client/src/interfaces/indexerClient.ts

@@ -1,5 +1,7 @@
 import { AnyIndexerFetchedEvent, ChainId } from "@grants-stack-indexer/shared";
 
+import { GetEventsFilters } from "../internal.js";
+
 /**
  * Interface for the indexer client
  */
@@ -17,4 +19,11 @@ export interface IIndexerClient {
         logIndex: number,
         limit?: number,
     ): Promise<AnyIndexerFetchedEvent[]>;
+
+    /**
+     * Get the events by filters from the indexer service
+     * @param params Filters to fetch events
+     * @returns Events fetched from the indexer service
+     */
+    getEvents(params: GetEventsFilters): Promise<AnyIndexerFetchedEvent[]>;
 }

packages/indexer-client/src/internal.ts

@@ -1,3 +1,4 @@
 export * from "./exceptions/index.js";
+export * from "./types/index.js";
 export * from "./interfaces/index.js";
 export * from "./providers/index.js";

packages/indexer-client/src/providers/envioIndexerClient.ts

@@ -3,7 +3,7 @@ import { gql, GraphQLClient } from "graphql-request";
 import { AnyIndexerFetchedEvent, ChainId, stringify } from "@grants-stack-indexer/shared";
 
 import { IndexerClientError, InvalidIndexerResponse } from "../exceptions/index.js";
-import { IIndexerClient } from "../internal.js";
+import { GetEventsFilters, IIndexerClient } from "../internal.js";
 
 /**
  * Indexer client for the Envio indexer service
@@ -73,4 +73,104 @@ export class EnvioIndexerClient implements IIndexerClient {
             throw new IndexerClientError(stringify(error, Object.getOwnPropertyNames(error)));
         }
     }
+
+    /** @inheritdoc */
+    async getEvents(params: GetEventsFilters): Promise<AnyIndexerFetchedEvent[]> {
+        try {
+            const { chainId, srcAddresses, from, to, limit = 100 } = params;
+
+            // Build the _and conditions array
+            const andConditions = [];
+            andConditions.push(`chain_id: { _eq: $chainId }`);
+            const vars: Record<string, unknown> = { chainId };
+
+            // Add srcAddresses filter if provided
+            if (srcAddresses && srcAddresses.length > 0) {
+                andConditions.push(`src_address: { _in: $srcAddresses }`);
+                vars["srcAddresses"] = srcAddresses;
+            }
+
+            if (from != undefined && from != null) {
+                andConditions.push(`
+                    _or: [
+                        { block_number: { _gt: $fromBlock } },
+                        {
+                            _and: [
+                                { block_number: { _eq: $fromBlock } },
+                                { log_index: { _gt: $fromLogIndex } }
+                            ]
+                        }
+                    ]
+                `);
+                vars["fromBlock"] = from.blockNumber;
+                vars["fromLogIndex"] = from.logIndex;
+            }
+
+            if (to != undefined && to != null) {
+                andConditions.push(`
+                    _or: [
+                        { block_number: { _lt: $toBlock } },
+                        {
+                            _and: [
+                                { block_number: { _eq: $toBlock } },
+                                { log_index: { _lte: $toLogIndex } }
+                            ]
+                        }
+                    ]
+                `);
+                vars["toBlock"] = to.blockNumber;
+                vars["toLogIndex"] = to.logIndex;
+            }
+
+            const whereClause =
+                andConditions.length > 1
+                    ? `_and: [{ ${andConditions.join(" }, { ")} }]`
+                    : andConditions[0];
+
+            const response = (await this.client.request(
+                gql`
+                    query getEvents(
+                        $chainId: Int!
+                        $srcAddresses: [String!]
+                        $fromBlock: Int
+                        $fromLogIndex: Int
+                        $toBlock: Int
+                        $toLogIndex: Int
+                        $limit: Int!
+                    ) {
+                        raw_events(
+                            order_by: [{ block_number: asc }, { log_index: asc }]
+                            where: { ${whereClause} }
+                            limit: $limit
+                        ) {
+                            blockNumber: block_number
+                            blockTimestamp: block_timestamp
+                            chainId: chain_id
+                            contractName: contract_name
+                            eventName: event_name
+                            logIndex: log_index
+                            params
+                            srcAddress: src_address
+                            transactionFields: transaction_fields
+                        }
+                    }
+                `,
+                {
+                    ...vars,
+                    limit,
+                },
+            )) as { raw_events: AnyIndexerFetchedEvent[] };
+
+            if (response?.raw_events) {
+                return response.raw_events;
+            } else {
+                throw new InvalidIndexerResponse(stringify(response));
+            }
+        } catch (error) {
+            if (error instanceof InvalidIndexerResponse) {
+                throw error;
+            }
+            throw new IndexerClientError(stringify(error, Object.getOwnPropertyNames(error)));
+        }
+    }
 }

packages/indexer-client/src/types/index.ts

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

packages/indexer-client/src/types/indexerClient.types.ts

@@ -0,0 +1,36 @@
+import { Address, ChainId } from "@grants-stack-indexer/shared";
+
+export type GetEventsFilters = {
+    /**
+     * Id of the chain to fetch events from
+     */
+    chainId: ChainId;
+    /**
+     * Src addresses to filter events by
+     */
+    srcAddresses?: Address[];
+    from?: {
+        /**
+         * Block number to start fetching events from
+         */
+        blockNumber: number;
+        /**
+         * Log index in the block
+         */
+        logIndex: number;
+    };
+    to?: {
+        /**
+         * Block number to end fetching events at
+         */
+        blockNumber: number;
+        /**
+         * Log index in the block
+         */
+        logIndex: number;
+    };
+    /**
+     * Limit of events to fetch
+     */
+    limit?: number;
+};