feat(webhooks): add postgres persistence to clients and requests (feature branch) (#108)

Author: daywiss

packages/indexer-api/src/main.ts

@@ -87,10 +87,10 @@ export async function Main(
   const redis = await initializeRedis(redisConfig, logger);
   const webhooks = Webhooks.WebhookFactory(
     {
-      requireApiKey: false,
       enabledWebhooks: [Webhooks.WebhookTypes.DepositStatus],
+      enabledWebhookRequestWorkers: false,
     },
-    { postgres, logger },
+    { postgres, logger, redis },
   );
 
   const allRouters: Record<string, Router> = {

packages/indexer-database/src/entities/WebhookClient.ts

@@ -0,0 +1,17 @@
+import { Entity, Column, PrimaryGeneratedColumn, Unique } from "typeorm";
+
+@Entity()
+@Unique("UK_webhook_client_api_key", ["apiKey"])
+export class WebhookClient {
+  @PrimaryGeneratedColumn()
+  id: number;
+
+  @Column()
+  name: string;
+
+  @Column()
+  apiKey: string;
+
+  @Column("jsonb")
+  domains: string[];
+}

packages/indexer-database/src/entities/WebhookRequest.ts

@@ -0,0 +1,28 @@
+import {
+  Entity,
+  PrimaryColumn,
+  Column,
+  Unique,
+  CreateDateColumn,
+  Index,
+} from "typeorm";
+
+@Entity()
+@Unique("UK_webhook_request_clientId_filter", ["clientId", "filter"])
+@Index("IX_webhook_request_filter", ["filter"])
+export class WebhookRequest {
+  @PrimaryColumn()
+  id: string;
+
+  @Column({ type: "integer" })
+  clientId: number;
+
+  @Column()
+  url: string;
+
+  @Column()
+  filter: string;
+
+  @CreateDateColumn()
+  createdAt: Date;
+}

packages/indexer-database/src/entities/index.ts

@@ -19,3 +19,6 @@ export * from "./BundleEvent";
 export * from "./BundleBlockRange";
 export * from "./RootBundleExecutedJoinTable";
 export * from "./RelayHashInfo";
+
+export * from "./WebhookRequest";
+export * from "./WebhookClient";

packages/indexer-database/src/main.ts

@@ -36,6 +36,9 @@ export const createDataSource = (config: DatabaseConfig): DataSource => {
       entities.RootBundleExecutedJoinTable,
       // Others
       entities.RelayHashInfo,
+      // Webhooks
+      entities.WebhookRequest,
+      entities.WebhookClient,
     ],
     migrationsTableName: "_migrations",
     migrations: ["migrations/*.ts"],

packages/indexer-database/src/migrations/1732297474910-WebhookClient.ts

@@ -0,0 +1,21 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class WebhookClient1732297474910 implements MigrationInterface {
+  name = "WebhookClient1732297474910";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      CREATE TABLE "webhook_client" (
+        "id" SERIAL NOT NULL, 
+        "name" character varying NOT NULL, 
+        "apiKey" character varying NOT NULL, 
+        "domains" jsonb NOT NULL, 
+        CONSTRAINT "UK_webhook_client_api_key" UNIQUE ("apiKey"), 
+        CONSTRAINT "PK_f7330fb3bdb2e19534eae691d44" PRIMARY KEY ("id")
+      )`);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`DROP TABLE "webhook_client"`);
+  }
+}

packages/indexer-database/src/migrations/1732297948190-WebhookRequest.ts

@@ -0,0 +1,22 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class WebhookRequest1732297948190 implements MigrationInterface {
+  name = "WebhookRequest1732297948190";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      CREATE TABLE "webhook_request" (
+        "id" character varying NOT NULL, 
+        "clientId" integer NOT NULL, 
+        "url" character varying NOT NULL, 
+        "filter" character varying NOT NULL, 
+        "createdAt" TIMESTAMP NOT NULL DEFAULT now(), 
+        CONSTRAINT "UK_webhook_request_clientId_filter" UNIQUE ("clientId", "filter"), 
+        CONSTRAINT "PK_67a7784045de2d1b7139b611b93" PRIMARY KEY ("id")
+    )`);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`DROP TABLE "webhook_request"`);
+  }
+}

packages/indexer-database/src/migrations/1732310112989-WebhookRequest.ts

@@ -0,0 +1,15 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class WebhookRequest1732310112989 implements MigrationInterface {
+  name = "WebhookRequest1732310112989";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `CREATE INDEX "IX_webhook_request_filter" ON "webhook_request" ("filter") `,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`DROP INDEX "public"."IX_webhook_request_filter"`);
+  }
+}

packages/indexer/src/data-indexing/service/AcrossIndexerManager.ts

@@ -1,6 +1,7 @@
 import { Logger } from "winston";
 
 import { DataSource } from "@repo/indexer-database";
+import { eventProcessorManager } from "@repo/webhooks";
 
 import { Config } from "../../parseEnv";
 import { HubPoolRepository } from "../../database/HubPoolRepository";
@@ -39,6 +40,7 @@ export class AcrossIndexerManager {
     private spokePoolRepository: SpokePoolRepository,
     private redisCache: RedisCache,
     private indexerQueuesService: IndexerQueuesService,
+    private webhookWriteFn?: eventProcessorManager.WebhookWriteFn,
   ) {}
 
   public async start() {
@@ -93,7 +95,12 @@ export class AcrossIndexerManager {
           this.hubPoolClientFactory,
           this.spokePoolClientFactory,
           this.spokePoolRepository,
-          new SpokePoolProcessor(this.postgres, this.logger, chainId),
+          new SpokePoolProcessor(
+            this.postgres,
+            this.logger,
+            chainId,
+            this.webhookWriteFn,
+          ),
           this.indexerQueuesService,
         );
         const spokePoolIndexer = new Indexer(

packages/indexer/src/main.ts

@@ -57,10 +57,10 @@ export async function Main(config: parseEnv.Config, logger: winston.Logger) {
   // Call write to kick off webhook calls
   const { write } = WebhookFactory(
     {
-      requireApiKey: false,
       enabledWebhooks: [WebhookTypes.DepositStatus],
+      enabledWebhookRequestWorkers: true,
     },
-    { postgres, logger },
+    { postgres, logger, redis },
   );
   // Retry providers factory
   const retryProvidersFactory = new RetryProvidersFactory(
@@ -96,6 +96,7 @@ export async function Main(config: parseEnv.Config, logger: winston.Logger) {
     new SpokePoolRepository(postgres, logger),
     redisCache,
     indexerQueuesService,
+    write,
   );
   const bundleServicesManager = new BundleServicesManager(
     config,