Refactor middleware processing (v2)

package.json

@@ -46,9 +46,11 @@
     "@slack/web-api": "^5.8.0",
     "@types/express": "^4.16.1",
     "@types/node": ">=10",
+    "@types/promise.allsettled": "^1.0.3",
     "axios": "^0.19.0",
     "express": "^4.16.4",
     "please-upgrade-node": "^3.2.0",
+    "promise.allsettled": "^1.0.2",
     "raw-body": "^2.3.3",
     "tsscmp": "^1.0.6"
   },

src/App.spec.ts

@@ -5,7 +5,7 @@ import { assert } from 'chai';
 import { Override, mergeOverrides, createFakeLogger, delay } from './test-helpers';
 import rewiremock from 'rewiremock';
 import { ErrorCode, UnknownError } from './errors';
-import { Receiver, ReceiverEvent, SayFn, NextMiddleware } from './types';
+import { Receiver, ReceiverEvent, SayFn, NextFn } from './types';
 import { ConversationStore } from './conversation-store';
 import { LogLevel } from '@slack/logger';
 import App, { ViewConstraints } from './App';
@@ -14,16 +14,17 @@ import { WebClientOptions, WebClient } from '@slack/web-api';
 // TODO: swap out rewiremock for proxyquire to see if it saves execution time
 // Utility functions
 const noop = (() => Promise.resolve(undefined));
-const noopMiddleware = async ({ next }: { next: NextMiddleware; }) => { await next(); };
+const noopMiddleware = async ({ next }: { next: NextFn; }) => { await next!(); };
 const noopAuthorize = (() => Promise.resolve({}));
 
 // Dummies (values that have no real behavior but pass through the system opaquely)
