spike: introduce generic workload interface

config.default.json

@@ -10,6 +10,10 @@
   "ACCEPT_IAC": "tf,yaml,yml,json,tpl",
   "ACCEPT_LARGE_MANIFESTS": true,
   "ACCEPT_CUSTOM_PR_TEMPLATES": true,
+  "REMOTE_WORKLOAD_NAME":"BrokerWorkload",
+  "REMOTE_WORKLOAD_MODULE_PATH":"../broker-workload/websocketRequests",
+  "CLIENT_WORKLOAD_NAME": "BrokerClientRequestWorkload",
+  "CLIENT_WORKLOAD_MODULE_PATH": "../broker-workload/clientRequests",
   "BROKER_SERVER_UNIVERSAL_CONFIG_ENABLED": false,
   "SUPPORTED_BROKER_TYPES": [
     "apprisk",

lib/broker-workload/clientRequest.ts → lib/broker-workload/clientRequests.ts

@@ -3,18 +3,24 @@ import { HybridClientRequestHandler } from '../hybrid-sdk/clientRequestHelpers';
 import { incrementHttpRequestsTotal } from '../common/utils/metrics';
 import { filterClientRequest } from './requestFiltering';
 import { log as logger } from '../logs/logger';
+import {
+  LocalClientWorkloadRuntimeParams,
+  Workload,
+  WorkloadType,
+} from '../hybrid-sdk/workloadFactory';
 
-export class BrokerClientRequestWorkload {
+export class BrokerClientRequestWorkload extends Workload<WorkloadType.localClient> {
   req: Request;
   res: Response;
   options;
   constructor(req, res, options) {
+    super('broker', WorkloadType['local-client']);
     this.req = req;
     this.res = res;
     this.options = options;
   }
 
-  async handler(makeRequestOverHttp = false) {
+  async handler(data: LocalClientWorkloadRuntimeParams) {
     const hybridClientRequestHandler = new HybridClientRequestHandler(
       this.req,
       this.res,
@@ -38,7 +44,7 @@ export class BrokerClientRequestWorkload {
     } else {
       hybridClientRequestHandler.makeRequest(
         filterResponse,
-        makeRequestOverHttp,
+        data.makeRequestOverHttp,
       );
       incrementHttpRequestsTotal(false, 'inbound-request');
     }

lib/common/relay/prepareRequest.ts → lib/broker-workload/prepareRequest.ts

@@ -1,37 +1,30 @@
 import { format, parse } from 'url';
-import { TestResult } from '../types/filter';
-import version from '../utils/version';
-import tryJSONParse from '../utils/try-json-parse';
-import { replace } from '../utils/replace-vars';
+import { TestResult } from '../common/types/filter';
+import version from '../common/utils/version';
+import tryJSONParse from '../common/utils/try-json-parse';
+import { replace } from '../common/utils/replace-vars';
 import undefsafe from 'undefsafe';
-import { log as logger } from '../../logs/logger';
+import { log as logger } from '../logs/logger';
 import {
   gitHubCommitSigningEnabled,
   gitHubTreeCheckNeeded,
   signGitHubCommit,
   validateGitHubTreePayload,
-} from '../../client/scm';
-import { getConfigForIdentifier } from '../config/universal';
-import { computeContentLength } from '../utils/content-length';
+} from '../client/scm';
+import { getConfigForIdentifier } from '../common/config/universal';
+import { computeContentLength } from './content-length';
 import {
   contentLengthHeader,
   contentTypeHeader,
   urlencoded,
-} from '../utils/headers-value-constants';
+} from './headers-value-constants';
+import { PostFilterPreparedRequest } from '../hybrid-sdk/types';
 
 export interface PostFilterPreparingRequestError {
   status: number;
   errorMsg: string;
 }
 
-export interface PostFilterPreparedRequest {
-  url: string;
-  headers: Object;
-  method: string;
-  body?: any;
-  timeoutMs?: number;
-}
-
 export const prepareRequestFromFilterResult = async (
   result: TestResult,
   payload,

lib/broker-workload/index.ts → lib/broker-workload/websocketRequests.ts

@@ -1,24 +1,29 @@
 import { runPreRequestPlugins } from '../client/brokerClientPlugins/pluginManager';
-import { prepareRequestFromFilterResult } from '../common/relay/prepareRequest';
+import { prepareRequestFromFilterResult } from './prepareRequest';
 import { ExtendedLogContext } from '../common/types/log';
-import { computeContentLength } from '../common/utils/content-length';
-import { contentLengthHeader } from '../common/utils/headers-value-constants';
+import { computeContentLength } from './content-length';
+import { contentLengthHeader } from './headers-value-constants';
 import {
   incrementWebSocketRequestsTotal,
   incrementHttpRequestsTotal,
 } from '../common/utils/metrics';
 import { maskToken, hashToken } from '../common/utils/token';
 import { log as logger } from '../logs/logger';
 import { HybridResponseHandler } from '../hybrid-sdk/responseSenders';
-import { getCorrelationDataFromHeaders } from '../common/utils/correlation-headers';
+import { getCorrelationDataFromHeaders } from './correlation-headers';
 import { filterRequest } from './requestFiltering';
 import {
   makeRequestToDownstream,
   makeStreamingRequestToDownstream,
 } from '../hybrid-sdk/http/request';
 import { logError } from '../logs/log';
+import {
+  RemoteServerWorkloadRuntimeParams,
+  Workload,
+  WorkloadType,
+} from '../hybrid-sdk/workloadFactory';
 
-export class BrokerWorkload {
+export class BrokerWorkload extends Workload<WorkloadType.remoteServer> {
   options;
   connectionIdentifier: string;
   websocketConnectionHandler;
@@ -27,12 +32,15 @@ export class BrokerWorkload {
     options,
     websocketConnectionHandler,
   ) {
+    super('broker', WorkloadType['remote-server']);
     this.options = options;
     this.connectionIdentifier = connectionIdentifier;
     this.websocketConnectionHandler = websocketConnectionHandler;
   }
 
-  async handler(payload, websocketResponseHandler) {
+  async handler(data: RemoteServerWorkloadRuntimeParams) {
+    const { payload, websocketHandler } = data;
+    const websocketResponseHandler = websocketHandler;
     if (this.options.config.universalBrokerEnabled) {
       payload.connectionIdentifier = this.connectionIdentifier;
     }

lib/client/auth/oauth.ts

@@ -1,5 +1,5 @@
 import { makeRequestToDownstream } from '../../hybrid-sdk/http/request';
-import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../hybrid-sdk/types';
 import { log as logger } from '../../logs/logger';
 interface tokenExchangeResponse {
   access_token: string;

lib/client/brokerClientPlugins/abstractBrokerPlugin.ts

@@ -3,7 +3,7 @@ import {
   HttpResponse,
   makeRequestToDownstream,
 } from '../../hybrid-sdk/http/request';
-import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../hybrid-sdk/types';
 import { log as logger } from '../../logs/logger';
 
 export default abstract class BrokerPlugin {

lib/client/brokerClientPlugins/pluginManager.ts

@@ -2,7 +2,7 @@ import { readdir } from 'fs/promises';
 import { log as logger } from '../../logs/logger';
 import BrokerPlugin from './abstractBrokerPlugin';
 import { existsSync } from 'fs';
-import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../hybrid-sdk/types';
 import { getPluginsConfig } from '../../common/config/pluginsConfig';
 
 export const loadPlugins = async (pluginsFolderPath: string, clientOpts) => {

lib/client/brokerClientPlugins/plugins/githubServerAppAuth.ts

@@ -3,7 +3,7 @@ import { existsSync, readFileSync } from 'node:fs';
 import BrokerPlugin from '../abstractBrokerPlugin';
 import { createPrivateKey } from 'node:crypto';
 import { sign } from 'jsonwebtoken';
-import { PostFilterPreparedRequest } from '../../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../../hybrid-sdk/types';
 import { makeRequestToDownstream } from '../../../hybrid-sdk/http/request';
 import { maskSCMToken } from '../../../common/utils/token';
 import { getPluginsConfig } from '../../../common/config/pluginsConfig';

lib/client/checks/config/brokerClientVersionCheck.ts

@@ -1,5 +1,5 @@
 import { makeSingleRawRequestToDownstream } from '../../../hybrid-sdk/http/request';
-import { PostFilterPreparedRequest } from '../../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../../hybrid-sdk/types';
 import version from '../../../common/utils/version';
 import { CheckOptions, CheckResult } from '../types';
 

lib/client/config/remoteConfig.ts

@@ -1,6 +1,6 @@
 import { readFileSync, writeFileSync } from 'fs';
 import { makeRequestToDownstream } from '../../hybrid-sdk/http/request';
-import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../hybrid-sdk/types';
 import { ClientOpts } from '../../common/types/options';
 import { BrokerConnectionApiResponse } from '../types/api';
 import { capitalizeKeys } from '../utils/configurations';

lib/client/dispatcher/client/api.ts

@@ -1,5 +1,5 @@
 import { makeRequestToDownstream } from '../../../hybrid-sdk/http/request';
-import { PostFilterPreparedRequest } from '../../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../../hybrid-sdk/types';
 import { log as logger } from '../../../logs/logger';
 import {
   CreateConnectionRequestData,

lib/client/utils/connectionValidation.ts

@@ -1,5 +1,5 @@
 import { makeSingleRawRequestToDownstream } from '../../hybrid-sdk/http/request';
-import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../hybrid-sdk/types';
 import version from '../../common/utils/version';
 import { ConnectionConfig } from '../types/config';
 import { log as logger } from '../../logs/logger';

lib/common/filter/utils.ts

@@ -1,7 +1,7 @@
 import path from 'node:path';
 import fs from 'fs';
 import { makeSingleRawRequestToDownstream } from '../../hybrid-sdk/http/request';
-import { PostFilterPreparedRequest } from '../relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../../hybrid-sdk/types';
 import version from '../utils/version';
 import { findProjectRoot } from '../config/config';
 import { log as logger } from '../../logs/logger';

lib/common/relay/forwardHttpRequest.ts

@@ -1,7 +1,12 @@
 import { Request, Response } from 'express';
 
 import { LoadedClientOpts, LoadedServerOpts } from '../types/options';
-import { BrokerClientRequestWorkload } from '../../broker-workload/clientRequest';
+import {
+  LocalClientWorkloadRuntimeParams,
+  Workload,
+  WorkloadType,
+} from '../../hybrid-sdk/workloadFactory';
+import { BrokerClientRequestWorkload } from '../../broker-workload/clientRequests';
 
 // 1. Request coming in over HTTP conn (logged)
 // 2. Filter for rule match (log and block if no match)
@@ -13,7 +18,19 @@ export const forwardHttpRequest = (
   makeHttpRequest = false,
 ) => {
   return async (req: Request, res: Response) => {
-    const workload = new BrokerClientRequestWorkload(req, res, options);
-    await workload.handler(makeHttpRequest);
+    const workloadName = options.config.clientWorkloadName;
+    const workloadModulePath = options.config.clientWorkloadModulePath;
+
+    const workload = (await Workload.instantiate(
+      workloadName,
+      workloadModulePath,
+      WorkloadType.localClient,
+      { req, res, options },
+    )) as BrokerClientRequestWorkload;
+
+    const data: LocalClientWorkloadRuntimeParams = {
+      makeRequestOverHttp: makeHttpRequest,
+    };
+    await workload.handler(data);
   };
 };

lib/common/relay/forwardWebsocketRequest.ts

@@ -1,7 +1,13 @@
 import { RequestPayload } from '../types/http';
 import { WebSocketConnection } from '../../client/types/client';
 import { LoadedClientOpts, LoadedServerOpts } from '../types/options';
-import { BrokerWorkload } from '../../broker-workload';
+import {
+  RemoteServerWorkloadRuntimeParams,
+  Workload,
+  WorkloadType,
+} from '../../hybrid-sdk/workloadFactory';
+import { BrokerWorkload } from '../../broker-workload/websocketRequests';
+// import { BrokerWorkload } from '../../broker-workload/websocketRequests';
 
 export const forwardWebSocketRequest = (
   options: LoadedClientOpts | LoadedServerOpts,
@@ -14,11 +20,19 @@ export const forwardWebSocketRequest = (
   // 5. Send response over websocket conn
 
   return (connectionIdentifier) => async (payload: RequestPayload, emit) => {
-    const workload = new BrokerWorkload(
-      connectionIdentifier,
-      options,
-      websocketConnectionHandler,
-    );
-    await workload.handler(payload, emit);
+    const workloadName = options.config.remoteWorkloadName;
+    const workloadModulePath = options.config.remoteWorkloadModulePath;
+    const workload = (await Workload.instantiate(
+      workloadName,
+      workloadModulePath,
+      WorkloadType.remoteServer,
+      { connectionIdentifier, options, websocketConnectionHandler },
+    )) as BrokerWorkload;
+
+    const data: RemoteServerWorkloadRuntimeParams = {
+      payload,
+      websocketHandler: emit,
+    };
+    await workload.handler(data);
   };
 };

lib/hybrid-sdk/http/request.ts

@@ -3,7 +3,7 @@ import https from 'https';
 import { getProxyForUrl } from 'proxy-from-env';
 import { bootstrap } from 'global-agent';
 import { log as logger } from '../../logs/logger';
-import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from '../types';
 import { getConfig } from '../../common/config/config';
 import { switchToInsecure } from './utils';
 export interface HttpResponse {

lib/hybrid-sdk/requestsHelper.ts

@@ -2,7 +2,7 @@ import {
   makeRequestToDownstream,
   makeStreamingRequestToDownstream,
 } from './http/request';
-import { PostFilterPreparedRequest } from '../common/relay/prepareRequest';
+import { PostFilterPreparedRequest } from './types';
 import { log as logger } from '../logs/logger';
 import { logError, logResponse } from '../logs/log';
 import { isJson } from '../common/utils/json';

lib/hybrid-sdk/types.ts

@@ -1,7 +1,14 @@
-import { CorrelationHeaders } from '../common/utils/correlation-headers';
+import { CorrelationHeaders } from '../broker-workload/correlation-headers';
 
 export type RequestMetadata = {
   connectionIdentifier: string;
   payloadStreamingId: string;
   //   streamResponse: boolean;
 } & CorrelationHeaders;
+export interface PostFilterPreparedRequest {
+  url: string;
+  headers: Object;
+  method: string;
+  body?: any;
+  timeoutMs?: number;
+}