feat: adds an API for integration with Express (#3)

Author: TENSIILE

package-lock.json

@@ -45,10 +45,10 @@
   },
   "lint-staged": {
     "*.{js,jsx,ts,tsx}": [
-      "npm run fix:prettier"
+      "npm run format:prettier"
     ],
     "*.{json,md}": [
-      "npm run fix:prettier"
+      "npm run format:prettier"
     ]
   }
 }

package.json

@@ -1 +1 @@
-export * from './abortable-service';
+export * from './request-interruption-service';

src/service/abortable-service.ts

@@ -0,0 +1,2 @@
+export const abortableMetaSymbol = Symbol('Saborter.meta');
+export const X_REQUEST_ID_HEADER = 'x-request-id';

src/service/index.ts

@@ -0,0 +1,47 @@
+import { Request, Response } from 'express';
+import { getRequestMeta } from './request-interruption-service.utils';
+import { WithAbortableConfig } from './request-interruption-service.types';
+
+export const getAbortSignal = (req: Request): AbortSignal | null => {
+  return getRequestMeta(req).signal ?? null;
+};
+
+export const withAbortable = (
+  handler: (req: Request, res: Response) => Promise<void>,
+  { isErrorNativeBehavior = false }: WithAbortableConfig = {}
+) => {
+  return async (req: Request, res: Response) => {
+    const signal = getAbortSignal(req);
+
+    if (!signal) return handler(req, res);
+
+    let currentReject: ((reason?: any) => void) | null = null;
+
+    const listener = () => currentReject?.(signal.reason);
+
+    try {
+      await Promise.race([
+        handler(req, res),
+        new Promise((_, reject) => {
+          currentReject = reject;
+
+          signal.addEventListener('abort', listener, {
+            once: true
+          });
+        })
+      ]);
+    } catch (err) {
+      if (isErrorNativeBehavior) {
+        throw err;
+      }
+    } finally {
+      signal.removeEventListener('abort', listener);
+    }
+  };
+};
+
+export const abort = (req: Request, reason?: any): void => {
+  const controller = getRequestMeta(req).controller ?? new AbortController();
+
+  controller.abort(reason);
+};

src/service/request-interruption-service.constants.ts

@@ -0,0 +1,74 @@
+import { Request, Response, NextFunction, Express } from 'express';
+import { AbortFunction } from './request-interruption-service.types';
+import { getRequestId, setRequestMeta } from './request-interruption-service.utils';
+
+class RequestInterruptionService {
+  private abortRegistries = new Map<string, AbortFunction | null>();
+
+  private registerAbortableFunction = (requestId: string, abortFn: AbortFunction): void => {
+    this.abortRegistries.set(requestId, abortFn);
+  };
+
+  public abort = (requestId: string, { deletable = true }: { deletable?: boolean } = {}): boolean => {
+    const abortFn = this.abortRegistries.get(requestId) ?? null;
+
+    if (abortFn) {
+      abortFn();
+
+      if (deletable) {
+        this.abortRegistries.delete(requestId);
+      }
+
+      return true;
+    }
+
+    return false;
+  };
+
+  public expressMiddleware = (req: Request, res: Response, next: NextFunction): void => {
+    const requestId = getRequestId(req);
+
+    if (!requestId) {
+      return next();
+    }
+
+    if (this.abortRegistries.has(requestId)) {
+      this.abort(requestId, { deletable: false });
+    }
+
+    const controller = new AbortController();
+
+    setRequestMeta(req, controller);
+
+    this.registerAbortableFunction(requestId, () => {
+      controller.abort(new Error('AbortError'));
+    });
+
+    req.on('close', () => {
+      if (!res.writableEnded) {
+        this.abort(requestId);
+      }
+    });
+
+    next();
+  };
+}
+
+export const initRequestInterruptionService = (
+  app: Express,
+  { basePath = '', endpointName = '/api/cancel' }: { basePath?: string; endpointName?: string } = {}
+) => {
+  const requestInterruptionService = new RequestInterruptionService();
+
+  app.use(requestInterruptionService.expressMiddleware);
+
+  app.post(`${basePath}${endpointName}`, (req, res) => {
+    const requestId = getRequestId(req);
+
+    if (requestId && requestInterruptionService.abort(requestId)) {
+      res.status(499).json({ cancelled: true });
+    } else {
+      res.status(404).json({ error: 'Request not found' });
+    }
+  });
+};

src/service/request-interruption-service.lib.ts

@@ -0,0 +1,11 @@
+export type AbortFunction = () => void;
+
+export interface InterruptibleRequestMeta {
+  requestId: string;
+  controller: AbortController;
+  signal: AbortSignal;
+}
+
+export interface WithAbortableConfig {
+  isErrorNativeBehavior?: boolean;
+}

src/service/request-interruption-service.ts

@@ -0,0 +1,23 @@
+import { Request } from 'express';
+import { InterruptibleRequestMeta } from './request-interruption-service.types';
+import * as Constants from './request-interruption-service.constants';
+
+export const getRequestId = (req: Request): string => {
+  const requestId = req.headers[Constants.X_REQUEST_ID_HEADER]?.toString().split(',');
+
+  return requestId?.[0] ?? '';
+};
+
+export const getRequestMeta = (req: Request): InterruptibleRequestMeta => {
+  return (req as any)[Constants.abortableMetaSymbol];
+};
+
+export const setRequestMeta = (req: Request, controller: AbortController): void => {
+  const requestId = getRequestId(req);
+
+  (req as any)[Constants.abortableMetaSymbol] = {
+    requestId,
+    controller,
+    signal: controller.signal
+  } as InterruptibleRequestMeta;
+};