Merge branch 'feat/optimize-metadata-bootstrap' into fix/pending-round-roles

Author: 0xkenj1

.env.example

@@ -69,10 +69,10 @@ INDEXER_ADMIN_SECRET=testing
 METADATA_SOURCE=public-gateway
 PUBLIC_GATEWAY_URLS=["https://ipfs.io","https://dweb.link","https://cloudflare-ipfs.com","https://gateway.pinata.cloud","https://ipfs.infura.io","https://ipfs.fleek.co","https://ipfs.eth.aragon.network","https://ipfs.jes.xxx","https://ipfs.lol","https://ipfs.mle.party"]
 
-RETRY_MAX_ATTEMPTS=100000
-RETRY_BASE_DELAY_MS=200
-RETRY_MAX_DELAY_MS=1000
-RETRY_FACTOR=1.5
+RETRY_MAX_ATTEMPTS = 10
+RETRY_BASE_DELAY_MS = 200
+RETRY_MAX_DELAY_MS = 1000
+RETRY_FACTOR = 1.5
 
 PRICING_SOURCE=coingecko
 COINGECKO_API_KEY={{YOUR_PRO_KEY}}

apps/processing/.env.example

@@ -18,4 +18,7 @@ COINGECKO_API_TYPE=demo
 RETRY_MAX_ATTEMPTS=3
 RETRY_BASE_DELAY_MS=3000
 RETRY_FACTOR=2
-RETRY_MAX_DELAY_MS=300000
+RETRY_MAX_DELAY_MS=300000
+
+NOTIFIER_PROVIDER=# 'slack' or 'null'
+SLACK_WEBHOOK_URL=

apps/processing/README.md

@@ -40,6 +40,8 @@ Available options:
 | `RETRY_BASE_DELAY_MS` | Base delay for retry attempts | 3000 | No | |
 | `RETRY_FACTOR` | Delay factor for retry attempts | 2 | No | |
 | `RETRY_MAX_DELAY_MS` | Maximum delay for retry attempts | 300000 | No | |
+| `NOTIFIER_PROVIDER` | Notifier provider (slack or null) | null | No | |
+| `SLACK_WEBHOOK_URL` | Slack webhook URL | N/A | Only if NOTIFIER_PROVIDER is slack | |
 
 ## Available Scripts
 

apps/processing/src/config/env.ts

@@ -39,6 +39,9 @@ const baseSchema = z.object({
     RETRY_BASE_DELAY_MS: z.coerce.number().int().min(1).default(3000), // 3 seconds
     RETRY_FACTOR: z.coerce.number().min(1).default(2),
     RETRY_MAX_DELAY_MS: z.coerce.number().int().min(1).optional(), // 5 minute
+    NOTIFIER_PROVIDER: z.enum(["slack", "null"]).default("null"),
+    SLACK_WEBHOOK_URL: z.string().url().optional(),
+    LOG_LEVEL: z.enum(["error", "warn", "info", "debug"]).default("info"),
 });
 
 const dummyPricingSchema = baseSchema.extend({
@@ -75,6 +78,15 @@ const validationSchema = z
             ...val,
         };
     })
