[CORL-649] Migrations (#2597)

* feat: added migration framework

* chore: added premod user status migration

* feat: enhanced error handling of migrations

* fix: added missing argument from abstract method

* fix: another templating blunder

* fix: removed debug code

* feat: enhanced migration tracking

* fix: remove skipping migrations

* feat: moved indexing to migration system

* fix: linting

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

@@ -17,16 +17,12 @@ import logger from "coral-server/logger";
 import {
   Connection,
   ConnectionInput,
-  createCollection,
-  createConnectionOrderVariants,
-  createIndexFactory,
   FilterQuery,
   Query,
   resolveConnection,
 } from "coral-server/models/helpers";
 import { TenantResource } from "coral-server/models/tenant";
-
-const collection = createCollection<CommentAction>("commentActions");
+import { commentActions as collection } from "coral-server/services/mongodb/collections";
 
 export enum ACTION_TYPE {
   /**
@@ -134,32 +130,6 @@ export interface CommentAction extends TenantResource {
   metadata?: Record<string, any>;
 }
 
-export async function createCommentActionIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ tenantID: 1, id: 1 }, { unique: true });
-
-  // { actionType, commentID, userID }
-  await createIndex(
-    { tenantID: 1, actionType: 1, commentID: 1, userID: 1 },
-    { background: true }
-  );
-
-  const variants = createConnectionOrderVariants<Readonly<CommentAction>>(
-    [{ createdAt: -1 }],
-    { background: true }
-  );
-
-  // Connection pagination.
-  // { ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    actionType: 1,
-    commentID: 1,
-  });
-}
-
 const ActionSchema = [
   // Flags
   {

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

@@ -6,33 +6,11 @@ import { GQLCOMMENT_STATUS } from "coral-server/graph/tenant/schema/__generated_
 import {
   Connection,
   ConnectionInput,
-  createCollection,
-  createConnectionOrderVariants,
-  createIndexFactory,
   Query,
   resolveConnection,
 } from "coral-server/models/helpers";
 import { TenantResource } from "coral-server/models/tenant";
-
-const collection = createCollection<CommentModerationAction>(
-  "commentModerationActions"
-);
-
-export async function createCommentModerationActionIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ tenantID: 1, id: 1 }, { unique: true });
-
-  const createVariants = createConnectionOrderVariants<
-    Readonly<CommentModerationAction>
-  >([{ createdAt: -1 }]);
-
-  // { moderatorID, ...connectionParams }
-  await createVariants(createIndex, {
-    moderatorID: 1,
-  });
-}
+import { commentModerationActions as collection } from "coral-server/services/mongodb/collections";
 
 /**
  * CommentModerationAction stores information around a moderation action that

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

@@ -19,10 +19,7 @@ import {
 } from "coral-server/models/action/comment";
 import {
   Connection,
-  createCollection,
   createConnection,
-  createConnectionOrderVariants,
-  createIndexFactory,
   doesNotContainNull,
   FilterQuery,
   nodesToEdges,
@@ -32,6 +29,7 @@ import {
   resolveConnection,
 } from "coral-server/models/helpers";
 import { TenantResource } from "coral-server/models/tenant";
+import { comments as collection } from "coral-server/services/mongodb/collections";
 
 import { PUBLISHED_STATUSES } from "./constants";
 import {
@@ -42,8 +40,6 @@ import {
 import { Revision } from "./revision";
 import { CommentTag } from "./tag";
 
-const collection = createCollection<Comment>("comments");
-
 /**
  * Comment's are created by User's on Stories. Each Comment contains a body, and
  * can be moderated by another Moderator or Admin User.
@@ -128,84 +124,6 @@ export interface Comment extends TenantResource {
   deletedAt?: Date;
 }
 
-export async function createCommentIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ tenantID: 1, id: 1 }, { unique: true });
-
-  // Facility for counting the tags against a story.
-  await createIndex(
-    {
-      tenantID: 1,
-      storyID: 1,
-      "tags.type": 1,
-      status: 1,
-    },
-    {
-      background: true,
-      partialFilterExpression: {
-        "tags.type": { $exists: true },
-      },
-    }
-  );
-
-  const variants = createConnectionOrderVariants<Readonly<Comment>>([
-    { createdAt: -1 },
-    { createdAt: 1 },
-    { childCount: -1, createdAt: -1 },
-    { "actionCounts.REACTION": -1, createdAt: -1 },
-  ]);
-
-  // Story based Comment Connection pagination.
-  // { storyID, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    storyID: 1,
-    status: 1,
-  });
-
-  // Moderation based Comment Connection pagination.
-  // { storyID, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    status: 1,
-  });
-
-  // Story based Comment Connection pagination that are flagged.
-  // { storyID, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    storyID: 1,
-    status: 1,
-    "actionCounts.FLAG": 1,
-  });
-
-  // Story + Reply based Comment Connection pagination.
-  // { storyID, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    storyID: 1,
-    parentID: 1,
-    status: 1,
-  });
-
-  // Author based Comment Connection pagination.
-  // { authorID, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    authorID: 1,
-    status: 1,
-  });
-
-  // Tag based Comment Connection pagination.
-  // { tags.type, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    "tags.type": 1,
-  });
-}
-
 export type CreateCommentInput = Omit<
   Comment,
   | "id"

src/core/server/models/helpers/index.ts

@@ -1,5 +1,4 @@
 export * from "./collection";
 export * from "./connection";
-export * from "./indexing";
 export { default as Query } from "./query";
 export * from "./query";

src/core/server/models/helpers/indexing.ts

@@ -1,79 +0,0 @@
-import { merge } from "lodash";
-import { Collection, IndexOptions } from "mongodb";
-
-import { Writable } from "coral-common/types";
-import logger from "coral-server/logger";
-
-type IndexType = 1 | -1 | "text";
-
-export type IndexSpecification<T> = {
-  [P in keyof Writable<Partial<T>>]: IndexType
-} &
-  Record<string, IndexType>;
-
-type IndexCreationFunction<T> = (
-  indexSpec: IndexSpecification<T>,
-  indexOptions?: IndexOptions
-) => Promise<string>;
-
-export function createIndexFactory<T>(
-  collection: Collection<T>
-): IndexCreationFunction<T> {
-  const log = logger.child(
-    {
-      collectionName: collection.collectionName,
-    },
-    true
-  );
-
-  return async (
-    indexSpec: IndexSpecification<T>,
-    indexOptions: IndexOptions = {}
-  ) => {
-    try {
-      // Try to create the index.
-      const indexName = await collection.createIndex(indexSpec, indexOptions);
-      log.debug({ indexName, indexSpec, indexOptions }, "index was created");
-
-      // Match the interface from the `createIndex` function by returning the
-      // index name.
-      return indexName;
-    } catch (err) {
-      log.error({ err, indexSpec, indexOptions }, "could not create index");
-
-      // Rethrow the error here.
-      throw err;
-    }
-  };
-}
-
-export function createConnectionOrderVariants<T>(
-  variants: Array<IndexSpecification<T>>,
-  indexOptions: IndexOptions = {}
-) {
-  return async (
-    createIndex: IndexCreationFunction<T>,
-    indexSpec: IndexSpecification<T>,
-    variantIndexOptions: IndexOptions = {}
-  ) => {
-    /**
-     * createIndexVariant will create a variant on the specified `indexSpec` that
-     * will include the new variation.
-     *
-     * @param variantSpec the spec that makes this variant different
-     */
-    const createIndexVariant = (variantSpec: IndexSpecification<T>) =>
-      createIndex(
-        merge({}, indexSpec, variantSpec),
-        merge({}, indexOptions, variantIndexOptions)
-      );
-
-    // Create a raw index without the variants applied.
-    await createIndex(indexSpec, merge({}, indexOptions, variantIndexOptions));
-
-    // Create all the variants.
-    for (const variant of variants) {
-      await createIndexVariant(variant);
-    }
-  };
-}

