initial subscription support

src/core/server/app/index.ts

@@ -1,99 +1,34 @@
-import express, { Express, Router } from 'express';
+import { Express } from 'express';
 import { Db } from 'mongodb';
 import http from 'http';
+import { Redis } from 'ioredis';
 
 import { Config } from 'talk-server/config';
-import { create } from 'talk-server/services/mongodb';
-import tenantGraphMiddleware from 'talk-server/graph/tenant/middleware';
-import managementGraphMiddleware from 'talk-server/graph/management/middleware';
+import { Schemas } from 'talk-server/graph/schemas';
+import { handleSubscriptions } from 'talk-server/graph/common/subscriptions/middleware';
 
+import { createRouter } from './router';
 import serveStatic from './middleware/serveStatic';
-import playground from './middleware/playground';
 import {
     access as accessLogger,
     error as errorLogger,
 } from './middleware/logging';
-import { Redis } from 'ioredis';
 
 export interface AppOptions {
+    parent: Express;
     config: Config;
     mongo: Db;
     redis: Redis;
-}
-
-async function createManagementRouter(opts: AppOptions): Promise<Router> {
-    const router = express.Router();
-
-    if (opts.config.get('env') === 'development') {
-        // GraphiQL
-        router.get(
-            '/graphiql',
-            playground({
-                endpoint: '/api/management/graphql',
-            })
-        );
-    }
-
-    // Management API
-    router.use(
-        '/graphql',
-        express.json(),
-        managementGraphMiddleware(opts.mongo)
-    );
-
-    return router;
-}
-
-async function createTenantRouter(opts: AppOptions): Promise<Router> {
-    const router = express.Router();
-
-    if (opts.config.get('env') === 'development') {
-        // GraphiQL
-        router.get(
-            '/graphiql',
-            playground({
-                endpoint: '/api/tenant/graphql',
-            })
-        );
-    }
-
-    // Tenant API
-    router.use('/graphql', express.json(), tenantGraphMiddleware(opts.mongo));
-
-    return router;
-}
-
-async function createAPIRouter(opts: AppOptions): Promise<Router> {
-    // Create a router.
-    const router = express.Router();
-
-    // Configure the tenant routes.
-    router.use('/tenant', await createTenantRouter(opts));
-
-    // Configure the management routes.
-    router.use('/management', await createManagementRouter(opts));
-
-    return router;
-}
-
-async function createRouter(opts: AppOptions): Promise<Router> {
-    // Create a router.
-    const router = express.Router();
-
-    router.use('/api', await createAPIRouter(opts));
-
-    return router;
+    schemas: Schemas;
 }
 
 /**
  * createApp will create a Talk Express app that can be used to handle requests.
- *
- * @param parent the root application to attach the Talk routes/middleware to.
  */
