[next] Concurrency (#2136)

* feat: added concurrency options

* fix: Default CONCURRENCY to 2 for development

src/core/server/config.ts

@@ -1,5 +1,6 @@
 import convict from "convict";
 import Joi from "joi";
+import os from "os";
 
 // Add custom format for the mongo uri scheme.
 convict.addFormat({
@@ -55,6 +56,13 @@ const config = convict({
     env: "ENABLE_GRAPHIQL",
     arg: "enableGraphiQL",
   },
+  concurrency: {
+    doc: "The number of worker nodes to spawn to handle traffic",
+    format: Number,
+    default: os.cpus().length,
+    env: "CONCURRENCY",
+    arg: "concurrency",
+  },
   port: {
     doc: "The port to bind.",
     format: "port",

src/core/server/index.ts

@@ -1,18 +1,22 @@
 import express, { Express } from "express";
 import http from "http";
+import { Db } from "mongodb";
 
+import {
+  attachSubscriptionHandlers,
+  createApp,
+  listenAndServe,
+} from "talk-server/app";
+import config, { Config } from "talk-server/config";
 import getManagementSchema from "talk-server/graph/management/schema";
 import { Schemas } from "talk-server/graph/schemas";
 import getTenantSchema from "talk-server/graph/tenant/schema";
-import { createQueue } from "talk-server/queue";
-import TenantCache from "talk-server/services/tenant/cache";
-
+import logger from "talk-server/logger";
+import { createQueue, TaskQueue } from "talk-server/queue";
 import { createJWTSigningConfig } from "talk-server/services/jwt";
-import { attachSubscriptionHandlers, createApp, listenAndServe } from "./app";
-import config, { Config } from "./config";
-import logger from "./logger";
-import { createMongoDB } from "./services/mongodb";
-import { createRedisClient } from "./services/redis";
+import { createMongoDB } from "talk-server/services/mongodb";
+import { AugmentedRedis, createRedisClient } from "talk-server/services/redis";
+import TenantCache from "talk-server/services/tenant/cache";
 
 export interface ServerOptions {
   config?: Config;
@@ -36,6 +40,24 @@ class Server {
   // the requested port.
   public httpServer: http.Server;
 
+  // queue stores a reference to the queues that can process operations.
+  private queue: TaskQueue;
+
+  // redis stores the redis connection used by the application.
+  private redis: AugmentedRedis;
+
+  // mongo stores the mongo connection used by the application.
+  private mongo: Db;
+
+  // tenantCache stores the tenant cache used by the application.
+  private tenantCache: TenantCache;
+
+  // connected when true, indicates that `connect()` was already called.
+  private connected: boolean = false;
+
+  // processing when true, indicates that `process()` was already called.
+  private processing: boolean = false;
+
   constructor(options: ServerOptions) {
     this.parentApp = express();
 
@@ -52,6 +74,51 @@ class Server {
     };
   }
 
+  public async connect() {
+    // Guard against double connecting.
+    if (this.connected) {
+      throw new Error("server has already connected");
+    }
+    this.connected = true;
+
+    // Setup MongoDB.
+    this.mongo = await createMongoDB(config);
+
+    // Setup Redis.
+    this.redis = createRedisClient(config);
+
+    // Create the TenantCache.
+    this.tenantCache = new TenantCache(
+      this.mongo,
+      await createRedisClient(this.config),
+      config
+    );
+
+    // Prime the tenant cache so it'll be ready to serve now.
+    await this.tenantCache.primeAll();
+
+    // Create the Job Queue.
+    this.queue = createQueue({
+      config: this.config,
+      mongo: this.mongo,
+      tenantCache: this.tenantCache,
+    });
+  }
+
+  /**
+   * process will start the job processors.
+   */
+  public async process() {
+    // Guard against double connecting.
+    if (this.processing) {
+      throw new Error("server has already processing");
+    }
+    this.processing = true;
+
+    this.queue.mailer.process();
+    this.queue.scraper.process();
+  }
+
   /**
    * start orchestrates the application by starting it and returning a promise
    * when the server has started.
@@ -59,43 +126,29 @@ class Server {
    * @param parent the optional express application to bind the server to.
    */
   public async start(parent?: Express) {
+    // Guard against not being connected.
+    if (!this.connected) {
+      throw new Error("server has not connected yet");
+    }
+
     const port = this.config.get("port");
 
     // Ensure we have an app to bind to.
     parent = parent ? parent : this.parentApp;
 
-    // Setup MongoDB.
-    const mongo = await createMongoDB(config);
-
-    // Setup Redis.
-    const redis = createRedisClient(config);
-
     // Create the signing config.
     const signingConfig = createJWTSigningConfig(this.config);
 
-    // Create the TenantCache.
-    const tenantCache = new TenantCache(
-      mongo,
-      await createRedisClient(config),
-      config
-    );
-
-    // Prime the tenant cache so it'll be ready to serve now.
-    await tenantCache.primeAll();
-
-    // Create the Job Queue.
-    const queue = createQueue({ config, mongo, tenantCache });
-
     // Create the Talk App, branching off from the parent app.
     const app: Express = await createApp({
       parent,
-      queue,
-      mongo,
-      redis,
+      mongo: this.mongo,
+      redis: this.redis,
+      signingConfig,
+      tenantCache: this.tenantCache,
+      queue: this.queue,
       config: this.config,
       schemas: this.schemas,
-      signingConfig,
-      tenantCache,
     });
 
     // Start the application and store the resulting http.Server.

src/core/server/logger.ts

@@ -1,5 +1,6 @@
 import bunyan, { LogLevelString, stdSerializers as serializers } from "bunyan";
 import PrettyStream from "bunyan-prettystream";
+import cluster from "cluster";
 
 import config from "talk-server/config";
 
@@ -19,6 +20,9 @@ function getStreams() {
 const logger = bunyan.createLogger({
   name: "talk",
 
+  // Attach the cluster node information to the log entries.
+  clusterNode: cluster.worker ? `worker.${cluster.worker.id}` : "master",
+
   // Include file references in log entries.
   src: true,
   serializers,

src/core/server/queue/Task.ts

@@ -18,9 +18,6 @@ export default class Task<T, U = any> {
     this.queue = new Queue(options.jobName, options.queue);
     this.options = options;
     this.log = logger.child({ jobName: options.jobName });
-
-    // Sets up and attaches the job processor to the queue.
-    this.setupAndAttachProcessor();
   }
 
   /**
@@ -39,7 +36,12 @@ export default class Task<T, U = any> {
     this.log.trace({ jobID: job.id }, "added job to queue");
     return job;
   }
-  private setupAndAttachProcessor() {
+
+  /**
+   * process will connect the queue to the processor so that it may process the
+   * job requests.
+   */
+  public process() {
     this.queue.process(async (job: Job<T>) => {
       const log = this.log.child({ jobID: job.id });
 

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

@@ -193,6 +193,13 @@ export class Mailer {
       },
     });
   }
+
+  /**
+   * process maps the interface to the task process function.
+   */
+  public process() {
+    return this.task.process();
+  }
 }
 
 export const createMailerTask = (