Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: retroactively handle strategy #45

Open
wants to merge 2 commits into
base: dev
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions apps/processing/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ Available scripts that can be run using `pnpm`:
| `start` | Run the compiled app from dist folder |
| `test` | Run tests using vitest |
| `test:cov` | Run tests with coverage report |
| `retroactive` | Run retroactive processing for all chains |

TODO: e2e tests
TODO: Docker image
1 change: 1 addition & 0 deletions apps/processing/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
"format:fix": "prettier --write \"{src,test}/**/*.{js,ts,json}\"",
"lint": "eslint \"{src,test}/**/*.{js,ts,json}\"",
"lint:fix": "pnpm lint --fix",
"retroactive": "tsx src/retroactiveHandleStrategies.ts",
"start": "node dist/index.js",
"test": "vitest run --config vitest.config.ts --passWithNoTests",
"test:cov": "vitest run --config vitest.config.ts --coverage --passWithNoTests"
Expand Down
33 changes: 33 additions & 0 deletions apps/processing/src/retroactiveHandleStrategies.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
import { inspect } from "util";

import { environment } from "./config/index.js";
import { ProcessingService } from "./services/processing.service.js";

let processor: ProcessingService;

const main = async (): Promise<void> => {
processor = await ProcessingService.initialize(environment);
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be good to add some way to communicate if it initialized successfully/unsuccessfully

Could either have an initialized log after success

Or could have a failed to initialize X service message in the initialize error handling

await processor.processRetroactiveEvents();
};

process.on("unhandledRejection", (reason, p) => {
console.error(`Unhandled Rejection at: \n${inspect(p, undefined, 100)}, \nreason: ${reason}`);
process.exit(1);
});

process.on("uncaughtException", (error: Error) => {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Type Error or unknown? 🤔

console.error(
`An uncaught exception occurred: ${error}\n` + `Exception origin: ${error.stack}`,
);
process.exit(1);
});

main()
.catch((err) => {
console.error(`Caught error in main handler: ${err}`);
process.exit(1);
})
// eslint-disable-next-line @typescript-eslint/no-misused-promises
.finally(async () => {
await processor?.releaseResources();
});
58 changes: 40 additions & 18 deletions apps/processing/src/services/processing.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import {
InMemoryCachedEventRegistry,
InMemoryCachedStrategyRegistry,
Orchestrator,
RetroactiveProcessor,
} from "@grants-stack-indexer/data-flow";
import { ChainId, Logger } from "@grants-stack-indexer/shared";

