[next] MongoDB Indexes (#2142)

* feat: added mongo indexing support

* fix: fixed typescript issue

* chore: better types

* fix: revert debug stuff

* fix: addressed ts error

* feat: added config option to disable auto-indexing

* chore: reordered imports

* refactor: cleaned up some filepaths

src/core/server/models/helpers/connection.spec.ts

@@ -0,0 +1,59 @@
+import { getPageInfo } from "./connection";
+
+it("handles when there is none requested", () => {
+  const pageInfo = getPageInfo({ first: 0 }, []);
+
+  expect(pageInfo).toEqual({
+    hasNextPage: false,
+    hasPreviousPage: false,
+    startCursor: null,
+    endCursor: null,
+  });
+});
+
+it("handles when there is no edges", () => {
+  const pageInfo = getPageInfo({ first: 10 }, []);
+
+  expect(pageInfo).toEqual({
+    hasNextPage: false,
+    hasPreviousPage: false,
+    startCursor: null,
+    endCursor: null,
+  });
+});
+
+it("handles when there is more edges than requested", () => {
+  const pageInfo = getPageInfo({ first: 1 }, [
+    {
+      node: null,
+      cursor: 1,
+    },
+    {
+      node: null,
+      cursor: 2,
+    },
+  ]);
+
+  expect(pageInfo).toEqual({
+    hasNextPage: true,
+    hasPreviousPage: false,
+    startCursor: null,
+    endCursor: 1,
+  });
+});
+
+it("handles when there is exactly as many edges as requested", () => {
+  const pageInfo = getPageInfo({ first: 1 }, [
+    {
+      node: null,
+      cursor: 1,
+    },
+  ]);
+
+  expect(pageInfo).toEqual({
+    hasNextPage: false,
+    hasPreviousPage: false,
+    startCursor: null,
+    endCursor: 1,
+  });
+});

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

@@ -0,0 +1,119 @@
+import { merge } from "lodash";
+
+import { FilterQuery } from "talk-server/models/helpers/query";
+
+export type Cursor = Date | number | string | null;
+
+export interface Edge<T> {
+  node: T;
+  cursor: Cursor;
+}
+
+export interface PageInfo {
+  hasNextPage?: boolean;
+  hasPreviousPage?: boolean;
+  startCursor?: Cursor;
+  endCursor?: Cursor;
+}
+
+export interface Connection<T> {
+  edges: Array<Edge<T>>;
+  pageInfo: PageInfo;
+}
+
+export interface ConnectionInput<T> {
+  /**
+   * first is the number of items to load for the connection. The returned
+   * amount of items may be less.
+   */
+  first: number;
+
+  /**
+   * after is an optional cursor that can be used to paginate the result set.
+   */
+  after?: Cursor;
+
+  /**
+   * filter is an optional query that can be used to constrain the result set.
+   */
+  filter?: FilterQuery<T>;
+}
+
+export interface OrderedConnectionInput<T, U> extends ConnectionInput<T> {
+  /**
+   * orderBy allows ordering of the returned connection.
+   */
+  orderBy: U;
+}
+
+/**
+ * createConnection will create a base Connection that can be used to satisfy
+ * the Connection<T> interface.
+ *
+ * @param connection the base connection to optionally merge with the default base
+ * connection details.
+ */
+export function createConnection<T>(
+  connection: Partial<Connection<T>> = {}
+): Connection<T> {
+  return merge(
+    {
+      edges: [],
+      pageInfo: {
+        hasNextPage: false,
+        hasPreviousPage: false,
+      },
+    },
+    connection
+  );
+}
+
+export interface PaginationArgs {
+  first: number;
+}
+
+export function getPageInfo<T>(args: PaginationArgs, edges: Array<Edge<T>>) {
+  const pageInfo: PageInfo = {
+    hasNextPage: false,
+    hasPreviousPage: false,
+    startCursor: null,
+    endCursor: null,
+  };
+
+  if (edges.length === 0) {
+    // If there are no edges, then there's nothing to paginate!
+    return pageInfo;
+  }
+
+  // The hasNextPage is always handled the same (ask for one more than we need,
+  // if there is one more, than there is more).
+  if (args.first >= 0 && edges.length > args.first) {
+    // There was one more than we expected! Set hasNextPage = true.
+    pageInfo.hasNextPage = true;
+  }
+
+  if (pageInfo.hasNextPage && edges.length > 1) {
+    // There was more than one expected, and there is two entries in the edges
+    // array. We should grab the second last one because the last one will have
+    // to be trimmed off after.
+    pageInfo.endCursor = edges[edges.length - 2].cursor;
+  } else {
+    // There was not more than expected, so we should just grab the last edge to
+    // get the endCursor.
+    pageInfo.endCursor = edges[edges.length - 1].cursor;
+  }
+
+  return pageInfo;
+}
+
+export type NodeToCursorTransformer<T> = (node: T, index: number) => Cursor;
+
+export function nodesToEdges<T>(
+  nodes: T[],
+  transformer: NodeToCursorTransformer<T>
+): Array<Edge<T>> {
+  return nodes.map((node, index) => ({
+    node,
+    cursor: transformer(node, index),
+  }));
+}

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

@@ -0,0 +1,164 @@
+import { merge } from "lodash";
+import {
+  Collection,
+  Cursor,
+  FilterQuery as MongoFilterQuery,
+  IndexOptions,
+} from "mongodb";
+
+import { Writeable } from "talk-common/types";
+import logger from "talk-server/logger";
+
+/**
+ * FilterQuery<T> ensures that given the type T, that the FilterQuery will be a
+ * writeable, partial set of properties while also including MongoDB specific
+ * properties (like $lt, or $gte).
+ */
+export type FilterQuery<T> = MongoFilterQuery<Writeable<Partial<T>>>;
+
+/**
+ * Query is a convenience class used to wrap the existing MongoDB driver to
+ * provide easier complex query management.
+ */
+export default class Query<T> {
+  public filter: FilterQuery<T>;
+
+  private collection: Collection<T>;
+  private skip?: number;
+  private limit?: number;
+  private sort?: object;
+
+  constructor(collection: Collection<T>) {
+    this.collection = collection;
+  }
+
+  /**
+   * where will merge the given filter into the existing query.
+   *
+   * @param filter the filter to merge into the existing query
+   */
+  public where(filter: FilterQuery<T>): Query<T> {
+    this.filter = merge({}, this.filter || {}, filter);
+    return this;
+  }
+
+  /**
+   * after will skip the indicated number of documents.
+   *
+   * @param skip the number of documents to skip
+   */
+  public after(skip: number): Query<T> {
+    this.skip = skip;
+    return this;
+  }
+
+  /**
+   * first will limit to the indicated number of documents.
+   *
+   * @param limit the number of documents to limit the result to
+   */
+  public first(limit: number): Query<T> {
+    this.limit = limit;
+    return this;
+  }
+
+  /**
+   * orderBy will apply sorting to the query filter when executed.
+   *
+   * @param sort the sorting option for the documents
+   */
+  public orderBy(sort: object): Query<T> {
+    this.sort = merge({}, this.sort || {}, sort);
+    return this;
+  }
+
+  /**
+   * exec will return a cursor to the query.
+   */
+  public async exec(): Promise<Cursor<T>> {
+    let cursor = await this.collection.find(this.filter);
+
+    if (this.limit) {
+      // Apply a limit if it exists.
+      cursor = cursor.limit(this.limit);
+    }
+
+    if (this.sort) {
+      // Apply a sort if it exists.
+      cursor = cursor.sort(this.sort);
+    }
+
+    if (this.skip) {
+      // Apply a skip if it exists.
+      cursor = cursor.skip(this.skip);
+    }
+
+    return cursor;
+  }
+}
+
+type IndexType = 1 | -1;
+
+export type IndexSpecification<T> = {
+  [P in keyof Writeable<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,
+  });
+
+  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>>
+) {
+  return async (
+    createIndex: IndexCreationFunction<T>,
+    indexSpec: IndexSpecification<T>
+  ) => {
+    /**
+     * 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));
+
+    // Create a raw index without the variants applied.
+    await createIndex(indexSpec);
+
+    // Create all the variants.
+    for (const variant of variants) {
+      await createIndexVariant(variant);
+    }
+  };
+}