-function createDummyReceiverEvent(): ReceiverEvent {
+function createDummyReceiverEvent(type: string = 'dummy_event_type'): ReceiverEvent {
   // NOTE: this is a degenerate ReceiverEvent that would successfully pass through the App. it happens to look like a
   // IncomingEventType.Event
   return {
     body: {
       event: {
+        type,
       },
     },
     ack: noop,
@@ -342,11 +343,13 @@ describe('App', () => {
         // Arrange
         app.use(fakeFirstMiddleware);
         app.use(async ({ next }) => {
-          await next();
-          await next();
+          await next!();
+          await next!();
         });
         app.use(fakeSecondMiddleware);
         app.error(fakeErrorHandler);
+
+        // Act
         await fakeReceiver.sendEvent(dummyReceiverEvent);
 
         // Assert
@@ -360,7 +363,7 @@ describe('App', () => {
           await delay(100);
           changed = true;
 
-          await next();
+          await next!();
         });
 
         await fakeReceiver.sendEvent(dummyReceiverEvent);
@@ -374,7 +377,7 @@ describe('App', () => {
 
         app.use(async ({ next }) => {
           try {
-            await next();
+            await next!();
           } catch (err) {
             caughtError = err;
           }
@@ -397,15 +400,17 @@ describe('App', () => {
         let middlewareCount = 0;
 
         /**
-         * Middleware that, when calls, asserts that it was called in the correct order
-         * @param orderDown The order it should be called when processing middleware down the stack
-         * @param orderUp The order it should be called when processing middleware up the stack
+         * Middleware that, when called, asserts that it was called in the correct order
+         * @param orderDown The order it should be called when processing middleware down the chain
+         * @param orderUp The order it should be called when processing middleware up the chain
          */
-        const assertOrderMiddleware = (orderDown: number, orderUp: number) => async ({ next }: any) => {
+        const assertOrderMiddleware = (orderDown: number, orderUp: number) => async ({ next }: { next?: NextFn }) => {
           await delay(100);
           middlewareCount += 1;
           assert.equal(middlewareCount, orderDown);
-          await next();
+          if (next !== undefined) {
+            await next();
+          }
           middlewareCount += 1;
           assert.equal(middlewareCount, orderUp);
         };
@@ -465,6 +470,36 @@ describe('App', () => {
       });
     });
 
+    describe('listener middleware', () => {
+      it('should aggregate multiple errors in listeners for the same incoming event', async () => {
+        // Arrange
+        const App = await importApp(); // tslint:disable-line:variable-name
+        const app = new App({
+          receiver: fakeReceiver,
+          authorize: sinon.fake.resolves(dummyAuthorizationResult),
+        });
+        const eventType = 'some_event_type';
+        const dummyReceiverEvent = createDummyReceiverEvent(eventType);
+        app.error(fakeErrorHandler);
+        const errorsToThrow = [new Error('first listener error'), new Error('second listener error')];
+        function createThrowingListener(toBeThrown: Error): () => Promise<void> {
+          return async () => { throw toBeThrown; };
+        }
+
+        // Act
+        app.event('some_event_type', createThrowingListener(errorsToThrow[0]));
+        app.event('some_event_type', createThrowingListener(errorsToThrow[1]));
+        await fakeReceiver.sendEvent(dummyReceiverEvent);
+
+        // Assert
+        assert(fakeErrorHandler.calledOnce);
+        const error = fakeErrorHandler.firstCall.args[0];
+        assert.instanceOf(error, Error);
+        assert(error.code === ErrorCode.MultipleListenerError);
+        assert.sameMembers(error.originals, errorsToThrow);
+      });
+    });
+
     describe('middleware and listener arguments', () => {
       let fakeErrorHandler: SinonSpy;
       const dummyChannelId = 'CHANNEL_ID';
@@ -655,7 +690,7 @@ describe('App', () => {
 
           app.use(async ({ next }) => {
             await ackFn();
-            await next();
+            await next!();
           });
           app.shortcut({ callback_id: 'message_action_callback_id' }, async ({ }) => { await shortcutFn(); });
           app.shortcut(
@@ -799,7 +834,7 @@ describe('App', () => {
           });
           app.use(async ({ logger, body, next }) => {
             logger.info(body);
-            await next();
+            await next!();
           });
 
           app.event('app_home_opened', async ({ logger, event }) => {
@@ -847,7 +882,7 @@ describe('App', () => {
           });
           app.use(async ({ logger, body, next }) => {
             logger.info(body);
-            await next();
+            await next!();
           });
 
           app.event('app_home_opened', async ({ logger, event }) => {
@@ -913,7 +948,7 @@ describe('App', () => {
           });
           app.use(async ({ client, next }) => {
             await client.auth.test();
-            await next();
+            await next!();
           });
           const clients: WebClient[] = [];
           app.event('app_home_opened', async ({ client }) => {

src/App.ts

@@ -53,8 +53,9 @@ import {
   CodedError,
   asCodedError,
   AppInitializationError,
+  MultipleListenerError,
 } from './errors';
-import { MiddlewareContext } from './types/middleware';
+import allSettled = require('promise.allsettled'); // tslint:disable-line:no-require-imports import-name
 const packageJson = require('../package.json'); // tslint:disable-line:no-require-imports no-var-requires
 
 /** App initialization options */
@@ -155,13 +156,13 @@ export default class App {
   /** Logger */
   private logger: Logger;
 
-  /**  */
+  /** Authorize */
   private authorize: Authorize;
 
-  /** Global middleware */
+  /** Global middleware chain */
   private middleware: Middleware<AnyMiddlewareArgs>[];
 
-  /** Listeners (and their middleware) */
+  /** Listener middleware chains */
   private listeners: Middleware<AnyMiddlewareArgs>[][];
 
   private errorHandler: ErrorHandler;
@@ -503,16 +504,6 @@ export default class App {
       };
     };
 
-    let listenerArgClient = this.client;
-    const token = selectToken(context);
-    if (typeof token !== 'undefined') {
-      let pool = this.clients[source.teamId];
-      if (typeof pool === 'undefined') {
-        pool = this.clients[source.teamId] = new WebClientPool();
-      }
-      listenerArgClient = pool.getOrCreate(token, this.clientOptions);
-    }
-
     // Set body and payload (this value will eventually conform to AnyMiddlewareArgs)
     // NOTE: the following doesn't work because... distributive?
     // const listenerArgs: Partial<AnyMiddlewareArgs> = {
@@ -523,13 +514,7 @@ export default class App {
       respond?: RespondFn,
       /** Ack function might be set below */
       ack?: AckFn<any>,
-      /** The logger for this Bolt app */
-      logger?: Logger,
-      /** WebClient with token  */
-      client?: WebClient,
     } = {
-      logger: this.logger,
-      client: listenerArgClient,
       body: bodyArg,
       payload:
         (type === IncomingEventType.Event) ?
@@ -595,30 +580,66 @@ export default class App {
       // Events API requests are acknowledged right away, since there's no data expected
       await ack();
     }
-    const middlewareChain = [...this.middleware];
-
-    if (this.listeners.length > 0) {
-      middlewareChain.push(async (ctx) => {
-        const { next } = ctx;
-
-        await Promise.all(this.listeners.map(
-            listener => processMiddleware(listener, ctx)));
 
-        await next();
-      });
+    // Get the client arg
+    let client = this.client;
+    const token = selectToken(context);
+    if (typeof token !== 'undefined') {
+      let pool = this.clients[source.teamId];
+      if (typeof pool === 'undefined') {
+        pool = this.clients[source.teamId] = new WebClientPool();
+      }
+      client = pool.getOrCreate(token, this.clientOptions);
     }
 
-    // Dispatch event through global middleware
+    // Dispatch even through the global middleware chain
     try {
-      await processMiddleware(middlewareChain, {
+      await processMiddleware(
+        this.middleware,
+        listenerArgs as AnyMiddlewareArgs,
         context,
-        ...(listenerArgs as MiddlewareContext<AnyMiddlewareArgs>),
-      });
+        client,
+        this.logger,
+        async () => {
+          // Dispatch the event through the listener middleware chains and aggregate their results
+          // TODO: change the name of this.middleware and this.listeners to help this make more sense
+          const listenerResults = this.listeners.map(async (origListenerMiddleware) => {
+            // Copy the array so modifications don't affect the original
+            const listenerMiddleware = [...origListenerMiddleware];
+
+            // Don't process the last item in the listenerMiddleware array - it shouldn't get a next fn
+            const listener = listenerMiddleware.pop();
+
+            if (listener !== undefined) {
+              return processMiddleware(
+                listenerMiddleware,
+                listenerArgs as AnyMiddlewareArgs,
+                context,
+                client,
+                this.logger,
+                async () =>
+                  // When the listener middleware chain is done processing, call the listener without a next fn
+                  listener({ ...listenerArgs as AnyMiddlewareArgs, context, client, logger: this.logger }),
+              );
+            }
+          });
+
+          const settledListenerResults = await allSettled(listenerResults);
+          const rejectedListenerResults =
+            settledListenerResults.filter(lr => lr.status === 'rejected') as allSettled.PromiseRejection<Error>[];
+          if (rejectedListenerResults.length === 1) {
+            throw rejectedListenerResults[0];
+          } else if (rejectedListenerResults.length > 1) {
+            throw new MultipleListenerError(rejectedListenerResults.map(rlr => rlr.reason));
+          }
+        },
+      );
     } catch (error) {
       return this.handleError(error);
     }
   }
 
+  // TODO: make the following method private if its no longer being used by Receiver
   /**
    * Global error handler. The final destination for all errors (hopefully).
    */

src/conversation-store.spec.ts

@@ -5,7 +5,7 @@ import sinon, { SinonSpy } from 'sinon';
 import { Override, createFakeLogger, delay, wrapToResolveOnFirstCall } from './test-helpers';
 import rewiremock from 'rewiremock';
 import { ConversationStore } from './conversation-store';
-import { AnyMiddlewareArgs, NextMiddleware, Context } from './types';
+import { AnyMiddlewareArgs, NextFn, Context } from './types';
 import { WebClient } from '@slack/web-api';
 import { Logger } from '@slack/logger';
 
@@ -193,7 +193,7 @@ describe('MemoryStore', () => {
 /* Testing Harness */
 
 type MiddlewareArgs = AnyMiddlewareArgs & {
-  next: NextMiddleware,
+  next: NextFn,
   context: Context,
   logger: Logger,
   client: WebClient,

src/conversation-store.ts

@@ -73,7 +73,8 @@ export function conversationContext<ConversationState = any>(
         .catch(error => logger.debug(error.message));
     } else {
       logger.debug('No conversation ID for incoming event');
-      await next();
+      // TODO: remove the non-null assertion operator
+      await next!();
     }
   };
 }

src/errors.ts

@@ -12,6 +12,8 @@ export enum ErrorCode {
   ReceiverAckTwiceError = 'slack_bolt_receiver_ack_twice_error',
   ReceiverAuthenticityError = 'slack_bolt_receiver_authenticity_error',
 
+  MultipleListenerError = 'slack_bolt_multiple_listener_error',
+
   /**
    * This value is used to assign to errors that occur inside the framework but do not have a code, to keep interfaces
    * in terms of CodedError.
@@ -68,6 +70,17 @@ export class ReceiverAuthenticityError extends Error implements CodedError {
   public code = ErrorCode.ReceiverAuthenticityError;
 }
 
+export class MultipleListenerError extends Error implements CodedError {
+  public code = ErrorCode.MultipleListenerError;
+  public originals: Error[];
+
+  constructor(originals: Error[]) {
+    super('Multiple errors occurred while handling several listeners. The `originals` property contains an array of each error.');
+
+    this.originals = originals;
+  }
+}
+
 export class UnknownError extends Error implements CodedError {
   public code = ErrorCode.UnknownError;
   public original: Error;

src/middleware/builtin.spec.ts

@@ -7,7 +7,7 @@ import { Override, wrapToResolveOnFirstCall, createFakeLogger } from '../test-he
 import rewiremock from 'rewiremock';
 import {
   SlackEventMiddlewareArgs,
-  NextMiddleware,
+  NextFn,
   Context,
   MessageEvent,
   SlackCommandMiddlewareArgs,
@@ -468,7 +468,7 @@ describe('matchCommandName', () => {
   const logger = createFakeLogger();
   const client = new WebClient(undefined, { logger, slackApiUrl: undefined });
 
-  function buildArgs(fakeNext: NextMiddleware): SlackCommandMiddlewareArgs & MiddlewareCommonArgs {
+  function buildArgs(fakeNext: NextFn): SlackCommandMiddlewareArgs & MiddlewareCommonArgs {
     const payload: SlashCommand = { ...validCommandPayload };
     return {
       payload,
@@ -652,7 +652,7 @@ interface DummyContext {
 }
 
 interface MiddlewareCommonArgs {
-  next: NextMiddleware;
+  next: NextFn;
   context: Context;
   logger: Logger;
   client: WebClient;