From 33def584aa5dad5d4d18d70fc9066c2158a3c73f Mon Sep 17 00:00:00 2001 From: nigiri <168690269+0xnigir1@users.noreply.github.com> Date: Mon, 16 Dec 2024 12:58:12 -0300 Subject: [PATCH 1/5] feat(wip): event registry repository & cached service impl --- .../src/services/processing.service.ts | 54 +++++++++++--- .../services/sharedDependencies.service.ts | 39 ++++------- .../unit/sharedDependencies.service.spec.ts | 7 +- packages/data-flow/src/eventsRegistry.ts | 40 ----------- packages/data-flow/src/external.ts | 2 + .../interfaces/eventsRegistry.interface.ts | 12 ++-- .../interfaces/strategyRegistry.interface.ts | 4 +- packages/data-flow/src/internal.ts | 2 +- .../registries/event/cachedEventRegistry.ts | 70 +++++++++++++++++++ .../src/registries/event/dbEventRegistry.ts | 28 ++++++++ .../src/registries/event/eventsRegistry.ts | 30 ++++++++ .../data-flow/src/registries/event/index.ts | 3 + packages/data-flow/src/registries/index.ts | 1 + .../test/unit/eventsRegistry.spec.ts | 2 +- packages/repository/src/db/connection.ts | 2 + packages/repository/src/external.ts | 3 + .../interfaces/eventsRepository.interface.ts | 19 +++++ packages/repository/src/interfaces/index.ts | 1 + .../kysely/eventRegistry.repository.ts | 33 +++++++++ .../src/repositories/kysely/index.ts | 1 + packages/repository/src/types/event.types.ts | 11 +++ packages/repository/src/types/index.ts | 1 + 22 files changed, 273 insertions(+), 92 deletions(-) delete mode 100644 packages/data-flow/src/eventsRegistry.ts create mode 100644 packages/data-flow/src/registries/event/cachedEventRegistry.ts create mode 100644 packages/data-flow/src/registries/event/dbEventRegistry.ts create mode 100644 packages/data-flow/src/registries/event/eventsRegistry.ts create mode 100644 packages/data-flow/src/registries/event/index.ts create mode 100644 packages/repository/src/interfaces/eventsRepository.interface.ts create mode 100644 packages/repository/src/repositories/kysely/eventRegistry.repository.ts create mode 100644 packages/repository/src/types/event.types.ts diff --git a/apps/processing/src/services/processing.service.ts b/apps/processing/src/services/processing.service.ts index cf75f58..5725a95 100644 --- a/apps/processing/src/services/processing.service.ts +++ b/apps/processing/src/services/processing.service.ts @@ -1,7 +1,13 @@ import { optimism } from "viem/chains"; import { EvmProvider } from "@grants-stack-indexer/chain-providers"; -import { Orchestrator } from "@grants-stack-indexer/data-flow"; +import { + DatabaseEventRegistry, + DatabaseStrategyRegistry, + InMemoryCachedEventRegistry, + InMemoryCachedStrategyRegistry, + Orchestrator, +} from "@grants-stack-indexer/data-flow"; import { ChainId, Logger } from "@grants-stack-indexer/shared"; import { Environment } from "../config/env.js"; @@ -23,34 +29,62 @@ export class ProcessingService { private readonly logger = new Logger({ className: "ProcessingService" }); private readonly kyselyDatabase: SharedDependencies["kyselyDatabase"]; - private constructor(env: Environment, sharedDependencies: SharedDependencies) { - const { CHAINS: chains } = env; - const { core, registries, indexerClient, kyselyDatabase } = sharedDependencies; + private constructor( + orchestrators: Map, + kyselyDatabase: SharedDependencies["kyselyDatabase"], + ) { + this.orchestrators = orchestrators; this.kyselyDatabase = kyselyDatabase; + } + + static async initialize(env: Environment): Promise { + const sharedDependencies = await SharedDependenciesService.initialize(env); + const { CHAINS: chains } = env; + const { core, registriesRepositories, indexerClient, kyselyDatabase } = sharedDependencies; + const { eventRegistryRepository, strategyRegistryRepository } = registriesRepositories; + const orchestrators: Map = new Map(); + + const strategyRegistry = await InMemoryCachedStrategyRegistry.initialize( + new Logger({ className: "InMemoryCachedStrategyRegistry" }), + new DatabaseStrategyRegistry( + new Logger({ className: "DatabaseStrategyRegistry" }), + strategyRegistryRepository, + ), + ); for (const chain of chains) { const chainLogger = new Logger({ chainId: chain.id as ChainId }); // Initialize EVM provider const evmProvider = new EvmProvider(chain.rpcUrls, optimism, chainLogger); - this.orchestrators.set( + // Initialize events registry + const eventsRegistry = await InMemoryCachedEventRegistry.initialize( + new Logger({ className: "InMemoryCachedEventRegistry" }), + new DatabaseEventRegistry( + new Logger({ className: "DatabaseEventRegistry" }), + eventRegistryRepository, + ), + [chain.id as ChainId], + ); + + orchestrators.set( chain.id as ChainId, new Orchestrator( chain.id as ChainId, { ...core, evmProvider }, indexerClient, - registries, + { + eventsRegistry, + strategyRegistry, + }, chain.fetchLimit, chain.fetchDelayMs, chainLogger, ), ); } - } - static async initialize(env: Environment): Promise { - const sharedDependencies = await SharedDependenciesService.initialize(env); - return new ProcessingService(env, sharedDependencies); + return new ProcessingService(orchestrators, kyselyDatabase); } /** diff --git a/apps/processing/src/services/sharedDependencies.service.ts b/apps/processing/src/services/sharedDependencies.service.ts index b2e05ae..033c386 100644 --- a/apps/processing/src/services/sharedDependencies.service.ts +++ b/apps/processing/src/services/sharedDependencies.service.ts @@ -1,19 +1,15 @@ -import { - CoreDependencies, - DatabaseStrategyRegistry, - IEventsRegistry, - InMemoryCachedStrategyRegistry, - InMemoryEventsRegistry, - IStrategyRegistry, -} from "@grants-stack-indexer/data-flow"; +import { CoreDependencies } from "@grants-stack-indexer/data-flow"; import { EnvioIndexerClient } from "@grants-stack-indexer/indexer-client"; import { IpfsProvider } from "@grants-stack-indexer/metadata"; import { PricingProviderFactory } from "@grants-stack-indexer/pricing"; import { createKyselyDatabase, + IEventRegistryRepository, + IStrategyRegistryRepository, KyselyApplicationPayoutRepository, KyselyApplicationRepository, KyselyDonationRepository, + KyselyEventRegistryRepository, KyselyProjectRepository, KyselyRoundRepository, KyselyStrategyRegistryRepository, @@ -24,9 +20,9 @@ import { Environment } from "../config/index.js"; export type SharedDependencies = { core: Omit; - registries: { - eventsRegistry: IEventsRegistry; - strategyRegistry: IStrategyRegistry; + registriesRepositories: { + eventRegistryRepository: IEventRegistryRepository; + strategyRegistryRepository: IStrategyRegistryRepository; }; indexerClient: EnvioIndexerClient; kyselyDatabase: ReturnType; @@ -68,20 +64,13 @@ export class SharedDependenciesService { new Logger({ className: "IpfsProvider" }), ); - // Initialize registries - const eventsRegistry = new InMemoryEventsRegistry( - new Logger({ className: "InMemoryEventsRegistry" }), - ); - const strategyRepository = new KyselyStrategyRegistryRepository( + const eventRegistryRepository = new KyselyEventRegistryRepository( kyselyDatabase, env.DATABASE_SCHEMA, ); - const strategyRegistry = await InMemoryCachedStrategyRegistry.initialize( - new Logger({ className: "InMemoryCachedStrategyRegistry" }), - new DatabaseStrategyRegistry( - new Logger({ className: "DatabaseStrategyRegistry" }), - strategyRepository, - ), + const strategyRegistryRepository = new KyselyStrategyRegistryRepository( + kyselyDatabase, + env.DATABASE_SCHEMA, ); // Initialize indexer client @@ -100,9 +89,9 @@ export class SharedDependenciesService { metadataProvider, applicationPayoutRepository, }, - registries: { - eventsRegistry, - strategyRegistry, + registriesRepositories: { + eventRegistryRepository, + strategyRegistryRepository, }, indexerClient, kyselyDatabase, diff --git a/apps/processing/test/unit/sharedDependencies.service.spec.ts b/apps/processing/test/unit/sharedDependencies.service.spec.ts index 9d309b9..e1b2ab0 100644 --- a/apps/processing/test/unit/sharedDependencies.service.spec.ts +++ b/apps/processing/test/unit/sharedDependencies.service.spec.ts @@ -112,10 +112,7 @@ describe("SharedDependenciesService", () => { expect(dependencies.core).toHaveProperty("applicationPayoutRepository"); // Verify registries - expect(dependencies.registries).toHaveProperty("eventsRegistry"); - expect(dependencies.registries).toHaveProperty("strategyRegistry"); - - // Verify InMemoryCachedStrategyRegistry initialization - expect(dependencies.registries.strategyRegistry).toBeDefined(); + expect(dependencies.registriesRepositories).toHaveProperty("eventRegistryRepository"); + expect(dependencies.registriesRepositories).toHaveProperty("strategyRegistryRepository"); }); }); diff --git a/packages/data-flow/src/eventsRegistry.ts b/packages/data-flow/src/eventsRegistry.ts deleted file mode 100644 index 42bed46..0000000 --- a/packages/data-flow/src/eventsRegistry.ts +++ /dev/null @@ -1,40 +0,0 @@ -import type { - AnyEvent, - ChainId, - ContractName, - ILogger, - ProcessorEvent, -} from "@grants-stack-indexer/shared"; -import { stringify } from "@grants-stack-indexer/shared"; - -import type { IEventsRegistry } from "./internal.js"; - -/** - * Class to store the last processed event in memory - */ -//TODO: Implement storage version to persist the last processed event. we need to store it by chainId -export class InMemoryEventsRegistry implements IEventsRegistry { - private lastProcessedEvent: Map> = new Map(); - - constructor(private logger: ILogger) {} - - /** - * @inheritdoc - */ - async getLastProcessedEvent( - chainId: ChainId, - ): Promise | undefined> { - return this.lastProcessedEvent.get(chainId); - } - - /** - * @inheritdoc - */ - async saveLastProcessedEvent( - chainId: ChainId, - event: ProcessorEvent, - ): Promise { - this.logger.debug(`Saving last processed event: ${stringify(event, undefined, 4)}`); - this.lastProcessedEvent.set(chainId, event); - } -} diff --git a/packages/data-flow/src/external.ts b/packages/data-flow/src/external.ts index 4a22f36..1d69f49 100644 --- a/packages/data-flow/src/external.ts +++ b/packages/data-flow/src/external.ts @@ -2,6 +2,8 @@ export { DataLoader, InMemoryEventsRegistry, InMemoryCachedStrategyRegistry, + InMemoryCachedEventRegistry, + DatabaseEventRegistry, DatabaseStrategyRegistry, Orchestrator, } from "./internal.js"; diff --git a/packages/data-flow/src/interfaces/eventsRegistry.interface.ts b/packages/data-flow/src/interfaces/eventsRegistry.interface.ts index 6c063e1..16e9fd7 100644 --- a/packages/data-flow/src/interfaces/eventsRegistry.interface.ts +++ b/packages/data-flow/src/interfaces/eventsRegistry.interface.ts @@ -1,4 +1,5 @@ -import { AnyEvent, ChainId, ContractName, ProcessorEvent } from "@grants-stack-indexer/shared"; +import { NewProcessedEvent, ProcessedEvent } from "@grants-stack-indexer/repository"; +import { ChainId } from "@grants-stack-indexer/shared"; /** * The events registry saves as a checkpoint to the last processed event by the system. @@ -10,16 +11,11 @@ export interface IEventsRegistry { * @param chainId - The chain id * @returns The last processed event or undefined if no event has been processed yet. */ - getLastProcessedEvent( - chainId: ChainId, - ): Promise | undefined>; + getLastProcessedEvent(chainId: ChainId): Promise; /** * Save the last processed event by the system * @param chainId - The chain id * @param event - The event to save. */ - saveLastProcessedEvent( - chainId: ChainId, - event: ProcessorEvent, - ): Promise; + saveLastProcessedEvent(chainId: ChainId, event: NewProcessedEvent): Promise; } diff --git a/packages/data-flow/src/interfaces/strategyRegistry.interface.ts b/packages/data-flow/src/interfaces/strategyRegistry.interface.ts index 965ad44..b75eabb 100644 --- a/packages/data-flow/src/interfaces/strategyRegistry.interface.ts +++ b/packages/data-flow/src/interfaces/strategyRegistry.interface.ts @@ -21,13 +21,13 @@ export interface IStrategyRegistry { * @param chainId - The chain id * @param strategyAddress - The strategy address * @param strategyId - The strategy id - * @param handled - Whether the strategy is handled + * @param handleable - Whether the strategy is handled */ saveStrategyId( chainId: ChainId, strategyAddress: Address, strategyId: Hex, - handled: boolean, + handleable: boolean, ): Promise; /** diff --git a/packages/data-flow/src/internal.ts b/packages/data-flow/src/internal.ts index cb9a1a9..ad35306 100644 --- a/packages/data-flow/src/internal.ts +++ b/packages/data-flow/src/internal.ts @@ -6,6 +6,6 @@ export * from "./utils/index.js"; export * from "./data-loader/index.js"; export * from "./eventsFetcher.js"; export * from "./registries/index.js"; -export * from "./eventsRegistry.js"; +export * from "./registries/event/eventsRegistry.js"; export * from "./eventsProcessor.js"; export * from "./orchestrator.js"; diff --git a/packages/data-flow/src/registries/event/cachedEventRegistry.ts b/packages/data-flow/src/registries/event/cachedEventRegistry.ts new file mode 100644 index 0000000..8f5bea8 --- /dev/null +++ b/packages/data-flow/src/registries/event/cachedEventRegistry.ts @@ -0,0 +1,70 @@ +import { NewProcessedEvent, ProcessedEvent } from "@grants-stack-indexer/repository"; +import { ChainId, ILogger, stringify } from "@grants-stack-indexer/shared"; + +import { IEventsRegistry } from "../../internal.js"; + +/** + * Proxy class to cache the events in memory or fallback to another event registry + */ +export class InMemoryCachedEventRegistry implements IEventsRegistry { + private cache: Map = new Map(); + + private constructor( + private readonly logger: ILogger, + private readonly eventRegistry: IEventsRegistry, + cache: Map, + ) { + this.cache = structuredClone(cache); + } + + /** @inheritdoc */ + async getLastProcessedEvent(chainId: ChainId): Promise { + const cachedEvent = this.cache.get(chainId); + if (cachedEvent) { + return cachedEvent; + } + const event = await this.eventRegistry.getLastProcessedEvent(chainId); + if (event) { + this.cache.set(chainId, event); + } + return event; + } + + /** @inheritdoc */ + async saveLastProcessedEvent(chainId: ChainId, event: NewProcessedEvent): Promise { + this.logger.debug(`Saving last processed event: ${stringify(event, undefined, 4)}`); + await this.eventRegistry.saveLastProcessedEvent(chainId, event); + this.cache.set(chainId, { ...event, chainId }); + } + + /** + /** + * Creates a new cached event registry instance. It will load the events into memory and cache them and + * fallback to the event registry if the event is not found in the cache. + * + * @param logger - The logger instance + * @param eventRegistry - The event registry instance + * @param chainIds - The chain ids to load the events for + * @returns The initialized cached event registry + */ + static async initialize( + logger: ILogger, + eventRegistry: IEventsRegistry, + chainIds: ChainId[], + ): Promise { + const events = await Promise.allSettled( + chainIds.map(async (chainId) => await eventRegistry.getLastProcessedEvent(chainId)), + ); + const cache = new Map(); + + logger.debug(`Loading events into memory...`); + + for (const event of events) { + if (event.status === "fulfilled" && event.value) { + cache.set(event.value.chainId, event.value); + } + } + + return new InMemoryCachedEventRegistry(logger, eventRegistry, cache); + } +} diff --git a/packages/data-flow/src/registries/event/dbEventRegistry.ts b/packages/data-flow/src/registries/event/dbEventRegistry.ts new file mode 100644 index 0000000..0196111 --- /dev/null +++ b/packages/data-flow/src/registries/event/dbEventRegistry.ts @@ -0,0 +1,28 @@ +import { + IEventRegistryRepository, + NewProcessedEvent, + ProcessedEvent, +} from "@grants-stack-indexer/repository"; +import { ChainId, ILogger } from "@grants-stack-indexer/shared"; + +import { IEventsRegistry } from "../../internal.js"; + +/** + * Class to store last processed event of a chain in Database + */ +export class DatabaseEventRegistry implements IEventsRegistry { + constructor( + private logger: ILogger, + private eventRepository: IEventRegistryRepository, + ) {} + + /** @inheritdoc */ + async getLastProcessedEvent(chainId: ChainId): Promise { + return this.eventRepository.getLastProcessedEvent(chainId); + } + + /** @inheritdoc */ + async saveLastProcessedEvent(chainId: ChainId, event: NewProcessedEvent): Promise { + return this.eventRepository.saveLastProcessedEvent(chainId, event); + } +} diff --git a/packages/data-flow/src/registries/event/eventsRegistry.ts b/packages/data-flow/src/registries/event/eventsRegistry.ts new file mode 100644 index 0000000..0bce55e --- /dev/null +++ b/packages/data-flow/src/registries/event/eventsRegistry.ts @@ -0,0 +1,30 @@ +import type { ChainId, ILogger } from "@grants-stack-indexer/shared"; +import { NewProcessedEvent, ProcessedEvent } from "@grants-stack-indexer/repository"; +import { stringify } from "@grants-stack-indexer/shared"; + +import type { IEventsRegistry } from "../../internal.js"; + +/** + * Class to store the last processed event in memory + */ +//TODO: Implement storage version to persist the last processed event. we need to store it by chainId +export class InMemoryEventsRegistry implements IEventsRegistry { + private lastProcessedEvent: Map = new Map(); + + constructor(private logger: ILogger) {} + + /** + * @inheritdoc + */ + async getLastProcessedEvent(chainId: ChainId): Promise { + return this.lastProcessedEvent.get(chainId); + } + + /** + * @inheritdoc + */ + async saveLastProcessedEvent(chainId: ChainId, event: NewProcessedEvent): Promise { + this.logger.debug(`Saving last processed event: ${stringify(event, undefined, 4)}`); + this.lastProcessedEvent.set(chainId, { ...event, chainId }); + } +} diff --git a/packages/data-flow/src/registries/event/index.ts b/packages/data-flow/src/registries/event/index.ts new file mode 100644 index 0000000..47653b0 --- /dev/null +++ b/packages/data-flow/src/registries/event/index.ts @@ -0,0 +1,3 @@ +export * from "./eventsRegistry.js"; +export * from "./cachedEventRegistry.js"; +export * from "./dbEventRegistry.js"; diff --git a/packages/data-flow/src/registries/index.ts b/packages/data-flow/src/registries/index.ts index c6319bf..36ea2a7 100644 --- a/packages/data-flow/src/registries/index.ts +++ b/packages/data-flow/src/registries/index.ts @@ -1 +1,2 @@ export * from "./strategy/index.js"; +export * from "./event/index.js"; diff --git a/packages/data-flow/test/unit/eventsRegistry.spec.ts b/packages/data-flow/test/unit/eventsRegistry.spec.ts index 1ddf8b7..be14f7a 100644 --- a/packages/data-flow/test/unit/eventsRegistry.spec.ts +++ b/packages/data-flow/test/unit/eventsRegistry.spec.ts @@ -2,7 +2,7 @@ import { describe, expect, it, vi } from "vitest"; import { ChainId, ILogger, ProcessorEvent } from "@grants-stack-indexer/shared"; -import { InMemoryEventsRegistry } from "../../src/eventsRegistry.js"; +import { InMemoryEventsRegistry } from "../../src/registries/event/eventsRegistry.js"; describe("InMemoryEventsRegistry", () => { const logger: ILogger = { diff --git a/packages/repository/src/db/connection.ts b/packages/repository/src/db/connection.ts index d9e5568..9a774bb 100644 --- a/packages/repository/src/db/connection.ts +++ b/packages/repository/src/db/connection.ts @@ -12,6 +12,7 @@ import { Application, ApplicationPayout, Donation as DonationTable, + ProcessedEvent as EventTable, MatchingDistribution, PendingProjectRole as PendingProjectRoleTable, PendingRoundRole as PendingRoundRoleTable, @@ -61,6 +62,7 @@ export interface Database { donations: DonationTable; applicationsPayouts: ApplicationPayoutTable; strategies: StrategyTable; + events: EventTable; } /** diff --git a/packages/repository/src/external.ts b/packages/repository/src/external.ts index fec981b..08222ec 100644 --- a/packages/repository/src/external.ts +++ b/packages/repository/src/external.ts @@ -9,6 +9,7 @@ export type { IDonationRepository, IApplicationPayoutRepository, IStrategyRegistryRepository, + IEventRegistryRepository, DatabaseConfig, } from "./internal.js"; @@ -37,6 +38,7 @@ export type { Donation, NewDonation } from "./types/index.js"; export type { NewApplicationPayout, ApplicationPayout } from "./types/index.js"; export type { Strategy, NewStrategy } from "./types/index.js"; +export type { ProcessedEvent, NewProcessedEvent } from "./types/index.js"; export type { Changeset, @@ -54,6 +56,7 @@ export { KyselyDonationRepository, KyselyApplicationPayoutRepository, KyselyStrategyRegistryRepository, + KyselyEventRegistryRepository, } from "./repositories/kysely/index.js"; export { diff --git a/packages/repository/src/interfaces/eventsRepository.interface.ts b/packages/repository/src/interfaces/eventsRepository.interface.ts new file mode 100644 index 0000000..5e3ee73 --- /dev/null +++ b/packages/repository/src/interfaces/eventsRepository.interface.ts @@ -0,0 +1,19 @@ +import { ChainId } from "@grants-stack-indexer/shared"; + +import { NewProcessedEvent, ProcessedEvent } from "../types/index.js"; + +export interface IEventRegistryRepository { + /** + * Get the last processed event for a given chain. + * @param chainId - The chain ID. + * @returns The last processed event or undefined if none exists. + */ + getLastProcessedEvent: (chainId: ChainId) => Promise; + + /** + * Save the last processed event for a given chain. + * @param chainId - The chain ID. + * @param event - The new processed event to save. + */ + saveLastProcessedEvent: (chainId: ChainId, event: NewProcessedEvent) => Promise; +} diff --git a/packages/repository/src/interfaces/index.ts b/packages/repository/src/interfaces/index.ts index 3562351..f1a27be 100644 --- a/packages/repository/src/interfaces/index.ts +++ b/packages/repository/src/interfaces/index.ts @@ -4,3 +4,4 @@ export * from "./applicationRepository.interface.js"; export * from "./donationRepository.interface.js"; export * from "./applicationPayoutRepository.interface.js"; export * from "./strategyRepository.interface.js"; +export * from "./eventsRepository.interface.js"; diff --git a/packages/repository/src/repositories/kysely/eventRegistry.repository.ts b/packages/repository/src/repositories/kysely/eventRegistry.repository.ts new file mode 100644 index 0000000..2b77858 --- /dev/null +++ b/packages/repository/src/repositories/kysely/eventRegistry.repository.ts @@ -0,0 +1,33 @@ +import { Kysely } from "kysely"; + +import { ChainId } from "@grants-stack-indexer/shared"; + +import { IEventRegistryRepository } from "../../interfaces/index.js"; +import { Database, NewProcessedEvent, ProcessedEvent } from "../../internal.js"; + +export class KyselyEventRegistryRepository implements IEventRegistryRepository { + constructor( + private readonly db: Kysely, + private readonly schemaName: string, + ) {} + + /** @inheritdoc */ + async getLastProcessedEvent(chainId: ChainId): Promise { + return this.db + .withSchema(this.schemaName) + .selectFrom("events") + .where("chainId", "=", chainId) + .selectAll() + .executeTakeFirst(); + } + + /** @inheritdoc */ + async saveLastProcessedEvent(chainId: ChainId, event: NewProcessedEvent): Promise { + await this.db + .withSchema(this.schemaName) + .insertInto("events") + .values({ ...event, chainId }) + .onConflict((oc) => oc.columns(["chainId"]).doUpdateSet({ ...event, chainId })) + .execute(); + } +} diff --git a/packages/repository/src/repositories/kysely/index.ts b/packages/repository/src/repositories/kysely/index.ts index ae4143a..66febbe 100644 --- a/packages/repository/src/repositories/kysely/index.ts +++ b/packages/repository/src/repositories/kysely/index.ts @@ -4,3 +4,4 @@ export * from "./application.repository.js"; export * from "./donation.repository.js"; export * from "./applicationPayout.repository.js"; export * from "./strategyRegistry.repository.js"; +export * from "./eventRegistry.repository.js"; diff --git a/packages/repository/src/types/event.types.ts b/packages/repository/src/types/event.types.ts new file mode 100644 index 0000000..3f1eb50 --- /dev/null +++ b/packages/repository/src/types/event.types.ts @@ -0,0 +1,11 @@ +import { AnyEvent, ChainId } from "@grants-stack-indexer/shared"; + +export type ProcessedEvent = { + chainId: ChainId; + blockNumber: number; + blockTimestamp: number; + logIndex: number; + rawEvent?: AnyEvent; +}; + +export type NewProcessedEvent = Omit; diff --git a/packages/repository/src/types/index.ts b/packages/repository/src/types/index.ts index c34de7d..9832417 100644 --- a/packages/repository/src/types/index.ts +++ b/packages/repository/src/types/index.ts @@ -5,3 +5,4 @@ export * from "./changeset.types.js"; export * from "./donation.types.js"; export * from "./applicationPayout.types.js"; export * from "./strategy.types.js"; +export * from "./event.types.js"; From 0f4214f7a64483fbd216cf73c62efdc15eae355d Mon Sep 17 00:00:00 2001 From: nigiri <168690269+0xnigir1@users.noreply.github.com> Date: Tue, 17 Dec 2024 11:27:38 -0300 Subject: [PATCH 2/5] chore: event registry table migration --- .../20241216T160549_event_registry.ts | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 scripts/migrations/src/migrations/20241216T160549_event_registry.ts diff --git a/scripts/migrations/src/migrations/20241216T160549_event_registry.ts b/scripts/migrations/src/migrations/20241216T160549_event_registry.ts new file mode 100644 index 0000000..0d57acc --- /dev/null +++ b/scripts/migrations/src/migrations/20241216T160549_event_registry.ts @@ -0,0 +1,27 @@ +import { Kysely } from "kysely"; + +/** + * The up function is called when you update your database schema to the next version and down when you go back to previous version. + * The only argument for the functions is an instance of Kysely. It's important to use Kysely and not Kysely. + * ref: https://kysely.dev/docs/migrations#migration-files + */ +// eslint-disable-next-line @typescript-eslint/no-explicit-any +export async function up(db: Kysely): Promise { + const CHAIN_ID_TYPE = "integer"; + + await db.schema + .createTable("events") + .addColumn("chainId", CHAIN_ID_TYPE) + .addColumn("blockNumber", "integer") + .addColumn("blockTimestamp", "integer") + .addColumn("logIndex", "integer") + .addColumn("rawEvent", "jsonb") + .addPrimaryKeyConstraint("events_pkey", ["chainId"]) + .execute(); +} + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +export async function down(db: Kysely): Promise { + // Drop everything in reverse order + await db.schema.dropTable("events").execute(); +} From c996d08259007cdaa58a3754ab9cc92baec094dc Mon Sep 17 00:00:00 2001 From: nigiri <168690269+0xnigir1@users.noreply.github.com> Date: Tue, 17 Dec 2024 15:22:19 -0300 Subject: [PATCH 3/5] feat: update processing service to use events registry and add tests --- .../src/services/processing.service.ts | 17 +-- .../services/sharedDependencies.service.ts | 2 +- .../test/unit/processing.service.spec.ts | 48 +++++++- .../unit/sharedDependencies.service.spec.ts | 16 ++- packages/data-flow/src/external.ts | 1 - packages/data-flow/src/internal.ts | 1 - packages/data-flow/src/orchestrator.ts | 5 +- .../src/registries/event/eventsRegistry.ts | 30 ----- .../data-flow/src/registries/event/index.ts | 1 - .../strategy/cachedStrategyRegistry.ts | 10 +- .../registries/cachedEventRegistry.spec.ts | 107 ++++++++++++++++++ .../test/registries/dbEventRegistry.spec.ts | 66 +++++++++++ .../test/unit/eventsRegistry.spec.ts | 105 ----------------- .../data-flow/test/unit/orchestrator.spec.ts | 19 +--- .../kysely/eventRegistry.repository.ts | 13 ++- packages/repository/src/types/event.types.ts | 4 +- 16 files changed, 273 insertions(+), 172 deletions(-) delete mode 100644 packages/data-flow/src/registries/event/eventsRegistry.ts create mode 100644 packages/data-flow/test/registries/cachedEventRegistry.spec.ts create mode 100644 packages/data-flow/test/registries/dbEventRegistry.spec.ts delete mode 100644 packages/data-flow/test/unit/eventsRegistry.spec.ts diff --git a/apps/processing/src/services/processing.service.ts b/apps/processing/src/services/processing.service.ts index 5725a95..0cbe94a 100644 --- a/apps/processing/src/services/processing.service.ts +++ b/apps/processing/src/services/processing.service.ts @@ -16,8 +16,10 @@ import { SharedDependencies, SharedDependenciesService } from "./index.js"; /** * Processor service application * - Initializes core dependencies (repositories, providers) via SharedDependenciesService + * - Initializes a StrategyRegistry and loads it with strategies from the database * For each chain: * - Sets up EVM provider with configured RPC endpoints + * - Instantiates an EventsRegistry and loads it with the last processed event for the chain * - Creates an Orchestrator instance to coordinate an specific chain: * - Fetching on-chain events via indexer client * - Processing events through registered handlers @@ -51,19 +53,20 @@ export class ProcessingService { strategyRegistryRepository, ), ); + const eventsRegistry = new DatabaseEventRegistry( + new Logger({ className: "DatabaseEventRegistry" }), + eventRegistryRepository, + ); for (const chain of chains) { const chainLogger = new Logger({ chainId: chain.id as ChainId }); // Initialize EVM provider const evmProvider = new EvmProvider(chain.rpcUrls, optimism, chainLogger); - // Initialize events registry - const eventsRegistry = await InMemoryCachedEventRegistry.initialize( + // Initialize events registry for the chain + const cachedEventsRegistry = await InMemoryCachedEventRegistry.initialize( new Logger({ className: "InMemoryCachedEventRegistry" }), - new DatabaseEventRegistry( - new Logger({ className: "DatabaseEventRegistry" }), - eventRegistryRepository, - ), + eventsRegistry, [chain.id as ChainId], ); @@ -74,7 +77,7 @@ export class ProcessingService { { ...core, evmProvider }, indexerClient, { - eventsRegistry, + eventsRegistry: cachedEventsRegistry, strategyRegistry, }, chain.fetchLimit, diff --git a/apps/processing/src/services/sharedDependencies.service.ts b/apps/processing/src/services/sharedDependencies.service.ts index 033c386..3fb29fe 100644 --- a/apps/processing/src/services/sharedDependencies.service.ts +++ b/apps/processing/src/services/sharedDependencies.service.ts @@ -31,7 +31,7 @@ export type SharedDependencies = { /** * Shared dependencies service * - Initializes core dependencies (repositories, providers) - * - Initializes registries + * - Initializes registries repositories * - Initializes indexer client */ export class SharedDependenciesService { diff --git a/apps/processing/test/unit/processing.service.spec.ts b/apps/processing/test/unit/processing.service.spec.ts index 6623fde..15b59ff 100644 --- a/apps/processing/test/unit/processing.service.spec.ts +++ b/apps/processing/test/unit/processing.service.spec.ts @@ -1,7 +1,13 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { EvmProvider } from "@grants-stack-indexer/chain-providers"; -import { Orchestrator } from "@grants-stack-indexer/data-flow"; +import { + DatabaseEventRegistry, + DatabaseStrategyRegistry, + InMemoryCachedEventRegistry, + InMemoryCachedStrategyRegistry, + Orchestrator, +} from "@grants-stack-indexer/data-flow"; import type { Environment } from "../../src/config/env.js"; import { ProcessingService } from "../../src/services/processing.service.js"; @@ -10,7 +16,7 @@ vi.mock("../../src/services/sharedDependencies.service.js", () => ({ SharedDependenciesService: { initialize: vi.fn(() => ({ core: {}, - registries: {}, + registriesRepositories: {}, indexerClient: {}, kyselyDatabase: { destroy: vi.fn(), @@ -23,6 +29,39 @@ vi.mock("@grants-stack-indexer/chain-providers", () => ({ EvmProvider: vi.fn(), })); +vi.mock("@grants-stack-indexer/data-flow", async (importOriginal) => { + const actual = await importOriginal(); + const mockStrategyRegistry = { + getStrategies: vi.fn(), + getStrategyId: vi.fn(), + saveStrategyId: vi.fn(), + }; + + const mockEventRegistry = { + getLastProcessedEvent: vi.fn(), + saveLastProcessedEvent: vi.fn(), + }; + + return { + ...actual, + InMemoryCachedStrategyRegistry: { + initialize: vi.fn().mockResolvedValue(mockStrategyRegistry), + }, + DatabaseStrategyRegistry: vi.fn().mockImplementation(() => ({ + getStrategies: vi.fn(), + getStrategyId: vi.fn(), + saveStrategyId: vi.fn(), + })), + DatabaseEventRegistry: vi.fn().mockImplementation(() => ({ + getLastProcessedEvent: vi.fn(), + saveLastProcessedEvent: vi.fn(), + })), + InMemoryCachedEventRegistry: { + initialize: vi.fn().mockResolvedValue(mockEventRegistry), + }, + }; +}); + vi.spyOn(Orchestrator.prototype, "run").mockImplementation(async function (signal: AbortSignal) { while (!signal.aborted) { await new Promise((resolve) => setTimeout(resolve, 100)); @@ -62,7 +101,12 @@ describe("ProcessingService", () => { }); it("initializes multiple orchestrators correctly", () => { + expect(InMemoryCachedStrategyRegistry.initialize).toHaveBeenCalledTimes(1); + expect(DatabaseStrategyRegistry).toHaveBeenCalledTimes(1); + expect(DatabaseEventRegistry).toHaveBeenCalledTimes(1); expect(EvmProvider).toHaveBeenCalledTimes(2); + expect(InMemoryCachedEventRegistry.initialize).toHaveBeenCalledTimes(2); + // Verify orchestrators were created with correct parameters expect(processingService["orchestrators"].size).toBe(2); diff --git a/apps/processing/test/unit/sharedDependencies.service.spec.ts b/apps/processing/test/unit/sharedDependencies.service.spec.ts index e1b2ab0..bff277e 100644 --- a/apps/processing/test/unit/sharedDependencies.service.spec.ts +++ b/apps/processing/test/unit/sharedDependencies.service.spec.ts @@ -22,6 +22,7 @@ vi.mock("@grants-stack-indexer/repository", () => ({ getStrategyId: vi.fn(), saveStrategyId: vi.fn(), })), + KyselyEventRegistryRepository: vi.fn(), })); vi.mock("@grants-stack-indexer/pricing", () => ({ @@ -45,8 +46,12 @@ vi.mock("@grants-stack-indexer/data-flow", () => { saveStrategyId: vi.fn(), }; + const mockEventRegistry = { + getLastProcessedEvent: vi.fn(), + saveLastProcessedEvent: vi.fn(), + }; + return { - InMemoryEventsRegistry: vi.fn(), InMemoryCachedStrategyRegistry: { initialize: vi.fn().mockResolvedValue(mockStrategyRegistry), }, @@ -55,6 +60,13 @@ vi.mock("@grants-stack-indexer/data-flow", () => { getStrategyId: vi.fn(), saveStrategyId: vi.fn(), })), + DatabaseEventRegistry: vi.fn().mockImplementation(() => ({ + getLastProcessedEvent: vi.fn(), + saveLastProcessedEvent: vi.fn(), + })), + InMemoryCachedEventRegistry: { + initialize: vi.fn().mockResolvedValue(mockEventRegistry), + }, }; }); @@ -98,7 +110,7 @@ describe("SharedDependenciesService", () => { // Verify structure of returned dependencies expect(dependencies).toHaveProperty("core"); - expect(dependencies).toHaveProperty("registries"); + expect(dependencies).toHaveProperty("registriesRepositories"); expect(dependencies).toHaveProperty("indexerClient"); expect(dependencies).toHaveProperty("kyselyDatabase"); diff --git a/packages/data-flow/src/external.ts b/packages/data-flow/src/external.ts index 1d69f49..954c0b7 100644 --- a/packages/data-flow/src/external.ts +++ b/packages/data-flow/src/external.ts @@ -1,6 +1,5 @@ export { DataLoader, - InMemoryEventsRegistry, InMemoryCachedStrategyRegistry, InMemoryCachedEventRegistry, DatabaseEventRegistry, diff --git a/packages/data-flow/src/internal.ts b/packages/data-flow/src/internal.ts index ad35306..9224340 100644 --- a/packages/data-flow/src/internal.ts +++ b/packages/data-flow/src/internal.ts @@ -6,6 +6,5 @@ export * from "./utils/index.js"; export * from "./data-loader/index.js"; export * from "./eventsFetcher.js"; export * from "./registries/index.js"; -export * from "./registries/event/eventsRegistry.js"; export * from "./eventsProcessor.js"; export * from "./orchestrator.js"; diff --git a/packages/data-flow/src/orchestrator.ts b/packages/data-flow/src/orchestrator.ts index 2eb2266..0a2f53e 100644 --- a/packages/data-flow/src/orchestrator.ts +++ b/packages/data-flow/src/orchestrator.ts @@ -110,7 +110,10 @@ export class Orchestrator { await delay(this.fetchDelayInMs); continue; } - await this.eventsRegistry.saveLastProcessedEvent(this.chainId, event); + await this.eventsRegistry.saveLastProcessedEvent(this.chainId, { + ...event, + rawEvent: event, + }); event = await this.enhanceStrategyId(event); if (this.isPoolCreated(event)) { diff --git a/packages/data-flow/src/registries/event/eventsRegistry.ts b/packages/data-flow/src/registries/event/eventsRegistry.ts deleted file mode 100644 index 0bce55e..0000000 --- a/packages/data-flow/src/registries/event/eventsRegistry.ts +++ /dev/null @@ -1,30 +0,0 @@ -import type { ChainId, ILogger } from "@grants-stack-indexer/shared"; -import { NewProcessedEvent, ProcessedEvent } from "@grants-stack-indexer/repository"; -import { stringify } from "@grants-stack-indexer/shared"; - -import type { IEventsRegistry } from "../../internal.js"; - -/** - * Class to store the last processed event in memory - */ -//TODO: Implement storage version to persist the last processed event. we need to store it by chainId -export class InMemoryEventsRegistry implements IEventsRegistry { - private lastProcessedEvent: Map = new Map(); - - constructor(private logger: ILogger) {} - - /** - * @inheritdoc - */ - async getLastProcessedEvent(chainId: ChainId): Promise { - return this.lastProcessedEvent.get(chainId); - } - - /** - * @inheritdoc - */ - async saveLastProcessedEvent(chainId: ChainId, event: NewProcessedEvent): Promise { - this.logger.debug(`Saving last processed event: ${stringify(event, undefined, 4)}`); - this.lastProcessedEvent.set(chainId, { ...event, chainId }); - } -} diff --git a/packages/data-flow/src/registries/event/index.ts b/packages/data-flow/src/registries/event/index.ts index 47653b0..547781d 100644 --- a/packages/data-flow/src/registries/event/index.ts +++ b/packages/data-flow/src/registries/event/index.ts @@ -1,3 +1,2 @@ -export * from "./eventsRegistry.js"; export * from "./cachedEventRegistry.js"; export * from "./dbEventRegistry.js"; diff --git a/packages/data-flow/src/registries/strategy/cachedStrategyRegistry.ts b/packages/data-flow/src/registries/strategy/cachedStrategyRegistry.ts index 16899cc..05cfec8 100644 --- a/packages/data-flow/src/registries/strategy/cachedStrategyRegistry.ts +++ b/packages/data-flow/src/registries/strategy/cachedStrategyRegistry.ts @@ -58,7 +58,11 @@ export class InMemoryCachedStrategyRegistry implements IStrategyRegistry { const strategy = await this.strategyRegistry.getStrategyId(chainId, strategyAddress); if (strategy) { - this.cache.get(chainId)?.set(strategyAddress, strategy); + if (!this.cache.has(strategy.chainId)) { + this.cache.set(strategy.chainId, new Map()); + } + + this.cache.get(strategy.chainId)?.set(strategyAddress, strategy); } return strategy; } @@ -79,6 +83,10 @@ export class InMemoryCachedStrategyRegistry implements IStrategyRegistry { ); await this.strategyRegistry.saveStrategyId(chainId, strategyAddress, strategyId, handled); + if (!this.cache.has(chainId)) { + this.cache.set(chainId, new Map()); + } + this.cache.get(chainId)?.set(strategyAddress, { address: strategyAddress, id: strategyId, diff --git a/packages/data-flow/test/registries/cachedEventRegistry.spec.ts b/packages/data-flow/test/registries/cachedEventRegistry.spec.ts new file mode 100644 index 0000000..b06de51 --- /dev/null +++ b/packages/data-flow/test/registries/cachedEventRegistry.spec.ts @@ -0,0 +1,107 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { NewProcessedEvent, ProcessedEvent } from "@grants-stack-indexer/repository"; +import { ChainId, ILogger } from "@grants-stack-indexer/shared"; + +import { IEventsRegistry } from "../../src/internal.js"; +import { InMemoryCachedEventRegistry } from "../../src/registries/event/cachedEventRegistry.js"; + +describe("InMemoryCachedEventRegistry", () => { + const logger: ILogger = { + debug: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + }; + + const mockEventRegistry: IEventsRegistry = { + getLastProcessedEvent: vi.fn(), + saveLastProcessedEvent: vi.fn(), + }; + + const chainId = 1 as ChainId; + const mockEvent: ProcessedEvent = { + chainId, + blockNumber: 100, + blockTimestamp: 1234567890, + logIndex: 1, + }; + + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("initialize with existing events", async () => { + vi.mocked(mockEventRegistry.getLastProcessedEvent).mockResolvedValue(mockEvent); + + const registry = await InMemoryCachedEventRegistry.initialize(logger, mockEventRegistry, [ + chainId, + ]); + + const cached = await registry.getLastProcessedEvent(chainId); + expect(cached).toEqual(mockEvent); + expect(mockEventRegistry.getLastProcessedEvent).toHaveBeenCalledTimes(1); + }); + + it("fetch from underlying registry when not in cache", async () => { + vi.mocked(mockEventRegistry.getLastProcessedEvent) + .mockResolvedValueOnce(undefined) // For initialization + .mockResolvedValueOnce(mockEvent); // For actual fetch + + const registry = await InMemoryCachedEventRegistry.initialize(logger, mockEventRegistry, [ + chainId, + ]); + + const result = await registry.getLastProcessedEvent(chainId); + expect(result).toEqual(mockEvent); + expect(mockEventRegistry.getLastProcessedEvent).toHaveBeenCalledTimes(2); + }); + + it("save event and update cache", async () => { + const registry = await InMemoryCachedEventRegistry.initialize(logger, mockEventRegistry, [ + chainId, + ]); + + const newEvent: NewProcessedEvent = { + blockNumber: 200, + blockTimestamp: 1234577890, + logIndex: 2, + }; + + await registry.saveLastProcessedEvent(chainId, newEvent); + + // Verify the event was saved to underlying registry + expect(mockEventRegistry.saveLastProcessedEvent).toHaveBeenCalledWith(chainId, newEvent); + + // Verify the cache was updated + const cached = await registry.getLastProcessedEvent(chainId); + expect(cached).toEqual({ + ...newEvent, + chainId, + }); + + // Verify no additional calls to underlying registry + expect(mockEventRegistry.getLastProcessedEvent).toHaveBeenCalledTimes(1); + }); + + it("initialize with multiple chain ids", async () => { + const chainId2 = 5 as ChainId; + const mockEvent2: ProcessedEvent = { ...mockEvent, chainId: chainId2 }; + + vi.mocked(mockEventRegistry.getLastProcessedEvent).mockImplementation(async (chain) => + chain === chainId ? mockEvent : mockEvent2, + ); + + const registry = await InMemoryCachedEventRegistry.initialize(logger, mockEventRegistry, [ + chainId, + chainId2, + ]); + + const cached1 = await registry.getLastProcessedEvent(chainId); + const cached2 = await registry.getLastProcessedEvent(chainId2); + + expect(cached1).toEqual(mockEvent); + expect(cached2).toEqual(mockEvent2); + expect(mockEventRegistry.getLastProcessedEvent).toHaveBeenCalledTimes(2); + }); +}); diff --git a/packages/data-flow/test/registries/dbEventRegistry.spec.ts b/packages/data-flow/test/registries/dbEventRegistry.spec.ts new file mode 100644 index 0000000..679c000 --- /dev/null +++ b/packages/data-flow/test/registries/dbEventRegistry.spec.ts @@ -0,0 +1,66 @@ +import { describe, expect, it, vi } from "vitest"; + +import { + IEventRegistryRepository, + NewProcessedEvent, + ProcessedEvent, +} from "@grants-stack-indexer/repository"; +import { ChainId, ILogger } from "@grants-stack-indexer/shared"; + +import { DatabaseEventRegistry } from "../../src/registries/event/dbEventRegistry.js"; + +describe("DatabaseEventRegistry", () => { + const logger: ILogger = { + debug: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + }; + + const mockEventRepository: IEventRegistryRepository = { + getLastProcessedEvent: vi.fn(), + saveLastProcessedEvent: vi.fn(), + }; + + const chainId = 1 as ChainId; + + it("return undefined for non-existent last processed event", async () => { + const registry = new DatabaseEventRegistry(logger, mockEventRepository); + + vi.mocked(mockEventRepository.getLastProcessedEvent).mockResolvedValue(undefined); + + const event = await registry.getLastProcessedEvent(chainId); + expect(event).toBeUndefined(); + expect(mockEventRepository.getLastProcessedEvent).toHaveBeenCalledWith(chainId); + }); + + it("return last processed event when it exists", async () => { + const registry = new DatabaseEventRegistry(logger, mockEventRepository); + const mockEvent: ProcessedEvent = { + chainId, + blockNumber: 100, + blockTimestamp: 1234567890, + logIndex: 1, + }; + + vi.mocked(mockEventRepository.getLastProcessedEvent).mockResolvedValue(mockEvent); + + const event = await registry.getLastProcessedEvent(chainId); + expect(event).toEqual(mockEvent); + expect(mockEventRepository.getLastProcessedEvent).toHaveBeenCalledWith(chainId); + }); + + it("save last processed event", async () => { + const registry = new DatabaseEventRegistry(logger, mockEventRepository); + const newEvent: NewProcessedEvent = { + blockNumber: 100, + blockTimestamp: 1234567890, + logIndex: 1, + }; + + vi.mocked(mockEventRepository.saveLastProcessedEvent).mockResolvedValue(); + + await registry.saveLastProcessedEvent(chainId, newEvent); + expect(mockEventRepository.saveLastProcessedEvent).toHaveBeenCalledWith(chainId, newEvent); + }); +}); diff --git a/packages/data-flow/test/unit/eventsRegistry.spec.ts b/packages/data-flow/test/unit/eventsRegistry.spec.ts deleted file mode 100644 index be14f7a..0000000 --- a/packages/data-flow/test/unit/eventsRegistry.spec.ts +++ /dev/null @@ -1,105 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; - -import { ChainId, ILogger, ProcessorEvent } from "@grants-stack-indexer/shared"; - -import { InMemoryEventsRegistry } from "../../src/registries/event/eventsRegistry.js"; - -describe("InMemoryEventsRegistry", () => { - const logger: ILogger = { - debug: vi.fn(), - error: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - }; - const chainId = 1 as ChainId; - it("return null when no event has been saved", async () => { - const registry = new InMemoryEventsRegistry(logger); - const lastEvent = await registry.getLastProcessedEvent(chainId); - expect(lastEvent).toBeUndefined(); - }); - - it("save and retrieve the last processed event", async () => { - const registry = new InMemoryEventsRegistry(logger); - const mockEvent: ProcessorEvent<"Allo", "PoolCreated"> = { - contractName: "Allo", - eventName: "PoolCreated", - blockNumber: 1, - blockTimestamp: 1234567890, - chainId: 1 as ChainId, - logIndex: 0, - srcAddress: "0x123", - strategyId: "0xstrategy", - params: { - poolId: "1", - profileId: "0x456", - strategy: "0x789", - token: "0xtoken", - amount: "0", - metadata: ["1", "0xmetadata"], - }, - transactionFields: { - hash: "0xabc", - transactionIndex: 0, - }, - }; - - await registry.saveLastProcessedEvent(chainId, mockEvent); - const retrievedEvent = await registry.getLastProcessedEvent(chainId); - - expect(retrievedEvent).toEqual(mockEvent); - }); - - it("updates the last processed event when saving multiple times", async () => { - const registry = new InMemoryEventsRegistry(logger); - - const firstEvent: ProcessorEvent<"Allo", "PoolCreated"> = { - contractName: "Allo", - eventName: "PoolCreated", - blockNumber: 1, - blockTimestamp: 1234567890, - chainId: 1 as ChainId, - logIndex: 0, - srcAddress: "0x123", - strategyId: "0xstrategy", - params: { - poolId: "1", - profileId: "0x456", - strategy: "0x789", - token: "0xtoken", - amount: "0", - metadata: ["1", "0xmetadata"], - }, - transactionFields: { - hash: "0xabc", - transactionIndex: 0, - }, - }; - - const secondEvent: ProcessorEvent<"Strategy", "RegisteredWithSender"> = { - contractName: "Strategy", - eventName: "RegisteredWithSender", - blockNumber: 1, - blockTimestamp: 1234567890, - chainId: 1 as ChainId, - logIndex: 0, - srcAddress: "0x123", - strategyId: "0xstrategy", - params: { - recipientId: "0xrecipient", - data: "0xdata", - sender: "0xsender", - }, - transactionFields: { - hash: "0xabc", - transactionIndex: 0, - }, - }; - - await registry.saveLastProcessedEvent(chainId, firstEvent); - await registry.saveLastProcessedEvent(chainId, secondEvent); - - const lastEvent = await registry.getLastProcessedEvent(chainId); - expect(lastEvent).toEqual(secondEvent); - expect(lastEvent).not.toEqual(firstEvent); - }); -}); diff --git a/packages/data-flow/test/unit/orchestrator.spec.ts b/packages/data-flow/test/unit/orchestrator.spec.ts index 7b492c2..7b90c04 100644 --- a/packages/data-flow/test/unit/orchestrator.spec.ts +++ b/packages/data-flow/test/unit/orchestrator.spec.ts @@ -178,14 +178,7 @@ describe("Orchestrator", { sequential: true }, () => { expect(eventsProcessorSpy).toHaveBeenCalledWith(mockEvents[0]); expect(eventsProcessorSpy).toHaveBeenCalledWith(mockEvents[1]); - expect(mockEventsRegistry.saveLastProcessedEvent).toHaveBeenCalledWith( - chainId, - mockEvents[0], - ); - expect(mockEventsRegistry.saveLastProcessedEvent).toHaveBeenCalledWith( - chainId, - mockEvents[1], - ); + expect(mockEventsRegistry.saveLastProcessedEvent).toHaveBeenCalledTimes(2); }); it("wait and keep polling on empty queue", async () => { @@ -274,10 +267,7 @@ describe("Orchestrator", { sequential: true }, () => { }); expect(orchestrator["dataLoader"].applyChanges).toHaveBeenCalledTimes(1); expect(orchestrator["dataLoader"].applyChanges).toHaveBeenCalledWith(changesets); - expect(mockEventsRegistry.saveLastProcessedEvent).toHaveBeenCalledWith( - chainId, - mockEvent, - ); + expect(mockEventsRegistry.saveLastProcessedEvent).toHaveBeenCalled(); }); it("save strategyId to registry on PoolCreated event", async () => { @@ -409,10 +399,7 @@ describe("Orchestrator", { sequential: true }, () => { ); expect(orchestrator["dataLoader"].applyChanges).toHaveBeenCalledTimes(1); expect(orchestrator["dataLoader"].applyChanges).toHaveBeenCalledWith(changesets); - expect(mockEventsRegistry.saveLastProcessedEvent).toHaveBeenCalledWith( - chainId, - mockEvent, - ); + expect(mockEventsRegistry.saveLastProcessedEvent).toHaveBeenCalled(); }); } diff --git a/packages/repository/src/repositories/kysely/eventRegistry.repository.ts b/packages/repository/src/repositories/kysely/eventRegistry.repository.ts index 2b77858..5a93502 100644 --- a/packages/repository/src/repositories/kysely/eventRegistry.repository.ts +++ b/packages/repository/src/repositories/kysely/eventRegistry.repository.ts @@ -23,11 +23,20 @@ export class KyselyEventRegistryRepository implements IEventRegistryRepository { /** @inheritdoc */ async saveLastProcessedEvent(chainId: ChainId, event: NewProcessedEvent): Promise { + const { blockNumber, blockTimestamp, logIndex, rawEvent } = event; // Extract only the fields from NewProcessedEvent await this.db .withSchema(this.schemaName) .insertInto("events") - .values({ ...event, chainId }) - .onConflict((oc) => oc.columns(["chainId"]).doUpdateSet({ ...event, chainId })) + .values({ blockNumber, blockTimestamp, logIndex, chainId, rawEvent }) + .onConflict((oc) => + oc.columns(["chainId"]).doUpdateSet({ + blockNumber, + blockTimestamp, + logIndex, + rawEvent, + chainId, + }), + ) .execute(); } } diff --git a/packages/repository/src/types/event.types.ts b/packages/repository/src/types/event.types.ts index 3f1eb50..1e0c5ca 100644 --- a/packages/repository/src/types/event.types.ts +++ b/packages/repository/src/types/event.types.ts @@ -1,11 +1,11 @@ -import { AnyEvent, ChainId } from "@grants-stack-indexer/shared"; +import { AnyEvent, ChainId, ContractName, ProcessorEvent } from "@grants-stack-indexer/shared"; export type ProcessedEvent = { chainId: ChainId; blockNumber: number; blockTimestamp: number; logIndex: number; - rawEvent?: AnyEvent; + rawEvent?: Partial>; }; export type NewProcessedEvent = Omit; From 66abdca4cc0f039aa4f743df41b0a6ff2a633861 Mon Sep 17 00:00:00 2001 From: nigiri <168690269+0xnigir1@users.noreply.github.com> Date: Thu, 19 Dec 2024 13:07:50 -0300 Subject: [PATCH 4/5] fix: change db table names --- packages/repository/src/db/connection.ts | 8 ++++---- .../src/repositories/kysely/eventRegistry.repository.ts | 4 ++-- .../repositories/kysely/strategyRegistry.repository.ts | 6 +++--- .../src/migrations/20241210T175001_strategy_registry.ts | 6 +++--- .../src/migrations/20241216T160549_event_registry.ts | 6 +++--- 5 files changed, 15 insertions(+), 15 deletions(-) diff --git a/packages/repository/src/db/connection.ts b/packages/repository/src/db/connection.ts index 9a774bb..d7398a4 100644 --- a/packages/repository/src/db/connection.ts +++ b/packages/repository/src/db/connection.ts @@ -12,7 +12,7 @@ import { Application, ApplicationPayout, Donation as DonationTable, - ProcessedEvent as EventTable, + ProcessedEvent as EventRegistryTable, MatchingDistribution, PendingProjectRole as PendingProjectRoleTable, PendingRoundRole as PendingRoundRoleTable, @@ -21,7 +21,7 @@ import { Round, RoundRole as RoundRoleTable, StatusSnapshot, - Strategy as StrategyTable, + Strategy as StrategyRegistryTable, } from "../internal.js"; const { Pool } = pg; @@ -61,8 +61,8 @@ export interface Database { applications: ApplicationTable; donations: DonationTable; applicationsPayouts: ApplicationPayoutTable; - strategies: StrategyTable; - events: EventTable; + strategiesRegistry: StrategyRegistryTable; + eventsRegistry: EventRegistryTable; } /** diff --git a/packages/repository/src/repositories/kysely/eventRegistry.repository.ts b/packages/repository/src/repositories/kysely/eventRegistry.repository.ts index 5a93502..5a88d8e 100644 --- a/packages/repository/src/repositories/kysely/eventRegistry.repository.ts +++ b/packages/repository/src/repositories/kysely/eventRegistry.repository.ts @@ -15,7 +15,7 @@ export class KyselyEventRegistryRepository implements IEventRegistryRepository { async getLastProcessedEvent(chainId: ChainId): Promise { return this.db .withSchema(this.schemaName) - .selectFrom("events") + .selectFrom("eventsRegistry") .where("chainId", "=", chainId) .selectAll() .executeTakeFirst(); @@ -26,7 +26,7 @@ export class KyselyEventRegistryRepository implements IEventRegistryRepository { const { blockNumber, blockTimestamp, logIndex, rawEvent } = event; // Extract only the fields from NewProcessedEvent await this.db .withSchema(this.schemaName) - .insertInto("events") + .insertInto("eventsRegistry") .values({ blockNumber, blockTimestamp, logIndex, chainId, rawEvent }) .onConflict((oc) => oc.columns(["chainId"]).doUpdateSet({ diff --git a/packages/repository/src/repositories/kysely/strategyRegistry.repository.ts b/packages/repository/src/repositories/kysely/strategyRegistry.repository.ts index a251f42..fe94580 100644 --- a/packages/repository/src/repositories/kysely/strategyRegistry.repository.ts +++ b/packages/repository/src/repositories/kysely/strategyRegistry.repository.ts @@ -18,7 +18,7 @@ export class KyselyStrategyRegistryRepository implements IStrategyRegistryReposi ): Promise { return this.db .withSchema(this.schemaName) - .selectFrom("strategies") + .selectFrom("strategiesRegistry") .where("chainId", "=", chainId) .where("address", "=", strategyAddress) .selectAll() @@ -29,7 +29,7 @@ export class KyselyStrategyRegistryRepository implements IStrategyRegistryReposi async saveStrategy(strategy: Strategy): Promise { await this.db .withSchema(this.schemaName) - .insertInto("strategies") + .insertInto("strategiesRegistry") .values(strategy) .onConflict((oc) => oc.columns(["chainId", "address"]).doUpdateSet(strategy)) .execute(); @@ -37,7 +37,7 @@ export class KyselyStrategyRegistryRepository implements IStrategyRegistryReposi /** @inheritdoc */ async getStrategies(filters?: { handled?: boolean; chainId?: ChainId }): Promise { - const query = this.db.withSchema(this.schemaName).selectFrom("strategies"); + const query = this.db.withSchema(this.schemaName).selectFrom("strategiesRegistry"); if (filters?.chainId) { query.where("chainId", "=", filters.chainId); diff --git a/scripts/migrations/src/migrations/20241210T175001_strategy_registry.ts b/scripts/migrations/src/migrations/20241210T175001_strategy_registry.ts index 45a5936..fbef0cb 100644 --- a/scripts/migrations/src/migrations/20241210T175001_strategy_registry.ts +++ b/scripts/migrations/src/migrations/20241210T175001_strategy_registry.ts @@ -11,17 +11,17 @@ export async function up(db: Kysely): Promise { const CHAIN_ID_TYPE = "integer"; await db.schema - .createTable("strategies") + .createTable("strategies_registry") .addColumn("address", ADDRESS_TYPE) .addColumn("id", "text") .addColumn("chainId", CHAIN_ID_TYPE) .addColumn("handled", "boolean") - .addPrimaryKeyConstraint("strategies_pkey", ["address", "chainId"]) + .addPrimaryKeyConstraint("strategies_registry_pkey", ["address", "chainId"]) .execute(); } // eslint-disable-next-line @typescript-eslint/no-explicit-any export async function down(db: Kysely): Promise { // Drop everything in reverse order - await db.schema.dropTable("strategies").execute(); + await db.schema.dropTable("strategies_registry").execute(); } diff --git a/scripts/migrations/src/migrations/20241216T160549_event_registry.ts b/scripts/migrations/src/migrations/20241216T160549_event_registry.ts index 0d57acc..3950876 100644 --- a/scripts/migrations/src/migrations/20241216T160549_event_registry.ts +++ b/scripts/migrations/src/migrations/20241216T160549_event_registry.ts @@ -10,18 +10,18 @@ export async function up(db: Kysely): Promise { const CHAIN_ID_TYPE = "integer"; await db.schema - .createTable("events") + .createTable("events_registry") .addColumn("chainId", CHAIN_ID_TYPE) .addColumn("blockNumber", "integer") .addColumn("blockTimestamp", "integer") .addColumn("logIndex", "integer") .addColumn("rawEvent", "jsonb") - .addPrimaryKeyConstraint("events_pkey", ["chainId"]) + .addPrimaryKeyConstraint("events_registry_pkey", ["chainId"]) .execute(); } // eslint-disable-next-line @typescript-eslint/no-explicit-any export async function down(db: Kysely): Promise { // Drop everything in reverse order - await db.schema.dropTable("events").execute(); + await db.schema.dropTable("events_registry").execute(); } From 60517009cae40ccd325267f6ab31f897fff9d1dd Mon Sep 17 00:00:00 2001 From: nigiri <168690269+0xnigir1@users.noreply.github.com> Date: Thu, 19 Dec 2024 13:09:09 -0300 Subject: [PATCH 5/5] test: when underlying registry throws for cache registry --- .../src/registries/event/cachedEventRegistry.ts | 1 - .../test/registries/cachedEventRegistry.spec.ts | 16 ++++++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/packages/data-flow/src/registries/event/cachedEventRegistry.ts b/packages/data-flow/src/registries/event/cachedEventRegistry.ts index 8f5bea8..0c3c344 100644 --- a/packages/data-flow/src/registries/event/cachedEventRegistry.ts +++ b/packages/data-flow/src/registries/event/cachedEventRegistry.ts @@ -37,7 +37,6 @@ export class InMemoryCachedEventRegistry implements IEventsRegistry { this.cache.set(chainId, { ...event, chainId }); } - /** /** * Creates a new cached event registry instance. It will load the events into memory and cache them and * fallback to the event registry if the event is not found in the cache. diff --git a/packages/data-flow/test/registries/cachedEventRegistry.spec.ts b/packages/data-flow/test/registries/cachedEventRegistry.spec.ts index b06de51..6e9caad 100644 --- a/packages/data-flow/test/registries/cachedEventRegistry.spec.ts +++ b/packages/data-flow/test/registries/cachedEventRegistry.spec.ts @@ -104,4 +104,20 @@ describe("InMemoryCachedEventRegistry", () => { expect(cached2).toEqual(mockEvent2); expect(mockEventRegistry.getLastProcessedEvent).toHaveBeenCalledTimes(2); }); + + it("throws error when underlying registry throws error", async () => { + vi.mocked(mockEventRegistry.saveLastProcessedEvent).mockRejectedValue( + new Error("Saving error"), + ); + + const registry = await InMemoryCachedEventRegistry.initialize(logger, mockEventRegistry, [ + chainId, + ]); + const cacheSetSpy = vi.spyOn(registry["cache"], "set"); + + await expect(registry.saveLastProcessedEvent(chainId, mockEvent)).rejects.toThrow( + "Saving error", + ); + expect(cacheSetSpy).not.toHaveBeenCalled(); + }); });