-export async function createApp(
-    parent: Express,
-    options: AppOptions
-): Promise<Express> {
+export async function createApp(options: AppOptions): Promise<Express> {
+    // Pull the parent out of the options.
+    const { parent } = options;
+
     // Logging
     parent.use(accessLogger);
 
@@ -110,13 +45,40 @@ export async function createApp(
 }
 
 /**
- * startApp will start the given express application.
+ * listenAndServe will start the given express application.
  *
- * @param port the port to listen on
  * @param app the express application to start
+ * @param port the port to listen on
  */
-export const startApp = (port: number, app: Express): Promise<http.Server> =>
+export const listenAndServe = (
+    app: Express,
+    port: number
+): Promise<http.Server> =>
     new Promise(async resolve => {
         // Listen on the designated port.
         const httpServer = app.listen(port, () => resolve(httpServer));
     });
+
+/**
+ * attachSubscriptionHandlers attaches all the handlers to the http.Server to
+ * handle websocket traffic by upgrading their http connections to websocket.
+ *
+ * @param schemas schemas for every schema this application handles
+ * @param server the http.Server to attach the websocket upgraders to
+ */
+export async function attachSubscriptionHandlers(
+    schemas: Schemas,
+    server: http.Server
+) {
+    // Setup the Management Subscription endpoint.
+    handleSubscriptions(server, {
+        schema: schemas.management,
+        path: '/api/management/live',
+    });
+
+    // Setup the Tenant Subscription endpoint.
+    handleSubscriptions(server, {
+        schema: schemas.tenant,
+        path: '/api/tenant/live',
+    });
+}

src/core/server/app/router.ts

@@ -0,0 +1,85 @@
+import express, { Router } from 'express';
+
+import tenantGraphMiddleware from 'talk-server/graph/tenant/middleware';
+import managementGraphMiddleware from 'talk-server/graph/management/middleware';
+
+import { AppOptions } from './index';
+import playground from './middleware/playground';
+
+async function createManagementRouter(opts: AppOptions) {
+    const router = express.Router();
+
+    if (opts.config.get('env') === 'development') {
+        // GraphiQL
+        router.get(
+            '/graphiql',
+            playground({
+                endpoint: '/api/management/graphql',
+                subscriptionEndpoint: '/api/management/live',
+            })
+        );
+    }
+
+    // Management API
+    router.use(
+        '/graphql',
+        express.json(),
+        await managementGraphMiddleware(
+            opts.schemas.management,
+            opts.config,
+            opts.mongo
+        )
+    );
+
+    return router;
+}
+
+async function createTenantRouter(opts: AppOptions) {
+    const router = express.Router();
+
+    if (opts.config.get('env') === 'development') {
+        // GraphiQL
+        router.get(
+            '/graphiql',
+            playground({
+                endpoint: '/api/tenant/graphql',
+                subscriptionEndpoint: '/api/tenant/live',
+            })
+        );
+    }
+
+    // Tenant API
+    router.use(
+        '/graphql',
+        express.json(),
+        await tenantGraphMiddleware(
+            opts.schemas.tenant,
+            opts.config,
+            opts.mongo
+        )
+    );
+
+    return router;
+}
+
+async function createAPIRouter(opts: AppOptions) {
+    // Create a router.
+    const router = express.Router();
+
+    // Configure the tenant routes.
+    router.use('/tenant', await createTenantRouter(opts));
+
+    // Configure the management routes.
+    router.use('/management', await createManagementRouter(opts));
+
+    return router;
+}
+
+export async function createRouter(opts: AppOptions) {
+    // Create a router.
+    const router = express.Router();
+
+    router.use('/api', await createAPIRouter(opts));
+
+    return router;
+}

src/core/server/config.ts

@@ -32,7 +32,7 @@ convict.addFormat({
     },
 });
 
-export const config = convict({
+const config = convict({
     env: {
         doc: 'The application environment.',
         format: ['production', 'development', 'test'],
@@ -67,6 +67,13 @@ export const config = convict({
         env: 'SECRET',
         arg: 'secret',
     },
+    logging_level: {
+        doc: 'The logging level to print to the console',
+        format: ['fatal', 'error', 'warn', 'info', 'debug', 'trace'],
+        default: 'info',
+        env: 'LOGGING_LEVEL',
+        arg: 'logging',
+    },
 });
 
 export type Config = typeof config;

src/core/server/graph/common/middleware/index.ts

@@ -0,0 +1,52 @@
+import {
+    graphqlExpress,
+    ExpressGraphQLOptionsFunction,
+    GraphQLOptions,
+} from 'apollo-server-express';
+import { GraphQLError, FieldDefinitionNode, ValidationContext } from 'graphql';
+import { resolveGraphqlOptions } from 'apollo-server-core';
+import { Config } from 'talk-server/config';
+
+// Sourced from: https://github.com/apollographql/apollo-server/blob/958846887598491fadea57b3f9373d129300f250/packages/apollo-server-core/src/ApolloServer.ts#L46-L57
+const NoIntrospection = (context: ValidationContext) => ({
+    Field(node: FieldDefinitionNode) {
+        if (node.name.value === '__schema' || node.name.value === '__type') {
+            context.reportError(
+                new GraphQLError(
+                    'GraphQL introspection is not allowed in production, but the query contained __schema or __type.',
+                    [node]
+                )
+            );
+        }
+    },
+});
+
+export const graphqlMiddleware = (
+    config: Config,
+    baseOptions: GraphQLOptions | ExpressGraphQLOptionsFunction
+) => {
+    // Generate the validation rules.
+    const validationRules: Array<(context: ValidationContext) => any> = [];
+
+    if (config.get('env') !== 'production') {
+        // Disable introspection in production.
+        validationRules.push(NoIntrospection);
+    }
+
+    // Generate the actual middleware.
+    return graphqlExpress(async (req, res) => {
+        // Resolve the base options.
+        const requestOptions = await resolveGraphqlOptions(
+            baseOptions,
+            req,
+            res
+        );
+
+        // Apply the validators, sourced from: https://github.com/apollographql/apollo-server/blob/958846887598491fadea57b3f9373d129300f250/packages/apollo-server-core/src/ApolloServer.ts#L104-L107
+        requestOptions.validationRules = requestOptions.validationRules
+            ? requestOptions.validationRules.concat(validationRules)
+            : validationRules;
+
+        return requestOptions;
+    });
+};

src/core/server/graph/common/schema/index.ts

@@ -0,0 +1,15 @@
+import { addResolveFunctionsToSchema, IResolvers } from 'graphql-tools';
+import { getGraphQLProjectConfig } from 'graphql-config';
+
+export default function loadSchema(projectName: string, resolvers: IResolvers) {
+    // Load the configuration from the provided `.graphqlconfig` file.
+    const config = getGraphQLProjectConfig(__dirname, projectName);
+
+    // Get the GraphQLSchema from the configuration.
+    const schema = config.getSchema();
+
+    // Attach the resolvers to the schema.
+    addResolveFunctionsToSchema({ schema, resolvers });
+
+    return schema;
+}

src/core/server/graph/common/subscriptions/middleware.ts

@@ -0,0 +1,29 @@
+import http from 'http';
+import { SubscriptionServer } from 'subscriptions-transport-ws';
+import { GraphQLSchema, execute, subscribe } from 'graphql';
+
+export interface SubscriptionMiddlewareOptions {
+    schema: GraphQLSchema;
+    path: string;
+}
+
+export function handleSubscriptions(
+    server: http.Server,
+    { schema, path }: SubscriptionMiddlewareOptions
+): SubscriptionServer {
+    // Configure some options for the subscription system.
+    const options = {
+        schema,
+        execute,
+        subscribe,
+    };
+
+    // Configure the socket options for the websocket server. It needs to handle
+    // upgrade requests on that route.
+    const socketOption = {
+        server,
+        path,
+    };
+
+    return new SubscriptionServer(options, socketOption);
+}

src/core/server/graph/common/subscriptions/pubsub.ts

@@ -0,0 +1,15 @@
+import { RedisPubSub } from 'graphql-redis-subscriptions';
+import { createRedisClient } from 'talk-server/services/redis';
+import { Config } from 'talk-server/config';
+
+export async function createPubSub(config: Config): Promise<RedisPubSub> {
+    // Create the Redis clients for the PubSub server.
+    const publisher = await createRedisClient(config);
+    const subscriber = await createRedisClient(config);
+
+    // Create the new PubSub manager.
+    return new RedisPubSub({
+        publisher,
+        subscriber,
+    });
+}

src/core/server/graph/management/middleware.ts

@@ -1,12 +1,13 @@
-import { graphqlExpress } from 'apollo-server-express';
-import schema from './schema';
-import Context from './context';
 import { Db } from 'mongodb';
+import { GraphQLSchema } from 'graphql';
 
-export default (db: Db) =>
-    graphqlExpress(async req => {
-        return {
-            schema,
-            context: new Context({ db }),
-        };
-    });
+import { graphqlMiddleware } from 'talk-server/graph/common/middleware';
+import { Config } from 'talk-server/config';
+
+import Context from './context';
+
+export default (schema: GraphQLSchema, config: Config, db: Db) =>
+    graphqlMiddleware(config, async () => ({
+        schema,
+        context: new Context({ db }),
+    }));

src/core/server/graph/management/schema/index.ts

@@ -1,15 +1,6 @@
-import { addResolveFunctionsToSchema } from 'graphql-tools';
-import { getGraphQLProjectConfig } from 'graphql-config';
+import loadSchema from 'talk-server/graph/common/schema';
+import resolvers from 'talk-server/graph/management/resolvers';
 
-import resolvers from '../resolvers';
-
-// Load the configuration from the provided `.graphqlconfig` file.
-const config = getGraphQLProjectConfig(__dirname, 'management');
-
-// Get the GraphQLSchema from the configuration.
-const schema = config.getSchema();
-
-// Attach the resolvers to the schema.
-addResolveFunctionsToSchema({ schema, resolvers });
-
-export default schema;
+export default function getManagementSchema() {
+    return loadSchema('management', resolvers);
+}

src/core/server/graph/schemas.ts

@@ -0,0 +1,6 @@
+import { GraphQLSchema } from 'graphql';
+
+export interface Schemas {
+    management: GraphQLSchema;
+    tenant: GraphQLSchema;
+}

src/core/server/graph/tenant/middleware.ts

@@ -1,13 +1,24 @@
-import { graphqlExpress } from 'apollo-server-express';
-import schema from './schema';
-import TenantContext from './context';
 import { Db } from 'mongodb';
-import { Tenant } from 'talk-server/models/tenant';
+import { GraphQLSchema } from 'graphql';
+
+import { retrieveByDomain } from 'talk-server/models/tenant';
+import { createPubSub } from 'talk-server/graph/common/subscriptions/pubsub';
+import { Config } from 'talk-server/config';
+import { graphqlMiddleware } from 'talk-server/graph/common/middleware';
+
+import TenantContext from './context';
+
+export default async (schema: GraphQLSchema, config: Config, db: Db) => {
+    // Configure the PubSub broker.
+    const pubsub = await createPubSub(config);
+
+    return graphqlMiddleware(config, async req => {
+        // TODO: replace with shared synced cache instead of direct db access.
+        const tenant = await retrieveByDomain(db, req.hostname);
 
-export default (db: Db) =>
-    graphqlExpress(async req => {
         return {
             schema,
-            context: new TenantContext({ db, tenant: { id: '1' } as Tenant }),
+            context: new TenantContext({ db, tenant }),
         };
     });
+};

src/core/server/graph/tenant/resolvers/query.ts

@@ -1,12 +1,13 @@
-import Context from 'talk-server/graph/tenant/context';
+import TenantContext from 'talk-server/graph/tenant/context';
 import { Asset } from 'talk-server/models/asset';
 
 export default {
     asset: async (
         _: any,
         { id, url }: { id?: string; url: string },
-        ctx: Context
+        ctx: TenantContext
     ): Promise<Asset> => {
         return ctx.loaders.Assets.asset.load(id);
     },
+    settings: async (parent: any, args: any, ctx: TenantContext) => ctx.tenant,
 };

src/core/server/graph/tenant/schema/index.ts

@@ -1,15 +1,6 @@
-import { addResolveFunctionsToSchema } from 'graphql-tools';
-import { getGraphQLProjectConfig } from 'graphql-config';
+import loadSchema from 'talk-server/graph/common/schema';
+import resolvers from 'talk-server/graph/tenant/resolvers';
 
-import resolvers from '../resolvers';
-
-// Load the configuration from the provided `.graphqlconfig` file.
-const config = getGraphQLProjectConfig(__dirname, 'tenant');
-
-// Get the GraphQLSchema from the configuration.
-const schema = config.getSchema();
-
-// Attach the resolvers to the schema.
-addResolveFunctionsToSchema({ schema, resolvers });
-
-export default schema;
+export default function getTenantSchema() {
+    return loadSchema('tenant', resolvers);
+}

src/core/server/index.ts

@@ -1,10 +1,14 @@
-import config, { Config } from './config';
 import express, { Express } from 'express';
 import http from 'http';
-import { createApp, startApp } from './app';
+
+import config, { Config } from './config';
+import { createApp, listenAndServe, attachSubscriptionHandlers } from './app';
 import logger from './logger';
-import { create as createMongoDB } from './services/mongodb';
-import { create as createRedis } from 'talk-server/services/redis';
+import { createMongoDB } from './services/mongodb';
+import { createRedisClient } from './services/redis';
+import getManagementSchema from 'talk-server/graph/management/schema';
+import getTenantSchema from 'talk-server/graph/tenant/schema';
+import { Schemas } from 'talk-server/graph/schemas';
 
 export interface ServerOptions {
     config?: Config;
@@ -17,6 +21,10 @@ class Server {
     // parentApp is the root application that the server will bind to.
     private parentApp: Express;
 
+    // schemas are the set of GraphQLSchema objects for each schema used by the
+    // server.
+    private schemas: Schemas;
+
     // config exposes application specific configuration.
     public config: Config;
 
@@ -29,6 +37,12 @@ class Server {
         this.config = config
             .load(options.config || {})
             .validate({ allowed: 'strict' });
+
+        // Load the graph schemas.
+        this.schemas = {
+            management: getManagementSchema(),
+            tenant: getTenantSchema(),
+        };
     }
 
     /**
@@ -47,17 +61,22 @@ class Server {
         const mongo = await createMongoDB(config);
 
         // Setup Redis.
-        const redis = await createRedis(config);
+        const redis = await createRedisClient(config);
 
         // Create the Talk App, branching off from the parent app.
-        const app = await createApp(parent, {
+        const app: Express = await createApp({
+            parent,
             mongo,
             redis,
             config: this.config,
+            schemas: this.schemas,
         });
 
-        // Start the application.
-        this.httpServer = await startApp(port, app);
+        // Start the application and store the resulting http.Server.
+        this.httpServer = await listenAndServe(app, port);
+
+        // Setup the websocket servers on the new http.Server.
+        attachSubscriptionHandlers(this.schemas, this.httpServer);
 
         logger.info({ port }, 'now listening');
     }

src/core/server/services/mongodb/index.ts

@@ -6,7 +6,7 @@ import { Config } from 'talk-server/config';
  *
  * @param config application configuration.
  */
-export async function create(config: Config): Promise<Db> {
+export async function createMongoDB(config: Config): Promise<Db> {
     // Connect and create a client for MongoDB.
     const client = await MongoClient.connect(config.get('mongodb'));
 

src/core/server/services/redis/index.ts

@@ -6,6 +6,6 @@ import { Config } from 'talk-server/config';
  *
  * @param config application configuration.
  */
-export async function create(config: Config): Promise<Redis> {
+export async function createRedisClient(config: Config): Promise<Redis> {
     return new RedisClient(config.get('redis'), {});
 }