diff --git a/src/core/server/graph/tenant/mutators/Actions.ts b/src/core/server/graph/tenant/mutators/Actions.ts
index fd63c3acc..8bc494d11 100644
--- a/src/core/server/graph/tenant/mutators/Actions.ts
+++ b/src/core/server/graph/tenant/mutators/Actions.ts
@@ -1,31 +1,32 @@
 import TenantContext from "coral-server/graph/tenant/context";
-import { hasTag } from "coral-server/models/comment";
-import { removeTag } from "coral-server/services/comments";
-import { approve, reject } from "coral-server/services/comments/moderation";
+import { approveComment, rejectComment } from "coral-server/stacks";
+
 import {
   GQLApproveCommentInput,
   GQLRejectCommentInput,
-  GQLTAG,
 } from "../schema/__generated__/types";
 
 export const Actions = (ctx: TenantContext) => ({
   approveComment: (input: GQLApproveCommentInput) =>
-    approve(ctx.mongo, ctx.redis, ctx.publisher, ctx.tenant, {
-      commentID: input.commentID,
-      commentRevisionID: input.commentRevisionID,
-      moderatorID: ctx.user!.id,
-    }),
+    approveComment(
+      ctx.mongo,
+      ctx.redis,
+      ctx.publisher,
+      ctx.tenant,
+      input.commentID,
+      input.commentRevisionID,
+      ctx.user!.id,
+      ctx.now
+    ),
   rejectComment: (input: GQLRejectCommentInput) =>
-    reject(ctx.mongo, ctx.redis, ctx.publisher, ctx.tenant, {
-      commentID: input.commentID,
-      commentRevisionID: input.commentRevisionID,
-      moderatorID: ctx.user!.id,
-    }).then(comment =>
-      // if a comment was previously featured
-      // and is now rejected, we need to remove the
-      // featured tag
-      hasTag(comment, GQLTAG.FEATURED)
-        ? removeTag(ctx.mongo, ctx.tenant, comment.id, GQLTAG.FEATURED)
-        : comment
+    rejectComment(
+      ctx.mongo,
+      ctx.redis,
+      ctx.publisher,
+      ctx.tenant,
+      input.commentID,
+      input.commentRevisionID,
+      ctx.user!.id,
+      ctx.now
     ),
 });
diff --git a/src/core/server/graph/tenant/mutators/Comments.ts b/src/core/server/graph/tenant/mutators/Comments.ts
index 3922d4f26..987e4f45e 100644
--- a/src/core/server/graph/tenant/mutators/Comments.ts
+++ b/src/core/server/graph/tenant/mutators/Comments.ts
@@ -2,6 +2,21 @@ import { ERROR_CODES } from "coral-common/errors";
 import { ADDITIONAL_DETAILS_MAX_LENGTH } from "coral-common/helpers/validate";
 import { mapFieldsetToErrorCodes } from "coral-server/graph/common/errors";
 import TenantContext from "coral-server/graph/tenant/context";
+import { addTag, removeTag } from "coral-server/services/comments";
+import {
+  createDontAgree,
+  createFlag,
+  createReaction,
+  removeDontAgree,
+  removeReaction,
+} from "coral-server/services/comments/actions";
+import { publishCommentFeatured } from "coral-server/services/events";
+import {
+  approveComment,
+  createComment,
+  editComment,
+} from "coral-server/stacks";
+
 import {
   GQLCOMMENT_STATUS,
   GQLCreateCommentDontAgreeInput,
@@ -16,22 +31,7 @@ import {
   GQLTAG,
   GQLUnfeatureCommentInput,
 } from "coral-server/graph/tenant/schema/__generated__/types";
-import {
-  addTag,
-  create,
-  edit,
-  removeTag,
-} from "coral-server/services/comments";
-import {
-  createDontAgree,
-  createFlag,
-  createReaction,
-  removeDontAgree,
-  removeReaction,
-} from "coral-server/services/comments/actions";
 
-import { approve } from "coral-server/services/comments/moderation";
-import { publishCommentFeatured } from "coral-server/services/events";
 import { validateMaximumLength, WithoutMutationID } from "./util";
 
 export const Comments = (ctx: TenantContext) => ({
@@ -41,7 +41,7 @@ export const Comments = (ctx: TenantContext) => ({
     ...comment
   }: GQLCreateCommentInput | GQLCreateCommentReplyInput) =>
     mapFieldsetToErrorCodes(
-      create(
+      createComment(
         ctx.mongo,
         ctx.redis,
         ctx.config,
@@ -64,7 +64,7 @@ export const Comments = (ctx: TenantContext) => ({
     ),
   edit: ({ commentID, body }: GQLEditCommentInput) =>
     mapFieldsetToErrorCodes(
-      edit(
+      editComment(
         ctx.mongo,
         ctx.redis,
         ctx.config,
@@ -170,11 +170,16 @@ export const Comments = (ctx: TenantContext) => ({
     )
       .then(comment =>
         comment.status !== GQLCOMMENT_STATUS.APPROVED
-          ? approve(ctx.mongo, ctx.redis, ctx.publisher, ctx.tenant, {
+          ? approveComment(
+              ctx.mongo,
+              ctx.redis,
+              ctx.publisher,
+              ctx.tenant,
               commentID,
               commentRevisionID,
-              moderatorID: ctx.user!.id,
-            })
+              ctx.user!.id,
+              ctx.now
+            )
           : comment
       )
       .then(comment => {
diff --git a/src/core/server/models/action/moderation/comment.ts b/src/core/server/models/action/moderation/comment.ts
index dc7fbf84e..c6071031f 100644
--- a/src/core/server/models/action/moderation/comment.ts
+++ b/src/core/server/models/action/moderation/comment.ts
@@ -56,7 +56,8 @@ export type CreateCommentModerationActionInput = Omit<
 export async function createCommentModerationAction(
   mongo: Db,
   tenantID: string,
-  input: CreateCommentModerationActionInput
+  input: CreateCommentModerationActionInput,
+  now: Date
 ) {
   // default are the properties set by the application when a new comment
   // moderation action is created.
@@ -66,7 +67,7 @@ export async function createCommentModerationAction(
   > = {
     id: uuid.v4(),
     tenantID,
-    createdAt: new Date(),
+    createdAt: now,
   };
 
   // Merge the defaults and the input together.
diff --git a/src/core/server/models/comment/comment.ts b/src/core/server/models/comment/comment.ts
index 9f0cd4783..6192a3a59 100644
--- a/src/core/server/models/comment/comment.ts
+++ b/src/core/server/models/comment/comment.ts
@@ -1,4 +1,4 @@
-import { isEmpty, merge } from "lodash";
+import { isEmpty } from "lodash";
 import { Db } from "mongodb";
 import performanceNow from "performance-now";
 import uuid from "uuid";
@@ -253,19 +253,19 @@ export function validateEditable(
 
 export interface EditComment {
   /**
-   * oldComment is the Comment that was previously set.
+   * before is the comment before the edit.
    */
-  oldComment: Comment;
+  before: Readonly<Comment>;
 
   /**
-   * editedComment is the Comment after the edit was performed.
+   * after is the comment after the edit.
    */
-  editedComment: Comment;
+  after: Readonly<Comment>;
 
   /**
-   * newRevision returns the new revision that was created in the Comment.
+   * revision is the new revision generated.
    */
-  newRevision: Revision;
+  revision: Readonly<Revision>;
 }
 
 /**
@@ -345,25 +345,25 @@ export async function editComment(
     throw new Error("comment edit failed for an unexpected reason");
   }
 
-  // Create a new "editedComment" where the same changes were applied to it as
+  // Create a new "after" where the same changes were applied to it as
   // we did to the MongoDB document.
-  const editedComment: Comment = merge({}, result.value, {
-    // Add in all the $set operations.
+  const after: Comment = {
+    ...result.value,
+    // $set status
     status,
-    metadata,
-    // Merge the actionCounts from the old Comment with the new actionCounts.
+    // $inc actionCounts
     actionCounts: mergeCommentActionCounts(
       result.value.actionCounts,
       actionCounts
     ),
-    // Add in the $push operations.
+    // $push revisions
     revisions: [...result.value.revisions, revision],
-  });
+  };
 
   return {
-    oldComment: result.value,
-    editedComment,
-    newRevision: revision,
+    before: result.value,
+    after,
+    revision,
   };
 }
 
@@ -709,14 +709,14 @@ function applyInputToQuery(
 
 export interface UpdateCommentStatus {
   /**
-   * comment is the updated Comment with the new status associated with it.
+   * before is the comment before editing the status.
    */
-  comment: Readonly<Comment>;
+  before: Readonly<Comment>;
 
   /**
-   * oldStatus is the previous status that the given Comment had.
+   * after is the comment after editing the status.
    */
-  oldStatus: GQLCOMMENT_STATUS;
+  after: Readonly<Comment>;
 }
 
 export async function updateCommentStatus(
@@ -748,15 +748,12 @@ export async function updateCommentStatus(
     return null;
   }
 
-  // Grab the old status.
-  const oldStatus = result.value.status;
-
   return {
-    comment: {
+    before: result.value,
+    after: {
       ...result.value,
       status,
     },
-    oldStatus,
   };
 }
 
diff --git a/src/core/server/models/story/counts/index.ts b/src/core/server/models/story/counts/index.ts
index 53417a1fd..3c08d4f4b 100644
--- a/src/core/server/models/story/counts/index.ts
+++ b/src/core/server/models/story/counts/index.ts
@@ -126,8 +126,6 @@ export const updateStoryActionCounts = (
   action: EncodedCommentActionCounts
 ) => updateStoryCounts(mongo, redis, tenantID, id, { action });
 
-export type StoryCounts = DeepPartial<StoryCommentCounts>;
-
 /**
  * updateStoryCounts will update the comment counts for the story indicated.
  *
@@ -142,7 +140,7 @@ export async function updateStoryCounts(
   redis: AugmentedRedis,
   tenantID: string,
   id: string,
-  commentCounts: StoryCounts
+  commentCounts: DeepPartial<StoryCommentCounts>
 ) {
   // Update all the specific comment moderation queue counts.
   const update: DeepPartial<Story> = { commentCounts };
diff --git a/src/core/server/models/story/counts/shared.ts b/src/core/server/models/story/counts/shared.ts
index 426ede297..02916fab0 100644
--- a/src/core/server/models/story/counts/shared.ts
+++ b/src/core/server/models/story/counts/shared.ts
@@ -2,10 +2,11 @@ import { flatten, flattenDeep, identity, isEmpty, pickBy } from "lodash";
 import { Db } from "mongodb";
 import ms from "ms";
 
+import { DeepPartial } from "coral-common/types";
 import logger from "coral-server/logger";
 import {
   CommentModerationCountsPerQueue,
-  StoryCounts,
+  StoryCommentCounts,
 } from "coral-server/models/story/counts";
 import { stories as collection } from "coral-server/services/mongodb/collections";
 import { AugmentedPipeline, AugmentedRedis } from "coral-server/services/redis";
@@ -180,7 +181,7 @@ export async function retrieveSharedModerationQueueQueuesCounts(
 export async function updateSharedCommentCounts(
   redis: AugmentedRedis,
   tenantID: string,
-  commentCounts: StoryCounts
+  commentCounts: DeepPartial<StoryCommentCounts>
 ) {
   const pipeline: AugmentedPipeline = redis.pipeline();
 
diff --git a/src/core/server/models/user/user.ts b/src/core/server/models/user/user.ts
index 837cd65e8..777869dfd 100644
--- a/src/core/server/models/user/user.ts
+++ b/src/core/server/models/user/user.ts
@@ -1,4 +1,5 @@
 import bcrypt from "bcryptjs";
+import { identity, isEmpty, pickBy } from "lodash";
 import { DateTime, DurationObject } from "luxon";
 import { Db, MongoError } from "mongodb";
 import uuid from "uuid";
@@ -19,16 +20,6 @@ import {
   UsernameAlreadySetError,
   UserNotFoundError,
 } from "coral-server/errors";
-import {
-  GQLBanStatus,
-  GQLDIGEST_FREQUENCY,
-  GQLPremodStatus,
-  GQLSuspensionStatus,
-  GQLTimeRange,
-  GQLUSER_ROLE,
-  GQLUsernameStatus,
-  GQLUserNotificationSettings,
-} from "coral-server/graph/tenant/schema/__generated__/types";
 import logger from "coral-server/logger";
 import {
   Connection,
@@ -40,6 +31,21 @@ import { TenantResource } from "coral-server/models/tenant";
 import { DigestibleTemplate } from "coral-server/queue/tasks/mailer/templates";
 import { users as collection } from "coral-server/services/mongodb/collections";
 
+import {
+  GQLBanStatus,
+  GQLDIGEST_FREQUENCY,
+  GQLPremodStatus,
+  GQLSuspensionStatus,
+  GQLTimeRange,
+  GQLUSER_ROLE,
+  GQLUsernameStatus,
+  GQLUserNotificationSettings,
+} from "coral-server/graph/tenant/schema/__generated__/types";
+
+import {
+  CommentStatusCounts,
+  createEmptyCommentStatusCounts,
+} from "../comment";
 import { getLocalProfile, hasLocalProfile } from "./helpers";
 
 export interface LocalProfile {
@@ -352,6 +358,10 @@ export interface Digest {
   createdAt: Date;
 }
 
+export interface UserCommentCounts {
+  status: CommentStatusCounts;
+}
+
 /**
  * User is someone that leaves Comments, and logs in.
  */
@@ -461,6 +471,8 @@ export interface User extends TenantResource {
    * deletedAt is the time that this user was deleted from our system.
    */
   deletedAt?: Date;
+
+  commentCounts: UserCommentCounts;
 }
 
 function hashPassword(password: string): Promise<string> {
@@ -514,6 +526,9 @@ async function findOrCreateUserInput(
     moderatorNotes: [],
     digests: [],
     createdAt: now,
+    commentCounts: {
+      status: createEmptyCommentStatusCounts(),
+    },
   };
 
   if (input.username) {
@@ -2446,3 +2461,32 @@ export async function deleteModeratorNote(
   }
   return result.value;
 }
+
+export async function updateUserCommentCounts(
+  mongo: Db,
+  tenantID: string,
+  id: string,
+  commentCounts: DeepPartial<UserCommentCounts>
+) {
+  // Update all the specific comment moderation queue counts.
+  const update: DeepPartial<User> = { commentCounts };
+  const $inc = pickBy(dotize(update), identity);
+  if (isEmpty($inc)) {
+    // Nothing needs to be incremented, just return the User.
+    return retrieveUser(mongo, tenantID, id);
+  }
+
+  logger.trace({ update: { $inc } }, "incrementing user comment counts");
+
+  const result = await collection(mongo).findOneAndUpdate(
+    { id, tenantID },
+    { $inc },
+    {
+      // False to return the updated document instead of the original
+      // document.
+      returnOriginal: false,
+    }
+  );
+
+  return result.value || null;
+}
diff --git a/src/core/server/services/comments/comments.ts b/src/core/server/services/comments/comments.ts
index b46452e49..f13cf9885 100644
--- a/src/core/server/services/comments/comments.ts
+++ b/src/core/server/services/comments/comments.ts
@@ -1,469 +1,16 @@
 import { DateTime } from "luxon";
 import { Db } from "mongodb";
 
-import { ERROR_TYPES } from "coral-common/errors";
-import { Omit } from "coral-common/types";
-import { Config } from "coral-server/config";
-import {
-  CommentNotFoundError,
-  CoralError,
-  StoryNotFoundError,
-} from "coral-server/errors";
+import { CommentNotFoundError } from "coral-server/errors";
 import { GQLTAG } from "coral-server/graph/tenant/schema/__generated__/types";
-import { Publisher } from "coral-server/graph/tenant/subscriptions/publisher";
-import logger from "coral-server/logger";
-import {
-  encodeActionCounts,
-  filterDuplicateActions,
-} from "coral-server/models/action/comment";
-import { createCommentModerationAction } from "coral-server/models/action/moderation/comment";
 import {
   addCommentTag,
-  createComment,
-  CreateCommentInput,
-  editComment,
-  EditCommentInput,
-  pushChildCommentIDOntoParent,
   removeCommentTag,
   retrieveComment,
-  validateEditable,
 } from "coral-server/models/comment";
-import {
-  getLatestRevision,
-  hasAncestors,
-  hasPublishedStatus,
-} from "coral-server/models/comment/helpers";
-import {
-  retrieveStory,
-  StoryCounts,
-  updateStoryCounts,
-} from "coral-server/models/story";
+import { getLatestRevision } from "coral-server/models/comment/helpers";
 import { Tenant } from "coral-server/models/tenant";
 import { User } from "coral-server/models/user";
-import {
-  publishCommentCreated,
-  publishCommentReplyCreated,
-  publishCommentStatusChanges,
-  publishModerationQueueChanges,
-} from "coral-server/services/events";
-import { AugmentedRedis } from "coral-server/services/redis";
-import { Request } from "coral-server/types/express";
-
-import { updateUserLastCommentID } from "../users";
-import { addCommentActions, CreateAction } from "./actions";
-import { calculateCounts, calculateCountsDiff } from "./moderation/counts";
-import { PhaseResult, processForModeration } from "./pipeline";
-
-export type CreateComment = Omit<
-  CreateCommentInput,
-  "status" | "metadata" | "ancestorIDs" | "actionCounts" | "tags"
->;
-
-export async function create(
-  mongo: Db,
-  redis: AugmentedRedis,
-  config: Config,
-  publish: Publisher,
-  tenant: Tenant,
-  author: User,
-  input: CreateComment,
-  nudge: boolean,
-  now = new Date(),
-  req?: Request
-) {
-  let log = logger.child(
-    {
-      authorID: author.id,
-      tenantID: tenant.id,
-      storyID: input.storyID,
-      parentID: input.parentID,
-      nudge,
-    },
-    true
-  );
-
-  log.trace("creating comment on story");
-
-  // Grab the story that we'll use to check moderation pieces with.
-  const story = await retrieveStory(mongo, tenant.id, input.storyID);
-  if (!story) {
-    throw new StoryNotFoundError(input.storyID);
-  }
-
-  const ancestorIDs: string[] = [];
-  if (input.parentID) {
-    // Check to see that the reference parent ID exists.
-    const parent = await retrieveComment(mongo, tenant.id, input.parentID);
-    if (!parent) {
-      throw new CommentNotFoundError(input.parentID);
-    }
-
-    // Check that the parent comment was visible.
-    if (!hasPublishedStatus(parent)) {
-      throw new CommentNotFoundError(parent.id);
-    }
-
-    ancestorIDs.push(input.parentID);
-    if (hasAncestors(parent)) {
-      // Push the parent's ancestors id's into the comment's ancestor id's.
-      ancestorIDs.push(...parent.ancestorIDs);
-    }
-
-    log.trace(
-      { ancestorIDs: ancestorIDs.length },
-      "pushed parent ancestorIDs into comment"
-    );
-  }
-
-  let result: PhaseResult;
-
-  try {
-    // Run the comment through the moderation phases.
-    result = await processForModeration({
-      action: "NEW",
-      log,
-      mongo,
-      redis,
-      config,
-      nudge,
-      story,
-      tenant,
-      comment: input,
-      author,
-      req,
-      now,
-    });
-  } catch (err) {
-    if (
-      err instanceof CoralError &&
-      err.type === ERROR_TYPES.MODERATION_NUDGE_ERROR
-    ) {
-      log.info({ err }, "detected pipeline nudge");
-    }
-
-    throw err;
-  }
-
-  const { actions, body, status, metadata, tags } = result;
-
-  // This is the first time this comment is being published.. So we need to
-  // ensure we don't run into any race conditions when we create the comment.
-  // One of the situations where we could encounter a race is when the comment
-  // is created, and does not have it's flag data associated with it. This would
-  // cause the comment to not be added to the flagged queue. If a flag is
-  // pending, and a user flags this comment before the next step can proceed,
-  // then we would end up double adding the comment to the flagged queue.
-  // Instead, we need to add the action metadata to the comment before we add it
-  // for the first time to ensure that the data is there for when the next flag
-  // is added, that it can already know that the comment is already in the
-  // queue.
-  let actionCounts = {};
-  if (actions.length > 0) {
-    // Determine the unique actions, we will use this to compute the comment
-    // action counts. This should match what is added below.
-    const deDuplicatedActions = filterDuplicateActions(actions);
-
-    // Encode the action counts.
-    actionCounts = encodeActionCounts(...deDuplicatedActions);
-  }
-
-  // Create the comment!
-  const comment = await createComment(
-    mongo,
-    tenant.id,
-    {
-      ...input,
-      tags,
-      body,
-      status,
-      ancestorIDs,
-      metadata,
-      actionCounts,
-    },
-    now
-  );
-
-  await updateUserLastCommentID(redis, tenant, author, comment.id);
-
-  // Pull the revision out.
-  const revision = getLatestRevision(comment);
-
-  log = log.child(
-    { commentID: comment.id, status, revisionID: revision.id },
-    true
-  );
-
-  log.trace("comment created");
-
-  if (input.parentID) {
-    // Push the child's ID onto the parent.
-    await pushChildCommentIDOntoParent(
-      mongo,
-      tenant.id,
-      input.parentID,
-      comment.id
-    );
-
-    log.trace("pushed child comment id onto parent");
-  }
-
-  if (actions.length > 0) {
-    // Actually add the actions to the database. This will not interact with the
-    // counts at all.
-    const upsertedActions = await addCommentActions(
-      mongo,
-      tenant,
-      actions.map(
-        (action): CreateAction => ({
-          ...action,
-          commentID: comment.id,
-          commentRevisionID: revision.id,
-
-          // Store the Story ID on the action.
-          storyID: story.id,
-        })
-      ),
-      now
-    );
-
-    log.trace({ actions: upsertedActions.length }, "added actions to comment");
-  }
-  const moderationQueue = calculateCounts(comment);
-
-  // Publish changes.
-  publishModerationQueueChanges(publish, moderationQueue, comment);
-
-  // If this is a reply, publish it.
-  if (input.parentID) {
-    publishCommentReplyCreated(publish, comment);
-  }
-
-  // If this comment is visible (and not a reply), publish it.
-  if (!input.parentID && hasPublishedStatus(comment)) {
-    publishCommentCreated(publish, comment);
-  }
-
-  // Compile the changes we want to apply to the story counts.
-  const storyCounts: Required<Omit<StoryCounts, "action">> = {
-    // This is a new comment, so we need to increment for this status.
-    status: { [status]: 1 },
-    // This comment is being created, so we can compute it raw from the comment
-    // that we created.
-    moderationQueue,
-  };
-
-  log.trace({ storyCounts }, "updating story status counts");
-
-  // Increment the status count for the particular status on the Story.
-  await updateStoryCounts(mongo, redis, tenant.id, story.id, storyCounts);
-
-  return comment;
-}
-
-export type EditComment = Omit<
-  EditCommentInput,
-  "status" | "authorID" | "lastEditableCommentCreatedAt" | "metadata"
->;
-
-export async function edit(
-  mongo: Db,
-  redis: AugmentedRedis,
-  config: Config,
-  publish: Publisher,
-  tenant: Tenant,
-  author: User,
-  input: EditComment,
-  now = new Date(),
-  req?: Request
-) {
-  let log = logger.child({ commentID: input.id, tenantID: tenant.id }, true);
-
-  // Get the comment that we're editing. This comment is considered stale,
-  // because it wasn't involved in the atomic transaction.
-  const originalStaleComment = await retrieveComment(
-    mongo,
-    tenant.id,
-    input.id
-  );
-  if (!originalStaleComment) {
-    throw new CommentNotFoundError(input.id);
-  }
-
-  // The editable time is based on the current time, and the edit window
-  // length. By subtracting the current date from the edit window length, we
-  // get the maximum value for the `createdAt` time that would be permitted
-  // for the comment edit to succeed.
-  const lastEditableCommentCreatedAt = getLastCommentEditableUntilDate(
-    tenant,
-    now
-  );
-
-  // Validate and potentially return with a more useful error.
-  validateEditable(originalStaleComment, {
-    authorID: author.id,
-    lastEditableCommentCreatedAt,
-  });
-
-  // Grab the story that we'll use to check moderation pieces with.
-  const story = await retrieveStory(
-    mongo,
-    tenant.id,
-    originalStaleComment.storyID
-  );
-  if (!story) {
-    throw new StoryNotFoundError(originalStaleComment.storyID);
-  }
-
-  // Run the comment through the moderation phases.
-  const { body, status, metadata, actions } = await processForModeration({
-    action: "EDIT",
-    log,
-    mongo,
-    redis,
-    config,
-    story,
-    tenant,
-    comment: input,
-    author,
-    req,
-    now,
-  });
-
-  let actionCounts = {};
-  if (actions.length > 0) {
-    // Encode the new action counts that are going to be added to the new
-    // revision.
-    actionCounts = encodeActionCounts(...filterDuplicateActions(actions));
-  }
-
-  log.trace(
-    { predictedActionCounts: actionCounts },
-    "associating action counts with comment"
-  );
-
-  // Perform the edit.
-  const result = await editComment(
-    mongo,
-    tenant.id,
-    {
-      id: input.id,
-      authorID: author.id,
-      body,
-      status,
-      metadata,
-      actionCounts,
-      lastEditableCommentCreatedAt,
-    },
-    now
-  );
-  if (!result) {
-    throw new CommentNotFoundError(input.id);
-  }
-
-  // Pull the old/edited comments out of the edit result.
-  const { oldComment, editedComment, newRevision } = result;
-
-  log = log.child({ revisionID: newRevision.id }, true);
-
-  if (actions.length > 0) {
-    // Insert and handle creating the actions.
-    const upsertedActions = await addCommentActions(
-      mongo,
-      tenant,
-      actions.map(
-        (action): CreateAction => ({
-          ...action,
-          commentID: editedComment.id,
-          commentRevisionID: newRevision.id,
-          storyID: story.id,
-        })
-      ),
-      now
-    );
-
-    log.trace(
-      {
-        actualActionCounts: encodeActionCounts(...upsertedActions),
-        actions: upsertedActions.length,
-      },
-      "added actions to comment"
-    );
-  }
-
-  // Compute the changes in queue counts. This looks at the action counts that
-  // are encoded, as well as the comment status's. We however may have had the
-  // comment status when we grabbed the updated comment after changing the
-  // action counts, so we extract the action counts out of the edited comment
-  // and use the status from the moderation decision.
-  const moderationQueue = calculateCountsDiff(oldComment, {
-    status,
-    actionCounts: editedComment.actionCounts,
-  });
-
-  // Compile the changes we want to apply to the story counts.
-  const storyCounts: Required<Omit<StoryCounts, "action">> = {
-    // Status is updated below if it has been changed.
-    status: {},
-    moderationQueue,
-  };
-
-  if (oldComment.status !== editedComment.status) {
-    // Increment the status count for the particular status on the Story, and
-    // decrement the status on the comment's previous status. The old comment
-    // status was only there before the atomic mutation. The new status is based
-    // on the moderation pipeline.
-    storyCounts.status[oldComment.status] = -1;
-    storyCounts.status[status] = 1;
-
-    // The comment status changed as a result of a pipeline operation, create a
-    // moderation action as a result.
-    await createCommentModerationAction(mongo, tenant.id, {
-      commentID: editedComment.id,
-      commentRevisionID: newRevision.id,
-      status: editedComment.status,
-      moderatorID: null,
-    });
-  }
-
-  log.trace({ storyCounts }, "updating story status counts");
-
-  // Update the story counts as a result.
-  await updateStoryCounts(mongo, redis, tenant.id, story.id, storyCounts);
-
-  // Publish changes.
-  publishModerationQueueChanges(publish, moderationQueue, editedComment);
-  publishCommentStatusChanges(
-    publish,
-    oldComment.status,
-    editedComment.status,
-    editedComment.id,
-    // This is a comment that was edited, so it should not present a moderator.
-    null
-  );
-
-  return editedComment;
-}
-
-/**
- * getLastCommentEditableUntilDate will return the `createdAt` date that will
- * represent the _oldest_ date that a comment could have been created on in
- * order to still be editable.
- *
- * @param tenant the tenant that contains settings related editing
- * @param now the date that is the base, defaulting to the current time
- */
-export function getLastCommentEditableUntilDate(
-  tenant: Pick<Tenant, "editCommentWindowLength">,
-  now = new Date()
-): Date {
-  return (
-    DateTime.fromJSDate(now)
-      // editCommentWindowLength is in seconds, so multiply by 1000 to get
-      // milliseconds.
-      .minus(tenant.editCommentWindowLength * 1000)
-      .toJSDate()
-  );
-}
 
 /**
  * getCommentEditableUntilDate will return the date that the given comment is
diff --git a/src/core/server/services/comments/moderation/index.ts b/src/core/server/services/comments/moderation/index.ts
index bd46c87d5..f6274822e 100644
--- a/src/core/server/services/comments/moderation/index.ts
+++ b/src/core/server/services/comments/moderation/index.ts
@@ -1,2 +1,2 @@
 export * from "./counts";
-export * from "./moderation";
+export { default as moderate } from "./moderate";
diff --git a/src/core/server/services/comments/moderation/moderate.ts b/src/core/server/services/comments/moderation/moderate.ts
new file mode 100644
index 000000000..89f47429b
--- /dev/null
+++ b/src/core/server/services/comments/moderation/moderate.ts
@@ -0,0 +1,46 @@
+import { Db } from "mongodb";
+
+import { CommentNotFoundError } from "coral-server/errors";
+import {
+  createCommentModerationAction,
+  CreateCommentModerationActionInput,
+} from "coral-server/models/action/moderation/comment";
+import { updateCommentStatus } from "coral-server/models/comment";
+import { Tenant } from "coral-server/models/tenant";
+
+export type Moderate = CreateCommentModerationActionInput;
+
+export default async function moderate(
+  mongo: Db,
+  tenant: Tenant,
+  input: Moderate,
+  now: Date
+) {
+  // TODO: wrap these operations in a transaction?
+
+  // Create the moderation action in the audit log.
+  const action = await createCommentModerationAction(
+    mongo,
+    tenant.id,
+    input,
+    now
+  );
+  if (!action) {
+    // TODO: wrap in better error?
+    throw new Error("could not create moderation action");
+  }
+
+  // Update the Comment's status.
+  const result = await updateCommentStatus(
+    mongo,
+    tenant.id,
+    input.commentID,
+    input.commentRevisionID,
+    input.status
+  );
+  if (!result) {
+    throw new CommentNotFoundError(input.commentID, input.commentRevisionID);
+  }
+
+  return result;
+}
diff --git a/src/core/server/services/comments/moderation/moderation.spec.ts b/src/core/server/services/comments/moderation/moderation.spec.ts
deleted file mode 100644
index acda37d06..000000000
--- a/src/core/server/services/comments/moderation/moderation.spec.ts
+++ /dev/null
@@ -1,194 +0,0 @@
-import { GQLCOMMENT_STATUS } from "coral-server/graph/tenant/schema/__generated__/types";
-import { calculateCountsDiff } from "./counts";
-
-it("allows transition from NONE to APPROVED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.APPROVED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: 0,
-      pending: 0,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from NONE to REJECTED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.REJECTED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: 0,
-      pending: 0,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from NONE to FLAGGED*", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: { FLAG: 1 } }
-    )
-  ).toEqual({
-    total: 0,
-    queues: {
-      reported: 1,
-      pending: 0,
-      unmoderated: 0,
-    },
-  });
-});
-
-it("allows transition from FLAGGED* to APPROVED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: { FLAG: 1 } },
-      { status: GQLCOMMENT_STATUS.APPROVED, actionCounts: { FLAG: 1 } }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: -1,
-      pending: 0,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from FLAGGED* to REJECTED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: { FLAG: 1 } },
-      { status: GQLCOMMENT_STATUS.REJECTED, actionCounts: { FLAG: 1 } }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: -1,
-      pending: 0,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from PREMOD to APPROVED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.PREMOD, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.APPROVED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: 0,
-      pending: -1,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from PREMOD to REJECTED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.PREMOD, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.REJECTED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: 0,
-      pending: -1,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from SYSTEM_WITHHELD to APPROVED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.SYSTEM_WITHHELD, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.APPROVED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: 0,
-      pending: -1,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from SYSTEM_WITHHELD to REJECTED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.SYSTEM_WITHHELD, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.REJECTED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: -1,
-    queues: {
-      reported: 0,
-      pending: -1,
-      unmoderated: -1,
-    },
-  });
-});
-
-it("allows transition from APPROVED to REJECTED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.APPROVED, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.REJECTED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: 0,
-    queues: {
-      reported: 0,
-      pending: 0,
-      unmoderated: 0,
-    },
-  });
-});
-
-it("allows transition from REJECTED to APPROVED", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.REJECTED, actionCounts: {} },
-      { status: GQLCOMMENT_STATUS.APPROVED, actionCounts: {} }
-    )
-  ).toEqual({
-    total: 0,
-    queues: {
-      reported: 0,
-      pending: 0,
-      unmoderated: 0,
-    },
-  });
-});
-
-it("allows no transition once a comment has been flagged more than once", () => {
-  expect(
-    calculateCountsDiff(
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: { FLAG: 1 } },
-      { status: GQLCOMMENT_STATUS.NONE, actionCounts: { FLAG: 2 } }
-    )
-  ).toEqual({
-    total: 0,
-    queues: {
-      reported: 0,
-      pending: 0,
-      unmoderated: 0,
-    },
-  });
-});
diff --git a/src/core/server/services/comments/moderation/moderation.ts b/src/core/server/services/comments/moderation/moderation.ts
deleted file mode 100644
index 7922689a0..000000000
--- a/src/core/server/services/comments/moderation/moderation.ts
+++ /dev/null
@@ -1,134 +0,0 @@
-import { Db } from "mongodb";
-
-import { Omit } from "coral-common/types";
-import { CommentNotFoundError, StoryNotFoundError } from "coral-server/errors";
-import { GQLCOMMENT_STATUS } from "coral-server/graph/tenant/schema/__generated__/types";
-import { Publisher } from "coral-server/graph/tenant/subscriptions/publisher";
-import logger from "coral-server/logger";
-import {
-  createCommentModerationAction,
-  CreateCommentModerationActionInput,
-} from "coral-server/models/action/moderation/comment";
-import { updateCommentStatus } from "coral-server/models/comment";
-import { updateStoryCounts } from "coral-server/models/story";
-import { Tenant } from "coral-server/models/tenant";
-import {
-  publishCommentReleased,
-  publishCommentStatusChanges,
-  publishModerationQueueChanges,
-} from "coral-server/services/events";
-import { AugmentedRedis } from "coral-server/services/redis";
-
-import { calculateCountsDiff } from "./counts";
-
-export type Moderate = Omit<CreateCommentModerationActionInput, "status">;
-
-const moderate = (
-  status: GQLCOMMENT_STATUS.APPROVED | GQLCOMMENT_STATUS.REJECTED
-) => async (
-  mongo: Db,
-  redis: AugmentedRedis,
-  publish: Publisher,
-  tenant: Tenant,
-  input: Moderate
-) => {
-  // TODO: wrap these operations in a transaction?
-
-  // Create the logger.
-  const log = logger.child(
-    {
-      ...input,
-      tenantID: tenant.id,
-      newStatus: status,
-    },
-    true
-  );
-
-  // Update the Comment's status.
-  const result = await updateCommentStatus(
-    mongo,
-    tenant.id,
-    input.commentID,
-    input.commentRevisionID,
-    status
-  );
-  if (!result) {
-    throw new CommentNotFoundError(input.commentID, input.commentRevisionID);
-  }
-
-  log.trace("updated comment status");
-
-  // Create the moderation action in the audit log.
-  const action = await createCommentModerationAction(mongo, tenant.id, {
-    ...input,
-    status,
-  });
-  if (!action) {
-    // TODO: wrap in better error?
-    throw new Error("could not create moderation action");
-  }
-
-  log.trace(
-    { commentModerationActionID: action.id },
-    "created the moderation action"
-  );
-
-  // Compute the queue difference as a result of the old status and the new
-  // status.
-  const moderationQueue = calculateCountsDiff(
-    {
-      status: result.oldStatus,
-      actionCounts: result.comment.actionCounts,
-    },
-    {
-      status,
-      actionCounts: result.comment.actionCounts,
-    }
-  );
-
-  // Update the story comment counts.
-  const story = await updateStoryCounts(
-    mongo,
-    redis,
-    tenant.id,
-    result.comment.storyID,
-    {
-      // Update the comment counts.
-      status: {
-        [result.oldStatus]: -1,
-        [status]: 1,
-      },
-
-      moderationQueue,
-    }
-  );
-  if (!story) {
-    throw new StoryNotFoundError(result.comment.storyID);
-  }
-
-  // Publish changes.
-  publishModerationQueueChanges(publish, moderationQueue, result.comment);
-  publishCommentStatusChanges(
-    publish,
-    result.oldStatus,
-    status,
-    result.comment.id,
-    input.moderatorID
-  );
-  if (
-    [GQLCOMMENT_STATUS.PREMOD, GQLCOMMENT_STATUS.SYSTEM_WITHHELD].includes(
-      result.oldStatus
-    ) &&
-    status === "APPROVED"
-  ) {
-    publishCommentReleased(publish, result.comment);
-  }
-
-  log.trace({ oldStatus: result.oldStatus }, "adjusted story comment counts");
-
-  return result.comment;
-};
-
-export const approve = moderate(GQLCOMMENT_STATUS.APPROVED);
-
-export const reject = moderate(GQLCOMMENT_STATUS.REJECTED);
diff --git a/src/core/server/services/migrate/migrations/1575649180_user_comment_counts.ts b/src/core/server/services/migrate/migrations/1575649180_user_comment_counts.ts
new file mode 100644
index 000000000..4cc454a3e
--- /dev/null
+++ b/src/core/server/services/migrate/migrations/1575649180_user_comment_counts.ts
@@ -0,0 +1,100 @@
+import { Db } from "mongodb";
+
+import {
+  CommentStatusCounts,
+  createEmptyCommentStatusCounts,
+} from "coral-server/models/comment";
+import Migration from "coral-server/services/migrate/migration";
+import collections from "coral-server/services/mongodb/collections";
+
+import { GQLCOMMENT_STATUS } from "coral-server/graph/tenant/schema/__generated__/types";
+
+const BATCH_SIZE = 500;
+
+export default class extends Migration {
+  public async up(mongo: Db, tenantID: string) {
+    const cursor = collections
+      .comments<{
+        _id: string;
+        statuses: { status: GQLCOMMENT_STATUS; sum: number }[];
+      }>(mongo)
+      .aggregate([
+        // Find all comments written by this Tenant.
+        {
+          $match: { tenantID },
+        },
+        // Group each comment into it's authorID and status.
+        {
+          $group: {
+            _id: {
+              authorID: "$authorID",
+              status: "$status",
+            },
+            sum: { $sum: 1 },
+          },
+        },
+        // Group the documents from the previous stage into a count of each
+        // author's status counts.
+        {
+          $group: {
+            _id: "$_id.authorID",
+            statuses: {
+              $push: {
+                status: "$_id.status",
+                sum: "$sum",
+              },
+            },
+          },
+        },
+      ]);
+
+    let updates: {
+      updateOne: {
+        filter: { tenantID: string; id: string };
+        update: { $set: { commentCounts: { status: CommentStatusCounts } } };
+      };
+    }[] = [];
+
+    while (await cursor.hasNext()) {
+      const doc = await cursor.next();
+      if (!doc) {
+        break;
+      }
+
+      // Reconstruct the comment status counts.
+      const statuses = createEmptyCommentStatusCounts();
+      for (const { status, sum } of doc.statuses) {
+        statuses[status] += sum;
+      }
+
+      // Push the update.
+      updates.push({
+        updateOne: {
+          filter: {
+            tenantID,
+            id: doc._id,
+          },
+          update: {
+            $set: {
+              commentCounts: {
+                status: statuses,
+              },
+            },
+          },
+        },
+      });
+
+      // Process updates if we are at the batch size.
+      if (updates.length >= BATCH_SIZE) {
+        await collections.users(mongo).bulkWrite(updates);
+        updates = [];
+      }
+    }
+
+    // Process any missed updates.
+    if (updates.length > 0) {
+      await collections.users(mongo).bulkWrite(updates);
+      updates = [];
+    }
+  }
+}
diff --git a/src/core/server/stacks/approveComment.ts b/src/core/server/stacks/approveComment.ts
new file mode 100644
index 000000000..3e2732721
--- /dev/null
+++ b/src/core/server/stacks/approveComment.ts
@@ -0,0 +1,52 @@
+import { Db } from "mongodb";
+
+import { Publisher } from "coral-server/graph/tenant/subscriptions/publisher";
+import { Tenant } from "coral-server/models/tenant";
+import { moderate } from "coral-server/services/comments/moderation";
+import { AugmentedRedis } from "coral-server/services/redis";
+
+import { GQLCOMMENT_STATUS } from "coral-server/graph/tenant/schema/__generated__/types";
+
+import { publishChanges, updateAllCounts } from "./helpers";
+
+const approveComment = async (
+  mongo: Db,
+  redis: AugmentedRedis,
+  publisher: Publisher,
+  tenant: Tenant,
+  commentID: string,
+  commentRevisionID: string,
+  moderatorID: string,
+  now: Date
+) => {
+  // Approve the comment.
+  const result = await moderate(
+    mongo,
+    tenant,
+    {
+      commentID,
+      commentRevisionID,
+      moderatorID,
+      status: GQLCOMMENT_STATUS.APPROVED,
+    },
+    now
+  );
+
+  // Update all the comment counts on stories and users.
+  const counts = await updateAllCounts(mongo, redis, {
+    tenant,
+    ...result,
+  });
+
+  // Publish changes to the event publisher.
+  await publishChanges(publisher, {
+    ...result,
+    ...counts,
+    moderatorID,
+  });
+
+  // Return the resulting comment.
+  return result.after;
+};
+
+export default approveComment;
diff --git a/src/core/server/stacks/createComment.ts b/src/core/server/stacks/createComment.ts
new file mode 100644
index 000000000..bb9124911
--- /dev/null
+++ b/src/core/server/stacks/createComment.ts
@@ -0,0 +1,247 @@
+import { Db } from "mongodb";
+
+import { ERROR_TYPES } from "coral-common/errors";
+import { Omit } from "coral-common/types";
+import { Config } from "coral-server/config";
+import {
+  CommentNotFoundError,
+  CoralError,
+  StoryNotFoundError,
+} from "coral-server/errors";
+import { Publisher } from "coral-server/graph/tenant/subscriptions/publisher";
+import logger from "coral-server/logger";
+import {
+  encodeActionCounts,
+  filterDuplicateActions,
+} from "coral-server/models/action/comment";
+import {
+  createComment,
+  CreateCommentInput,
+  pushChildCommentIDOntoParent,
+  retrieveComment,
+} from "coral-server/models/comment";
+import {
+  getLatestRevision,
+  hasAncestors,
+  hasPublishedStatus,
+} from "coral-server/models/comment/helpers";
+import { retrieveStory } from "coral-server/models/story";
+import { Tenant } from "coral-server/models/tenant";
+import { User } from "coral-server/models/user";
+import {
+  addCommentActions,
+  CreateAction,
+} from "coral-server/services/comments/actions";
+import {
+  PhaseResult,
+  processForModeration,
+} from "coral-server/services/comments/pipeline";
+import {
+  publishCommentCreated,
+  publishCommentReplyCreated,
+} from "coral-server/services/events";
+import { AugmentedRedis } from "coral-server/services/redis";
+import { updateUserLastCommentID } from "coral-server/services/users";
+import { Request } from "coral-server/types/express";
+
+import { publishChanges, updateAllCounts } from "./helpers";
+
+export type CreateComment = Omit<
+  CreateCommentInput,
+  "status" | "metadata" | "ancestorIDs" | "actionCounts" | "tags"
+>;
+
+export default async function create(
+  mongo: Db,
+  redis: AugmentedRedis,
+  config: Config,
+  publisher: Publisher,
+  tenant: Tenant,
+  author: User,
+  input: CreateComment,
+  nudge: boolean,
+  now = new Date(),
+  req?: Request
+) {
+  let log = logger.child(
+    {
+      authorID: author.id,
+      tenantID: tenant.id,
+      storyID: input.storyID,
+      parentID: input.parentID,
+      nudge,
+    },
+    true
+  );
+
+  log.trace("creating comment on story");
+
+  // Grab the story that we'll use to check moderation pieces with.
+  const story = await retrieveStory(mongo, tenant.id, input.storyID);
+  if (!story) {
+    throw new StoryNotFoundError(input.storyID);
+  }
+
+  const ancestorIDs: string[] = [];
+  if (input.parentID) {
+    // Check to see that the reference parent ID exists.
+    const parent = await retrieveComment(mongo, tenant.id, input.parentID);
+    if (!parent) {
+      throw new CommentNotFoundError(input.parentID);
+    }
+
+    // Check that the parent comment was visible.
+    if (!hasPublishedStatus(parent)) {
+      throw new CommentNotFoundError(parent.id);
+    }
+
+    ancestorIDs.push(input.parentID);
+    if (hasAncestors(parent)) {
+      // Push the parent's ancestors id's into the comment's ancestor id's.
+      ancestorIDs.push(...parent.ancestorIDs);
+    }
+
+    log.trace(
+      { ancestorIDs: ancestorIDs.length },
+      "pushed parent ancestorIDs into comment"
+    );
+  }
+
+  let result: PhaseResult;
+  try {
+    // Run the comment through the moderation phases.
+    result = await processForModeration({
+      action: "NEW",
+      log,
+      mongo,
+      redis,
+      config,
+      nudge,
+      story,
+      tenant,
+      comment: input,
+      author,
+      req,
+      now,
+    });
+  } catch (err) {
+    if (
+      err instanceof CoralError &&
+      err.type === ERROR_TYPES.MODERATION_NUDGE_ERROR
+    ) {
+      log.info({ err }, "detected pipeline nudge");
+    }
+
+    throw err;
+  }
+
+  const { actions, body, status, metadata, tags } = result;
+
+  // This is the first time this comment is being published.. So we need to
+  // ensure we don't run into any race conditions when we create the comment.
+  // One of the situations where we could encounter a race is when the comment
+  // is created, and does not have it's flag data associated with it. This would
+  // cause the comment to not be added to the flagged queue. If a flag is
+  // pending, and a user flags this comment before the next step can proceed,
+  // then we would end up double adding the comment to the flagged queue.
+  // Instead, we need to add the action metadata to the comment before we add it
+  // for the first time to ensure that the data is there for when the next flag
+  // is added, that it can already know that the comment is already in the
+  // queue.
+  let actionCounts = {};
+  if (actions.length > 0) {
+    // Determine the unique actions, we will use this to compute the comment
+    // action counts. This should match what is added below.
+    const deDuplicatedActions = filterDuplicateActions(actions);
+
+    // Encode the action counts.
+    actionCounts = encodeActionCounts(...deDuplicatedActions);
+  }
+
+  // Create the comment!
+  const comment = await createComment(
+    mongo,
+    tenant.id,
+    {
+      ...input,
+      tags,
+      body,
+      status,
+      ancestorIDs,
+      metadata,
+      actionCounts,
+    },
+    now
+  );
+
+  await updateUserLastCommentID(redis, tenant, author, comment.id);
+
+  // Pull the revision out.
+  const revision = getLatestRevision(comment);
+
+  log = log.child(
+    { commentID: comment.id, status, revisionID: revision.id },
+    true
+  );
+
+  log.trace("comment created");
+
+  if (input.parentID) {
+    // Push the child's ID onto the parent.
+    await pushChildCommentIDOntoParent(
+      mongo,
+      tenant.id,
+      input.parentID,
+      comment.id
+    );
+
+    log.trace("pushed child comment id onto parent");
+  }
+
+  if (actions.length > 0) {
+    // Actually add the actions to the database. This will not interact with the
+    // counts at all.
+    const upsertedActions = await addCommentActions(
+      mongo,
+      tenant,
+      actions.map(
+        (action): CreateAction => ({
+          ...action,
+          commentID: comment.id,
+          commentRevisionID: revision.id,
+
+          // Store the Story ID on the action.
+          storyID: story.id,
+        })
+      ),
+      now
+    );
+
+    log.trace({ actions: upsertedActions.length }, "added actions to comment");
+  }
+
+  // Update all the comment counts on stories and users.
+  const counts = await updateAllCounts(mongo, redis, {
+    tenant,
+    after: comment,
+  });
+
+  // Publish changes to the event publisher.
+  await publishChanges(publisher, {
+    ...counts,
+    after: comment,
+    moderatorID: null,
+  });
+
+  // If this is a reply, publish it.
+  if (input.parentID) {
+    publishCommentReplyCreated(publisher, comment);
+  }
+
+  // If this comment is visible (and not a reply), publish it.
+  if (!input.parentID && hasPublishedStatus(comment)) {
+    publishCommentCreated(publisher, comment);
+  }
+
+  return comment;
+}
diff --git a/src/core/server/stacks/editComment.ts b/src/core/server/stacks/editComment.ts
new file mode 100644
index 000000000..dee56eb2e
--- /dev/null
+++ b/src/core/server/stacks/editComment.ts
@@ -0,0 +1,212 @@
+import { DateTime } from "luxon";
+import { Db } from "mongodb";
+
+import { Omit } from "coral-common/types";
+import { Config } from "coral-server/config";
+import { CommentNotFoundError, StoryNotFoundError } from "coral-server/errors";
+import { Publisher } from "coral-server/graph/tenant/subscriptions/publisher";
+import logger from "coral-server/logger";
+import {
+  encodeActionCounts,
+  filterDuplicateActions,
+} from "coral-server/models/action/comment";
+import { createCommentModerationAction } from "coral-server/models/action/moderation/comment";
+import {
+  editComment,
+  EditCommentInput,
+  retrieveComment,
+  validateEditable,
+} from "coral-server/models/comment";
+import { retrieveStory } from "coral-server/models/story";
+import { Tenant } from "coral-server/models/tenant";
+import { User } from "coral-server/models/user";
+import {
+  addCommentActions,
+  CreateAction,
+} from "coral-server/services/comments/actions";
+import { processForModeration } from "coral-server/services/comments/pipeline";
+import { AugmentedRedis } from "coral-server/services/redis";
+import { Request } from "coral-server/types/express";
+
+import { publishChanges, updateAllCounts } from "./helpers";
+
+/**
+ * getLastCommentEditableUntilDate will return the `createdAt` date that will
+ * represent the _oldest_ date that a comment could have been created on in
+ * order to still be editable.
+ *
+ * @param tenant the tenant that contains settings related editing
+ * @param now the date that is the base, defaulting to the current time
+ */
+function getLastCommentEditableUntilDate(
+  tenant: Pick<Tenant, "editCommentWindowLength">,
+  now = new Date()
+): Date {
+  return (
+    DateTime.fromJSDate(now)
+      // editCommentWindowLength is in seconds, so multiply by 1000 to get
+      // milliseconds.
+      .minus(tenant.editCommentWindowLength * 1000)
+      .toJSDate()
+  );
+}
+
+export type EditComment = Omit<
+  EditCommentInput,
+  "status" | "authorID" | "lastEditableCommentCreatedAt" | "metadata"
+>;
+
+export default async function edit(
+  mongo: Db,
+  redis: AugmentedRedis,
+  config: Config,
+  publisher: Publisher,
+  tenant: Tenant,
+  author: User,
+  input: EditComment,
+  now = new Date(),
+  req?: Request
+) {
+  let log = logger.child({ commentID: input.id, tenantID: tenant.id }, true);
+
+  // Get the comment that we're editing. This comment is considered stale,
+  // because it wasn't involved in the atomic transaction.
+  const originalStaleComment = await retrieveComment(
+    mongo,
+    tenant.id,
+    input.id
+  );
+  if (!originalStaleComment) {
+    throw new CommentNotFoundError(input.id);
+  }
+
+  // The editable time is based on the current time, and the edit window
+  // length. By subtracting the current date from the edit window length, we
+  // get the maximum value for the `createdAt` time that would be permitted
+  // for the comment edit to succeed.
+  const lastEditableCommentCreatedAt = getLastCommentEditableUntilDate(
+    tenant,
+    now
+  );
+
+  // Validate and potentially return with a more useful error.
+  validateEditable(originalStaleComment, {
+    authorID: author.id,
+    lastEditableCommentCreatedAt,
+  });
+
+  // Grab the story that we'll use to check moderation pieces with.
+  const story = await retrieveStory(
+    mongo,
+    tenant.id,
+    originalStaleComment.storyID
+  );
+  if (!story) {
+    throw new StoryNotFoundError(originalStaleComment.storyID);
+  }
+
+  // Run the comment through the moderation phases.
+  const { body, status, metadata, actions } = await processForModeration({
+    action: "EDIT",
+    log,
+    mongo,
+    redis,
+    config,
+    story,
+    tenant,
+    comment: input,
+    author,
+    req,
+    now,
+  });
+
+  let actionCounts = {};
+  if (actions.length > 0) {
+    // Encode the new action counts that are going to be added to the new
+    // revision.
+    actionCounts = encodeActionCounts(...filterDuplicateActions(actions));
+  }
+
+  log.trace(
+    { predictedActionCounts: actionCounts },
+    "associating action counts with comment"
+  );
+
+  // Perform the edit.
+  const result = await editComment(
+    mongo,
+    tenant.id,
+    {
+      id: input.id,
+      authorID: author.id,
+      body,
+      status,
+      metadata,
+      actionCounts,
+      lastEditableCommentCreatedAt,
+    },
+    now
+  );
+  if (!result) {
+    throw new CommentNotFoundError(input.id);
+  }
+
+  log = log.child({ revisionID: result.revision.id }, true);
+
+  if (actions.length > 0) {
+    // Insert and handle creating the actions.
+    const upsertedActions = await addCommentActions(
+      mongo,
+      tenant,
+      actions.map(
+        (action): CreateAction => ({
+          ...action,
+          commentID: result.after.id,
+          commentRevisionID: result.revision.id,
+          storyID: story.id,
+        })
+      ),
+      now
+    );
+
+    log.trace(
+      {
+        actualActionCounts: encodeActionCounts(...upsertedActions),
+        actions: upsertedActions.length,
+      },
+      "added actions to comment"
+    );
+  }
+
+  // If the comment status changed as a result of a pipeline operation, create a
+  // moderation action.
+  if (result.before.status !== result.after.status) {
+    await createCommentModerationAction(
+      mongo,
+      tenant.id,
+      {
+        commentID: result.after.id,
+        commentRevisionID: result.revision.id,
+        status: result.after.status,
+        moderatorID: null,
+      },
+      now
+    );
+  }
+
+  // Update all the comment counts on stories and users.
+  const counts = await updateAllCounts(mongo, redis, {
+    tenant,
+    ...result,
+  });
+
+  // Publish changes to the event publisher.
+  await publishChanges(publisher, {
+    ...result,
+    ...counts,
+    moderatorID: null,
+  });
+
+  // Return the resulting comment.
+  return result.after;
+}
diff --git a/src/core/server/stacks/helpers/index.ts b/src/core/server/stacks/helpers/index.ts
new file mode 100644
index 000000000..97c4e8d71
--- /dev/null
+++ b/src/core/server/stacks/helpers/index.ts
@@ -0,0 +1,2 @@
+export { default as publishChanges } from "./publishChanges";
+export { default as updateAllCounts } from "./updateAllCounts";
diff --git a/src/core/server/stacks/helpers/publishChanges.ts b/src/core/server/stacks/helpers/publishChanges.ts
new file mode 100644
index 000000000..a42595f54
--- /dev/null
+++ b/src/core/server/stacks/helpers/publishChanges.ts
@@ -0,0 +1,43 @@
+import { Publisher } from "coral-server/graph/tenant/subscriptions/publisher";
+import {
+  Comment,
+  hasModeratorStatus,
+  hasPublishedStatus,
+} from "coral-server/models/comment";
+import { CommentModerationQueueCounts } from "coral-server/models/story";
+import {
+  publishCommentReleased,
+  publishCommentStatusChanges,
+  publishModerationQueueChanges,
+} from "coral-server/services/events";
+
+interface PublishChangesInput {
+  before?: Readonly<Comment>;
+  after: Readonly<Comment>;
+  moderationQueue: CommentModerationQueueCounts;
+  moderatorID: string | null;
+}
+
+export default async function publishChanges(
+  publish: Publisher,
+  input: PublishChangesInput
+) {
+  // Publish changes.
+  publishModerationQueueChanges(publish, input.moderationQueue, input.after);
+
+  // If this was a change, and it has a "before" state for the comment, process
+  // those updates too.
+  if (input.before) {
+    publishCommentStatusChanges(
+      publish,
+      input.before.status,
+      input.after.status,
+      input.after.id,
+      input.moderatorID
+    );
+
+    if (hasModeratorStatus(input.before) && hasPublishedStatus(input.after)) {
+      publishCommentReleased(publish, input.after);
+    }
+  }
+}
diff --git a/src/core/server/stacks/helpers/updateAllCounts.ts b/src/core/server/stacks/helpers/updateAllCounts.ts
new file mode 100644
index 000000000..803f4cfd8
--- /dev/null
+++ b/src/core/server/stacks/helpers/updateAllCounts.ts
@@ -0,0 +1,86 @@
+import { Db } from "mongodb";
+
+import { Comment, CommentStatusCounts } from "coral-server/models/comment";
+import {
+  CommentModerationQueueCounts,
+  updateStoryCounts,
+} from "coral-server/models/story";
+import { Tenant } from "coral-server/models/tenant";
+import { updateUserCommentCounts } from "coral-server/models/user";
+import {
+  calculateCounts,
+  calculateCountsDiff,
+} from "coral-server/services/comments/moderation";
+import { AugmentedRedis } from "coral-server/services/redis";
+
+interface UpdateAllCountsInput {
+  tenant: Readonly<Tenant>;
+  before?: Readonly<Comment>;
+  after: Readonly<Comment>;
+}
+
+function calculateModerationQueue(
+  input: UpdateAllCountsInput
+): CommentModerationQueueCounts {
+  if (input.before) {
+    return calculateCountsDiff(input.before, input.after);
+  }
+
+  return calculateCounts(input.after);
+}
+
+function calculateStatus(
+  input: UpdateAllCountsInput
+): Partial<CommentStatusCounts> {
+  if (input.before) {
+    if (input.before.status !== input.after.status) {
+      return {
+        [input.before.status]: -1,
+        [input.after.status]: 1,
+      };
+    }
+
+    return {};
+  }
+
+  return {
+    [input.after.status]: 1,
+  };
+}
+
+export default async function updateAllCounts(
+  mongo: Db,
+  redis: AugmentedRedis,
+  input: UpdateAllCountsInput
+) {
+  // Compute the queue difference as a result of the old status and the new
+  // status and the action counts.
+  const moderationQueue = calculateModerationQueue(input);
+
+  // Compute the status changes as a result of the change to the comment status.
+  const status = calculateStatus(input);
+
+  // Pull out some params from the input for easier usage.
+  const {
+    tenant,
+    after: { storyID, authorID },
+  } = input;
+
+  // Update the story comment counts.
+  await updateStoryCounts(mongo, redis, tenant.id, storyID, {
+    status,
+    moderationQueue,
+  });
+
+  if (authorID) {
+    // Update the user comment counts.
+    await updateUserCommentCounts(mongo, tenant.id, authorID, {
+      status,
+    });
+  }
+
+  return {
+    status,
+    moderationQueue,
+  };
+}
diff --git a/src/core/server/stacks/index.ts b/src/core/server/stacks/index.ts
new file mode 100644
index 000000000..6daf4527a
--- /dev/null
+++ b/src/core/server/stacks/index.ts
@@ -0,0 +1,4 @@
+export { default as approveComment } from "./approveComment";
+export { default as createComment } from "./createComment";
+export { default as editComment } from "./editComment";
+export { default as rejectComment } from "./rejectComment";
diff --git a/src/core/server/stacks/rejectComment.ts b/src/core/server/stacks/rejectComment.ts
new file mode 100644
index 000000000..e77c1eae1
--- /dev/null
+++ b/src/core/server/stacks/rejectComment.ts
@@ -0,0 +1,62 @@
+import { Db } from "mongodb";
+
+import { Publisher } from "coral-server/graph/tenant/subscriptions/publisher";
+import { hasTag } from "coral-server/models/comment";
+import { Tenant } from "coral-server/models/tenant";
+import { removeTag } from "coral-server/services/comments";
+import { moderate } from "coral-server/services/comments/moderation";
+import { AugmentedRedis } from "coral-server/services/redis";
+
+import {
+  GQLCOMMENT_STATUS,
+  GQLTAG,
+} from "coral-server/graph/tenant/schema/__generated__/types";
+
+import { publishChanges, updateAllCounts } from "./helpers";
+
+const rejectComment = async (
+  mongo: Db,
+  redis: AugmentedRedis,
+  publisher: Publisher,
+  tenant: Tenant,
+  commentID: string,
+  commentRevisionID: string,
+  moderatorID: string,
+  now: Date
+) => {
+  // Reject the comment.
+  const result = await moderate(
+    mongo,
+    tenant,
+    {
+      commentID,
+      commentRevisionID,
+      moderatorID,
+      status: GQLCOMMENT_STATUS.REJECTED,
+    },
+    now
+  );
+
+  // Update all the comment counts on stories and users.
+  const counts = await updateAllCounts(mongo, redis, {
+    tenant,
+    ...result,
+  });
+
+  // Publish changes to the event publisher.
+  await publishChanges(publisher, {
+    ...result,
+    ...counts,
+    moderatorID,
+  });
+
+  // If there was a featured tag on this comment, remove it.
+  if (hasTag(result.after, GQLTAG.FEATURED)) {
+    return removeTag(mongo, tenant, result.after.id, GQLTAG.FEATURED);
+  }
+
+  // Return the resulting comment.
+  return result.after;
+};
+
+export default rejectComment;