feat: slack notifications

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/src/config/env.ts

@@ -40,6 +40,9 @@ const baseSchema = z.object({
     RETRY_BASE_DELAY_MS: z.coerce.number().int().min(1).default(3000), // 3 seconds
     RETRY_FACTOR: z.coerce.number().int().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({
@@ -66,6 +69,15 @@ const validationSchema = z
             apiType: val.COINGECKO_API_TYPE,
             ...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"],
+            });
+        }
     });
 
 const env = validationSchema.safeParse(process.env);

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(
             {
@@ -146,6 +158,22 @@ export class SharedDependenciesService {
             kyselyDatabase,
             retryStrategy,
             logger,
+            notifier,
+        };
+    }
+
+    static getNotifierOptions(env: Environment): NotifierConfig<NotifierProvider> {
+        if (env.NOTIFIER_PROVIDER === "slack") {
+            return {
+                notifierProvider: env.NOTIFIER_PROVIDER,
+                opts: {
+                    webhookUrl: env.SLACK_WEBHOOK_URL!,
+                },
+            };
+        }
+
+        return {
+            notifierProvider: "null",
         };
     }
 }

packages/data-flow/src/orchestrator.ts

@@ -22,6 +22,7 @@ import {
     getToken,
     Hex,
     ILogger,
+    INotifier,
     isAlloEvent,
     isStrategyEvent,
     ProcessorEvent,
@@ -69,8 +70,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>>;
@@ -103,6 +102,7 @@ export class Orchestrator {
         private fetchDelayInMs: number = 10000,
         private retryStrategy: RetryStrategy,
         private logger: ILogger,
+        private notifier: INotifier,
     ) {
         this.eventsFetcher = new EventsFetcher(this.indexerClient);
         this.eventsProcessor = new EventsProcessor(this.chainId, {
@@ -180,7 +180,6 @@ export class Orchestrator {
                         rawEvent: event,
                     });
                 }
-                // TODO: notify
                 if (
                     error instanceof UnsupportedStrategy ||
                     error instanceof InvalidEvent ||
@@ -202,6 +201,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,
@@ -213,6 +217,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(
@@ -222,6 +231,13 @@ export class Orchestrator {
                                 chainId: this.chainId,
                             },
                         );
+                        void this.notifier.send(
+                            `Error processing event: ${stringify(event)} ${error}`,
+                            {
+                                chainId: this.chainId,
+                                event: event!,
+                            },
+                        );
                     }
                 }
             }

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

