[CORL-1077] Rudderstack Integration (#2987)

* feat: added rudderstack support with events

- Added events for
  - Comment Moderated
  - Comment Created
  - Story Created
  - Comment Reaction Created
  - User Flagged Comment

* fix: added logging to event tracking

src/core/server/app/index.ts

@@ -196,6 +196,9 @@ function configureApplicationViews(options: AppOptions) {
     // caching.
     watch: options.config.get("env") === "development",
     noCache: options.config.get("env") === "development",
+    // Trim whitespace in templates.
+    trimBlocks: true,
+    lstripBlocks: true,
   });
 
   // assign the nunjucks engine to .njk and .html files.

src/core/server/app/router/client.ts

@@ -1,5 +1,4 @@
 import express, { Router } from "express";
-import { minify } from "html-minifier";
 import { Db } from "mongodb";
 import path from "path";
 
@@ -15,6 +14,29 @@ import { RequestHandler } from "coral-server/types/express";
 import Entrypoints, { Entrypoint } from "../helpers/entrypoints";
 
 export interface ClientTargetHandlerOptions {
+  /**
+   * analytics contains configuration for frontend analytics from RudderStack.
+   */
+  analytics: {
+    /**
+     * key is the Write Key for the frontend integration.
+     */
+    key: string;
+
+    /**
+     * url is the URL to the data plane for your RudderStack deployment.
+     */
+    url: string;
+
+    /**
+     * sdk is the URL to the JS SDK for the RudderStack deployment.
+     */
+    sdk: string;
+  };
+
+  /**
+   * defaultLocale is the configured fallback locale for this installation.
+   */
   defaultLocale: LanguageCode;
 
   /**
@@ -62,6 +84,11 @@ function createClientTargetRouter(options: ClientTargetHandlerOptions) {
 }
 
 interface MountClientRouteOptions {
+  analytics: {
+    key: string;
+    url: string;
+    sdk: string;
+  };
   defaultLocale: LanguageCode;
   tenantCache: TenantCache;
   staticURI: string;
@@ -69,6 +96,7 @@ interface MountClientRouteOptions {
 }
 
 const clientHandler = ({
+  analytics,
   staticURI,
   entrypoint,
   enableCustomCSS,
@@ -85,28 +113,19 @@ const clientHandler = ({
     locale = req.coral.tenant.locale;
   }
 
-  res.render(
-    "client",
-    { staticURI, entrypoint, enableCustomCSS, locale, config },
-    (err, html) => {
-      if (err) {
-        return next(err);
-      }
-
-      // Send back the HTML minified.
-      res.send(
-        minify(html, {
-          removeComments: true,
-          collapseWhitespace: true,
-        })
-      );
-    }
-  );
+  res.render("client", {
+    analytics,
+    staticURI,
+    entrypoint,
+    enableCustomCSS,
+    locale,
+    config,
+  });
 };
 
 export function mountClientRoutes(
   router: Router,
-  { staticURI, tenantCache, defaultLocale, mongo }: MountClientRouteOptions
+  { tenantCache, ...options }: MountClientRouteOptions
 ) {
   // TODO: (wyattjoh) figure out a better way of referencing paths.
   // Load the entrypoint manifest.
@@ -141,31 +160,25 @@ export function mountClientRoutes(
   router.use(
     "/embed/stream",
     createClientTargetRouter({
-      staticURI,
+      ...options,
       enableCustomCSS: true,
       entrypoint: entrypoints.get("stream"),
-      defaultLocale,
-      mongo,
     })
   );
   router.use(
     "/embed/auth/callback",
     createClientTargetRouter({
-      staticURI,
+      ...options,
       cacheDuration: false,
       entrypoint: entrypoints.get("authCallback"),
-      defaultLocale,
-      mongo,
     })
   );
   router.use(
     "/embed/auth",
     createClientTargetRouter({
-      staticURI,
+      ...options,
       cacheDuration: false,
       entrypoint: entrypoints.get("auth"),
-      defaultLocale,
-      mongo,
     })
   );
 
@@ -175,11 +188,9 @@ export function mountClientRoutes(
     // If we aren't already installed, redirect the user to the install page.
     installedMiddleware(),
     createClientTargetRouter({
-      staticURI,
+      ...options,
       cacheDuration: false,
       entrypoint: entrypoints.get("account"),
-      defaultLocale,
-      mongo,
     })
   );
   // Add the standalone targets.
@@ -188,11 +199,9 @@ export function mountClientRoutes(
     // If we aren't already installed, redirect the user to the install page.
     installedMiddleware(),
     createClientTargetRouter({
-      staticURI,
+      ...options,
       cacheDuration: false,
       entrypoint: entrypoints.get("admin"),
-      defaultLocale,
-      mongo,
     })
   );
   router.use(
@@ -203,11 +212,9 @@ export function mountClientRoutes(
       redirectURL: "/admin",
     }),
     createClientTargetRouter({
-      staticURI,
+      ...options,
       cacheDuration: false,
       entrypoint: entrypoints.get("install"),
-      defaultLocale,
-      mongo,
     })
   );
 

src/core/server/app/router/index.ts

@@ -27,6 +27,11 @@ export function createRouter(app: AppOptions, options: RouterOptions) {
 
   if (!options.disableClientRoutes) {
     mountClientRoutes(router, {
+      analytics: {
+        key: app.config.get("analytics_frontend_key"),
+        url: app.config.get("analytics_data_plane_url"),
+        sdk: app.config.get("analytics_frontend_sdk_url"),
+      },
       defaultLocale: app.config.get("default_locale") as LanguageCode,
       // When mounting client routes, we need to provide a staticURI even when
       // not provided to the default current domain relative "/".

src/core/server/app/views/client.html

@@ -13,6 +13,29 @@
   {% if enableCustomCSS and tenant and tenant.customCSSURL %}
     {{ macros.preload(tenant.customCSSURL, "style") }}
   {% endif %}
+  {% if analytics and analytics.key and analytics.url and analytics.sdk %}
+    {# If analytics is enabled and available, configure the rudderstack analytics to load #}
+    <script type="text/javascript">
+      try {
+        rudderanalytics = window.rudderanalytics = [];
+        var methods = ["load", "page", "track", "alias", "group", "identify", "ready", "reset"];
+        for (var i = 0; i < methods.length; i++) {
+          var method = methods[i];
+          rudderanalytics[method] = function(d) {
+            return function() {
+              rudderanalytics.push([d, ...arguments])
+            }
+          }(method);
+        }
+
+        rudderanalytics.load("{{ analytics.key }}", "{{ analytics.url }}");
+        rudderanalytics.page();
+      } catch (err) {
+        console.error(err);
+      }
+    </script>
+    <script src="{{ analytics.sdk }}"></script>
+  {% endif %}
   {% if entrypoint.js %}
     {% for asset in entrypoint.js %}
       {{ macros.preload(asset.src, "script", prefix = staticURI) }}

src/core/server/config.ts

@@ -259,6 +259,30 @@ const config = convict({
     default: false,
     env: "DISABLE_JOB_PROCESSORS",
   },
+  analytics_frontend_key: {
+    doc: "Analytics write key from RudderStack for the Javascript client.",
+    format: String,
+    default: "",
+    env: "ANALYTICS_FRONTEND_KEY",
+  },
+  analytics_backend_key: {
+    doc: "Analytics write key from RudderStack for the Node server.",
+    format: String,
+    default: "",
+    env: "ANALYTICS_BACKEND_KEY",
+  },
+  analytics_frontend_sdk_url: {
+    doc: "Analytics URL to the RudderStack Frontend JS SDK. Defaults to the ",
+    format: "url",
+    default: "https://cdn.rudderlabs.com/v1/rudder-analytics.min.js",
+    env: "ANALYTICS_FRONTEND_SDK_URL",
+  },
+  analytics_data_plane_url: {
+    doc: "Analytics URL to the RudderStack data plane instance.",
+    format: "optional-url",
+    default: "",
+    env: "ANALYTICS_DATA_PLANE_URL",
+  },
 });
 
 export type Config = typeof config;

src/core/server/events/events.ts

@@ -7,6 +7,7 @@ import {
   CommentReplyCreatedInput,
   CommentStatusUpdatedInput,
 } from "coral-server/graph/resolvers/Subscription";
+import { FLAG_REASON } from "coral-server/models/action/comment";
 
 import { CoralEventPayload, createCoralEvent } from "./event";
 import { CoralEventType } from "./types";
@@ -20,6 +21,39 @@ export const CommentEnteredModerationQueueCoralEvent = createCoralEvent<
   CommentEnteredModerationQueueCoralEventPayload
 >(CoralEventType.COMMENT_ENTERED_MODERATION_QUEUE);
 
+export type CommentReactionCreatedCoralEventPayload = CoralEventPayload<
+  CoralEventType.COMMENT_REACTION_CREATED,
+  {
+    commentID: string;
+    commentRevisionID: string;
+    commentParentID?: string;
+    actionUserID: string;
+    storyID: string;
+    siteID: string;
+  }
+>;
+
+export const CommentReactionCreatedCoralEvent = createCoralEvent<
+  CommentReactionCreatedCoralEventPayload
+>(CoralEventType.COMMENT_REACTION_CREATED);
+
+export type CommentFlagCreatedCoralEventPayload = CoralEventPayload<
+  CoralEventType.COMMENT_FLAG_CREATED,
+  {
+    commentID: string;
+    commentRevisionID: string;
+    commentParentID?: string;
+    flagReason: FLAG_REASON;
+    actionUserID: string;
+    storyID: string;
+    siteID: string;
+  }
+>;
+
+export const CommentFlagCreatedCoralEvent = createCoralEvent<
+  CommentFlagCreatedCoralEventPayload
+>(CoralEventType.COMMENT_FLAG_CREATED);
+
 export type CommentLeftModerationQueueCoralEventPayload = CoralEventPayload<
   CoralEventType.COMMENT_LEFT_MODERATION_QUEUE,
   CommentLeftModerationQueueInput

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

@@ -0,0 +1,221 @@
+import Analytics, { Event } from "@rudderstack/rudder-sdk-node";
+
+import { Config } from "coral-server/config";
+import GraphContext from "coral-server/graph/context";
+import { relativeTo } from "coral-server/helpers";
+import logger from "coral-server/logger";
+
+import { GQLCOMMENT_STATUS } from "coral-server/graph/schema/__generated__/types";
+
+import {
+  CommentCreatedCoralEventPayload,
+  CommentFlagCreatedCoralEventPayload,
+  CommentReactionCreatedCoralEventPayload,
+  CommentReplyCreatedCoralEventPayload,
+  CommentStatusUpdatedCoralEventPayload,
+  StoryCreatedCoralEventPayload,
+} from "../events";
+import { CoralEventListener, CoralEventPublisherFactory } from "../publisher";
+import { CoralEventType } from "../types";
+
+type AnalyticsCoralEventListenerPayloads =
+  | CommentStatusUpdatedCoralEventPayload
+  | CommentCreatedCoralEventPayload
+  | CommentReplyCreatedCoralEventPayload
+  | CommentReactionCreatedCoralEventPayload
+  | CommentFlagCreatedCoralEventPayload
+  | StoryCreatedCoralEventPayload;
+
+export class AnalyticsCoralEventListener
+  implements CoralEventListener<AnalyticsCoralEventListenerPayloads> {
+  public readonly name = "analytics";
+  public readonly events = [
+    CoralEventType.COMMENT_CREATED,
+    CoralEventType.COMMENT_REPLY_CREATED,
+    CoralEventType.COMMENT_STATUS_UPDATED,
+    CoralEventType.COMMENT_REACTION_CREATED,
+    CoralEventType.COMMENT_FLAG_CREATED,
+    CoralEventType.STORY_CREATED,
+  ];
+
+  public readonly disabled: boolean = false;
+  private readonly analytics: Analytics;
+
+  constructor(config: Config) {
+    const key = config.get("analytics_backend_key");
+    const url = config.get("analytics_data_plane_url");
+    if (!key || !url) {
+      this.disabled = true;
+
+      return;
+    }
+
+    this.analytics = new Analytics(key, relativeTo("/v1/batch", url));
+  }
+
+  private filter(event: AnalyticsCoralEventListenerPayloads): boolean {
+    switch (event.type) {
+      case CoralEventType.COMMENT_CREATED:
+      case CoralEventType.COMMENT_REPLY_CREATED:
+      case CoralEventType.COMMENT_REACTION_CREATED:
+      case CoralEventType.COMMENT_FLAG_CREATED:
+      case CoralEventType.STORY_CREATED:
+        return true;
+      case CoralEventType.COMMENT_STATUS_UPDATED:
+        // We only record when a comment has been rejected/approved.
+        if (
+          event.data.newStatus !== GQLCOMMENT_STATUS.APPROVED &&
+          event.data.newStatus !== GQLCOMMENT_STATUS.REJECTED
+        ) {
+          return false;
+        }
+
+        return true;
+      default:
+        return false;
+    }
+  }
+
+  private async create(
+    ctx: GraphContext,
+    event: AnalyticsCoralEventListenerPayloads
+  ): Promise<Pick<Event, "event" | "properties"> | undefined> {
+    switch (event.type) {
+      case CoralEventType.COMMENT_CREATED:
+      case CoralEventType.COMMENT_REPLY_CREATED: {
+        const [comment, story] = await Promise.all([
+          ctx.loaders.Comments.comment.load(event.data.commentID),
+          ctx.loaders.Stories.story.load(event.data.storyID),
+        ]);
+        if (!comment || !story || !comment.authorID) {
+          return;
+        }
+
+        return {
+          event: "Comment Created",
+          properties: {
+            siteID: comment.siteID,
+            storyID: comment.storyID,
+            storyURL: story.url,
+            commentID: comment.id,
+            commentStatus: comment.status,
+            commentIsReply: !!comment.parentID,
+            commentAuthorID: comment.authorID,
+          },
+        };
+      }
+
+      case CoralEventType.STORY_CREATED: {
+        return {
+          event: "Story Created",
+          properties: {
+            siteID: event.data.siteID,
+            storyID: event.data.storyID,
+            storyURL: event.data.storyURL,
+          },
+        };
+      }
+
+      case CoralEventType.COMMENT_REACTION_CREATED: {
+        const story = await ctx.loaders.Stories.story.load(event.data.storyID);
+        if (!story) {
+          return;
+        }
+
+        return {
+          event: "Comment Reaction Created",
+          properties: {
+            siteID: story.siteID,
+            storyID: story.id,
+            storyURL: story.url,
+            commentID: event.data.commentID,
+            commentIsReply: !!event.data.commentParentID,
+            actionUserID: event.data.actionUserID,
+          },
+        };
+      }
+
+      case CoralEventType.COMMENT_FLAG_CREATED: {
+        const story = await ctx.loaders.Stories.story.load(event.data.storyID);
+        if (!story) {
+          return;
+        }
+
+        return {
+          event: "User Flagged Comment",
+          properties: {
+            siteID: story.siteID,
+            storyID: story.id,
+            storyURL: story.url,
+            commentID: event.data.commentID,
+            commentIsReply: !!event.data.commentParentID,
+            actionUserID: event.data.actionUserID,
+            flagReason: event.data.flagReason,
+          },
+        };
+      }
+
+      case CoralEventType.COMMENT_STATUS_UPDATED: {
+        const story = await ctx.loaders.Stories.story.load(event.data.storyID);
+        if (!story) {
+          return;
+        }
+
+        return {
+          event: "Comment Moderated",
+          properties: {
+            siteID: story.siteID,
+            storyID: story.id,
+            storyURL: story.url,
+            commentID: event.data.commentID,
+            commentStatus: event.data.newStatus,
+            commentPreviousStatus: event.data.oldStatus,
+          },
+        };
+      }
+    }
+  }
+
+  public initialize: CoralEventPublisherFactory<
+    AnalyticsCoralEventListenerPayloads
+  > = (ctx) => {
+    return async (event) => {
+      // Check to see if we should process this event.
+      if (!this.filter(event)) {
+        // The event should not be processed.
+        return;
+      }
+
+      // Create the event payload.
+      const details = await this.create(ctx, event);
+      if (!details) {
+        return;
+      }
+
+      // Pull some properties out of the context.
+      const {
+        // Sometimes, the user isn't defined (an anonymous request), so default
+        // so we don't get any spread errors.
+        user: { id: userId } = {},
+        tenant: { id: tenantID, domain: tenantDomain },
+      } = ctx;
+
+      // Assemble the track payload.
+      const payload: Event = {
+        event: details.event,
+        userId,
+        properties: {
+          ...details.properties,
+          tenantID,
+          tenantDomain,
+        },
+        timestamp: event.createdAt,
+      };
+
+      logger.debug({ payload }, "sending analytics event");
+
+      // Send the event payload to analytics.
+      return this.analytics.track(payload);
+    };
+  };
+}

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

@@ -1,3 +1,4 @@
+export * from "./analytics";
 export * from "./notifier";
 export * from "./perspective";
 export * from "./slack";

src/core/server/events/publisher.ts

@@ -20,6 +20,11 @@ export abstract class CoralEventListener<T extends CoralEventPayload = any> {
    */
   public abstract readonly name: string;
 
+  /**
+   * disabled if true will disable the event listener from handling requests.
+   */
+  public abstract readonly disabled?: boolean;
+
   /**
    * events is the array of event types that this listener should listen for.
    */
@@ -109,6 +114,11 @@ export default class CoralEventListenerBroker {
       return;
     }
 
+    if (listener.disabled) {
+      logger.warn({ listenerName: listener.name }, "listener was disabled");
+      return;
+    }
+
     logger.trace(
       { listenerName: listener.name, listenerEvents: listener.events },
       "registering listener for events"

src/core/server/events/types.ts

@@ -7,4 +7,6 @@ export enum CoralEventType {
   COMMENT_FEATURED = "COMMENT_FEATURED",
   COMMENT_RELEASED = "COMMENT_RELEASED",
   STORY_CREATED = "STORY_CREATED",
+  COMMENT_REACTION_CREATED = "COMMENT_REACTION_CREATED",
+  COMMENT_FLAG_CREATED = "COMMENT_FLAG_CREATED",
 }

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

@@ -10,6 +10,7 @@ import {
 export interface CommentReplyCreatedInput extends SubscriptionPayload {
   ancestorIDs: string[];
   commentID: string;
+  storyID: string;
 }
 
 export type CommentReplyCreatedSubscription = SubscriptionType<

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

@@ -16,6 +16,7 @@ export interface CommentStatusUpdatedInput extends SubscriptionPayload {
   moderatorID: string | null;
   commentID: string;
   commentRevisionID: string;
+  storyID: string;
 }
 
 export type CommentStatusUpdatedSubscription = SubscriptionType<

src/core/server/helpers/index.ts

@@ -1 +1,2 @@
 export { default as createTimer } from "./createTimer";
+export { default as relativeTo } from "./relativeTo";

src/core/server/helpers/relativeTo.spec.ts

@@ -0,0 +1,7 @@
+import relativeTo from "./relativeTo";
+
+it("strips the leading / from urls", () => {
+  expect(
+    relativeTo("/root/test", "https://coralproject.net/another/path/")
+  ).toEqual("https://coralproject.net/another/path/root/test");
+});

src/core/server/helpers/relativeTo.ts

@@ -0,0 +1,7 @@
+import { URL } from "url";
+
+function relativeTo(input: string, base: string): string {
+  return new URL(input.startsWith("/") ? input.slice(1) : input, base).href;
+}
+
+export default relativeTo;

src/core/server/index.ts

@@ -36,6 +36,7 @@ import {
 import { TenantCache } from "coral-server/services/tenant/cache";
 
 import {
+  AnalyticsCoralEventListener,
   NotifierCoralEventListener,
   PerspectiveCoralEventListener,
   SlackCoralEventListener,
@@ -64,10 +65,10 @@ class Server {
   private parentApp: Express;
 
   // schema is the GraphQL Schema that relates to the given Tenant.
-  private schema: GraphQLSchema;
+  private readonly schema: GraphQLSchema;
 
   // config exposes application specific configuration.
-  public config: Config;
+  public readonly config: Config;
 
   // httpServer is the running instance of the HTTP server that will bind to
   // the requested port.
@@ -102,10 +103,10 @@ class Server {
   private processing = false;
 
   // i18n is the server reference to the i18n framework.
-  private i18n: I18n;
+  private readonly i18n: I18n;
 
   // signingConfig is the server reference to the signing configuration.
-  private signingConfig: JWTSigningConfig;
+  private readonly signingConfig: JWTSigningConfig;
 
   // persistedQueryCache is the cache of persisted queries used by the GraphQL
   // server to handle persisted queries.
@@ -216,6 +217,7 @@ class Server {
 
     // Setup the broker.
     this.broker = new CoralEventListenerBroker();
+    this.broker.register(new AnalyticsCoralEventListener(this.config));
     this.broker.register(new NotifierCoralEventListener(this.tasks.notifier));
     this.broker.register(new SlackCoralEventListener());
     this.broker.register(new SubscriptionCoralEventListener());

src/core/server/models/action/comment.spec.ts

@@ -1,4 +1,3 @@
-import { GQLCOMMENT_FLAG_REASON } from "coral-server/graph/schema/__generated__/types";
 import {
   ACTION_TYPE,
   CommentAction,
@@ -9,6 +8,8 @@ import {
   validateAction,
 } from "coral-server/models/action/comment";
 
+import { GQLCOMMENT_FLAG_REASON } from "coral-server/graph/schema/__generated__/types";
+
 describe("#encodeActionCounts", () => {
   it("generates the action counts correctly", () => {
     const actions: Array<Partial<CommentAction>> = [

src/core/server/services/comments/actions.ts

@@ -4,6 +4,7 @@ import { CommentNotFoundError } from "coral-server/errors";
 import { CoralEventPublisherBroker } from "coral-server/events/publisher";
 import {
   ACTION_TYPE,
+  CommentAction,
   CreateActionInput,
   createActions,
   encodeActionCounts,
@@ -28,6 +29,10 @@ import {
 } from "coral-server/stacks/helpers";
 
 import { GQLCOMMENT_FLAG_REPORTED_REASON } from "coral-server/graph/schema/__generated__/types";
+import {
+  publishCommentFlagCreated,
+  publishCommentReactionCreated,
+} from "../events";
 
 export type CreateAction = CreateActionInput;
 
@@ -72,6 +77,11 @@ export async function addCommentActionCounts(
   return updatedComment;
 }
 
+interface AddCommentAction {
+  comment: Readonly<Comment>;
+  action?: CommentAction;
+}
+
 async function addCommentAction(
   mongo: Db,
   redis: AugmentedRedis,
@@ -79,7 +89,7 @@ async function addCommentAction(
   tenant: Tenant,
   input: Omit<CreateActionInput, "storyID" | "siteID">,
   now = new Date()
-): Promise<Readonly<Comment>> {
+): Promise<AddCommentAction> {
   const oldComment = await retrieveComment(mongo, tenant.id, input.commentID);
   if (!oldComment) {
     throw new CommentNotFoundError(input.commentID);
@@ -95,6 +105,9 @@ async function addCommentAction(
   // Update the actions for the comment.
   const commentActions = await addCommentActions(mongo, tenant, [action], now);
   if (commentActions.length > 0) {
+    // Get the comment action.
+    const [commentAction] = commentActions;
+
     // Compute the action counts.
     const actionCounts = encodeActionCounts(...commentActions);
 
@@ -122,10 +135,10 @@ async function addCommentAction(
       commentRevisionID: input.commentRevisionID,
     });
 
-    return updatedComment;
+    return { comment: updatedComment, action: commentAction };
   }
 
-  return oldComment;
+  return { comment: oldComment };
 }
 
 export async function removeCommentAction(
@@ -218,7 +231,7 @@ export async function createReaction(
   input: CreateCommentReaction,
   now = new Date()
 ) {
-  return addCommentAction(
+  const { comment, action } = await addCommentAction(
     mongo,
     redis,
     broker,
@@ -231,6 +244,17 @@ export async function createReaction(
     },
     now
   );
+  if (action) {
+    // A comment reaction was created! Publish it.
+    publishCommentReactionCreated(
+      broker,
+      comment,
+      input.commentRevisionID,
+      action
+    );
+  }
+
+  return comment;
 }
 
 export type RemoveCommentReaction = Pick<RemoveActionInput, "commentID">;
@@ -264,7 +288,7 @@ export async function createDontAgree(
   input: CreateCommentDontAgree,
   now = new Date()
 ) {
-  return addCommentAction(
+  const { comment } = await addCommentAction(
     mongo,
     redis,
     broker,
@@ -278,6 +302,8 @@ export async function createDontAgree(
     },
     now
   );
+
+  return comment;
 }
 
 export type RemoveCommentDontAgree = Pick<RemoveActionInput, "commentID">;
@@ -313,7 +339,7 @@ export async function createFlag(
   input: CreateCommentFlag,
   now = new Date()
 ) {
-  return addCommentAction(
+  const { comment, action } = await addCommentAction(
     mongo,
     redis,
     broker,
@@ -328,4 +354,10 @@ export async function createFlag(
     },
     now
   );
+  if (action) {
+    // A action was created! Publish the event.
+    publishCommentFlagCreated(broker, comment, input.commentRevisionID, action);
+  }
+
+  return comment;
 }

src/core/server/services/events/comments.ts

@@ -2,12 +2,15 @@ import {
   CommentCreatedCoralEvent,
   CommentEnteredModerationQueueCoralEvent,
   CommentFeaturedCoralEvent,
+  CommentFlagCreatedCoralEvent,
   CommentLeftModerationQueueCoralEvent,
+  CommentReactionCreatedCoralEvent,
   CommentReleasedCoralEvent,
   CommentReplyCreatedCoralEvent,
   CommentStatusUpdatedCoralEvent,
 } from "coral-server/events";
 import { CoralEventPublisherBroker } from "coral-server/events/publisher";
+import { CommentAction } from "coral-server/models/action/comment";
 import {
   Comment,
   CommentModerationQueueCounts,
@@ -26,6 +29,7 @@ export async function publishCommentStatusChanges(
   newStatus: GQLCOMMENT_STATUS,
   commentID: string,
   commentRevisionID: string,
+  storyID: string,
   moderatorID: string | null
 ) {
   if (oldStatus !== newStatus) {
@@ -34,6 +38,7 @@ export async function publishCommentStatusChanges(
       oldStatus,
       commentID,
       commentRevisionID,
+      storyID,
       moderatorID,
     });
   }
@@ -41,12 +46,13 @@ export async function publishCommentStatusChanges(
 
 export async function publishCommentReplyCreated(
   broker: CoralEventPublisherBroker,
-  comment: Pick<Comment, "id" | "status" | "ancestorIDs">
+  comment: Pick<Comment, "id" | "status" | "storyID" | "ancestorIDs">
 ) {
   if (getDepth(comment) > 0 && hasPublishedStatus(comment)) {
     await CommentReplyCreatedCoralEvent.publish(broker, {
       ancestorIDs: comment.ancestorIDs,
       commentID: comment.id,
+      storyID: comment.storyID,
     });
   }
 }
@@ -75,6 +81,46 @@ export async function publishCommentReleased(
   }
 }
 
+export async function publishCommentReactionCreated(
+  broker: CoralEventPublisherBroker,
+  comment: Pick<Comment, "id" | "storyID" | "siteID" | "parentID">,
+  commentRevisionID: string,
+  { userID }: Pick<CommentAction, "userID">
+) {
+  // We only publish reaction created events for reactions created by users.
+  if (userID) {
+    await CommentReactionCreatedCoralEvent.publish(broker, {
+      commentID: comment.id,
+      commentRevisionID,
+      commentParentID: comment.parentID,
+      actionUserID: userID,
+      storyID: comment.storyID,
+      siteID: comment.siteID,
+    });
+  }
+}
+
+export async function publishCommentFlagCreated(
+  broker: CoralEventPublisherBroker,
+  comment: Pick<Comment, "id" | "storyID" | "siteID" | "parentID">,
+  commentRevisionID: string,
+  { userID, reason }: Pick<CommentAction, "reason" | "userID">
+) {
+  // We only publish flag created events for flags created by the system with
+  // a reason.
+  if (userID && reason) {
+    await CommentFlagCreatedCoralEvent.publish(broker, {
+      commentID: comment.id,
+      commentRevisionID,
+      commentParentID: comment.parentID,
+      actionUserID: userID,
+      flagReason: reason,
+      storyID: comment.storyID,
+      siteID: comment.siteID,
+    });
+  }
+}
+
 export async function publishCommentFeatured(
   broker: CoralEventPublisherBroker,
   comment: Pick<Comment, "id" | "status" | "storyID">

src/core/server/stacks/helpers/publishChanges.ts

@@ -37,6 +37,7 @@ export default async function publishChanges(
       input.after.status,
       input.after.id,
       input.commentRevisionID,
+      input.after.storyID,
       input.moderatorID || null
     );