feat(ts-client): add persisted predicates and periodic health check

.vscode/launch.json

@@ -0,0 +1,18 @@
+{
+  "version": "0.2.0",
+  "configurations": [
+    {
+        "type": "node",
+        "request": "launch",
+        "name": "ts client: jest",
+        "localRoot": "${workspaceFolder}/components/client/typescript",
+        "program": "${workspaceFolder}/components/client/typescript/node_modules/jest/bin/jest",
+        "args": [
+          "--runInBand",
+          "--no-cache",
+        ],
+        "outputCapture": "std",
+        "console": "integratedTerminal",
+      },
+  ]
+}

components/client/typescript/.gitignore

@@ -1,2 +1,3 @@
 dist/
 node_modules/
+tmp/

components/client/typescript/jest.config.js

@@ -0,0 +1,15 @@
+module.exports = {
+  collectCoverageFrom: [
+    "src/**/*.ts",
+  ],
+  coverageProvider: "v8",
+  // globalSetup: './tests/setup.ts',
+  preset: 'ts-jest',
+  rootDir: '',
+  testPathIgnorePatterns: [
+    "/node_modules/",
+    "/dist/"
+  ],
+  transform: {},
+  transformIgnorePatterns: ["./dist/.+\\.js"]
+};

components/client/typescript/package.json

