[CORL-166] Live Updates on Mod Queues (#2368)

* feat: client implementation of subscriptions and modqueue live counts

* fix: unit tests

* feat: live status update in moderation

* feat: live update of new comments in moderation

* chore: View New instead of View More

* feat: fade in transition for new comments

* chore: turn websocket proxy back on

* feat: initial server impl

* fix: make it work :-)

* fix: add box shadow

* chore: make test subscriptions only support 1 top level field following the spec

* fix: linting

* feat: support clientID

* fix: linting

* feat: support commentStatusUpdated subscription

* fix: disabled styles for approve and reject button

* feat: show moderated by system and update flags

* feat: support metrics recording on websocket connections

* fix: handle when same comment enters but leaves again

src/core/server/graph/common/context.ts

@@ -1,3 +1,4 @@
+import { Db } from "mongodb";
 import uuid from "uuid";
 
 import { LanguageCode } from "coral-common/helpers/i18n/locales";
@@ -5,7 +6,9 @@ import { Config } from "coral-server/config";
 import logger, { Logger } from "coral-server/logger";
 import { User } from "coral-server/models/user";
 import { I18n } from "coral-server/services/i18n";
+import { AugmentedRedis } from "coral-server/services/redis";
 import { Request } from "coral-server/types/express";
+import { RedisPubSub } from "graphql-redis-subscriptions";
 
 export interface CommonContextOptions {
   id?: string;
@@ -14,8 +17,12 @@ export interface CommonContextOptions {
   req?: Request;
   logger?: Logger;
   lang?: LanguageCode;
+  disableCaching?: boolean;
   config: Config;
   i18n: I18n;
+  pubsub: RedisPubSub;
+  mongo: Db;
+  redis: AugmentedRedis;
 }
 
 export default class CommonContext {
@@ -27,6 +34,10 @@ export default class CommonContext {
   public readonly lang: LanguageCode;
   public readonly now: Date;
   public readonly logger: Logger;
+  public readonly pubsub: RedisPubSub;
+  public readonly mongo: Db;
+  public readonly redis: AugmentedRedis;
+  public readonly disableCaching: boolean;
 
   constructor({
     id = uuid.v1(),
@@ -37,6 +48,10 @@ export default class CommonContext {
     config,
     i18n,
     lang = i18n.getDefaultLang(),
+    pubsub,
+    mongo,
+    redis,
+    disableCaching = false,
   }: CommonContextOptions) {
     this.id = id;
     this.logger = log.child({
@@ -49,5 +64,9 @@ export default class CommonContext {
     this.config = config;
     this.i18n = i18n;
     this.lang = lang;
+    this.pubsub = pubsub;
+    this.mongo = mongo;
+    this.redis = redis;
+    this.disableCaching = disableCaching;
   }
 }

src/core/server/graph/common/extensions/LoggerExtension.ts

@@ -1,4 +1,4 @@
-import { ExecutionArgs, GraphQLError } from "graphql";
+import { DocumentNode, ExecutionArgs, GraphQLError } from "graphql";
 import {
   EndHandler,
   GraphQLExtension,
@@ -13,6 +13,20 @@ export function logError(ctx: CommonContext, err: GraphQLError) {
   ctx.logger.error({ err }, "graphql query error");
 }
 
+export function logQuery(
+  ctx: CommonContext,
+  document: DocumentNode,
+  responseTime?: number
+) {
+  ctx.logger.debug(
+    {
+      responseTime,
+      ...getOperationMetadata(document),
+    },
+    "graphql query"
+  );
+}
+
 export class LoggerExtension implements GraphQLExtension<CommonContext> {
   public executionDidStart(o: {
     executionArgs: ExecutionArgs;
@@ -27,12 +41,10 @@ export class LoggerExtension implements GraphQLExtension<CommonContext> {
         const responseTime = Math.round(now() - startTime);
 
         // Log out the details of the request.
-        o.executionArgs.contextValue.logger.debug(
-          {
-            responseTime,
-            ...getOperationMetadata(o.executionArgs.document),
-          },
-          "graphql query"
+        logQuery(
+          o.executionArgs.contextValue,
+          o.executionArgs.document,
+          responseTime
         );
       };
     }

src/core/server/graph/common/extensions/MetricsExtension.ts

@@ -1,22 +1,13 @@
+import CommonContext from "coral-server/graph/common/context";
+import { Metrics } from "coral-server/services/metrics";
 import { ExecutionArgs } from "graphql";
 import { EndHandler, GraphQLExtension } from "graphql-extensions";
 import now from "performance-now";
-import { Counter, Histogram } from "prom-client";
 
-import CommonContext from "coral-server/graph/common/context";
 import { getOperationMetadata } from "./helpers";
 
-export interface MetricsExtensionOptions {
-  executedGraphQueriesTotalCounter: Counter;
-  graphQLExecutionTimingsHistogram: Histogram;
-}
-
 export class MetricsExtension implements GraphQLExtension<CommonContext> {
-  private options: MetricsExtensionOptions;
-
-  constructor(options: MetricsExtensionOptions) {
-    this.options = options;
-  }
+  constructor(private metrics: Metrics) {}
 
   public executionDidStart(o: {
     executionArgs: ExecutionArgs;
@@ -37,11 +28,11 @@ export class MetricsExtension implements GraphQLExtension<CommonContext> {
 
         if (operation && operationName) {
           // Increment the graph query value, tagging with the name of the query.
-          this.options.executedGraphQueriesTotalCounter
+          this.metrics.executedGraphQueriesTotalCounter
             .labels(operation, operationName)
             .inc();
 
-          this.options.graphQLExecutionTimingsHistogram
+          this.metrics.graphQLExecutionTimingsHistogram
             .labels(operation, operationName)
             .observe(responseTime);
         }

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

@@ -0,0 +1,12 @@
+import { RedisPubSub } from "graphql-redis-subscriptions";
+import { Redis } from "ioredis";
+
+export function createPubSubClient(
+  publisher: Redis,
+  subscriber: Redis
+): RedisPubSub {
+  return new RedisPubSub({
+    publisher,
+    subscriber,
+  });
+}

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

@@ -1,52 +1,56 @@
-import { Db } from "mongodb";
-
 import CommonContext, {
   CommonContextOptions,
 } from "coral-server/graph/common/context";
+import {
+  createPublisher,
+  Publisher,
+} from "coral-server/graph/tenant/subscriptions/publisher";
 import { Tenant } from "coral-server/models/tenant";
 import { User } from "coral-server/models/user";
 import { MailerQueue } from "coral-server/queue/tasks/mailer";
 import { ScraperQueue } from "coral-server/queue/tasks/scraper";
 import { JWTSigningConfig } from "coral-server/services/jwt";
-import { AugmentedRedis } from "coral-server/services/redis";
 import TenantCache from "coral-server/services/tenant/cache";
 
 import loaders from "./loaders";
 import mutators from "./mutators";
 
 export interface TenantContextOptions extends CommonContextOptions {
-  mongo: Db;
-  redis: AugmentedRedis;
   tenant: Tenant;
   tenantCache: TenantCache;
   mailerQueue: MailerQueue;
   scraperQueue: ScraperQueue;
   signingConfig?: JWTSigningConfig;
+  clientID?: string;
 }
 
 export default class TenantContext extends CommonContext {
   public readonly tenant: Tenant;
   public readonly tenantCache: TenantCache;
-  public readonly mongo: Db;
-  public readonly redis: AugmentedRedis;
+
   public readonly mailerQueue: MailerQueue;
   public readonly scraperQueue: ScraperQueue;
-  public readonly loaders: ReturnType<typeof loaders>;
-  public readonly mutators: ReturnType<typeof mutators>;
+  public readonly publisher: Publisher;
   public readonly user?: User;
   public readonly signingConfig?: JWTSigningConfig;
+  public readonly clientID?: string;
+  public readonly loaders: ReturnType<typeof loaders>;
+  public readonly mutators: ReturnType<typeof mutators>;
 
   constructor(options: TenantContextOptions) {
     super({ ...options, lang: options.tenant.locale });
 
     this.tenant = options.tenant;
     this.tenantCache = options.tenantCache;
-    this.user = options.user;
-    this.mongo = options.mongo;
-    this.redis = options.redis;
     this.scraperQueue = options.scraperQueue;
     this.mailerQueue = options.mailerQueue;
     this.signingConfig = options.signingConfig;
+    this.clientID = options.clientID;
+    this.publisher = createPublisher(
+      this.pubsub,
+      this.tenant.id,
+      this.clientID
+    );
     this.loaders = loaders(this);
     this.mutators = mutators(this);
   }

src/core/server/graph/tenant/loaders/Auth.ts

@@ -8,7 +8,13 @@ export default (ctx: TenantContext) => ({
   discoverOIDCConfiguration: new DataLoader<
     string,
     GQLDiscoveredOIDCConfiguration | null
-  >(issuers =>
-    Promise.all(issuers.map(issuer => discoverOIDCConfiguration(issuer)))
+  >(
+    issuers =>
+      Promise.all(issuers.map(issuer => discoverOIDCConfiguration(issuer))),
+    {
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cache: !ctx.disableCaching,
+    }
   ),
 });

src/core/server/graph/tenant/loaders/Comments.ts

@@ -50,10 +50,12 @@ const tagFilter = (tag?: GQLTAG): CommentConnectionInput["filter"] => {
 const primeCommentsFromConnection = (ctx: Context) => (
   connection: Readonly<Connection<Readonly<Comment>>>
 ) => {
-  // For each of the nodes, prime the comment loader.
-  connection.nodes.forEach(comment => {
-    ctx.loaders.Comments.comment.prime(comment.id, comment);
-  });
+  if (!ctx.disableCaching) {
+    // For each of the nodes, prime the comment loader.
+    connection.nodes.forEach(comment => {
+      ctx.loaders.Comments.comment.prime(comment.id, comment);
+    });
+  }
 
   return connection;
 };
@@ -96,10 +98,16 @@ const mapVisibleComments = (user?: Pick<User, "role">) => (
 ): Array<Readonly<Comment> | null> => comments.map(mapVisibleComment(user));
 
 export default (ctx: Context) => ({
-  comment: new DataLoader((ids: string[]) =>
-    retrieveManyComments(ctx.mongo, ctx.tenant.id, ids).then(
-      mapVisibleComments(ctx.user)
-    )
+  comment: new DataLoader(
+    (ids: string[]) =>
+      retrieveManyComments(ctx.mongo, ctx.tenant.id, ids).then(
+        mapVisibleComments(ctx.user)
+      ),
+    {
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cache: !ctx.disableCaching,
+    }
   ),
   forFilter: ({
     first = 10,
@@ -217,13 +225,19 @@ export default (ctx: Context) => ({
       // The cursor passed here is always going to be a number.
       before: before as number,
     }).then(primeCommentsFromConnection(ctx)),
-  sharedModerationQueueQueuesCounts: new SingletonResolver(() =>
-    retrieveSharedModerationQueueQueuesCounts(
-      ctx.mongo,
-      ctx.redis,
-      ctx.tenant.id,
-      ctx.now
-    )
+  sharedModerationQueueQueuesCounts: new SingletonResolver(
+    () =>
+      retrieveSharedModerationQueueQueuesCounts(
+        ctx.mongo,
+        ctx.redis,
+        ctx.tenant.id,
+        ctx.now
+      ),
+    {
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cacheable: !ctx.disableCaching,
+    }
   ),
   tagCounts: new DataLoader((storyIDs: string[]) =>
     retrieveStoryCommentTagCounts(ctx.mongo, ctx.tenant.id, storyIDs)

src/core/server/graph/tenant/loaders/Stories.ts

@@ -56,10 +56,12 @@ const queryFilter = (query?: string): StoryConnectionInput["filter"] => {
 const primeStoriesFromConnection = (ctx: TenantContext) => (
   connection: Readonly<Connection<Readonly<Story>>>
 ) => {
-  // For each of these nodes, prime the story loader.
-  connection.nodes.forEach(story => {
-    ctx.loaders.Stories.story.prime(story.id, story);
-  });
+  if (!ctx.disableCaching) {
+    // For each of these nodes, prime the story loader.
+    connection.nodes.forEach(story => {
+      ctx.loaders.Stories.story.prime(story.id, story);
+    });
+  }
 
   return connection;
 };
@@ -72,6 +74,9 @@ export default (ctx: TenantContext) => ({
     {
       // TODO: (wyattjoh) see if there's something we can do to improve the cache key
       cacheKeyFn: (input: FindOrCreateStory) => `${input.id}:${input.url}`,
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cache: !ctx.disableCaching,
     }
   ),
   find: new DataLoader(
@@ -81,10 +86,18 @@ export default (ctx: TenantContext) => ({
     {
       // TODO: (wyattjoh) see if there's something we can do to improve the cache key
       cacheKeyFn: (input: FindStory) => `${input.id}:${input.url}`,
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cache: !ctx.disableCaching,
     }
   ),
-  story: new DataLoader<string, Story | null>(ids =>
-    retrieveManyStories(ctx.mongo, ctx.tenant.id, ids)
+  story: new DataLoader<string, Story | null>(
+    ids => retrieveManyStories(ctx.mongo, ctx.tenant.id, ids),
+    {
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cache: !ctx.disableCaching,
+    }
   ),
   connection: ({ first = 10, after, status, query }: QueryToStoriesArgs) =>
     retrieveStoryConnection(ctx.mongo, ctx.tenant.id, {
@@ -99,6 +112,11 @@ export default (ctx: TenantContext) => ({
       },
     }).then(primeStoriesFromConnection(ctx)),
   debugScrapeMetadata: new DataLoader(
-    createManyBatchLoadFn((url: string) => scraper.scrape(url))
+    createManyBatchLoadFn((url: string) => scraper.scrape(url)),
+    {
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cache: !ctx.disableCaching,
+    }
   ),
 });

src/core/server/graph/tenant/loaders/Users.ts

@@ -82,20 +82,27 @@ const statusFilter = (
 const primeUsersFromConnection = (ctx: Context) => (
   connection: Readonly<Connection<Readonly<User>>>
 ) => {
-  // For each of the nodes, prime the user loader.
-  connection.nodes.forEach(user => {
-    ctx.loaders.Users.user.prime(user.id, user);
-  });
+  if (!ctx.disableCaching) {
+    // For each of the nodes, prime the user loader.
+    connection.nodes.forEach(user => {
+      ctx.loaders.Users.user.prime(user.id, user);
+    });
+  }
 
   return connection;
 };
 
 export default (ctx: Context) => {
-  const user = new DataLoader<string, User | null>(ids =>
-    retrieveManyUsers(ctx.mongo, ctx.tenant.id, ids)
+  const user = new DataLoader<string, User | null>(
+    ids => retrieveManyUsers(ctx.mongo, ctx.tenant.id, ids),
+    {
+      // Disable caching for the DataLoader if the Context is designed to be
+      // long lived.
+      cache: !ctx.disableCaching,
+    }
   );
 
-  if (ctx.user) {
+  if (ctx.user && !ctx.disableCaching) {
     // Prime the current logged in user in the dataloader cache.
     user.prime(ctx.user.id, ctx.user);
   }

src/core/server/graph/tenant/loaders/util.ts

@@ -1,15 +1,28 @@
+export interface SingletonResolverOptions {
+  cacheable?: boolean;
+}
+
 /**
  * SingletonResolver is a cached loader for a single result.
  */
 export class SingletonResolver<T> {
   private cache: Promise<T> | null = null;
   private resolver: () => Promise<T>;
+  private cacheable: boolean;
 
-  constructor(resolver: () => Promise<T>) {
+  constructor(
+    resolver: () => Promise<T>,
+    { cacheable = true }: SingletonResolverOptions = {}
+  ) {
     this.resolver = resolver;
+    this.cacheable = cacheable;
   }
 
   public load() {
+    if (!this.cacheable) {
+      return this.resolver();
+    }
+
     if (this.cache) {
       return this.cache;
     }

src/core/server/graph/tenant/mutators/Actions.ts

@@ -7,13 +7,13 @@ import {
 
 export const Actions = (ctx: TenantContext) => ({
   approveComment: (input: GQLApproveCommentInput) =>
-    approve(ctx.mongo, ctx.redis, ctx.tenant, {
+    approve(ctx.mongo, ctx.redis, ctx.publisher, ctx.tenant, {
       commentID: input.commentID,
       commentRevisionID: input.commentRevisionID,
       moderatorID: ctx.user!.id,
     }),
   rejectComment: (input: GQLRejectCommentInput) =>
-    reject(ctx.mongo, ctx.redis, ctx.tenant, {
+    reject(ctx.mongo, ctx.redis, ctx.publisher, ctx.tenant, {
       commentID: input.commentID,
       commentRevisionID: input.commentRevisionID,
       moderatorID: ctx.user!.id,

src/core/server/graph/tenant/mutators/Comments.ts

@@ -43,6 +43,7 @@ export const Comments = (ctx: TenantContext) => ({
       create(
         ctx.mongo,
         ctx.redis,
+        ctx.publisher,
         ctx.tenant,
         ctx.user!,
         { authorID: ctx.user!.id, ...comment },
@@ -64,6 +65,7 @@ export const Comments = (ctx: TenantContext) => ({
       edit(
         ctx.mongo,
         ctx.redis,
+        ctx.publisher,
         ctx.tenant,
         ctx.user!,
         {
@@ -87,6 +89,7 @@ export const Comments = (ctx: TenantContext) => ({
     createReaction(
       ctx.mongo,
       ctx.redis,
+      ctx.publisher,
       ctx.tenant,
       ctx.user!,
       {
@@ -107,6 +110,7 @@ export const Comments = (ctx: TenantContext) => ({
     createDontAgree(
       ctx.mongo,
       ctx.redis,
+      ctx.publisher,
       ctx.tenant,
       ctx.user!,
       {
@@ -133,6 +137,7 @@ export const Comments = (ctx: TenantContext) => ({
     createFlag(
       ctx.mongo,
       ctx.redis,
+      ctx.publisher,
       ctx.tenant,
       ctx.user!,
       {
@@ -161,7 +166,7 @@ export const Comments = (ctx: TenantContext) => ({
       ctx.now
     ).then(comment =>
       comment.status !== GQLCOMMENT_STATUS.APPROVED
-        ? approve(ctx.mongo, ctx.redis, ctx.tenant, {
+        ? approve(ctx.mongo, ctx.redis, ctx.publisher, ctx.tenant, {
             commentID,
             commentRevisionID,
             moderatorID: ctx.user!.id,

src/core/server/graph/tenant/resolvers/Subscription/commentEnteredModerationQueue.ts

@@ -0,0 +1,38 @@
+import {
+  GQLMODERATION_QUEUE,
+  SubscriptionToCommentEnteredModerationQueueResolver,
+} from "coral-server/graph/tenant/schema/__generated__/types";
+
+import { createIterator } from "./helpers";
+import { SUBSCRIPTION_CHANNELS, SubscriptionPayload } from "./types";
+
+export interface CommentEnteredModerationQueueInput
+  extends SubscriptionPayload {
+  queue: GQLMODERATION_QUEUE;
+  commentID: string;
+  storyID: string;
+}
+
+export const commentEnteredModerationQueue: SubscriptionToCommentEnteredModerationQueueResolver<
+  CommentEnteredModerationQueueInput
+> = createIterator(SUBSCRIPTION_CHANNELS.COMMENT_ENTERED_MODERATION_QUEUE, {
+  filter: (source, { storyID, queue }) => {
+    // If we're filtering by storyID, then only send back comments with the
+    // specific storyID.
+    if (storyID && source.storyID !== storyID) {
+      return false;
+    }
+
+    // If we're filtering by queue, then only send back comments from the
+    // specific queue.
+    if (queue && source.queue !== queue) {
+      return false;
+    }
+
+    return true;
+  },
+  resolve: ({ queue, commentID }, args, ctx) => ({
+    queue: () => queue,
+    comment: () => ctx.loaders.Comments.comment.load(commentID),
+  }),
+});

src/core/server/graph/tenant/resolvers/Subscription/commentLeftModerationQueue.ts

@@ -0,0 +1,37 @@
+import {
+  GQLMODERATION_QUEUE,
+  SubscriptionToCommentLeftModerationQueueResolver,
+} from "coral-server/graph/tenant/schema/__generated__/types";
+
+import { createIterator } from "./helpers";
+import { SUBSCRIPTION_CHANNELS, SubscriptionPayload } from "./types";
+
+export interface CommentLeftModerationQueueInput extends SubscriptionPayload {
+  queue: GQLMODERATION_QUEUE;
+  commentID: string;
+  storyID: string;
+}
+
+export const commentLeftModerationQueue: SubscriptionToCommentLeftModerationQueueResolver<
+  CommentLeftModerationQueueInput
+> = createIterator(SUBSCRIPTION_CHANNELS.COMMENT_LEFT_MODERATION_QUEUE, {
+  filter: (source, { storyID, queue }) => {
+    // If we're filtering by storyID, then only send back comments with the
+    // specific storyID.
+    if (storyID && source.storyID !== storyID) {
+      return false;
+    }
+
+    // If we're filtering by queue, then only send back comments from the
+    // specific queue.
+    if (queue && source.queue !== queue) {
+      return false;
+    }
+
+    return true;
+  },
+  resolve: ({ queue, commentID }, args, ctx) => ({
+    queue: () => queue,
+    comment: () => ctx.loaders.Comments.comment.load(commentID),
+  }),
+});

src/core/server/graph/tenant/resolvers/Subscription/commentStatusUpdated.ts

@@ -0,0 +1,40 @@
+import {
+  GQLCOMMENT_STATUS,
+  SubscriptionToCommentStatusUpdatedResolver,
+} from "coral-server/graph/tenant/schema/__generated__/types";
+
+import { createIterator } from "./helpers";
+import { SUBSCRIPTION_CHANNELS, SubscriptionPayload } from "./types";
+
+export interface CommentStatusUpdatedInput extends SubscriptionPayload {
+  newStatus: GQLCOMMENT_STATUS;
+  oldStatus: GQLCOMMENT_STATUS;
+  moderatorID: string | null;
+  commentID: string;
+}
+
+export const commentStatusUpdated: SubscriptionToCommentStatusUpdatedResolver<
+  CommentStatusUpdatedInput
+> = createIterator(SUBSCRIPTION_CHANNELS.COMMENT_STATUS_UPDATED, {
+  filter: (source, { id }) => {
+    // If we're filtering by id, then only send back updates for the specified
+    // comment.
+    if (id && source.commentID !== id) {
+      return false;
+    }
+
+    return true;
+  },
+  resolve: ({ newStatus, oldStatus, moderatorID, commentID }, args, ctx) => ({
+    newStatus: () => newStatus,
+    oldStatus: () => oldStatus,
+    moderator: () => {
+      if (moderatorID) {
+        return ctx.loaders.Users.user.load(moderatorID);
+      }
+
+      return null;
+    },
+    comment: () => ctx.loaders.Comments.comment.load(commentID),
+  }),
+});

src/core/server/graph/tenant/resolvers/Subscription/helpers.ts

@@ -0,0 +1,108 @@
+import { GraphQLResolveInfo } from "graphql";
+import { withFilter } from "graphql-subscriptions";
+
+import TenantContext from "../../context";
+import { SUBSCRIPTION_CHANNELS, SubscriptionPayload } from "./types";
+
+type FilterFn<TParent, TArgs, TContext> = (
+  parent: TParent,
+  args: TArgs,
+  context: TContext,
+  info: GraphQLResolveInfo
+) => boolean | Promise<boolean>;
+
+type Resolver<TParent, TArgs, TResult> = (
+  source: TParent,
+  args: TArgs,
+  ctx: TenantContext,
+  info: GraphQLResolveInfo
+) => TResult;
+
+interface SubscriptionResolver<TParent, TArgs, TResult> {
+  subscribe: Resolver<TParent, TArgs, AsyncIterator<TResult>>;
+  resolve?: Resolver<TParent, TArgs, TResult>;
+}
+
+export function createTenantAsyncIterator<TParent, TArgs, TResult>(
+  channel: SUBSCRIPTION_CHANNELS
+): Resolver<TParent, TArgs, AsyncIterator<TResult>> {
+  return (source, args, ctx) =>
+    ctx.pubsub.asyncIterator<TResult>(
+      createSubscriptionChannelName(ctx.tenant.id, channel)
+    );
+}
+
+export function createSubscriptionChannelName(
+  tenantID: string,
+  channel: SUBSCRIPTION_CHANNELS
+): string {
+  return `TENANT[${tenantID}][${channel}]`;
+}
+
+/**
+ * defaultFilterFn will perform filtering operations on the subscription
+ * responses to ensure that mutations issued by one user is not sent back as a
+ * subscription to the same requesting User, as they already implement the
+ * update via the mutation response.
+ *
+ * @param source the source for the document passed down, we don't actually need
+ *               it here.
+ * @param args the arguments for the specific subscription operation, we don't
+ *             actually need it here.
+ * @param ctx the context for the request, this contains the references we'll
+ *            need to determine eligibility to send the subscription back or
+ *            not.
+ */
+export function defaultFilterFn<TParent extends SubscriptionPayload, TArgs>(
+  source: TParent,
+  args: TArgs,
+  ctx: TenantContext
+): boolean {
+  if (source.clientID && ctx.clientID && source.clientID === ctx.clientID) {
+    return false;
+  }
+
+  return true;
+}
+
+/**
+ * Ensure that even when we're provided with a domain specific filtering
+ * function we respect the subscription id that is sent back with the request to
+ * prevent double responses.
+ */
+export function createFilterFn<TParent, TArgs>(
+  filter?: FilterFn<TParent, TArgs, TenantContext>
+): FilterFn<TParent, TArgs, TenantContext> {
+  return filter
+    ? // Combine the filters, preferring the defaultFilterFn first.
+      (source, args, ctx, info) => {
+        if (!defaultFilterFn(source, args, ctx)) {
+          return false;
+        }
+
+        return filter(source, args, ctx, info);
+      }
+    : defaultFilterFn;
+}
+
+export interface CreateIteratorInput<TParent, TArgs, TResult> {
+  filter?: FilterFn<TParent, TArgs, TenantContext>;
+  resolve?: Resolver<TParent, TArgs, TResult>;
+}
+
+export function createIterator<
+  TParent extends SubscriptionPayload,
+  TArgs,
+  TResult
+>(
+  channel: SUBSCRIPTION_CHANNELS,
+  { filter, resolve }: CreateIteratorInput<TParent, TArgs, TResult> = {}
+): SubscriptionResolver<TParent, TArgs, TResult> {
+  return {
+    subscribe: withFilter(
+      createTenantAsyncIterator(channel),
+      createFilterFn(filter)
+    ),
+    resolve,
+  };
+}

src/core/server/graph/tenant/resolvers/Subscription/index.ts

@@ -0,0 +1,11 @@
+import { GQLSubscriptionTypeResolver } from "coral-server/graph/tenant/schema/__generated__/types";
+
+import { commentEnteredModerationQueue } from "./commentEnteredModerationQueue";
+import { commentLeftModerationQueue } from "./commentLeftModerationQueue";
+import { commentStatusUpdated } from "./commentStatusUpdated";
+
+export const Subscription: GQLSubscriptionTypeResolver = {
+  commentEnteredModerationQueue,
+  commentLeftModerationQueue,
+  commentStatusUpdated,
+};

src/core/server/graph/tenant/resolvers/Subscription/types.ts

@@ -0,0 +1,35 @@
+import { CommentEnteredModerationQueueInput } from "./commentEnteredModerationQueue";
+import { CommentLeftModerationQueueInput } from "./commentLeftModerationQueue";
+import { CommentStatusUpdatedInput } from "./commentStatusUpdated";
+
+export enum SUBSCRIPTION_CHANNELS {
+  COMMENT_ENTERED_MODERATION_QUEUE = "COMMENT_ENTERED_MODERATION_QUEUE",
+  COMMENT_LEFT_MODERATION_QUEUE = "COMMENT_LEFT_MODERATION_QUEUE",
+  COMMENT_STATUS_UPDATED = "COMMENT_STATUS_UPDATED",
+}
+
+export interface SubscriptionPayload {
+  clientID?: string;
+}
+
+export interface SubscriptionType<
+  TChannel extends SUBSCRIPTION_CHANNELS,
+  TPayload extends SubscriptionPayload
+> {
+  channel: TChannel;
+  payload: TPayload;
+}
+
+export type SUBSCRIPTION_INPUT =
+  | SubscriptionType<
+      SUBSCRIPTION_CHANNELS.COMMENT_ENTERED_MODERATION_QUEUE,
+      CommentEnteredModerationQueueInput
+    >
+  | SubscriptionType<
+      SUBSCRIPTION_CHANNELS.COMMENT_LEFT_MODERATION_QUEUE,
+      CommentLeftModerationQueueInput
+    >
+  | SubscriptionType<
+      SUBSCRIPTION_CHANNELS.COMMENT_STATUS_UPDATED,
+      CommentStatusUpdatedInput
+    >;

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

@@ -25,6 +25,7 @@ import { Query } from "./Query";
 import { RejectCommentPayload } from "./RejectCommentPayload";
 import { Story } from "./Story";
 import { StorySettings } from "./StorySettings";
+import { Subscription } from "./Subscription";
 import { SuspensionStatus } from "./SuspensionStatus";
 import { SuspensionStatusHistory } from "./SuspensionStatusHistory";
 import { Tag } from "./Tag";
@@ -56,6 +57,7 @@ const Resolvers: GQLResolver = {
   RejectCommentPayload,
   Story,
   StorySettings,
+  Subscription,
   SuspensionStatus,
   SuspensionStatusHistory,
   Tag,

src/core/server/graph/tenant/schema/schema.graphql

@@ -4459,3 +4459,121 @@ type Mutation {
   removeUserIgnore(input: RemoveUserIgnoreInput!): RemoveUserIgnorePayload!
     @auth(permit: [SUSPENDED, BANNED])
 }
+
+##################
+## Subscriptions
+##################
+
+"""
+CommentStatusUpdatedPayload is returned when a Comment has it's status updated
+after it was created.
+"""
+type CommentStatusUpdatedPayload {
+  """
+  newStatus  is the new status assigned to the Comment. This status may not
+  match the status provided by `comment.status` due to race conditions in the
+  data loaders.
+  """
+  newStatus: COMMENT_STATUS!
+
+  """
+  oldStatus is the old status that was previously assigned to the Comment.
+  """
+  oldStatus: COMMENT_STATUS!
+
+  """
+  moderator is the User that updated the Comment's status. If null, then the
+  system assigned the new Comment status (for example, when a comment is edited
+  by the author, and now contains a banned word).
+  """
+  moderator: User
+
+  """
+  comment is the updated Comment after the status has been updated.
+  """
+  comment: Comment!
+}
+
+"""
+MODERATION_QUEUE references the specific ModerationQueue that a given Comment
+can be associated with.
+"""
+enum MODERATION_QUEUE {
+  """
+  UNMODERATED refers to the ModerationQueue for all Comments that have not been
+  moderated yet.
+  """
+  UNMODERATED
+
+  """
+  REPORTED refers to the ModerationQueue for all Comments that have been
+  published, have not been moderated by a human yet, and have been reported by
+  a User via a flag.
+  """
+  REPORTED
+
+  """
+  PENDING refers to the ModerationQueue for all Comments that were held back by
+  the system and require moderation in order to be published.
+  """
+  PENDING
+}
+
+"""
+CommentEnteredModerationQueuePayload is returned when a Comment enters a
+specific ModerationQueue.
+"""
+type CommentEnteredModerationQueuePayload {
+  """
+  queue refers to the specific ModerationQueue that a given Comment entered.
+  """
+  queue: MODERATION_QUEUE!
+
+  """
+  comment is the Comment that entered the ModerationQueue.
+  """
+  comment: Comment!
+}
+
+"""
+CommentLeftModerationQueuePayload is returned when a Comment leaves a specific
+ModerationQueue.
+"""
+type CommentLeftModerationQueuePayload {
+  """
+  queue refers to the specific ModerationQueue that a given Comment left.
+  """
+  queue: MODERATION_QUEUE!
+
+  """
+  comment is the Comment that left the ModerationQueue.
+  """
+  comment: Comment!
+}
+
+type Subscription {
+  """
+  commentEnteredModerationQueue returns when a Comment enters a ModerationQueue.
+  Note that a Comment may enter multiple moderation queues.
+  """
+  commentEnteredModerationQueue(
+    storyID: ID
+    queue: MODERATION_QUEUE
+  ): CommentEnteredModerationQueuePayload! @auth(roles: [MODERATOR, ADMIN])
+
+  """
+  commentLeftModerationQueue returns when a Comment leaves a ModerationQueue.
+  Note that a Comment may leave multiple moderation queues.
+  """
+  commentLeftModerationQueue(
+    storyID: ID
+    queue: MODERATION_QUEUE
+  ): CommentLeftModerationQueuePayload! @auth(roles: [MODERATOR, ADMIN])
+
+  """
+  commentStatusUpdated returns when a Comment has it's status changed after
+  being created.
+  """
+  commentStatusUpdated(id: ID): CommentStatusUpdatedPayload!
+    @auth(roles: [MODERATOR, ADMIN])
+}

src/core/server/graph/tenant/subscriptions/publisher.ts

@@ -0,0 +1,28 @@
+import { RedisPubSub } from "graphql-redis-subscriptions";
+
+import { createSubscriptionChannelName } from "coral-server/graph/tenant/resolvers/Subscription/helpers";
+import { SUBSCRIPTION_INPUT } from "coral-server/graph/tenant/resolvers/Subscription/types";
+import logger from "coral-server/logger";
+
+export type Publisher = (input: SUBSCRIPTION_INPUT) => Promise<void>;
+
+/**
+ * createPublisher will create a new Publisher that can be used to send events
+ * over the pubsub broker to facilitate live updates.
+ *
+ * @param pubsub the pubsub broker to be used to facilitate the publish action
+ * @param tenantID the ID of the Tenant where the event will be published with
+ * @param clientID the ID of the client to de-duplicate mutation responses
+ */
+export const createPublisher = (
+  pubsub: RedisPubSub,
+  tenantID: string,
+  clientID?: string
+): Publisher => async ({ channel, payload }) => {
+  logger.trace({ channel, tenantID, clientID }, "publishing event");
+
+  return pubsub.publish(createSubscriptionChannelName(tenantID, channel), {
+    ...payload,
+    clientID,
+  });
+};

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

@@ -0,0 +1,222 @@
+import {
+  execute,
+  ExecutionResult,
+  GraphQLSchema,
+  parse,
+  subscribe,
+} from "graphql";
+import http, { IncomingMessage } from "http";
+import {
+  ConnectionContext,
+  ExecutionParams,
+  OperationMessagePayload,
+  SubscriptionServer,
+} from "subscriptions-transport-ws";
+
+import { ACCESS_TOKEN_PARAM, CLIENT_ID_PARAM } from "coral-common/constants";
+import { Omit } from "coral-common/types";
+import { AppOptions } from "coral-server/app";
+import { getHostname } from "coral-server/app/helpers/hostname";
+import {
+  createVerifiers,
+  verifyAndRetrieveUser,
+} from "coral-server/app/middleware/passport/strategies/jwt";
+import {
+  CoralError,
+  InternalError,
+  TenantNotFoundError,
+} from "coral-server/errors";
+import {
+  enrichError,
+  logError,
+  logQuery,
+} from "coral-server/graph/common/extensions";
+import { getOperationMetadata } from "coral-server/graph/common/extensions/helpers";
+import logger from "coral-server/logger";
+import { extractTokenFromRequest } from "coral-server/services/jwt";
+
+import TenantContext, { TenantContextOptions } from "../context";
+
+type OnConnectFn = (
+  params: OperationMessagePayload,
+  socket: any,
+  context: ConnectionContext
+) => Promise<TenantContext>;
+
+export function extractTokenFromWSRequest(
+  connectionParams: OperationMessagePayload,
+  req: IncomingMessage
+): string | null {
+  // Try to grab the token from the connection params if available.
+  if (
+    typeof connectionParams[ACCESS_TOKEN_PARAM] === "string" &&
+    connectionParams[ACCESS_TOKEN_PARAM].length > 0
+  ) {
+    return connectionParams[ACCESS_TOKEN_PARAM];
+  }
+
+  // Try to get the access token from the request.
+  return extractTokenFromRequest(req);
+}
+
+export function extractClientID(connectionParams: OperationMessagePayload) {
+  if (
+    typeof connectionParams[CLIENT_ID_PARAM] === "string" &&
+    connectionParams[CLIENT_ID_PARAM].length > 0
+  ) {
+    return connectionParams[CLIENT_ID_PARAM];
+  }
+
+  return null;
+}
+
+export type OnConnectOptions = Omit<
+  TenantContextOptions,
+  "tenant" | "signingConfig" | "disableCaching"
+> &
+  Required<Pick<TenantContextOptions, "signingConfig">>;
+
+export function onConnect(options: OnConnectOptions): OnConnectFn {
+  // Create the JWT verifiers that will be used to verify all the requests
+  // coming in.
+  const verifiers = createVerifiers(options);
+
+  // Return the per-connection operation.
+  return async (connectionParams, socket) => {
+    try {
+      // Pull the upgrade request off of the connection.
+      const req: IncomingMessage = socket.upgradeReq;
+
+      // Get the hostname of the request.
+      const hostname = getHostname(req);
+      if (!hostname) {
+        throw new Error("could not detect hostname");
+      }
+
+      // Get the Tenant for this hostname.
+      const tenant = await options.tenantCache.retrieveByDomain(hostname);
+      if (!tenant) {
+        throw new TenantNotFoundError(hostname);
+      }
+
+      // Create some new options to store the tenant context details inside.
+      const opts: TenantContextOptions = {
+        ...options,
+        // Disable caching with this Context to ensure that every call (besides)
+        // to the tenant, is not cached, and is instead fresh.
+        disableCaching: true,
+        tenant,
+      };
+
+      // If the token is available, try to get the user.
+      const tokenString = extractTokenFromWSRequest(connectionParams, req);
+      if (tokenString) {
+        const user = await verifyAndRetrieveUser(
+          verifiers,
+          tenant,
+          tokenString
+        );
+        if (user) {
+          opts.user = user;
+        }
+      }
+
+      // Extract the users clientID from the request.
+      const clientID = extractClientID(connectionParams);
+      if (clientID) {
+        opts.clientID = clientID;
+      }
+
+      return new TenantContext(opts);
+    } catch (err) {
+      logger.error({ err }, "could not setup websocket connection");
+
+      if (!(err instanceof CoralError)) {
+        err = new InternalError(err, "could not setup websocket connection");
+      }
+      const { message } = err.serializeExtensions(
+        options.i18n.getDefaultBundle()
+      );
+
+      throw { message };
+    }
+  };
+}
+
+export type FormatResponseOptions = Pick<AppOptions, "metrics">;
+
+export function formatResponse({ metrics }: FormatResponseOptions) {
+  return (
+    value: ExecutionResult,
+    { context, query }: ExecutionParams<TenantContext>
+  ) => {
+    // Parse the query in order to extract operation metadata.
+    if (typeof query === "string") {
+      query = parse(query);
+    }
+
+    // Log out the query.
+    logQuery(context, query);
+
+    // Increment the metrics if enabled.
+    if (metrics) {
+      // Get the request metadata.
+      const { operation, operationName } = getOperationMetadata(query);
+      if (operation && operationName) {
+        // Increment the graph query value, tagging with the name of the query.
+        metrics.executedGraphQueriesTotalCounter
+          .labels(operation, operationName)
+          .inc();
+      }
+    }
+
+    if (value.errors && value.errors.length > 0) {
+      return {
+        ...value,
+        errors: value.errors.map(err => {
+          const enriched = enrichError(context, err);
+
+          // Log the error out.
+          logError(context, enriched);
+
+          return enriched;
+        }),
+      };
+    }
+
+    return value;
+  };
+}
+
+export type OnOperationOptions = FormatResponseOptions;
+
+export function onOperation(options: OnOperationOptions) {
+  return (message: any, params: ExecutionParams<TenantContext>) => {
+    // Attach the response formatter.
+    params.formatResponse = formatResponse(options);
+
+    return params;
+  };
+}
+
+export type Options = OnConnectOptions & OnOperationOptions;
+
+export function createSubscriptionServer(
+  server: http.Server,
+  schema: GraphQLSchema,
+  options: Options
+) {
+  return SubscriptionServer.create(
+    {
+      schema,
+      execute,
+      subscribe,
+      onConnect: onConnect(options),
+      onOperation: onOperation(options),
+    },
+    {
+      server,
+      path: "/api/graphql/live",
+    }
+  );
+}