feat: 🎸 persist webhook state with repository

add Local and Postgres repositories for subscriptions and notifications

README.md

@@ -178,7 +178,7 @@ If using the project's compose file, an Artemis console will be exposed on `:818
 
 ### Webhooks (alpha)
 
-Normally the endpoints that create transactions wait for block finalization before returning a response, which normally takes around 15 seconds. When processMode `submitAndCallback` is used the `webhookUrl` param must also be provided. The server will respond after submitting the transaction to the mempool with 202 (Accepted) status code instead of the usual 201 (Created).
+Normally the endpoints that create transactions wait for block finalization before returning a response, which normally takes around 15 seconds. When processMode `submitWithCallback` is used the `webhookUrl` param must also be provided. The server will respond after submitting the transaction to the mempool with 202 (Accepted) status code instead of the usual 201 (Created).
 
 Before sending any information to the endpoint the service will first make a request with the header `x-hook-secret` set to a value. The endpoint should return a `200` response with this header copied into the response headers.
 
@@ -224,9 +224,6 @@ To implement a new repo for a service, first define an abstract class describing
 
 To implement a new datastore create a new module in `~/datastores` and create a set of `Repos` that will implement the abstract classes. You will then need to set up the `DatastoreModule` to export the module when it is configured. For testing, each implemented Repo should be able to pass the `test` method defined on the abstract class it is implementing.
 
-
-
-
 ### With docker
 
 To pass in the env variables you can use `-e` to pass them individually, or use a file with `--env-file`.

src/common/decorators/swagger.ts