+    .superRefine((data, ctx) => {
+        if (data.NOTIFIER_PROVIDER === "slack" && !data.SLACK_WEBHOOK_URL) {
+            ctx.addIssue({
+                code: "custom",
+                message: "SLACK_WEBHOOK_URL is required when NOTIFIER_PROVIDER is slack",
+                path: ["SLACK_WEBHOOK_URL"],
+            });
+        }
+    })
     .and(
         z
             .discriminatedUnion("METADATA_SOURCE", [

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

@@ -51,6 +51,7 @@ export class ProcessingService {
             kyselyDatabase,
             logger,
             retryStrategy,
+            notifier,
         } = sharedDependencies;
         const {
             eventRegistryRepository,
@@ -84,6 +85,7 @@ export class ProcessingService {
                 chain.fetchDelayMs,
                 retryStrategy,
                 logger,
+                notifier,
             );
             const retroactiveProcessor = new RetroactiveProcessor(
                 chain.id as ChainId,

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

@@ -21,7 +21,16 @@ import {
     KyselyStrategyRegistryRepository,
     KyselyTransactionManager,
 } from "@grants-stack-indexer/repository";
-import { ExponentialBackoff, ILogger, Logger, RetryStrategy } from "@grants-stack-indexer/shared";
+import {
+    ExponentialBackoff,
+    ILogger,
+    INotifier,
+    Logger,
+    NotifierConfig,
+    NotifierFactory,
+    NotifierProvider,
+    RetryStrategy,
+} from "@grants-stack-indexer/shared";
 
 import { Environment } from "../config/index.js";
 
@@ -36,6 +45,7 @@ export type SharedDependencies = {
     kyselyDatabase: ReturnType<typeof createKyselyDatabase>;
     retryStrategy: RetryStrategy;
     logger: ILogger;
+    notifier: INotifier;
 };
 
 /**
@@ -48,6 +58,8 @@ export class SharedDependenciesService {
     static async initialize(env: Environment): Promise<SharedDependencies> {
         const logger = Logger.getInstance();
 
+        const notifier = NotifierFactory.create(this.getNotifierOptions(env), logger);
+
         // Initialize repositories
         const kyselyDatabase = createKyselyDatabase(
             {
@@ -151,6 +163,26 @@ export class SharedDependenciesService {
             kyselyDatabase,
             retryStrategy,
             logger,
+            notifier,
+        };
+    }
+
+    private static getNotifierOptions(env: Environment): NotifierConfig<NotifierProvider> {
+        if (env.NOTIFIER_PROVIDER === "slack") {
+            if (!env.SLACK_WEBHOOK_URL) {
+                throw new Error("SLACK_WEBHOOK_URL is required when NOTIFIER_PROVIDER is 'slack'");
+            }
+
+            return {
+                notifierProvider: env.NOTIFIER_PROVIDER,
+                opts: {
+                    webhookUrl: env.SLACK_WEBHOOK_URL,
+                },
+            };
+        }
+
+        return {
+            notifierProvider: "null",
         };
     }
 }

packages/data-flow/src/constants.ts

@@ -1,18 +1,3 @@
-/**
- * Maximum number of retries for bulk fetching metadata.
- */
-export const MAX_BULK_FETCH_METADATA_RETRIES = 10;
-
-/**
- * Base delay in milliseconds for bulk fetching metadata retries.
- */
-export const METADATA_BULK_FETCH_BASE_DELAY_MS = 1000;
-
-/**
- * Backoff factor for bulk fetching metadata retries.
- */
-export const METADATA_BULK_FETCH_BACKOFF_FACTOR = 1.5;
-
 /**
  * Maximum concurrency for bulk fetching metadata.
  */

packages/data-flow/src/orchestrator.ts

@@ -1,7 +1,5 @@
 import { isNativeError } from "util/types";
 import pMap from "p-map";
-import { retryAsyncUntilDefined } from "ts-retry";
-import { DelayParameters } from "ts-retry/lib/cjs/retry/options.js";
 
 import { IIndexerClient } from "@grants-stack-indexer/indexer-client";
 import { TokenPrice } from "@grants-stack-indexer/pricing";
@@ -24,6 +22,7 @@ import {
     ContractName,
     Hex,
     ILogger,
+    INotifier,
     isAlloEvent,
     isStrategyEvent,
     ProcessorEvent,
@@ -38,12 +37,7 @@ import {
 } from "@grants-stack-indexer/shared";
 
 import type { IEventsFetcher, IStrategyRegistry } from "./interfaces/index.js";
-import {
-    MAX_BULK_FETCH_METADATA_CONCURRENCY,
-    MAX_BULK_FETCH_METADATA_RETRIES,
-    METADATA_BULK_FETCH_BACKOFF_FACTOR,
-    METADATA_BULK_FETCH_BASE_DELAY_MS,
-} from "./constants.js";
+import { MAX_BULK_FETCH_METADATA_CONCURRENCY } from "./constants.js";
 import { EventsFetcher } from "./eventsFetcher.js";
 import { EventsProcessor } from "./eventsProcessor.js";
 import { InvalidEvent } from "./exceptions/index.js";
@@ -79,8 +73,6 @@ type TokenWithTimestamps = {
  * - Retry handling with exponential backoff for transient failures
  * - Comprehensive error handling and logging for various failure scenarios
  * - Registry tracking of supported/unsupported strategies and events
- *
- * TODO: Enhance logging and observability
  */
 export class Orchestrator {
     private readonly eventsQueue: IQueue<ProcessorEvent<ContractName, AnyEvent>>;
@@ -113,6 +105,7 @@ export class Orchestrator {
         private fetchDelayInMs: number = 10000,
         private retryStrategy: RetryStrategy,
         private logger: ILogger,
+        private notifier: INotifier,
         private environment: "development" | "staging" | "production" = "development",
     ) {
         this.eventsFetcher = new EventsFetcher(this.indexerClient);
@@ -199,7 +192,6 @@ export class Orchestrator {
                         rawEvent: event,
                     });
                 }
-                // TODO: notify
                 if (
                     error instanceof UnsupportedStrategy ||
                     error instanceof InvalidEvent ||
@@ -221,6 +213,11 @@ export class Orchestrator {
                             className: Orchestrator.name,
                             chainId: this.chainId,
                         });
+                        void this.notifier.send(error.message, {
+                            chainId: this.chainId,
+                            event: event!,
+                            stack: error.getFullStack(),
+                        });
                     } else if (error instanceof Error || isNativeError(error)) {
                         const shouldIgnoreError = this.shouldIgnoreTimestampsUpdatedError(
                             error,
@@ -232,6 +229,11 @@ export class Orchestrator {
                                 className: Orchestrator.name,
                                 chainId: this.chainId,
                             });
+                            void this.notifier.send(error.message, {
+                                chainId: this.chainId,
+                                event: event!,
+                                stack: error.stack,
+                            });
                         }
                     } else {
                         this.logger.error(
@@ -241,6 +243,13 @@ export class Orchestrator {
                                 chainId: this.chainId,
                             },
                         );
+                        void this.notifier.send(
+                            `Error processing event: ${stringify(event)} ${error}`,
+                            {
+                                chainId: this.chainId,
+                                event: event!,
+                            },
+                        );
                     }
                 }
             }
@@ -355,18 +364,8 @@ export class Orchestrator {
             metadataIds,
             async (id) => {
                 try {
-                    const result = await retryAsyncUntilDefined(
-                        () => this.dependencies.metadataProvider.getMetadata<unknown>(id),
-                        {
-                            maxTry: MAX_BULK_FETCH_METADATA_RETRIES,
-                            delay: (params: DelayParameters<unknown>) => {
-                                return (
-                                    METADATA_BULK_FETCH_BASE_DELAY_MS *
-                                    METADATA_BULK_FETCH_BACKOFF_FACTOR ** params.currentTry
-                                );
-                            },
-                        },
-                    );
+                    const result =
+                        await this.dependencies.metadataProvider.getMetadata<unknown>(id);
                     return { status: "fulfilled", value: result };
                 } catch (error) {
                     return { status: "rejected", reason: error };