src/core/server/models/invite.ts

@@ -3,23 +3,8 @@ import uuid from "uuid";
 
 import { Omit, Sub } from "coral-common/types";
 import { GQLUSER_ROLE } from "coral-server/graph/tenant/schema/__generated__/types";
-import {
-  createCollection,
-  createIndexFactory,
-} from "coral-server/models/helpers";
 import { TenantResource } from "coral-server/models/tenant";
-
-const collection = createCollection<Readonly<Invite>>("invites");
-
-export async function createInviteIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ tenantID: 1, id: 1 }, { unique: true });
-
-  // UNIQUE { email }
-  await createIndex({ tenantID: 1, email: 1 }, { unique: true });
-}
+import { invites as collection } from "coral-server/services/mongodb/collections";
 
 export interface Invite extends TenantResource {
   readonly id: string;

src/core/server/models/migration/index.ts

@@ -0,0 +1 @@
+export * from "./migration";

src/core/server/models/migration/migration.ts

@@ -0,0 +1,96 @@
+import { Db } from "mongodb";
+
+import { migrations as collection } from "coral-server/services/mongodb/collections";
+
+export enum MIGRATION_STATE {
+  STARTED = "STARTED",
+  FAILED = "FAILED",
+  FINISHED = "FINISHED",
+}
+
+export interface MigrationRecord {
+  id: number;
+  state: MIGRATION_STATE;
+  clientID?: string;
+  createdAt: Date;
+  updatedAt?: Date;
+}
+
+export async function startMigration(
+  mongo: Db,
+  id: number,
+  clientID: string,
+  now = new Date()
+) {
+  const result = await collection(mongo).findOneAndUpdate(
+    { id },
+    {
+      $setOnInsert: {
+        id,
+        clientID,
+        state: MIGRATION_STATE.STARTED,
+        createdAt: now,
+      },
+    },
+    {
+      // False to return the updated document instead of the original document.
+      returnOriginal: false,
+      upsert: true,
+    }
+  );
+  if (!result.value) {
+    throw new Error("an unexpected error occurred");
+  }
+
+  return result.value;
+}
+
+/**
+ * updateMigrationState will update the state of a migration record to reflect
+ * the new state as well as un-setting the client ID from the records.
+ *
+ * @param mongo the database to interact on
+ * @param id the migration version to update
+ * @param state the state to switch the record to
+ * @param now the current date
+ */
+async function updateMigrationState(
+  mongo: Db,
+  id: number,
+  state: MIGRATION_STATE.FINISHED | MIGRATION_STATE.FAILED,
+  now: Date
+) {
+  const result = await collection(mongo).findOneAndUpdate(
+    { id },
+    {
+      $set: {
+        state,
+        updatedAt: now,
+      },
+      $unset: {
+        clientID: "",
+      },
+    },
+    {
+      // False to return the updated document instead of the original document.
+      returnOriginal: false,
+    }
+  );
+
+  return result.value || null;
+}
+
+export async function finishMigration(mongo: Db, id: number, now = new Date()) {
+  return updateMigrationState(mongo, id, MIGRATION_STATE.FINISHED, now);
+}
+
+export async function failMigration(mongo: Db, id: number, now = new Date()) {
+  return updateMigrationState(mongo, id, MIGRATION_STATE.FAILED, now);
+}
+
+export async function retrieveAllMigrationRecords(mongo: Db) {
+  const cursor = await collection(mongo)
+    .find({})
+    .sort({ id: 1 });
+  return cursor.toArray();
+}

src/core/server/models/queries/queries.ts

@@ -3,12 +3,7 @@ import { Db, MongoError } from "mongodb";
 
 import { waitFor } from "coral-common/helpers";
 import logger from "coral-server/logger";
-import {
-  createCollection,
-  createIndexFactory,
-} from "coral-server/models/helpers";
-
-const collection = createCollection<Readonly<PersistedQuery>>("queries");
+import { queries as collection } from "coral-server/services/mongodb/collections";
 
 export interface PersistedQuery {
   id: string;
@@ -19,13 +14,6 @@ export interface PersistedQuery {
   version: string;
 }
 
-export async function createQueriesIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ id: 1 }, { unique: true });
-}
-
 export async function primeQueries(
   mongo: Db,
   queries: PersistedQuery[],

src/core/server/models/story/counts/index.ts

@@ -13,28 +13,12 @@ import {
   CommentStatusCounts,
   createEmptyCommentStatusCounts,
 } from "coral-server/models/comment/helpers";
-import {
-  createCollection,
-  createIndexFactory,
-} from "coral-server/models/helpers";
 import { retrieveStory, Story } from "coral-server/models/story";
+import { stories as collection } from "coral-server/services/mongodb/collections";
 import { AugmentedRedis } from "coral-server/services/redis";
 
 import { updateSharedCommentCounts } from "./shared";
 
-/**
- * collection provides a reference to the stories collection used by the
- * counting system.
- */
-const collection = createCollection<Story>("stories");
-
-export async function createStoryCountIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // { createdAt }
-  await createIndex({ tenantID: 1, createdAt: 1 }, { background: true });
-}
-
 /**
  * CommentModerationCountsPerQueue stores the number of Comments that exist in
  * each of the Moderation Queues.

src/core/server/models/story/counts/shared.ts

@@ -8,12 +8,11 @@ import {
   CommentStatusCounts,
   createEmptyCommentStatusCounts,
 } from "coral-server/models/comment/helpers";
-import { createCollection } from "coral-server/models/helpers";
-import { Story } from "coral-server/models/story";
 import {
   CommentModerationCountsPerQueue,
   StoryCounts,
 } from "coral-server/models/story/counts";
+import { stories as collection } from "coral-server/services/mongodb/collections";
 import { AugmentedPipeline, AugmentedRedis } from "coral-server/services/redis";
 
 import {
@@ -46,12 +45,6 @@ const commentCountsModerationQueueTotalKey = (tenantID: string) =>
 const commentCountsModerationQueueQueuesKey = (tenantID: string) =>
   `${tenantID}:commentCounts:moderationQueue:queues`;
 
-/**
- * collection provides a reference to the stories collection used by the
- * counting system.
- */
-const collection = createCollection<Story>("stories");
-
 /**
  * recalculateSharedModerationQueueQueueCounts will reset the counts stored for
  * this Tenant.

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

@@ -14,16 +14,14 @@ import {
 import {
   Connection,
   ConnectionInput,
-  createConnectionOrderVariants,
-  createIndexFactory,
   Query,
   resolveConnection,
 } from "coral-server/models/helpers";
 import { GlobalModerationSettings } from "coral-server/models/settings";
 import { TenantResource } from "coral-server/models/tenant";
+import { stories as collection } from "coral-server/services/mongodb/collections";
 
 import { createEmptyCommentStatusCounts } from "../comment/helpers";
-import { createCollection } from "../helpers/collection";
 import {
   createEmptyCommentModerationQueueCounts,
   StoryCommentCounts,
@@ -32,8 +30,6 @@ import {
 export * from "./counts";
 export * from "./helpers";
 
-const collection = createCollection<Story>("stories");
-
 export type StorySettings = DeepPartial<
   Pick<GQLStorySettings, "messageBox"> & GlobalModerationSettings
 >;
@@ -81,40 +77,6 @@ export interface Story extends TenantResource {
   createdAt: Date;
 }
 
-export async function createStoryIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ tenantID: 1, id: 1 }, { unique: true });
-
-  // UNIQUE { url }
-  await createIndex({ tenantID: 1, url: 1 }, { unique: true });
-
-  // TEXT { $**, createdAt }
-  await createIndex(
-    { tenantID: 1, "$**": "text", createdAt: -1 },
-    { background: true }
-  );
-
-  const variants = createConnectionOrderVariants<Readonly<Story>>(
-    [{ createdAt: -1 }],
-    { background: true }
-  );
-
-  // Story Connection pagination.
-  // { ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-  });
-
-  // Closed At ordered Story Connection pagination.
-  // { closedAt, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    closedAt: 1,
-  });
-}
-
 export interface UpsertStoryInput {
   id?: string;
   url: string;

src/core/server/models/tenant/tenant.ts

@@ -9,15 +9,11 @@ import {
   GQLMODERATION_MODE,
   GQLSettings,
 } from "coral-server/graph/tenant/schema/__generated__/types";
-import {
-  createCollection,
-  createIndexFactory,
-} from "coral-server/models/helpers";
 import { Settings } from "coral-server/models/settings";
 import { I18n } from "coral-server/services/i18n";
-import { generateSSOKey, getDefaultReactionConfiguration } from "./helpers";
+import { tenants as collection } from "coral-server/services/mongodb/collections";
 
-const collection = createCollection<Tenant>("tenants");
+import { generateSSOKey, getDefaultReactionConfiguration } from "./helpers";
 
 /**
  * TenantResource references a given resource that should be owned by a specific
@@ -46,16 +42,6 @@ export interface TenantSettings
  */
 export type Tenant = Settings & TenantSettings;
 