@@ -28,6 +28,7 @@ import {
     Hex,
     ICacheable,
     ILogger,
+    INotifier,
     ProcessorEvent,
     RateLimitError,
     StrategyEvent,
@@ -76,6 +77,9 @@ describe("Orchestrator", { sequential: true }, () => {
         info: vi.fn(),
         warn: vi.fn(),
     };
+    const notifier: INotifier = {
+        send: vi.fn(),
+    };
 
     beforeEach(() => {
         // Setup mock implementations
@@ -135,6 +139,7 @@ describe("Orchestrator", { sequential: true }, () => {
             mockFetchDelay,
             new ExponentialBackoff({ baseDelay: 10, maxAttempts: 3, factor: 2 }),
             logger,
+            notifier,
         );
     });
 

packages/shared/README.md

@@ -39,3 +39,13 @@ You can import the package in your TypeScript files as follows:
 ```typescript
 import { ILogger, ProcessorEvent, stringify } from "@grants-stack-indexer/shared";
 ```
+
+### Configuring the Notifier
+
+#### Slack
+
+Visit [Slack API docs](https://api.slack.com/messaging/webhooks) to create a Slack app and get a webhook URL. Follow the instructions for setting up a Slack app and get a webhook URL. Also, you can further customize the bot name, icons and scopes.
+
+#### Null
+
+The null notifier is a no-op notifier that does nothing. It is useful when you want to disable notifications.

packages/shared/package.json

@@ -28,6 +28,7 @@
         "test:cov": "vitest run --config vitest.config.ts --coverage"
     },
     "dependencies": {
+        "@slack/webhook": "7.0.4",
         "bignumber.js": "9.1.2",
         "viem": "2.21.19",
         "winston": "3.15.0"

packages/shared/src/external.ts

@@ -29,3 +29,11 @@ export { ExponentialBackoff, RetryHandler } from "./internal.js";
 export type { RetryStrategy, RetryStrategyOptions } from "./internal.js";
 
 export type { ICacheable } from "./internal.js";
+
+export type {
+    INotifier,
+    NotifierConfig,
+    NotifierProvider,
+    SlackNotifierConfig,
+} from "./internal.js";
+export { SlackNotifier, NullNotifier, NotifierFactory } from "./internal.js";

packages/shared/src/internal.ts

@@ -9,3 +9,4 @@ export * from "./tokens/tokens.js";
 export * from "./exceptions/index.js";
 export * from "./retry/index.js";
 export * from "./caching/index.js";
+export * from "./notifier/index.js";

packages/shared/src/notifier/index.ts

@@ -0,0 +1,5 @@
+export * from "./notifier.interface.js";
+export * from "./notifierConfig.interface.js";
+export * from "./slack.notifier.js";
+export * from "./null.notifier.js";
+export * from "./notifier.factory.js";

packages/shared/src/notifier/notifier.factory.ts

@@ -0,0 +1,29 @@
+import { ILogger, NullNotifier, SlackNotifier } from "../internal.js";
+import { INotifier } from "./notifier.interface.js";
+import { NotifierConfig, NotifierProvider } from "./notifierConfig.interface.js";
+
+/**
+ * Factory class to create instances of INotifier based on configuration.
+ */
+export class NotifierFactory {
+    /**
+     * Creates an instance of INotifier based on configuration.
+     *
+     * @param {NotifierConfig<NotifierProvider>} options - The configuration for the notifier.
+     * @returns {INotifier} An instance of INotifier.
+     */
+    static create(options: NotifierConfig<NotifierProvider>, logger: ILogger): INotifier {
+        let notifier: INotifier;
+
+        switch (options.notifierProvider) {
+            case "slack":
+                notifier = new SlackNotifier(options.opts, logger);
+                break;
+            // Can add more notification services here, like TelegramNotifier, etc.
+            default:
+                notifier = new NullNotifier();
+        }
+
+        return notifier;
+    }
+}

packages/shared/src/notifier/notifier.interface.ts

@@ -0,0 +1,15 @@
+import { AnyIndexerFetchedEvent, ChainId } from "../internal.js";
+
+export type NotifierContext = {
+    event?: AnyIndexerFetchedEvent;
+    chainId: ChainId;
+    stack?: string;
+};
+
+export interface INotifier {
+    /**
+     * Sends a notification to the notifier provider.
+     * @param message - The message to notify.
+     */
+    send(message: string, context: NotifierContext): Promise<void>;
+}

packages/shared/src/notifier/notifierConfig.interface.ts

@@ -0,0 +1,14 @@
+import { SlackOptions } from "./slack.notifier.js";
+
+export type NotifierProvider = "slack" | "null";
+
+export type SlackNotifierConfig = {
+    notifierProvider: "slack";
+    opts: SlackOptions;
+};
+
+export type NotifierConfig<Source extends NotifierProvider> = Source extends "slack"
+    ? SlackNotifierConfig
+    : {
+          notifierProvider: "null";
+      };

packages/shared/src/notifier/null.notifier.ts

@@ -0,0 +1,10 @@
+import { INotifier, NotifierContext } from "../internal.js";
+
+/**
+ * A null implementation of INotifier that performs no actions but allows the application to have an undefined notifier if desired
+ */
+export class NullNotifier implements INotifier {
+    async send(_message: string, _context: NotifierContext): Promise<void> {
+        // Do nothing
+    }
+}