Expand All @@ -27,12 +28,12 @@ import { SharedDependencies, SharedDependenciesService } from "./index.js";
* - Manages graceful shutdown on termination signals
*/
export class ProcessingService {
private readonly orchestrators: Map<ChainId, Orchestrator> = new Map();
private readonly orchestrators: Map<ChainId, [Orchestrator, RetroactiveProcessor]> = new Map();
private readonly logger = new Logger({ className: "ProcessingService" });
private readonly kyselyDatabase: SharedDependencies["kyselyDatabase"];

private constructor(
orchestrators: Map<ChainId, Orchestrator>,
orchestrators: Map<ChainId, [Orchestrator, RetroactiveProcessor]>,
kyselyDatabase: SharedDependencies["kyselyDatabase"],
) {
this.orchestrators = orchestrators;
Expand All @@ -43,8 +44,12 @@ export class ProcessingService {
const sharedDependencies = await SharedDependenciesService.initialize(env);
const { CHAINS: chains } = env;
const { core, registriesRepositories, indexerClient, kyselyDatabase } = sharedDependencies;
const { eventRegistryRepository, strategyRegistryRepository } = registriesRepositories;
const orchestrators: Map<ChainId, Orchestrator> = new Map();
const {
eventRegistryRepository,
strategyRegistryRepository,
strategyProcessingCheckpointRepository,
} = registriesRepositories;
const orchestrators: Map<ChainId, [Orchestrator, RetroactiveProcessor]> = new Map();

const strategyRegistry = await InMemoryCachedStrategyRegistry.initialize(
new Logger({ className: "InMemoryCachedStrategyRegistry" }),
Expand All @@ -70,21 +75,32 @@ export class ProcessingService {
[chain.id as ChainId],
);

orchestrators.set(
const orchestrator = new Orchestrator(
chain.id as ChainId,
{ ...core, evmProvider },
indexerClient,
{
eventsRegistry: cachedEventsRegistry,
strategyRegistry,
},
chain.fetchLimit,
chain.fetchDelayMs,
chainLogger,
);
const retroactiveProcessor = new RetroactiveProcessor(
chain.id as ChainId,
new Orchestrator(
chain.id as ChainId,
{ ...core, evmProvider },
indexerClient,
{
eventsRegistry: cachedEventsRegistry,
strategyRegistry,
},
chain.fetchLimit,
chain.fetchDelayMs,
chainLogger,
),
{ ...core, evmProvider },
indexerClient,
{
eventsRegistry: cachedEventsRegistry,
strategyRegistry,
checkpointRepository: strategyProcessingCheckpointRepository,
},
chain.fetchLimit,
chainLogger,
);

orchestrators.set(chain.id as ChainId, [orchestrator, retroactiveProcessor]);
}

return new ProcessingService(orchestrators, kyselyDatabase);
Expand Down Expand Up @@ -114,7 +130,7 @@ export class ProcessingService {
});

try {
for (const orchestrator of this.orchestrators.values()) {
for (const [orchestrator, _] of this.orchestrators.values()) {
this.logger.info(`Starting orchestrator for chain ${orchestrator.chainId}...`);
orchestratorProcesses.push(orchestrator.run(abortController.signal));
}
Expand All @@ -126,6 +142,12 @@ export class ProcessingService {
}
}

async processRetroactiveEvents(): Promise<void> {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

docs here?

for (const [_, retroactiveProcessor] of this.orchestrators.values()) {
await retroactiveProcessor.processRetroactiveStrategies();
}
}

/**
* Call this function when the processor service is terminated
* - Releases database resources
Expand Down
7 changes: 7 additions & 0 deletions apps/processing/src/services/sharedDependencies.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,15 @@ import { PricingProviderFactory } from "@grants-stack-indexer/pricing";
import {
createKyselyDatabase,
IEventRegistryRepository,
IStrategyProcessingCheckpointRepository,
IStrategyRegistryRepository,
KyselyApplicationPayoutRepository,
KyselyApplicationRepository,
KyselyDonationRepository,
KyselyEventRegistryRepository,
KyselyProjectRepository,
KyselyRoundRepository,
KyselyStrategyProcessingCheckpointRepository,
KyselyStrategyRegistryRepository,
} from "@grants-stack-indexer/repository";
import { Logger } from "@grants-stack-indexer/shared";
Expand All @@ -23,6 +25,7 @@ export type SharedDependencies = {
registriesRepositories: {
eventRegistryRepository: IEventRegistryRepository;
strategyRegistryRepository: IStrategyRegistryRepository;
strategyProcessingCheckpointRepository: IStrategyProcessingCheckpointRepository;
};
indexerClient: EnvioIndexerClient;
kyselyDatabase: ReturnType<typeof createKyselyDatabase>;
Expand Down Expand Up @@ -73,6 +76,9 @@ export class SharedDependenciesService {
env.DATABASE_SCHEMA,
);

const strategyProcessingCheckpointRepository =
new KyselyStrategyProcessingCheckpointRepository(kyselyDatabase, env.DATABASE_SCHEMA);

// Initialize indexer client
const indexerClient = new EnvioIndexerClient(
env.INDEXER_GRAPHQL_URL,
Expand All @@ -92,6 +98,7 @@ export class SharedDependenciesService {
registriesRepositories: {
eventRegistryRepository,
strategyRegistryRepository,
strategyProcessingCheckpointRepository,
},
indexerClient,
kyselyDatabase,
Expand Down
Loading
Loading