-export async function createTenantIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ id: 1 }, { unique: true });
-
-  // UNIQUE { domain }
-  await createIndex({ domain: 1 }, { unique: true });
-}
-
 /**
  * CreateTenantInput is the set of properties that can be set when a given
  * Tenant is created. The remainder of the properties are set from defaults and

src/core/server/models/user/user.ts

@@ -33,19 +33,15 @@ import logger from "coral-server/logger";
 import {
   Connection,
   ConnectionInput,
-  createCollection,
-  createConnectionOrderVariants,
-  createIndexFactory,
   Query,
   resolveConnection,
 } from "coral-server/models/helpers";
 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 { getLocalProfile, hasLocalProfile } from "./helpers";
 
-const collection = createCollection<User>("users");
-
 export interface LocalProfile {
   type: "local";
   id: string;
@@ -297,10 +293,8 @@ export interface UserStatus {
   /**
    * premod stores whether a user is set to mandatory premod and history of
    * premod status.
-   *
-   * FIXME: (wyattjoh) set defaults during migration
    */
-  premod?: PremodStatus;
+  premod: PremodStatus;
 }
 
 /**
@@ -436,80 +430,6 @@ export interface User extends TenantResource {
   deletedAt?: Date;
 }
 
-export async function createUserIndexes(mongo: Db) {
-  const createIndex = createIndexFactory(collection(mongo));
-
-  // UNIQUE { id }
-  await createIndex({ tenantID: 1, id: 1 }, { unique: true });
-
-  // UNIQUE - PARTIAL { email }
-  await createIndex(
-    { tenantID: 1, email: 1 },
-    { unique: true, partialFilterExpression: { email: { $exists: true } } }
-  );
-
-  // UNIQUE { profiles.type, profiles.id }
-  await createIndex(
-    { tenantID: 1, "profiles.type": 1, "profiles.id": 1 },
-    {
-      unique: true,
-      partialFilterExpression: { "profiles.id": { $exists: true } },
-    }
-  );
-
-  // { profiles }
-  await createIndex(
-    { tenantID: 1, profiles: 1, email: 1 },
-    {
-      partialFilterExpression: { profiles: { $exists: true } },
-      background: true,
-    }
-  );
-
-  // TEXT { id, username, email, createdAt }
-  await createIndex(
-    {
-      tenantID: 1,
-      id: "text",
-      username: "text",
-      email: "text",
-      createdAt: -1,
-    },
-    { background: true }
-  );
-
-  const variants = createConnectionOrderVariants<Readonly<User>>(
-    [{ createdAt: -1 }],
-    { background: true }
-  );
-
-  // User Connection pagination.
-  // { ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-  });
-
-  // Role based User Connection pagination.
-  // { role, ...connectionParams }
-  await variants(createIndex, {
-    tenantID: 1,
-    role: 1,
-  });
-
-  // Suspension based User Connection pagination.
-  await variants(createIndex, {
-    tenantID: 1,
-    "status.suspension.history.from.start": 1,
-    "status.suspension.history.from.finish": 1,
-  });
-
-  // Ban based User Connection pagination.
-  await variants(createIndex, {
-    tenantID: 1,
-    "status.ban.active": 1,
-  });
-}
-
 function hashPassword(password: string): Promise<string> {
   return bcrypt.hash(password, 10);
 }
@@ -1460,8 +1380,7 @@ export async function premodUser(
 
     // Check to see if the user is already banned.
     const premod = consolidateUserPremodStatus(user.status.premod);
-    // FIXME: (wyattjoh) once migration has been performed, remove check
-    if (premod && premod.active) {
+    if (premod.active) {
       throw new UserAlreadyPremoderated();
     }
 
@@ -1873,8 +1792,7 @@ export function consolidateUserSuspensionStatus(
 export interface ConsolidatedUserStatus {
   suspension: ConsolidatedSuspensionStatus;
   ban: ConsolidatedBanStatus;
-  // FIXME: (wyattjoh) once migration has been performed, make required
-  premod?: ConsolidatedPremodStatus;
+  premod: ConsolidatedPremodStatus;
 }
 
 export function consolidateUserStatus(