Add webhook step

Author: pimterry

src/rules/requests/request-rule-builder.ts

@@ -20,7 +20,10 @@ import {
     JsonRpcResponseStep,
     ResetConnectionStep,
     CallbackResponseMessageResult,
-    DelayStep
+    DelayStep,
+    WebhookStep,
+    WaitForRequestBodyStep,
+    RequestWebhookEvents
 } from "./request-step-definitions";
 import { byteLength } from "../../util/util";
 import { BaseRuleBuilder } from "../base-rule-builder";
@@ -99,6 +102,32 @@ export class RequestRuleBuilder extends BaseRuleBuilder {
         return this;
     }
 
+    /**
+     * Wait until the request body has been fully received before continuing.
+     *
+     * Without this, other handlers like `thenReply` will react immediately, e.g. sending a
+     * response as soon as the headers are received, before the body has arrived. That is
+     * perfectly valid and will probably work fine, but could cause strange behaviour
+     * in some edge cases, and is not representative of how real server responses would
+     * generally behave.
+     */
+    waitForRequestBody(): this {
+        this.steps.push(new WaitForRequestBodyStep());
+        return this;
+    }
+
+    /**
+     * Register a webhook for the given events. The provided URL will receive a POST request
+     * with a JSON body containing the details of the configured events when they occur. If
+     * no event list is specified then it defaults to `['request', 'response']`.
+     *
+     * The JSON body will contain `{ eventType: string, eventData: object }`.
+     */
+    addWebhook(url: string, events?: RequestWebhookEvents[]): this {
+        this.steps.push(new WebhookStep(url, events ?? ['request', 'response']));
+        return this;
+    }
+
     /**
      * Reply to matched requests with a given status code and (optionally) status message,
      * body, headers & trailers.

src/rules/requests/request-step-definitions.ts

@@ -6,7 +6,7 @@ import {
     Operation as JsonPatchOperation,
     validate as validateJsonPatch
 } from 'fast-json-patch';
-import { MaybePromise } from '@httptoolkit/util';
+import { MaybePromise, joinAnd } from '@httptoolkit/util';
 
 import {
     Headers,
@@ -19,6 +19,7 @@ import {
 
 import { Replace } from '../../util/type-utils';
 import { asBuffer } from '../../util/buffer-utils';
+import { isAbsoluteUrl } from '../../util/url';
 import {
     MatchReplacePairs,
     SerializedMatchReplacePairs,
@@ -1077,6 +1078,47 @@ export class DelayStep extends Serializable implements RequestStepDefinition {
 
 }
 
+export class WaitForRequestBodyStep extends Serializable implements RequestStepDefinition {
+
+    readonly type = 'wait-for-request-body'
+    static readonly isFinal = false;
+
+    explain(): string {
+        return 'wait for the full request body to be received';
+    }
+
+}
+
+export type RequestWebhookEvents =
+    | 'request'
+    | 'response';
+
+export class WebhookStep extends Serializable implements RequestStepDefinition {
+
+    readonly type = 'webhook';
+    static readonly isFinal = false;
+
+    constructor(
+        public readonly url: string,
+        public readonly events: RequestWebhookEvents[]
+    ) {
+        super();
+
+        if (!isAbsoluteUrl(url)) {
+            throw new Error(`Webhook URL "${url}" must be absolute`);
+        }
+    }
+
+    explain(): string {
+        // We actively support sending no events to make it easier to quickly toggle
+        // settings here during debugging without breaking anything unnecessarily.
+        return `use ${this.url} as a webhook for ${
+            this.events?.length ? joinAnd(this.events) : 'no'
+        } events`;
+    }
+
+}
+
 export const StepDefinitionLookup = {
     'simple': FixedResponseStep,
     'callback': CallbackStep,
@@ -1087,5 +1129,7 @@ export const StepDefinitionLookup = {
     'reset-connection': ResetConnectionStep,
     'timeout': TimeoutStep,
     'json-rpc-response': JsonRpcResponseStep,
-    'delay': DelayStep
+    'delay': DelayStep,
+    'wait-for-request-body': WaitForRequestBodyStep,
+    'webhook': WebhookStep
 }

src/rules/requests/request-step-impls.ts

@@ -29,7 +29,8 @@ import {
     shouldKeepAlive,
     isHttp2,
     writeHead,
-    encodeBodyBuffer
+    encodeBodyBuffer,
+    waitForCompletedResponse
 } from '../../util/request-utils';
 import {
     h1HeadersToH2,
@@ -115,7 +116,9 @@ import {
     FixedResponseStep,
     StreamStep,
     TimeoutStep,
-    DelayStep
+    DelayStep,
+    WebhookStep,
+    WaitForRequestBodyStep
 } from './request-step-definitions';
 
 // Re-export various type definitions. This is mostly for compatibility with external
@@ -1386,6 +1389,94 @@ export class DelayStepImpl extends DelayStep {
     }
 }
 
+export class WaitForRequestBodyImpl extends WaitForRequestBodyStep {
+    async handle(request: OngoingRequest): Promise<{ continue: true }> {
+        await request.body.asBuffer();
+        return { continue: true };
+    }
+}
+
+const encodeWebhookBody = (body: Buffer) => {
+    return {
+        format: 'base64',
+        data: body.toString('base64')
+    };
+};
+
+export class WebhookStepImpl extends WebhookStep {
+
+    private sendEvent(data: {
+        eventType: string;
+        eventData: {};
+    }) {
+        const content = JSON.stringify(data);
+        const req = http.request(this.url, {
+            method: 'POST',
+            headers: {
+                'Content-Type': 'application/json',
+                'Content-Length': Buffer.byteLength(content)
+            }
+        }).end(content);
+
+        req.on('error', (e) => {
+            console.warn(`Error sending webhook to ${this.url}:`, e);
+        });
+
+        req.on('response', (res) => {
+            if (res.statusCode !== 200) {
+                console.warn(`Received unexpected ${res.statusCode} response from webhook ${this.url} for ${data.eventType}`);
+            }
+
+            res.on('error', () => {});
+            res.resume();
+        });
+    }
+
+    async handle(request: OngoingRequest, response: OngoingResponse) {
+        if (this.events.includes('request')) {
+            waitForCompletedRequest(request).then((completedReq) => {
+                const eventData = {
+                    ..._.pick(completedReq, [
+                        'id',
+                        'method',
+                        'url',
+                        'headers',
+                        'trailers'
+                    ]),
+                    body: encodeWebhookBody(completedReq.body.buffer)
+                }
+
+                this.sendEvent({
+                    eventType: 'request',
+                    eventData: eventData
+                });
+            }).catch(() => {});
+        }
+
+        if (this.events.includes('response')) {
+            waitForCompletedResponse(response).then((completedRes) => {
+                const eventData = {
+                    ..._.pick(completedRes, [
+                        'id',
+                        'statusCode',
+                        'statusMessage',
+                        'headers',
+                        'trailers'
+                    ]),
+                    body: encodeWebhookBody(completedRes.body.buffer)
+                }
+
+                this.sendEvent({
+                    eventType: 'response',
+                    eventData: eventData
+                });
+            }).catch(() => {});
+        }
+
+        return { continue: true };
+    }
+}
+
 export const StepLookup: typeof StepDefinitionLookup = {
     'simple': FixedResponseStepImpl,
     'callback': CallbackStepImpl,
@@ -1396,5 +1487,7 @@ export const StepLookup: typeof StepDefinitionLookup = {
     'reset-connection': ResetConnectionStepImpl,
     'timeout': TimeoutStepImpl,
     'json-rpc-response': JsonRpcResponseStepImpl,
-    'delay': DelayStepImpl
+    'delay': DelayStepImpl,
+    'wait-for-request-body': WaitForRequestBodyImpl,
+    'webhook': WebhookStepImpl
 }