[CORL-687] Webhooks (#2738)

* feat: initial webhook impl

* feat: added support for key rotation

* feat: harmonized fetcher

* feat: added expired secrets cleaning

* feat: event system refactor

* feat: added story event

* feat: simplfiied webhook handler

* feat: added ref's to locations where user events can be added

* feat: added UI to support webhooks

* fix: renaming some Webhook -> WebhookEndpoint

* fix: review comments to adjuist flow

* feat: added localizations

* fix: linting, updated snapshots

* fix: adapted for new fluent

* fix: rearranged folders

* fix: linting

* feat: added webhooks documentation

* feat: improved toc generation

* feat: added some tests to webhooks

* fix: chain transition hooks

* feat: added tests around webhook ui

* fix: renamed events

* fix: adjusted circle markdown linting

* fix: adjusted doctoc script call

* review: review fixes

* review: review comments

* review: adjusted signing secret confirmation

* review: adjusted styles to harmonize button usage

* fix: updated snapshots and tests

* review: move form out of webhooks

Moved the form out of the webhooks by relocating the layout used for the
route associated with the configure routes.

* fix: fixed bugs and snapshots with tests

* feat: revised slack message format to use block api

* fix: fixed a small text bug

Co-authored-by: Vinh <vinh@vinh.tech>
Co-authored-by: Kim Gardner <kgardnr@gmail.com>

src/core/server/queue/Task.ts

@@ -4,9 +4,11 @@ import Logger from "bunyan";
 import TIME from "coral-common/time";
 import logger from "coral-server/logger";
 
-export interface TaskOptions<T, U = any> {
+export type JobProcessor<T, U = void> = (job: Job<T>) => Promise<U>;
+
+export interface TaskOptions<T, U = void> {
   jobName: string;
-  jobProcessor: (job: Job<T>) => Promise<U>;
+  jobProcessor: JobProcessor<T, U>;
   jobOptions?: Queue.JobOptions;
   queue: Queue.QueueOptions;
 }

src/core/server/queue/index.ts

@@ -1,4 +1,5 @@
 import Queue from "bull";
+import { Redis } from "ioredis";
 import { Db } from "mongodb";
 
 import { Config } from "coral-server/config";
@@ -10,6 +11,7 @@ import TenantCache from "coral-server/services/tenant/cache";
 import { createMailerTask, MailerQueue } from "./tasks/mailer";
 import { createNotifierTask, NotifierQueue } from "./tasks/notifier";
 import { createScraperTask, ScraperQueue } from "./tasks/scraper";
+import { createWebhookTask, WebhookQueue } from "./tasks/webhook";
 
 const createQueueOptions = async (
   config: Config
@@ -47,12 +49,14 @@ export interface QueueOptions {
   tenantCache: TenantCache;
   i18n: I18n;
   signingConfig: JWTSigningConfig;
+  redis: Redis;
 }
 
 export interface TaskQueue {
   mailer: MailerQueue;
   scraper: ScraperQueue;
   notifier: NotifierQueue;
+  webhook: WebhookQueue;
 }
 
 export async function createQueue(options: QueueOptions): Promise<TaskQueue> {
@@ -67,11 +71,13 @@ export async function createQueue(options: QueueOptions): Promise<TaskQueue> {
     mailerQueue: mailer,
     ...options,
   });
+  const webhook = createWebhookTask(queueOptions, options);
 
   // Return the tasks + client.
   return {
     mailer,
     scraper,
     notifier,
+    webhook,
   };
 }

src/core/server/queue/tasks/notifier/index.ts

@@ -1,10 +1,8 @@
 import Queue from "bull";
-import { groupBy } from "lodash";
 import { Db } from "mongodb";
 
 import { Config } from "coral-server/config";
-import { SUBSCRIPTION_CHANNELS } from "coral-server/graph/resolvers/Subscription/types";
-import logger from "coral-server/logger";
+import { CoralEventType } from "coral-server/events";
 import Task from "coral-server/queue/Task";
 import { MailerQueue } from "coral-server/queue/tasks/mailer";
 import { JWTSigningConfig } from "coral-server/services/jwt";
@@ -29,34 +27,32 @@ interface Options {
  * that could be sent to users.
  */
 export class NotifierQueue {
-  private registry: Record<SUBSCRIPTION_CHANNELS, NotificationCategory[]>;
   private task: Task<NotifierData>;
 
   constructor(queue: Queue.QueueOptions, options: Options) {
+    const registry = new Map<CoralEventType, NotificationCategory[]>();
+
     // Notification categories have been grouped by their event name so that
     // each event emitted need only access the associated notification once.
-    this.registry = groupBy(categories, "event") as Record<
-      SUBSCRIPTION_CHANNELS,
-      NotificationCategory[]
-    >;
+    for (const category of categories) {
+      for (const event of category.events as CoralEventType[]) {
+        let handlers = registry.get(event);
+        if (!handlers) {
+          handlers = [];
+        }
+        handlers.push(category);
+        registry.set(event, handlers);
+      }
+    }
+
     this.task = new Task({
       jobName: JOB_NAME,
-      jobProcessor: createJobProcessor({ registry: this.registry, ...options }),
+      jobProcessor: createJobProcessor({ registry, ...options }),
       queue,
     });
   }
 
   public async add(data: NotifierData) {
-    // Get all the handlers that are active for this channel.
-    const c = this.registry[data.input.channel];
-    if (!c || c.length === 0) {
-      logger.debug(
-        { channel: data.input.channel },
-        "no notifications registered on this channel"
-      );
-      return;
-    }
-
     return this.task.add(data);
   }
 

src/core/server/queue/tasks/notifier/messages.ts

@@ -1,10 +1,11 @@
-import { SUBSCRIPTION_INPUT } from "coral-server/graph/resolvers/Subscription/types";
-import { GQLDIGEST_FREQUENCY } from "coral-server/graph/schema/__generated__/types";
+import { CoralEventPayload } from "coral-server/events/event";
 import logger from "coral-server/logger";
 import { NotificationCategory } from "coral-server/services/notifications/categories";
 import NotificationContext from "coral-server/services/notifications/context";
 import { Notification } from "coral-server/services/notifications/notification";
 
+import { GQLDIGEST_FREQUENCY } from "coral-server/graph/schema/__generated__/types";
+
 import { MailerQueue } from "../mailer";
 import { DigestibleTemplate } from "../mailer/templates";
 import { CategoryNotification } from "./processor";
@@ -52,11 +53,11 @@ export const filterSuperseded = (
 export const handleHandlers = async (
   ctx: NotificationContext,
   categories: NotificationCategory[],
-  input: SUBSCRIPTION_INPUT
+  payload: CoralEventPayload
 ): Promise<CategoryNotification[]> => {
   const notifications: Array<CategoryNotification | null> = await Promise.all(
     categories.map(async category => {
-      const notification = await category.process(ctx, input.payload);
+      const notification = await category.process(ctx, payload);
       if (!notification) {
         return null;
       }

src/core/server/queue/tasks/notifier/processor.ts

@@ -2,10 +2,8 @@ import { Job } from "bull";
 import { Db } from "mongodb";
 
 import { Config } from "coral-server/config";
-import {
-  SUBSCRIPTION_CHANNELS,
-  SUBSCRIPTION_INPUT,
-} from "coral-server/graph/resolvers/Subscription/types";
+import { CoralEventType } from "coral-server/events";
+import { NotifierCoralEventListenerPayloads } from "coral-server/events/listeners/notifier";
 import logger from "coral-server/logger";
 import { MailerQueue } from "coral-server/queue/tasks/mailer";
 import { JWTSigningConfig } from "coral-server/services/jwt";
@@ -27,14 +25,14 @@ export const JOB_NAME = "notifications";
  */
 export interface NotifierData {
   tenantID: string;
-  input: SUBSCRIPTION_INPUT;
+  input: NotifierCoralEventListenerPayloads;
 }
 
 interface Options {
   mailerQueue: MailerQueue;
   mongo: Db;
   config: Config;
-  registry: Record<SUBSCRIPTION_CHANNELS, NotificationCategory[]>;
+  registry: Map<CoralEventType, NotificationCategory[]>;
   tenantCache: TenantCache;
   signingConfig: JWTSigningConfig;
 }
@@ -83,7 +81,7 @@ export const createJobProcessor = ({
 
     try {
       // Get all the handlers that are active for this channel.
-      const categories = registry[input.channel];
+      const categories = registry.get(input.type);
       if (!categories || categories.length === 0) {
         return;
       }

src/core/server/queue/tasks/webhook/index.ts

@@ -0,0 +1,22 @@
+import Queue from "bull";
+
+import Task from "coral-server/queue/Task";
+
+import {
+  createJobProcessor,
+  JOB_NAME,
+  WebhookData,
+  WebhookProcessorOptions,
+} from "./processor";
+
+export type WebhookQueue = Task<WebhookData>;
+
+export const createWebhookTask = (
+  queue: Queue.QueueOptions,
+  options: WebhookProcessorOptions
+) =>
+  new Task({
+    jobName: JOB_NAME,
+    jobProcessor: createJobProcessor(options),
+    queue,
+  });

src/core/server/queue/tasks/webhook/processor.ts

@@ -0,0 +1,262 @@
+import crypto from "crypto";
+import { Redis } from "ioredis";
+import { Db } from "mongodb";
+import getNow from "performance-now";
+
+import { Config } from "coral-server/config";
+import { CoralEventPayload } from "coral-server/events/event";
+import logger from "coral-server/logger";
+import {
+  filterActiveSecrets,
+  filterExpiredSecrets,
+} from "coral-server/models/settings";
+import {
+  deleteEndpointSecrets,
+  Endpoint,
+  getWebhookEndpoint,
+} from "coral-server/models/tenant";
+import { JobProcessor } from "coral-server/queue/Task";
+import { createFetch, FetchOptions } from "coral-server/services/fetch";
+import { disableWebhookEndpoint } from "coral-server/services/tenant";
+import TenantCache from "coral-server/services/tenant/cache";
+
+export const JOB_NAME = "webhook";
+
+// The count of failures on a webhook delivery before we disable the endpoint.
+const MAXIMUM_FAILURE_COUNT = 10;
+
+// The number of webhook attempts that should be retained for debugging.
+const MAXIMUM_EVENT_ATTEMPTS_LOG_SIZE = 50;
+
+export interface WebhookProcessorOptions {
+  config: Config;
+  mongo: Db;
+  redis: Redis;
+  tenantCache: TenantCache;
+}
+
+export interface WebhookData {
+  contextID: string;
+  endpointID: string;
+  tenantID: string;
+  event: CoralEventPayload;
+}
+
+export interface WebhookDelivery {
+  id: string;
+  name: string;
+  success: boolean;
+  status: number;
+  statusText: string;
+  request: string;
+  response: string;
+  createdAt: Date;
+}
+
+/**
+ * generateSignature will generate a signature used to assist clients to
+ * validate that the request came from Coral.
+ *
+ * @param secret the secret used to sign the body with
+ * @param body the body to use when signing
+ */
+export function generateSignature(secret: string, body: string) {
+  return crypto
+    .createHmac("sha256", secret)
+    .update(body)
+    .digest()
+    .toString("hex");
+}
+
+export function generateSignatures(
+  endpoint: Pick<Endpoint, "signingSecrets">,
+  body: string,
+  now: Date
+) {
+  // For each of the signatures, we only want to sign the body with secrets that
+  // are still active.
+  return endpoint.signingSecrets
+    .filter(filterActiveSecrets(now))
+    .map(({ secret }) => generateSignature(secret, body))
+    .map(signature => `sha256=${signature}`)
+    .join(",");
+}
+
+export function generateFetchOptions(
+  endpoint: Pick<Endpoint, "signingSecrets">,
+  data: CoralEventPayload,
+  now: Date
+): FetchOptions {
+  // Serialize the body and signature to include in the request.
+  const body = JSON.stringify(data, null, 2);
+  const signature = generateSignatures(endpoint, body, now);
+
+  const headers: Record<string, any> = {
+    "Content-Type": "application/json",
+    "X-Coral-Event": data.type,
+    "X-Coral-Signature": signature,
+  };
+
+  return {
+    method: "POST",
+    headers,
+    body,
+  };
+}
+
+export function createJobProcessor({
+  mongo,
+  tenantCache,
+  redis,
+}: WebhookProcessorOptions): JobProcessor<WebhookData> {
+  // Create the fetcher that will orchestrate sending the actual webhooks.
+  const fetch = createFetch({ name: "Webhook" });
+
+  return async job => {
+    const { tenantID, endpointID, contextID, event } = job.data;
+
+    const log = logger.child(
+      {
+        eventID: event.id,
+        contextID,
+        jobID: job.id,
+        jobName: JOB_NAME,
+        tenantID,
+        endpointID,
+      },
+      true
+    );
+
+    // Get the referenced tenant so we can get the endpoint details.
+    const tenant = await tenantCache.retrieveByID(tenantID);
+    if (!tenant) {
+      log.error("referenced tenant was not found");
+      return;
+    }
+
+    // Get the referenced endpoint.
+    const endpoint = getWebhookEndpoint(tenant, endpointID);
+    if (!endpoint) {
+      log.error("referenced endpoint was not found");
+      return;
+    }
+
+    // If the endpoint is disabled, don't bother processing it.
+    if (!endpoint.enabled) {
+      log.warn("endpoint was disabled, skipping sending");
+      return;
+    }
+
+    // Get the current date.
+    const now = new Date();
+
+    // Get the fetch options.
+    const options = generateFetchOptions(endpoint, event, now);
+
+    // Send the request.
+    const startedSendingAt = getNow();
+    const res = await fetch(endpoint.url, options);
+    const took = getNow() - startedSendingAt;
+    if (res.ok) {
+      log.info(
+        { took, responseStatus: res.status },
+        "finished sending webhook"
+      );
+    } else {
+      log.warn(
+        { took, responseStatus: res.status },
+        "failed to deliver webhook"
+      );
+    }
+
+    // Grab the response from the webhook, we'll want to save this in the recent
+    // attempts.
+    const response = await res.text();
+
+    // Collect the delivery information.
+    const delivery: WebhookDelivery = {
+      id: event.id,
+      name: event.type,
+      success: res.ok,
+      status: res.status,
+      statusText: res.statusText,
+      // We only serialize the body as a string.
+      request: options.body as string,
+      response,
+      createdAt: new Date(),
+    };
+
+    // Record the delivery.
+    const endpointDeliveriesKey = `${tenantID}:endpointDeliveries:${endpointID}`;
+    const endpointFailuresKey = `${tenantID}:endpointFailures:${endpointID}`;
+    let [, , [, failuresString]] = await redis
+      .multi()
+      // Push the attempt into the list.
+      .rpush(endpointDeliveriesKey, JSON.stringify(delivery))
+      // Trim the list to the 50 most recent attempts.
+      .ltrim(endpointDeliveriesKey, 0, MAXIMUM_EVENT_ATTEMPTS_LOG_SIZE - 1)
+      // Get the current failure count.
+      .get(endpointFailuresKey)
+      // Execute the queued operations.
+      .exec();
+
+    let failures = failuresString ? parseInt(failuresString, 10) : null;
+    if (res.ok && failures && failures > 0) {
+      // The webhook delivery was a success, and there were previous failures.
+      // Remove the failures record.
+      await redis.del(endpointFailuresKey);
+    } else if (!res.ok) {
+      // Record the failed attempt.
+      failuresString = await redis.incr(endpointFailuresKey);
+
+      // If the failure count is higher than the allowed maximum, disable the
+      // endpoint.
+      failures = failuresString ? parseInt(failuresString, 10) : null;
+      if (failures && failures >= MAXIMUM_FAILURE_COUNT) {
+        log.warn(
+          { failures, maxFailures: MAXIMUM_FAILURE_COUNT },
+          "maximum failures reached, disabling endpoint"
+        );
+
+        await disableWebhookEndpoint(
+          mongo,
+          redis,
+          tenantCache,
+          tenant,
+          endpointID
+        );
+      } else {
+        // TODO: (wyattjoh) maybe schedule a retry?
+      }
+    }
+
+    // Remove the expired secrets in the next tick so that it does not affect
+    // the sending performance of this job, and errors do not impact the
+    // sending.
+    const expiredSigningSecrets = endpoint.signingSecrets.filter(
+      filterExpiredSecrets(now)
+    );
+    if (expiredSigningSecrets.length > 0) {
+      process.nextTick(() => {
+        deleteEndpointSecrets(
+          mongo,
+          tenantID,
+          endpoint.id,
+          expiredSigningSecrets.map(s => s.kid)
+        )
+          .then(() => {
+            log.info(
+              { secrets: expiredSigningSecrets.length },
+              "removed expired secrets from endpoint"
+            );
+          })
+          .catch(err => {
+            log.error(
+              { err },
+              "an error occurred when trying to remove expired secrets"
+            );
+          });
+      });
+    }
+  };
+}