feat: initial impl

src/core/client/account/index.tsx

@@ -19,6 +19,7 @@ async function main() {
   const ManagedCoralContextProvider = await createManaged({
     initLocalState,
     localesData,
+    bundle: "account",
   });
 
   const Index: FunctionComponent = () => (

src/core/client/admin/index.tsx

@@ -20,6 +20,7 @@ async function main() {
   const ManagedCoralContextProvider = await createManaged({
     initLocalState,
     localesData,
+    bundle: "admin",
   });
 
   const Index: FunctionComponent = () => (

src/core/client/auth/index.tsx

@@ -19,6 +19,7 @@ async function main() {
   const ManagedCoralContextProvider = await createManaged({
     initLocalState,
     localesData,
+    bundle: "auth",
   });
 
   const Index: FunctionComponent = () => (

src/core/client/framework/lib/bootstrap/createManaged.tsx

@@ -59,6 +59,12 @@ interface CreateContextArguments {
 
   /** Supports emitting and listening to events. */
   eventEmitter?: EventEmitter2;
+
+  /** bundle is the specific source of the connection */
+  bundle: string;
+
+  /** bundleConfig is the configuration parameters for this bundle */
+  bundleConfig?: Record<string, string>;
 }
 
 /** websocketURL points to our live graphql server */
@@ -271,6 +277,8 @@ export default async function createManaged({
   localesData,
   pym,
   eventEmitter = new EventEmitter2({ wildcard: true, maxListeners: 20 }),
+  bundle,
+  bundleConfig = {},
 }: CreateContextArguments): Promise<ComponentType> {
   // Listen for outside clicks.
   let registerClickFarAway: ClickFarAwayRegister | undefined;
@@ -318,7 +326,9 @@ export default async function createManaged({
 
   const subscriptionClient = createManagedSubscriptionClient(
     websocketURL,
-    clientID
+    clientID,
+    bundle,
+    bundleConfig
   );
 
   const { environment, accessTokenProvider } = createRelayEnvironment(

src/core/client/framework/lib/network/createManagedSubscriptionClient.ts

@@ -9,7 +9,12 @@ import {
   SubscriptionClient,
 } from "subscriptions-transport-ws";
 
-import { ACCESS_TOKEN_PARAM, CLIENT_ID_PARAM } from "coral-common/constants";
+import {
+  ACCESS_TOKEN_PARAM,
+  BUNDLE_CONFIG_PARAM,
+  BUNDLE_ID_PARAM,
+  CLIENT_ID_PARAM,
+} from "coral-common/constants";
 import { ERROR_CODES } from "coral-common/errors";
 
 /**
@@ -61,7 +66,9 @@ export interface ManagedSubscriptionClient {
  */
 export default function createManagedSubscriptionClient(
   url: string,
-  clientID: string
+  clientID: string,
+  bundle: string,
+  bundleConfig: Record<string, string>
 ): ManagedSubscriptionClient {
   const requests: SubscriptionRequest[] = [];
   let subscriptionClient: SubscriptionClient | null = null;
@@ -114,6 +121,8 @@ export default function createManagedSubscriptionClient(
           connectionParams: {
             [ACCESS_TOKEN_PARAM]: accessToken,
             [CLIENT_ID_PARAM]: clientID,
+            [BUNDLE_ID_PARAM]: bundle,
+            [BUNDLE_CONFIG_PARAM]: bundleConfig,
           },
         });
       }

src/core/client/install/index.tsx

@@ -19,6 +19,7 @@ async function main() {
   const ManagedCoralContextProvider = await createManaged({
     localesData,
     initLocalState,
+    bundle: "install",
   });
 
   const Index: FunctionComponent = () => (

src/core/client/stream/index.tsx

@@ -2,6 +2,7 @@ import { Child as PymChild } from "pym.js";
 import React, { FunctionComponent } from "react";
 import ReactDOM from "react-dom";
 
+import { parseQuery } from "coral-common/utils";
 import injectConditionalPolyfills from "coral-framework/helpers/injectConditionalPolyfills";
 import potentiallyInjectAxe from "coral-framework/helpers/potentiallyInjectAxe";
 import { createManaged } from "coral-framework/lib/bootstrap";
@@ -13,6 +14,11 @@ import localesData from "./locales";
 // Import css variables.
 import "coral-ui/theme/stream.css";
 
+function extractBundleConfig() {
+  const { storyID, storyURL } = parseQuery(location.search);
+  return { storyID, storyURL } as Record<string, string>;
+}
+
 async function main() {
   const pym = new PymChild({
     polling: 100,
@@ -20,10 +26,17 @@ async function main() {
   await injectConditionalPolyfills();
   // Potentially inject react-axe for runtime a11y checks.
   await potentiallyInjectAxe(pym.parentUrl);
+
+  // Detect and extract the storyID and storyURL from the current page so we can
+  // add it to the managed provider.
+  const bundleConfig = extractBundleConfig();
+
   const ManagedCoralContextProvider = await createManaged({
     initLocalState,
     localesData,
     pym,
+    bundle: "stream",
+    bundleConfig,
   });
 
   const Index: FunctionComponent = () => (

src/core/client/stream/tabs/Comments/Stream/StreamContainer.tsx

@@ -54,6 +54,7 @@ import StoryClosedTimeoutContainer from "./StoryClosedTimeout";
 import { SuspendedInfoContainer } from "./SuspendedInfo/index";
 import UnansweredCommentsTab from "./UnansweredCommentsTab";
 import useCommentCountEvent from "./useCommentCountEvent";
+import ViewersWatchingContainer from "./ViewersWatchingContainer";
 import WarningContainer from "./Warning";
 
 import styles from "./StreamContainer.css";
@@ -227,6 +228,10 @@ export const StreamContainer: FunctionComponent<Props> = (props) => {
             settings={props.settings}
           />
         )}
+        <ViewersWatchingContainer
+          story={props.story}
+          settings={props.settings}
+        />
         <HorizontalGutter spacing={4} className={styles.tabBarContainer}>
           <Flex
             direction="row"
@@ -416,6 +421,7 @@ const enhanced = withFragmentContainer<Props>({
       ...CreateCommentReplyMutation_story
       ...CreateCommentMutation_story
       ...ModerateStreamContainer_story
+      ...ViewersWatchingContainer_story
       id
       url
       settings {
@@ -457,6 +463,7 @@ const enhanced = withFragmentContainer<Props>({
       ...AnnouncementContainer_settings
       ...ModerateStreamContainer_settings
       ...WarningContainer_settings
+      ...ViewersWatchingContainer_settings
     }
   `,
 })(StreamContainer);

src/core/client/stream/tabs/Comments/Stream/ViewersWatchingContainer.css

@@ -0,0 +1,23 @@
+$start-color: var(--palette-success-500);
+$end-color: var(--palette-success-400);
+
+@keyframes color {
+  0% {
+    color: $start-color;
+  }
+  50% {
+    color: $end-color;
+  }
+  100% {
+    color: $start-color;
+  }
+}
+
+.title,
+.icon {
+  animation: color 1s ease-in-out infinite;
+}
+
+.icon {
+  flex-basis: calc(var(--spacing-3) + var(--spacing-1) + 8px);
+}

src/core/client/stream/tabs/Comments/Stream/ViewersWatchingContainer.tsx

@@ -0,0 +1,67 @@
+import { Localized } from "@fluent/react/compat";
+import React, { FunctionComponent } from "react";
+import { graphql } from "react-relay";
+
+import { useLive } from "coral-framework/hooks";
+import { withFragmentContainer } from "coral-framework/lib/relay";
+import { Icon } from "coral-ui/components/v2";
+import { CallOut } from "coral-ui/components/v3";
+
+import { ViewersWatchingContainer_settings } from "coral-stream/__generated__/ViewersWatchingContainer_settings.graphql";
+import { ViewersWatchingContainer_story } from "coral-stream/__generated__/ViewersWatchingContainer_story.graphql";
+
+import styles from "./ViewersWatchingContainer.css";
+
+interface Props {
+  story: ViewersWatchingContainer_story;
+  settings: ViewersWatchingContainer_settings;
+}
+
+const ViewersWatchingContainer: FunctionComponent<Props> = ({
+  story,
+  settings,
+}) => {
+  const live = useLive({ story, settings });
+  if (!live) {
+    return null;
+  }
+
+  // We always add one for the current viewer!
+  const viewerCount = story.viewerCount + 1;
+
+  return (
+    <CallOut
+      classes={{ icon: styles.icon, title: styles.title }}
+      icon={<Icon size="md">play_circle_filled</Icon>}
+      title={
+        <Localized id="comments-watchers" $count={viewerCount}>
+          <span>{viewerCount} people are here</span>
+        </Localized>
+      }
+      titleWeight="semiBold"
+    />
+  );
+};
+
+const enhanced = withFragmentContainer<Props>({
+  story: graphql`
+    fragment ViewersWatchingContainer_story on Story {
+      viewerCount
+      isClosed
+      settings {
+        live {
+          enabled
+        }
+      }
+    }
+  `,
+  settings: graphql`
+    fragment ViewersWatchingContainer_settings on Settings {
+      disableCommenting {
+        enabled
+      }
+    }
+  `,
+})(ViewersWatchingContainer);
+
+export default enhanced;

src/core/common/constants.ts

@@ -13,6 +13,19 @@ export const CLIENT_ID_HEADER = "X-Coral-Client-ID";
  */
 export const CLIENT_ID_PARAM = "clientID";
 
+/**
+ * BUNDLE_ID_PARAM references the name of the param used ot send the ID of the
+ * bundle via connectionParams when connecting via a websocket connection.
+ */
+export const BUNDLE_ID_PARAM = "bundleID";
+
+/**
+ * BUNDLE_CONFIG_PARAM references the name of the param used to send the
+ * parameters of the bundle via connectionParams when connecting via a websocket
+ * connection.
+ */
+export const BUNDLE_CONFIG_PARAM = "bundleConfig";
+
 /**
  * ACCESS_TOKEN_PARAM references the name of the param used to send the access
  * token in connectionParams when authenticating a websocket connection.

src/core/server/app/handlers/api/graphql.ts

@@ -64,8 +64,8 @@ export const graphQLHandler = ({
       // Add the clientID if there is one on the request.
       const clientID = req.get(CLIENT_ID_HEADER);
       if (clientID) {
-        // TODO: (wyattjoh) validate length
-        opts.clientID = clientID;
+        // Limit the clientID to 36 characters (the length of a UUID).
+        opts.clientID = clientID.slice(0, 36);
       }
 
       return {

src/core/server/config.ts

@@ -262,6 +262,13 @@ const config = convict({
     default: ms("800 milliseconds"),
     env: "PERSPECTIVE_TIMEOUT",
   },
+  story_viewer_timeout: {
+    doc:
+      "The length of time (in ms) that a user should be considered active on a story without interaction.",
+    format: "ms",
+    default: ms("2 minutes"),
+    env: "STORY_VIEWER_TIMEOUT",
+  },
   force_ssl: {
     doc:
       "Forces SSL in production by redirecting all HTTP requests to HTTPS, and sending HSTS headers.",

src/core/server/events/events.ts

@@ -6,6 +6,7 @@ import {
   CommentReleasedInput,
   CommentReplyCreatedInput,
   CommentStatusUpdatedInput,
+  LiveStoryViewersUpdateInput,
 } from "coral-server/graph/resolvers/Subscription";
 import { FLAG_REASON } from "coral-server/models/action/comment";
 
@@ -120,3 +121,12 @@ export type StoryCreatedCoralEventPayload = CoralEventPayload<
 export const StoryCreatedCoralEvent = createCoralEvent<
   StoryCreatedCoralEventPayload
 >(CoralEventType.STORY_CREATED);
+
+export type LiveStoryViewersUpdateEventPayload = CoralEventPayload<
+  CoralEventType.LIVE_STORY_VIEWERS_UPDATE,
+  LiveStoryViewersUpdateInput
+>;
+
+export const LiveStoryViewersUpdateEvent = createCoralEvent<
+  LiveStoryViewersUpdateEventPayload
+>(CoralEventType.LIVE_STORY_VIEWERS_UPDATE);

src/core/server/events/listeners/index.ts

@@ -3,4 +3,5 @@ export * from "./notifier";
 export * from "./perspective";
 export * from "./slack";
 export * from "./subscription";
+export * from "./viewers";
 export * from "./webhook";

src/core/server/events/listeners/viewers.ts

@@ -0,0 +1,35 @@
+import { touchStoryViewer } from "coral-server/models/story/viewers";
+
+import {
+  CommentCreatedCoralEventPayload,
+  CommentReactionCreatedCoralEventPayload,
+  CommentReplyCreatedCoralEventPayload,
+} from "../events";
+import { CoralEventListener, CoralEventPublisherFactory } from "../publisher";
+import { CoralEventType } from "../types";
+
+type ViewersCoralEventListenerPayloads =
+  | CommentReplyCreatedCoralEventPayload
+  | CommentCreatedCoralEventPayload
+  | CommentReactionCreatedCoralEventPayload;
+
+export class ViewersCoralEventListener
+  implements CoralEventListener<ViewersCoralEventListenerPayloads> {
+  public readonly name = "viewers";
+  public readonly events = [
+    CoralEventType.COMMENT_REPLY_CREATED,
+    CoralEventType.COMMENT_CREATED,
+    CoralEventType.COMMENT_REACTION_CREATED,
+  ];
+
+  public initialize: CoralEventPublisherFactory<
+    ViewersCoralEventListenerPayloads
+  > = ({ clientID, mongo, now }) => async () => {
+    if (!clientID) {
+      return;
+    }
+
+    // NOTE: (wyattjoh) maybe replace this with the create instead?
+    await touchStoryViewer(mongo, clientID, now);
+  };
+}

src/core/server/events/types.ts

@@ -9,4 +9,5 @@ export enum CoralEventType {
   STORY_CREATED = "STORY_CREATED",
   COMMENT_REACTION_CREATED = "COMMENT_REACTION_CREATED",
   COMMENT_FLAG_CREATED = "COMMENT_FLAG_CREATED",
+  LIVE_STORY_VIEWERS_UPDATE = "LIVE_STORY_VIEWERS_UPDATE",
 }

src/core/server/graph/resolvers/LiveStoryViewersUpdatePayload.ts

@@ -0,0 +1,4 @@
+import { GQLLiveStoryViewersUpdatePayloadTypeResolver } from "coral-server/graph/schema/__generated__/types";
+import { LiveStoryViewersUpdateInput } from "./Subscription";
+
+export const LiveStoryViewersUpdatePayload: GQLLiveStoryViewersUpdatePayloadTypeResolver<LiveStoryViewersUpdateInput> = {};

src/core/server/graph/resolvers/Story.ts

@@ -2,6 +2,7 @@ import { defaultsDeep } from "lodash";
 
 import { decodeActionCounts } from "coral-server/models/action/comment";
 import * as story from "coral-server/models/story";
+import { countStoryViewers } from "coral-server/models/story/viewers";
 import { hasFeatureFlag } from "coral-server/models/tenant";
 import {
   canModerate,
@@ -59,4 +60,14 @@ export const Story: GQLStoryTypeResolver<story.Story> = {
     ),
   moderationQueues: storyModerationInputResolver,
   site: (s, input, ctx) => ctx.loaders.Sites.site.load(s.siteID),
+  viewerCount: (s, input, ctx) =>
+    // TODO: (wyattjoh) return 0 when live updates are disabled
+    countStoryViewers(
+      ctx.mongo,
+      ctx.tenant.id,
+      s.siteID,
+      s.id,
+      ctx.config.get("story_viewer_timeout"),
+      ctx.now
+    ),
 };

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

@@ -7,8 +7,9 @@ import { commentLeftModerationQueue } from "./commentLeftModerationQueue";
 import { commentReleased } from "./commentReleased";
 import { commentReplyCreated } from "./commentReplyCreated";
 import { commentStatusUpdated } from "./commentStatusUpdated";
+import { liveStoryViewersUpdate } from "./liveStoryViewersUpdate";
 
-export const Subscription: GQLSubscriptionTypeResolver = {
+export const Subscription: Required<GQLSubscriptionTypeResolver> = {
   commentCreated,
   commentEnteredModerationQueue,
   commentLeftModerationQueue,
@@ -16,6 +17,7 @@ export const Subscription: GQLSubscriptionTypeResolver = {
   commentStatusUpdated,
   commentFeatured,
   commentReleased,
+  liveStoryViewersUpdate,
 };
 
 export { CommentFeaturedInput } from "./commentFeatured";
@@ -25,3 +27,4 @@ export { CommentLeftModerationQueueInput } from "./commentLeftModerationQueue";
 export { CommentReleasedInput } from "./commentReleased";
 export { CommentReplyCreatedInput } from "./commentReplyCreated";
 export { CommentStatusUpdatedInput } from "./commentStatusUpdated";
+export { LiveStoryViewersUpdateInput } from "./liveStoryViewersUpdate";

src/core/server/graph/resolvers/Subscription/liveStoryViewersUpdate.ts

@@ -0,0 +1,31 @@
+import { SubscriptionToLiveStoryViewersUpdateResolver } from "coral-server/graph/schema/__generated__/types";
+
+import { createIterator } from "./helpers";
+import {
+  SUBSCRIPTION_CHANNELS,
+  SubscriptionPayload,
+  SubscriptionType,
+} from "./types";
+
+export interface LiveStoryViewersUpdateInput extends SubscriptionPayload {
+  viewerCount: number;
+  storyID: string;
+}
+
+export type LiveStoryViewersUpdateSubscription = SubscriptionType<
+  SUBSCRIPTION_CHANNELS.LIVE_STORY_VIEWERS_UPDATE,
+  LiveStoryViewersUpdateInput
+>;
+
+export const liveStoryViewersUpdate: SubscriptionToLiveStoryViewersUpdateResolver<LiveStoryViewersUpdateInput> = createIterator(
+  SUBSCRIPTION_CHANNELS.LIVE_STORY_VIEWERS_UPDATE,
+  {
+    filter: (source, { storyID }) => {
+      if (source.storyID !== storyID) {
+        return false;
+      }
+
+      return true;
+    },
+  }
+);

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

@@ -5,6 +5,7 @@ import { CommentLeftModerationQueueSubscription } from "./commentLeftModerationQ
 import { CommentReleasedSubscription } from "./commentReleased";
 import { CommentReplyCreatedSubscription } from "./commentReplyCreated";
 import { CommentStatusUpdatedSubscription } from "./commentStatusUpdated";
+import { LiveStoryViewersUpdateSubscription } from "./liveStoryViewersUpdate";
 
 export enum SUBSCRIPTION_CHANNELS {
   COMMENT_ENTERED_MODERATION_QUEUE = "COMMENT_ENTERED_MODERATION_QUEUE",
@@ -14,6 +15,7 @@ export enum SUBSCRIPTION_CHANNELS {
   COMMENT_CREATED = "COMMENT_CREATED",
   COMMENT_FEATURED = "COMMENT_FEATURED",
   COMMENT_RELEASED = "COMMENT_RELEASED",
+  LIVE_STORY_VIEWERS_UPDATE = "LIVE_STORY_VIEWERS_UPDATE",
 }
 
 export interface SubscriptionPayload {
@@ -35,4 +37,5 @@ export type SUBSCRIPTION_INPUT =
   | CommentReplyCreatedSubscription
   | CommentCreatedSubscription
   | CommentFeaturedSubscription
-  | CommentReleasedSubscription;
+  | CommentReleasedSubscription
+  | LiveStoryViewersUpdateSubscription;

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

@@ -29,6 +29,7 @@ import { GiphyMediaConfiguration } from "./GiphyMediaConfiguration";
 import { GoogleAuthIntegration } from "./GoogleAuthIntegration";
 import { Invite } from "./Invite";
 import { LiveConfiguration } from "./LiveConfiguration";
+import { LiveStoryViewersUpdatePayload } from "./LiveStoryViewersUpdatePayload";
 import { MediaConfiguration } from "./MediaConfiguration";
 import { ModerationQueue } from "./ModerationQueue";
 import { ModerationQueues } from "./ModerationQueues";
@@ -92,6 +93,7 @@ const Resolvers: GQLResolver = {
   GoogleAuthIntegration,
   Invite,
   LiveConfiguration,
+  LiveStoryViewersUpdatePayload,
   Locale,
   MediaConfiguration,
   ModerationQueue,

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

@@ -3448,6 +3448,11 @@ type Story {
   site is the site associated with the story
   """
   site: Site!
+
+  """
+  viewerCount is the number of viewers active on this Story.
+  """
+  viewerCount: Int!
 }
 
 """
@@ -7965,6 +7970,17 @@ type CommentReleasedPayload {
   comment: Comment!
 }
 
+"""
+LiveStoryViewersUpdatePayload is returned when the viewer count on a story has
+been updated.
+"""
+type LiveStoryViewersUpdatePayload {
+  """
+  viewerCount is the number of viewers on the selected story.
+  """
+  viewerCount: Int!
+}
+
 type Subscription {
   """
   commentEnteredModerationQueue returns when a Comment enters a ModerationQueue.
@@ -8017,4 +8033,9 @@ type Subscription {
   """
   commentFeatured(storyID: ID!): CommentFeaturedPayload!
     @auth(roles: [MODERATOR, ADMIN])
+
+  """
+  liveStoryViewersUpdate returns when the number of viewers on a story changes.
+  """
+  liveStoryViewersUpdate(storyID: ID!): LiveStoryViewersUpdatePayload!
 }

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

@@ -14,7 +14,12 @@ import {
   SubscriptionServer,
 } from "subscriptions-transport-ws";
 
-import { ACCESS_TOKEN_PARAM, CLIENT_ID_PARAM } from "coral-common/constants";
+import {
+  ACCESS_TOKEN_PARAM,
+  BUNDLE_CONFIG_PARAM,
+  BUNDLE_ID_PARAM,
+  CLIENT_ID_PARAM,
+} from "coral-common/constants";
 import { RequireProperty } from "coral-common/types";
 import { AppOptions } from "coral-server/app";
 import { getHostname } from "coral-server/app/helpers/hostname";
@@ -34,8 +39,13 @@ import { getOperationMetadata } from "coral-server/graph/extensions/helpers";
 import { getPersistedQuery } from "coral-server/graph/persisted";
 import logger from "coral-server/logger";
 import { PersistedQuery } from "coral-server/models/queries";
+import {
+  createStoryViewer,
+  removeStoryViewer,
+} from "coral-server/models/story/viewers";
 import { hasStaffRole } from "coral-server/models/user/helpers";
 import { extractTokenFromRequest } from "coral-server/services/jwt";
+import { find } from "coral-server/services/stories";
 
 import { GQLUSER_ROLE } from "coral-server/graph/schema/__generated__/types";
 
@@ -47,6 +57,8 @@ type OnConnectFn = (
   context: ConnectionContext
 ) => Promise<GraphContext>;
 
+type OnDisconnectFn = (socket: any, context: ConnectionContext) => void;
+
 export function extractTokenFromWSRequest(
   connectionParams: OperationMessagePayload,
   req: IncomingMessage
@@ -68,12 +80,44 @@ export function extractClientID(connectionParams: OperationMessagePayload) {
     typeof connectionParams[CLIENT_ID_PARAM] === "string" &&
     connectionParams[CLIENT_ID_PARAM].length > 0
   ) {
-    return connectionParams[CLIENT_ID_PARAM];
+    // Limit the clientID to 36 characters (the length of a UUID).
+    return connectionParams[CLIENT_ID_PARAM].slice(0, 36);
   }
 
   return null;
 }
 
+export function extractBundleID(
+  connectionParams: OperationMessagePayload
+): string | null {
+  if (
+    typeof connectionParams[BUNDLE_ID_PARAM] === "string" &&
+    connectionParams[BUNDLE_ID_PARAM].length > 0
+  ) {
+    return connectionParams[BUNDLE_ID_PARAM];
+  }
+
+  return null;
+}
+
+export function extractBundleConfig(
+  connectionParams: OperationMessagePayload
+): null | Record<string, string> {
+  if (typeof connectionParams[BUNDLE_CONFIG_PARAM] === "object") {
+    return connectionParams[BUNDLE_CONFIG_PARAM];
+  }
+
+  return null;
+}
+
+function hasClientID(socket: any): socket is { clientID: string } {
+  if (typeof socket.clientID === "string" && socket.clientID.length > 0) {
+    return true;
+  }
+
+  return false;
+}
+
 export type OnConnectOptions = RequireProperty<
   Omit<GraphContextOptions, "tenant" | "disableCaching">,
   "signingConfig"
@@ -86,6 +130,8 @@ export function onConnect(options: OnConnectOptions): OnConnectFn {
 
   // Return the per-connection operation.
   return async (connectionParams, socket) => {
+    logger.trace("a socket has connected");
+
     try {
       // Pull the upgrade request off of the connection.
       const req: IncomingMessage = socket.upgradeReq;
@@ -141,7 +187,50 @@ export function onConnect(options: OnConnectOptions): OnConnectFn {
         opts.clientID = clientID;
       }
 
-      return new GraphContext(opts);
+      // Create the GraphContext.
+      const ctx = new GraphContext(opts);
+
+      // Get the bundleID and bundleConfig.
+      const bundleID = extractBundleID(connectionParams);
+      const bundleConfig = extractBundleConfig(connectionParams);
+
+      if (
+        // If the request has a clientID...
+        clientID &&
+        // And it's from the stream...
+        bundleID === "stream" &&
+        // And it has a bundle config...
+        bundleConfig &&
+        // And we have either a storyID or storyURL on the config...
+        (bundleConfig.storyID || bundleConfig.storyURL)
+      ) {
+        // TODO: (wyattjoh) validate the bundle config.
+
+        // Then we need to create a new storyViewerf for the request!
+        const story = await find(options.mongo, tenant, {
+          id: bundleConfig.storyID,
+          url: bundleConfig.storyURL,
+        });
+        if (story) {
+          // Attach the clientID to the socket so the disconnect handler can use
+          // it to disconnect this clientID.
+          socket.clientID = clientID;
+
+          // Create the viewer entry!
+          await createStoryViewer(
+            options.mongo,
+            {
+              tenantID: tenant.id,
+              siteID: story.siteID,
+              storyID: story.id,
+              clientID,
+            },
+            ctx.now
+          );
+        }
+      }
+
+      return ctx;
     } catch (err) {
       if (err instanceof LiveUpdatesDisabled) {
         logger.info({ err }, "websocket connection rejected");
@@ -165,6 +254,23 @@ export function onConnect(options: OnConnectOptions): OnConnectFn {
   };
 }
 
+export type OnDisconnectOptions = RequireProperty<
+  Omit<GraphContextOptions, "tenant" | "disableCaching">,
+  "redis" | "pubsub"
+>;
+
+function onDisconnect(options: OnDisconnectOptions): OnDisconnectFn {
+  return async (socket) => {
+    logger.trace("a socket has disconnected");
+
+    // If the socket has a clientID attached, then remove the story viewer
+    // entry.
+    if (hasClientID(socket)) {
+      await removeStoryViewer(options.mongo, socket.clientID);
+    }
+  };
+}
+
 export type FormatResponseOptions = Pick<AppOptions, "metrics">;
 
 export function formatResponse(
@@ -252,7 +358,9 @@ export function onOperation(options: OnOperationOptions) {
   };
 }
 
-export type Options = OnConnectOptions & OnOperationOptions;
+export type Options = OnConnectOptions &
+  OnDisconnectOptions &
+  OnOperationOptions;
 
 export function createSubscriptionServer(
   server: http.Server,
@@ -267,6 +375,7 @@ export function createSubscriptionServer(
       execute,
       subscribe,
       onConnect: onConnect(options),
+      onDisconnect: onDisconnect(options),
       onOperation: onOperation(options),
       keepAlive,
     },

src/core/server/index.ts

@@ -41,6 +41,7 @@ import {
   PerspectiveCoralEventListener,
   SlackCoralEventListener,
   SubscriptionCoralEventListener,
+  ViewersCoralEventListener,
   WebhookCoralEventListener,
 } from "./events/listeners";
 import CoralEventListenerBroker from "./events/publisher";
@@ -221,6 +222,7 @@ class Server {
     this.broker.register(new NotifierCoralEventListener(this.tasks.notifier));
     this.broker.register(new SlackCoralEventListener());
     this.broker.register(new SubscriptionCoralEventListener());
+    this.broker.register(new ViewersCoralEventListener());
     this.broker.register(new WebhookCoralEventListener(this.tasks.webhook));
     this.broker.register(new PerspectiveCoralEventListener());
 

src/core/server/models/story/viewers.ts

@@ -0,0 +1,86 @@
+import { DateTime } from "luxon";
+import { Db } from "mongodb";
+
+import { storyViewer as collection } from "coral-server/services/mongodb/collections";
+
+import { TenantResource } from "../tenant";
+
+export interface StoryViewer extends TenantResource {
+  storyID: string;
+  siteID: string;
+  clientID: string;
+  lastInteractedAt: Date;
+}
+
+interface CreateStoryViewer {
+  storyID: string;
+  siteID: string;
+  tenantID: string;
+  clientID: string;
+}
+
+export async function createStoryViewer(
+  mongo: Db,
+  { clientID, ...input }: CreateStoryViewer,
+  now: Date
+) {
+  await collection(mongo).findOneAndUpdate(
+    { clientID },
+    {
+      $set: {
+        ...input,
+        lastInteractedAt: now,
+      },
+    },
+    {
+      upsert: true,
+    }
+  );
+}
+
+export async function removeStoryViewer(mongo: Db, clientID: string) {
+  await collection(mongo).deleteOne({ clientID });
+}
+
+export async function touchStoryViewer(mongo: Db, clientID: string, now: Date) {
+  await collection(mongo).updateOne(
+    { clientID },
+    { $set: { lastInteractedAt: now } }
+  );
+}
+
+export async function countStoryViewers(
+  mongo: Db,
+  tenantID: string,
+  siteID: string,
+  storyID: string,
+  timeout: number,
+  now: Date
+) {
+  const start = DateTime.fromJSDate(now).minus({ second: timeout }).toJSDate();
+
+  const results = await collection<{ count: number }>(mongo)
+    .aggregate([
+      {
+        $match: {
+          tenantID,
+          siteID,
+          storyID,
+          lastInteractedAt: { $gt: start, $lte: now },
+        },
+      },
+      {
+        $group: {
+          _id: null,
+          count: { $sum: 1 },
+        },
+      },
+    ])
+    .toArray();
+
+  if (results.length === 0) {
+    return 0;
+  }
+
+  return results[0].count;
+}

src/core/server/services/migrate/indexing.ts

@@ -95,4 +95,5 @@ export const createIndexesFactory = (mongo: Db) => ({
   queries: createIndexFactory(collections.queries(mongo)),
   migrations: createIndexFactory(collections.migrations(mongo)),
   sites: createIndexFactory(collections.sites(mongo)),
+  storyViewer: createIndexFactory(collections.storyViewer(mongo)),
 });

src/core/server/services/migrate/migrations/1596782282871_story_viewers.ts

@@ -0,0 +1,20 @@
+import { Db } from "mongodb";
+
+import Migration from "coral-server/services/migrate/migration";
+import collections from "coral-server/services/mongodb/collections";
+
+import { createIndexFactory } from "../indexing";
+
+export default class extends Migration {
+  // Remove the following line once the migration is ready, otherwise the
+  // migration will not be ran!
+  public static disabled = true;
+
+  public async indexes(mongo: Db) {
+    const indexer = createIndexFactory(collections.storyViewer(mongo));
+
+    await indexer({ storyID: 1 });
+    await indexer({ clientID: 1 }, { unique: true });
+    await indexer({ lastInteractedAt: 1 }, { expireAfterSeconds: 30 * 60 });
+  }
+}

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

@@ -7,6 +7,7 @@ import { MigrationRecord } from "coral-server/models/migration";
 import { PersistedQuery } from "coral-server/models/queries";
 import { Site } from "coral-server/models/site";
 import { Story } from "coral-server/models/story";
+import { StoryViewer } from "coral-server/models/story/viewers";
 import { Tenant } from "coral-server/models/tenant";
 import { User } from "coral-server/models/user";
 
@@ -32,6 +33,8 @@ export const queries = createCollection<PersistedQuery>("queries");
 
 export const migrations = createCollection<MigrationRecord>("migrations");
 
+export const storyViewer = createCollection<StoryViewer>("storyViewers");
+
 const collections = {
   users,
   invites,
@@ -43,6 +46,7 @@ const collections = {
   queries,
   migrations,
   sites,
+  storyViewer,
 };
 
 export default collections;

src/core/server/services/stories/viewers.ts

@@ -0,0 +1,13 @@
+import { LiveStoryViewersUpdateEvent } from "coral-server/events";
+import { CoralEventPublisherBroker } from "coral-server/events/publisher";
+
+export async function publishViewersUpdate(
+  broker: CoralEventPublisherBroker,
+  storyID: string,
+  viewerCount: number
+) {
+  void LiveStoryViewersUpdateEvent.publish(broker, {
+    storyID,
+    viewerCount,
+  });
+}

src/locales/en-US/stream.ftl

@@ -51,6 +51,11 @@ comment-count-text =
 comments-allCommentsTab = All Comments
 comments-featuredTab = Featured
 comments-counter-shortNum = { SHORT_NUMBER($count) }
+comments-watchers = { SHORT_NUMBER($count) } {
+  $count ->
+    [one] person is here
+    *[other] people are here
+}
 comments-featuredCommentTooltip-how = How is a comment featured?
 comments-featuredCommentTooltip-handSelectedComments =
   Comments are chosen by our team as worth reading.