@@ -1,12 +1,12 @@
 {
   "name": "@hirosystems/chainhook-client",
-  "version": "1.12.0",
+  "version": "2.0.0",
   "description": "Chainhook TypeScript client",
   "main": "./dist/index.js",
   "typings": "./dist/index.d.ts",
   "scripts": {
     "build": "rimraf ./dist && tsc --project tsconfig.build.json",
-    "test": "jest",
+    "test": "jest --runInBand",
     "prepublishOnly": "npm run build"
   },
   "files": [

components/client/typescript/src/index.ts

@@ -1,11 +1,94 @@
 import { FastifyInstance } from 'fastify';
-import {
-  ServerOptions,
-  ChainhookNodeOptions,
-  ServerPredicate,
-  OnEventCallback,
-  buildServer,
-} from './server';
+import { buildServer } from './server';
+import { predicateHealthCheck } from './predicates';
+import { Payload } from './schemas/payload';
+import { Static, Type } from '@fastify/type-provider-typebox';
+import { BitcoinIfThisOptionsSchema, BitcoinIfThisSchema } from './schemas/bitcoin/if_this';
+import { StacksIfThisOptionsSchema, StacksIfThisSchema } from './schemas/stacks/if_this';
+import { logger } from './util/logger';
+
+const EventObserverOptionsSchema = Type.Object({
+  /** Event observer host name (usually '0.0.0.0') */
+  hostname: Type.String(),
+  /** Event observer port */
+  port: Type.Integer(),
+  /** Authorization token for all Chainhook payloads */
+  auth_token: Type.String(),
+  /** Base URL that will be used by Chainhook to send all payloads to this event observer */
+  external_base_url: Type.String(),
+  /** Wait for the chainhook node to be available before submitting predicates */
+  wait_for_chainhook_node: Type.Optional(Type.Boolean({ default: true })),
+  /** Validate the JSON schema of received chainhook payloads and report errors when invalid */
+  validate_chainhook_payloads: Type.Optional(Type.Boolean({ default: false })),
+  /** Validate the authorization token sent by the server is correct. */
+  validate_token_authorization: Type.Optional(Type.Boolean({ default: true })),
+  /** Size limit for received chainhook payloads (default 40MB) */
+  body_limit: Type.Optional(Type.Number({ default: 41943040 })),
+  /** Node type: `chainhook` or `ordhook` */
+  node_type: Type.Optional(
+    Type.Union([Type.Literal('chainhook'), Type.Literal('ordhook')], {
+      default: 'chainhook',
+    })
+  ),
+  /**
+   * Directory where registered predicates will be persisted to disk so they can be recalled on
+   * restarts.
+   */
+  predicate_disk_file_path: Type.String(),
+  /**
+   * How often we should check with the Chainhook server to make sure our predicates are active and
+   * up to date. If they become obsolete, we will attempt to re-register them.
+   */
+  predicate_health_check_interval_ms: Type.Optional(Type.Integer({ default: 5000 })),
+});
+/** Chainhook event observer configuration options */
+export type EventObserverOptions = Static<typeof EventObserverOptionsSchema>;
+
+const ChainhookNodeOptionsSchema = Type.Object({
+  /** Base URL where the Chainhook node is located */
+  base_url: Type.String(),
+});
+/** Chainhook node connection options */
+export type ChainhookNodeOptions = Static<typeof ChainhookNodeOptionsSchema>;
+
+/**
+ * Callback that will receive every single payload sent by Chainhook as a result of any predicates
+ * that have been registered.
+ */
+export type OnPredicatePayloadCallback = (payload: Payload) => Promise<void>;
+
+const IfThisThenNothingSchema = Type.Union([
+  Type.Composite([
+    BitcoinIfThisOptionsSchema,
+    Type.Object({
+      if_this: BitcoinIfThisSchema,
+    }),
+  ]),
+  Type.Composite([
+    StacksIfThisOptionsSchema,
+    Type.Object({
+      if_this: StacksIfThisSchema,
+    }),
+  ]),
+]);
+export const EventObserverPredicateSchema = Type.Composite([
+  Type.Object({
+    name: Type.String(),
+    version: Type.Integer(),
+    chain: Type.String(),
+  }),
+  Type.Object({
+    networks: Type.Object({
+      mainnet: Type.Optional(IfThisThenNothingSchema),
+      testnet: Type.Optional(IfThisThenNothingSchema),
+    }),
+  }),
+]);
+/**
+ * Partial predicate definition that allows users to build the core parts of a predicate and let the
+ * event observer fill in the rest.
+ */
+export type EventObserverPredicate = Static<typeof EventObserverPredicateSchema>;
 
 /**
  * Local web server that registers predicates and receives events from a Chainhook node. It handles
@@ -20,29 +103,43 @@ import {
  */
 export class ChainhookEventObserver {
   private fastify?: FastifyInstance;
-  private serverOpts: ServerOptions;
-  private chainhookOpts: ChainhookNodeOptions;
+  private observer: EventObserverOptions;
+  private chainhook: ChainhookNodeOptions;
+  private healthCheckTimer?: NodeJS.Timer;
 
-  constructor(serverOpts: ServerOptions, chainhookOpts: ChainhookNodeOptions) {
-    this.serverOpts = serverOpts;
-    this.chainhookOpts = chainhookOpts;
+  constructor(observer: EventObserverOptions, chainhook: ChainhookNodeOptions) {
+    this.observer = observer;
+    this.chainhook = chainhook;
   }
 
   /**
-   * Start the Chainhook event server.
-   * @param predicates - Predicates to register
+   * Starts the Chainhook event observer.
+   * @param predicates - Predicates to register. If `predicates_disk_file_path` is enabled in the
+   * observer, predicates stored on disk will take precedent over those specified here.
    * @param callback - Function to handle every Chainhook event payload sent by the node
    */
-  async start(predicates: ServerPredicate[], callback: OnEventCallback): Promise<void> {
+  async start(
+    predicates: EventObserverPredicate[],
+    callback: OnPredicatePayloadCallback
+  ): Promise<void> {
     if (this.fastify) return;
-    this.fastify = await buildServer(this.serverOpts, this.chainhookOpts, predicates, callback);
-    await this.fastify.listen({ host: this.serverOpts.hostname, port: this.serverOpts.port });
+    this.fastify = await buildServer(this.observer, this.chainhook, predicates, callback);
+    await this.fastify.listen({ host: this.observer.hostname, port: this.observer.port });
+    if (this.observer.predicate_health_check_interval_ms && this.healthCheckTimer === undefined) {
+      this.healthCheckTimer = setInterval(() => {
+        predicateHealthCheck(this.observer, this.chainhook).catch(err =>
+          logger.error(err, `ChainhookEventObserver predicate health check error`)
+        );
+      }, this.observer.predicate_health_check_interval_ms);
+    }
   }
 
   /**
    * Stop the Chainhook event server gracefully.
    */
   async close(): Promise<void> {
+    if (this.healthCheckTimer) clearInterval(this.healthCheckTimer);
+    this.healthCheckTimer = undefined;
     await this.fastify?.close();
     this.fastify = undefined;
   }