@@ -181,7 +181,7 @@ export const ApiPropertyOneOf = ({
 };
 
 /**
- * A helper that functions like `ApiCreatedResponse`, that also adds an `ApiAccepted` response in case "submitAndCallback" is used and `ApiOKResponse` if "offline" mode is used
+ * A helper that functions like `ApiCreatedResponse`, that also adds an `ApiAccepted` response in case "submitWithCallback" is used and `ApiOKResponse` if "offline" mode is used
  *
  * @param options - these will be passed to the `ApiCreatedResponse` decorator
  */
@@ -197,7 +197,7 @@ export function ApiTransactionResponse(
     ApiCreatedResponse(options),
     ApiAcceptedResponse({
       description:
-        'Returned if `"processMode": "submitAndCallback"` is passed in `options`. A response will be returned after the transaction has been validated. The result will be posted to the `webhookUrl` given when the transaction is completed',
+        'Returned if `"processMode": "submitWithCallback"` is passed in `options`. A response will be returned after the transaction has been validated. The result will be posted to the `webhookUrl` given when the transaction is completed',
       type: NotificationPayloadModel,
     })
   );

src/datastore/local-store/local-store.module.ts

@@ -5,11 +5,15 @@ import { ConfigModule } from '@nestjs/config';
 
 import { ApiKeyRepo } from '~/auth/repos/api-key.repo';
 import { LocalApiKeysRepo } from '~/datastore/local-store/repos/api-key.repo';
+import { LocalNotificationRepo } from '~/datastore/local-store/repos/notification.repo';
 import { LocalOfflineEventRepo } from '~/datastore/local-store/repos/offline-event.repo';
 import { LocalOfflineTxRepo } from '~/datastore/local-store/repos/offline-tx.repo';
+import { LocalSubscriptionRepo } from '~/datastore/local-store/repos/subscription.repo';
 import { LocalUserRepo } from '~/datastore/local-store/repos/users.repo';
+import { NotificationRepo } from '~/notifications/repo/notifications.repo';
 import { OfflineEventRepo } from '~/offline-recorder/repo/offline-event.repo';
 import { OfflineTxRepo } from '~/offline-submitter/repos/offline-tx.repo';
+import { SubscriptionRepo } from '~/subscriptions/repo/subscription.repo';
 import { UsersRepo } from '~/users/repo/user.repo';
 
 /**
@@ -28,7 +32,16 @@ import { UsersRepo } from '~/users/repo/user.repo';
       useClass: LocalOfflineEventRepo,
     },
     { provide: OfflineTxRepo, useClass: LocalOfflineTxRepo },
+    { provide: SubscriptionRepo, useClass: LocalSubscriptionRepo },
+    { provide: NotificationRepo, useClass: LocalNotificationRepo },
+  ],
+  exports: [
+    ApiKeyRepo,
+    UsersRepo,
+    OfflineEventRepo,
+    OfflineTxRepo,
+    SubscriptionRepo,
+    NotificationRepo,
   ],
-  exports: [ApiKeyRepo, UsersRepo, OfflineEventRepo, OfflineTxRepo],
 })
 export class LocalStoreModule {}

src/datastore/local-store/repos/__snapshots__/notification.repo.spec.ts.snap

@@ -0,0 +1,23 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`LocalNotificationRepo Notification test suite method: create should create a notification 1`] = `
+{
+  "eventId": 0,
+  "id": 1,
+  "nonce": 1,
+  "status": "active",
+  "subscriptionId": 1,
+  "triesLeft": 10,
+}
+`;
+
+exports[`LocalNotificationRepo Notification test suite method: findById should find the created notification 1`] = `
+{
+  "eventId": 0,
+  "id": 1,
+  "nonce": 1,
+  "status": "active",
+  "subscriptionId": 1,
+  "triesLeft": 10,
+}
+`;

src/datastore/local-store/repos/__snapshots__/subscription.repo.spec.ts.snap

@@ -0,0 +1,30 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`LocalSubscriptionRepo Subscription test suite method: create should create a subscription 1`] = `
+SubscriptionModel {
+  "createdAt": 1987-10-14T00:00:00.000Z,
+  "eventScope": "",
+  "eventType": "transaction.update",
+  "id": 1,
+  "legitimacySecret": "someSecret",
+  "nextNonce": 3,
+  "status": "inactive",
+  "triesLeft": 10,
+  "ttl": 10,
+  "webhookUrl": "http://example.com",
+}
+`;
+
+exports[`LocalSubscriptionRepo Subscription test suite method: findById should find the created notification 1`] = `
+{
+  "eventScope": "",
+  "eventType": "transaction.update",
+  "id": 1,
+  "legitimacySecret": "someSecret",
+  "nextNonce": 3,
+  "status": "inactive",
+  "triesLeft": 10,
+  "ttl": 10,
+  "webhookUrl": "http://example.com",
+}
+`;

src/datastore/local-store/repos/notification.repo.spec.ts

@@ -0,0 +1,8 @@
+import { LocalNotificationRepo } from '~/datastore/local-store/repos/notification.repo';
+import { NotificationRepo } from '~/notifications/repo/notifications.repo';
+
+describe(`LocalNotificationRepo ${NotificationRepo.type} test suite`, () => {
+  const repo = new LocalNotificationRepo();
+
+  NotificationRepo.test(repo);
+});

src/datastore/local-store/repos/notification.repo.ts

@@ -0,0 +1,46 @@
+import { Injectable } from '@nestjs/common';
+
+import { NotificationModel } from '~/notifications/model/notification.model';
+import { NotificationRepo } from '~/notifications/repo/notifications.repo';
+import { NotificationParams, NotificationStatus } from '~/notifications/types';
+
+const triesLeft = 10;
+
+@Injectable()
+export class LocalNotificationRepo extends NotificationRepo {
+  private currentId = 0;
+  private notifications: Record<string, NotificationModel> = {};
+
+  public async create(params: NotificationParams): Promise<NotificationModel> {
+    this.currentId += 1;
+
+    const model = new NotificationModel({
+      id: this.currentId,
+      ...params,
+      triesLeft,
+      status: NotificationStatus.Active,
+      createdAt: new Date(),
+    });
+
+    this.notifications[model.id] = model;
+
+    return model;
+  }
+
+  public async update(id: number, params: NotificationParams): Promise<NotificationModel> {
+    const model = this.notifications[id];
+
+    const updated = {
+      ...model,
+      ...params,
+    };
+
+    this.notifications[id] = updated;
+
+    return updated;
+  }
+
+  public async findById(id: number): Promise<NotificationModel> {
+    return this.notifications[id];
+  }
+}

src/datastore/local-store/repos/subscription.repo.spec.ts

@@ -0,0 +1,8 @@
+import { LocalSubscriptionRepo } from '~/datastore/local-store/repos/subscription.repo';
+import { SubscriptionRepo } from '~/subscriptions/repo/subscription.repo';
+
+describe(`LocalSubscriptionRepo ${SubscriptionRepo.type} test suite`, () => {
+  const repo = new LocalSubscriptionRepo();
+
+  SubscriptionRepo.test(repo);
+});

src/datastore/local-store/repos/subscription.repo.ts

@@ -0,0 +1,58 @@
+import { Injectable } from '@nestjs/common';
+
+import { AppNotFoundError } from '~/common/errors';
+import { SubscriptionModel } from '~/subscriptions/models/subscription.model';
+import { SubscriptionRepo } from '~/subscriptions/repo/subscription.repo';
+import { SubscriptionParams } from '~/subscriptions/types';
+
+@Injectable()
+export class LocalSubscriptionRepo implements SubscriptionRepo {
+  private currentId = 0;
+  private subscriptions: Record<string, SubscriptionModel> = {};
+
+  public async create(params: SubscriptionParams): Promise<SubscriptionModel> {
+    this.currentId += 1;
+
+    const model = new SubscriptionModel({
+      id: this.currentId,
+      ...params,
+    });
+
+    this.subscriptions[model.id] = model;
+
+    return model;
+  }
+
+  public async update(id: number, params: SubscriptionParams): Promise<SubscriptionModel> {
+    const model = this.subscriptions[id];
+
+    const updated = new SubscriptionModel({
+      ...model,
+      ...params,
+    });
+
+    this.subscriptions[id] = updated;
+
+    return updated;
+  }
+
+  public async findById(id: number): Promise<SubscriptionModel> {
+    const subscription = this.subscriptions[id];
+
+    if (!subscription) {
+      throw new AppNotFoundError(id.toString(), 'notification');
+    }
+
+    return subscription;
+  }
+
+  public async findAll(): Promise<SubscriptionModel[]> {
+    return Object.values(this.subscriptions);
+  }
+
+  public async incrementNonces(ids: number[]): Promise<void> {
+    ids.forEach(id => {
+      this.subscriptions[id].nextNonce += 1;
+    });
+  }
+}

src/datastore/postgres/entities/notification.entity.ts

@@ -0,0 +1,39 @@
+/* istanbul ignore file */
+
+import { FactoryProvider } from '@nestjs/common';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { Column, DataSource, Entity, PrimaryGeneratedColumn, Repository } from 'typeorm';
+
+import { NotificationStatus } from '~/notifications/types';
+
+@Entity()
+export class Notification {
+  @PrimaryGeneratedColumn('increment')
+  public id: number;
+
+  @Column({ type: 'int' })
+  public subscriptionId: number;
+
+  @Column({ type: 'int' })
+  public eventId: number;
+
+  @Column({ type: 'int' })
+  public triesLeft: number;
+
+  @Column({ type: 'text' })
+  public status: NotificationStatus;
+
+  @Column({ type: 'int' })
+  public nonce: number;
+
+  @Column({ type: 'timestamptz', default: () => 'CURRENT_TIMESTAMP' })
+  public createdAt: Date;
+}
+
+export const notificationRepoProvider: FactoryProvider = {
+  provide: getRepositoryToken(Notification),
+  useFactory: async (dataSource: DataSource): Promise<Repository<Notification>> => {
+    return dataSource.getRepository(Notification);
+  },
+  inject: [DataSource],
+};

src/datastore/postgres/entities/subscription.entity.ts

@@ -0,0 +1,49 @@
+/* istanbul ignore file */
+
+import { FactoryProvider } from '@nestjs/common';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { Column, DataSource, Entity, PrimaryGeneratedColumn, Repository } from 'typeorm';
+
+import { EventType } from '~/events/types';
+import { SubscriptionStatus } from '~/subscriptions/types';
+
+@Entity()
+export class Subscription {
+  @PrimaryGeneratedColumn('increment')
+  public id: number;
+
+  @Column({ type: 'text' })
+  public eventType: EventType;
+
+  @Column({ type: 'text' })
+  public eventScope: string;
+
+  @Column({ type: 'text' })
+  public webhookUrl: string;
+
+  @Column({ type: 'int' })
+  public ttl: number;
+
+  @Column({ type: 'text' })
+  public status: SubscriptionStatus;
+
+  @Column({ type: 'int' })
+  public triesLeft: number;
+
+  @Column({ type: 'int' })
+  public nextNonce: number;
+
+  @Column({ type: 'text' })
+  public legitimacySecret: string;
+
+  @Column({ type: 'timestamptz', default: () => 'CURRENT_TIMESTAMP' })
+  public createdAt: Date;
+}
+
+export const subscriptionRepoProvider: FactoryProvider = {
+  provide: getRepositoryToken(Subscription),
+  useFactory: async (dataSource: DataSource): Promise<Repository<Subscription>> => {
+    return dataSource.getRepository(Subscription);
+  },
+  inject: [DataSource],
+};

src/datastore/postgres/migrations/1718398114107-hooks.ts

@@ -0,0 +1,15 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class Hooks1718398114107 implements MigrationInterface {
+  name = 'Hooks1718398114107';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      'CREATE TABLE "subscription" ("id" SERIAL NOT NULL, "eventType" text NOT NULL, "eventScope" text NOT NULL, "webhookUrl" text NOT NULL, "ttl" integer NOT NULL, "status" text NOT NULL, "triesLeft" integer NOT NULL, "nextNonce" integer NOT NULL, "legitimacySecret" text NOT NULL, "createdAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), CONSTRAINT "PK_8c3e00ebd02103caa1174cd5d9d" PRIMARY KEY ("id"))'
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query('DROP TABLE "subscription"');
+  }
+}

src/datastore/postgres/migrations/1719003936380-notifications.ts

@@ -0,0 +1,19 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class Notifications1719003936380 implements MigrationInterface {
+  name = 'Notifications1719003936380';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query('DROP INDEX "public"."idx_address_nonce"');
+    await queryRunner.query(
+      'CREATE TABLE "notification" ("id" SERIAL NOT NULL, "subscriptionId" integer NOT NULL, "eventId" integer NOT NULL, "triesLeft" integer NOT NULL, "status" text NOT NULL, "nonce" integer NOT NULL, "createdAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), CONSTRAINT "PK_705b6c7cdf9b2c2ff7ac7872cb7" PRIMARY KEY ("id"))'
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query('DROP TABLE "notification"');
+    await queryRunner.query(
+      'CREATE INDEX "idx_address_nonce" ON "offline_tx" ("address", "nonce") '
+    );
+  }
+}