[next] Perspective API Integration (#1797)

* feat: initial toxic comments impl

* feat: improved logging

* feat: tenant cache adapter

* feat: move more types into graphql

package.json

@@ -57,6 +57,8 @@
     "lodash": "^4.17.10",
     "luxon": "^1.3.1",
     "mongodb": "^3.1.1",
+    "ms": "^2.1.1",
+    "node-fetch": "^2.2.0",
     "nunjucks": "^3.1.3",
     "passport": "^0.4.0",
     "passport-local": "^1.0.0",
@@ -101,7 +103,9 @@
     "@types/luxon": "^0.5.3",
     "@types/mini-css-extract-plugin": "^0.2.0",
     "@types/mongodb": "^3.1.1",
+    "@types/ms": "^0.7.30",
     "@types/node": "^10.5.2",
+    "@types/node-fetch": "^2.1.2",
     "@types/nunjucks": "^3.0.0",
     "@types/passport": "^0.4.6",
     "@types/passport-local": "^1.0.33",

scripts/generateSchemaTypes.js

@@ -66,7 +66,7 @@ async function main() {
     file.types = await generateTSTypesAsString(schema, {
       tabSpaces: 2,
       typePrefix: "GQL",
-      strictNulls: true,
+      strictNulls: false,
       ...file.config,
     });
   }

src/core/common/config.ts

@@ -86,6 +86,14 @@ const config = convict({
     env: "LOGGING_LEVEL",
     arg: "logging",
   },
+  disable_tenant_caching: {
+    doc:
+      "Disables the tenant caching, all tenants will be loaded from MongoDB each time it's needed",
+    format: Boolean,
+    default: false,
+    env: "DISABLE_TENANT_CACHING",
+    arg: "disableTenantCaching",
+  },
 });
 
 export type Config = typeof config;

src/core/server/app/index.ts

@@ -14,6 +14,7 @@ import { handleSubscriptions } from "talk-server/graph/common/subscriptions/midd
 import { Schemas } from "talk-server/graph/schemas";
 import TenantCache from "talk-server/services/tenant/cache";
 
+import { errorHandler } from "talk-server/app/middleware/error";
 import { accessLogger, errorLogger } from "./middleware/logging";
 import serveStatic from "./middleware/serveStatic";
 import { createRouter } from "./router";
@@ -57,6 +58,7 @@ export async function createApp(options: AppOptions): Promise<Express> {
   // Error Handling
   parent.use(notFoundMiddleware);
   parent.use(errorLogger);
+  parent.use(errorHandler);
 
   return parent;
 }

src/core/server/app/middleware/error.ts

@@ -4,3 +4,20 @@ export const apiErrorHandler: ErrorRequestHandler = (err, req, res, next) => {
   // TODO: handle better when we improve errors.
   res.status(500).json({ error: err.message });
 };
+
+export const errorHandler: ErrorRequestHandler = (err, req, res, next) => {
+  // TODO: handle better when we improve errors.
+  if (err.message === "not found") {
+    // TODO: handle better when we improve errors.
+    res
+      .status(404)
+      .send(err.message)
+      .end();
+  } else {
+    // TODO: handle better when we improve errors.
+    res
+      .status(500)
+      .send(err.message)
+      .end();
+  }
+};

src/core/server/app/middleware/logging.ts

@@ -39,6 +39,10 @@ export const accessLogger: RequestHandler = (req, res, next) => {
 };
 
 export const errorLogger: ErrorRequestHandler = (err, req, res, next) => {
-  logger.error(err, "http error");
+  // TODO: handle better when we improve errors.
+  if (err.message !== "not found") {
+    logger.error({ err }, "http error");
+  }
+
   next(err);
 };

src/core/server/app/middleware/passport/index.ts

@@ -2,6 +2,7 @@ import { NextFunction, RequestHandler, Response } from "express";
 import { Db } from "mongodb";
 import passport, { Authenticator } from "passport";
 
+import { Config } from "talk-common/config";
 import {
   createJWTStrategy,
   JWTSigningConfig,
@@ -22,6 +23,7 @@ export type VerifyCallback = (
 ) => void;
 
 export interface PassportOptions {
+  config: Config;
   mongo: Db;
   signingConfig: JWTSigningConfig;
   tenantCache: TenantCache;

src/core/server/app/middleware/passport/oidc.ts

@@ -5,6 +5,7 @@ import { Db } from "mongodb";
 import { Strategy as OAuth2Strategy, VerifyCallback } from "passport-oauth2";
 import { Strategy } from "passport-strategy";
 
+import { Config } from "talk-common/config";
 import { validate } from "talk-server/app/request/body";
 import { reconstructURL } from "talk-server/app/url";
 import { GQLUSER_ROLE } from "talk-server/graph/tenant/schema/__generated__/types";
@@ -12,6 +13,7 @@ import { OIDCAuthIntegration } from "talk-server/models/settings";
 import { Tenant } from "talk-server/models/tenant";
 import { OIDCProfile, retrieveUserWithProfile } from "talk-server/models/user";
 import TenantCache from "talk-server/services/tenant/cache";
+import { TenantCacheAdapter } from "talk-server/services/tenant/cache/adapter";
 import { upsert } from "talk-server/services/users";
 import { Request } from "talk-server/types/express";
 
@@ -177,25 +179,23 @@ const OIDC_SCOPE = "openid email profile";
 export interface OIDCStrategyOptions {
   mongo: Db;
   tenantCache: TenantCache;
+  config: Config;
 }
 
 export default class OIDCStrategy extends Strategy {
   public name = "oidc";
 
   private mongo: Db;
-  private cache = new Map<string, StrategyItem>();
+  private cache: TenantCacheAdapter<StrategyItem>;
 
-  constructor({ mongo, tenantCache }: OIDCStrategyOptions) {
+  constructor({ mongo, tenantCache, config }: OIDCStrategyOptions) {
     super();
 
     this.mongo = mongo;
+    this.cache = new TenantCacheAdapter(tenantCache, config);
 
-    // Subscribe to updates with Tenants.
-    tenantCache.subscribe(tenant => {
-      // Delete the tenant cache item when the tenant changes. The refreshed
-      // Tenant will come in with the request.
-      this.cache.delete(tenant.id);
-    });
+    // Connect the cache adapter.
+    this.cache.subscribe();
   }
 
   private lookupJWKSClient(

src/core/server/graph/tenant/resolvers/auth_integrations.ts

@@ -1,7 +1,10 @@
 import { GQLAuthIntegrationsTypeResolver } from "talk-server/graph/tenant/schema/__generated__/types";
-import { AuthIntegration, AuthIntegrations } from "talk-server/models/settings";
+import {
+  AuthIntegrations,
+  EnableableIntegration,
+} from "talk-server/models/settings";
 
-const disabled: AuthIntegration = { enabled: false };
+const disabled: EnableableIntegration = { enabled: false };
 
 const AuthIntegrations: GQLAuthIntegrationsTypeResolver<AuthIntegrations> = {
   local: auth => auth.local || disabled,

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

@@ -38,6 +38,16 @@ enum ACTION_ITEM_TYPE {
   COMMENTS
 }
 
+enum ACTION_GROUP {
+  SPAM_COMMENT
+  TOXIC_COMMENT
+  BODY_COUNT
+  TRUST
+  LINKS
+  BANNED_WORD
+  SUSPECT_WORD
+}
+
 ################################################################################
 ## Settings
 ################################################################################
@@ -57,9 +67,9 @@ enum MODERATION_MODE {
 }
 
 """
-WordlistSettings describes all the available wordlists.
+Wordlist describes all the available wordlists.
 """
-type WordlistSettings {
+type Wordlist {
   """
   banned words will by default reject the comment if it is found.
   """
@@ -177,6 +187,111 @@ type AuthSettings {
   integrations: AuthIntegrations!
 }
 
+################################################################################
+## ExternalIntegrations
+################################################################################
+
+type AkismetExternalIntegration {
+  """
+  enabled when True will enable the integration.
+  """
+  enabled: Boolean!
+
+  """
+  The key for the Akismet integration.
+  """
+  key: String
+
+  """
+  The site (blog) for the Akismet integration.
+  """
+  site: String
+}
+
+type PerspectiveExternalIntegration {
+  """
+  enabled when True will enable the integration.
+  """
+  enabled: Boolean!
+
+  """
+  The endpoint that Talk should use to communicate with the perspective API.
+  """
+  endpoint: String
+
+  """
+  The key for the Perspective API integration.
+  """
+  key: String
+
+  """
+  The threshold that given a specific toxic comment score, the comment will
+  be marked by Talk as toxic.
+  """
+  threshold: Float
+
+  """
+  When True, comments sent will not be stored by the Google Perspective API.
+  """
+  doNotStore: Boolean
+}
+
+type ExternalIntegrations {
+  """
+  akismet provides integration with the Akismet Spam detection service.
+  """
+  akismet: AkismetExternalIntegration!
+
+  """
+  perspective provides integration with the Perspective API comment analysis
+  platform.
+  """
+  perspective: PerspectiveExternalIntegration!
+}
+
+################################################################################
+## Karma
+################################################################################
+
+"""
+KarmaThreshold defines the bounds for which a User will become unreliable or
+reliable based on their karma score. If the score is equal or less than the
+unreliable value, they are unreliable. If the score is equal or more than the
+reliable value, they are reliable. If they are neither reliable or unreliable
+then they are neutral.
+"""
+type KarmaThreshold {
+  reliable: Int!
+  unreliable: Int!
+}
+
+type KarmaThresholds {
+  """
+  flag represents karma settings in relation to how well a User's flagging
+  ability aligns with the moderation decicions made by moderators.
+  """
+  flag: KarmaThreshold!
+
+  """
+  comment represents the karma setting in relation to how well a User's comments are moderated.
+  """
+  comment: KarmaThreshold!
+}
+
+type Karma {
+  """
+  When true, checks will be completed to ensure that the Karma checks are
+  completed.
+  """
+  enabled: Boolean!
+
+  """
+  karmaThresholds contains the currently set thresholds for triggering Trust
+  beheviour.
+  """
+  thresholds: KarmaThresholds!
+}
+
 ################################################################################
 ## Settings
 ################################################################################
@@ -295,7 +410,7 @@ type Settings {
   """
   wordlist will return a given list of words.
   """
-  wordlist: WordlistSettings @auth(roles: [ADMIN, MODERATOR])
+  wordlist: Wordlist @auth(roles: [ADMIN, MODERATOR])
 
   """
   domains will return a given list of whitelisted domains.
@@ -306,6 +421,17 @@ type Settings {
   auth contains all the settings related to authentication and authorization.
   """
   auth: AuthSettings!
+
+  """
+  integrations contains all the external integrations that can be enabled.
+  """
+  integrations: ExternalIntegrations @auth(roles: [ADMIN])
+
+  """
+  karma is the set of settings related to how user Trust and Karma are
+  handled.
+  """
+  karma: Karma @auth(roles: [ADMIN, MODERATOR])
 }
 
 ################################################################################
@@ -728,7 +854,7 @@ input SettingsInput {
   charCount: Int
   organizationName: String
   organizationContactEmail: String
-  # wordlist: WordlistSettings @auth(roles: [ADMIN, MODERATOR])
+  # wordlist: Wordlist @auth(roles: [ADMIN, MODERATOR])
   domains: [String!]
   # auth: AuthSettings!
 }

src/core/server/index.ts

@@ -70,7 +70,11 @@ class Server {
     const signingConfig = createJWTSigningConfig(this.config);
 
     // Create the TenantCache.
-    const tenantCache = new TenantCache(mongo, await createRedisClient(config));
+    const tenantCache = new TenantCache(
+      mongo,
+      await createRedisClient(config),
+      config
+    );
 
     // Prime the tenant cache so it'll be ready to serve now.
     await tenantCache.primeAll();

src/core/server/models/actions.ts

@@ -1,4 +1,5 @@
 import {
+  GQLACTION_GROUP,
   GQLACTION_ITEM_TYPE,
   GQLACTION_TYPE,
 } from "talk-server/graph/tenant/schema/__generated__/types";
@@ -10,7 +11,7 @@ export interface Action {
   action_type: GQLACTION_TYPE;
   item_type: GQLACTION_ITEM_TYPE;
   item_id: string;
-  group_id?: string;
+  group_id?: GQLACTION_GROUP;
   user_id?: string;
   created_at: Date;
   metadata?: Record<string, any>;

src/core/server/models/comment.ts

@@ -34,7 +34,7 @@ export interface StatusHistoryItem {
 
 export interface Comment extends TenantResource {
   readonly id: string;
-  parent_id: string | null;
+  parent_id?: string;
   author_id: string;
   asset_id: string;
   body: string;

src/core/server/models/settings.ts

@@ -1,13 +1,11 @@
 import {
+  GQLExternalIntegrations,
+  GQLKarma,
   GQLMODERATION_MODE,
   GQLUSER_ROLE,
+  GQLWordlist,
 } from "talk-server/graph/tenant/schema/__generated__/types";
 
-export interface Wordlist {
-  banned: string[];
-  suspect: string[];
-}
-
 export interface EmailDomainRuleCondition {
   /**
    * emailDomain is the domain name component of the email addresses that should
@@ -49,7 +47,7 @@ export interface AuthRules {
   restrictTo?: EmailDomainRuleCondition[];
 }
 
-export interface AuthIntegration {
+export interface EnableableIntegration {
   enabled: boolean;
 }
 
@@ -63,7 +61,7 @@ export interface DisplayNameAuthIntegration {
  * embed to allow single sign on.
  */
 export interface SSOAuthIntegration
-  extends AuthIntegration,
+  extends EnableableIntegration,
     DisplayNameAuthIntegration {
   key: string;
 }
@@ -73,7 +71,7 @@ export interface SSOAuthIntegration
  * will be used in the admin to provide staff logins for users.
  */
 export interface OIDCAuthIntegration
-  extends AuthIntegration,
+  extends EnableableIntegration,
     DisplayNameAuthIntegration {
   clientID: string;
   clientSecret: string;
@@ -83,17 +81,17 @@ export interface OIDCAuthIntegration
   tokenURL: string;
 }
 
-export interface FacebookAuthIntegration extends AuthIntegration {
+export interface FacebookAuthIntegration extends EnableableIntegration {
   clientID: string;
   clientSecret: string;
 }
 
-export interface GoogleAuthIntegration extends AuthIntegration {
+export interface GoogleAuthIntegration extends EnableableIntegration {
   clientID: string;
   clientSecret: string;
 }
 
-export type LocalAuthIntegration = AuthIntegration;
+export type LocalAuthIntegration = EnableableIntegration;
 
 /**
  * AuthIntegrations describes all of the possible auth integration
@@ -130,33 +128,6 @@ export interface Auth {
   integrations: AuthIntegrations;
 }
 
-/**
- * Akismet provides integration with the Akismet Spam detection service.
- */
-export interface AkismetIntegration {
-  /**
-   * When true, it will enable comments to be checked by Akismet.
-   */
-  enabled: boolean;
-
-  /**
-   * The key for the Akismet integration.
-   */
-  key?: string;
-
-  /**
-   * The site (blog) for the Akismet integration.
-   */
-  site?: string;
-}
-
-export interface ExternalIntegrations {
-  /**
-   * akismet provides integration with the Akismet Spam detection service.
-   */
-  akismet: AkismetIntegration;
-}
-
 export interface ModerationSettings {
   moderation: GQLMODERATION_MODE;
   requireEmailConfirmation: boolean;
@@ -175,45 +146,6 @@ export interface ModerationSettings {
   charCount?: number;
 }
 
-/**
- * KarmaThreshold defines the bounds for which a User will become unreliable or
- * reliable based on their karma score. If the score is equal or less than the
- * unreliable value, they are unreliable. If the score is equal or more than the
- * reliable value, they are reliable. If they are neither reliable or unreliable
- * then they are neutral.
- */
-export interface KarmaThreshold {
-  reliable: number;
-  unreliable: number;
-}
-
-export interface KarmaThresholds {
-  /**
-   * flag represents karma settings in relation to how well a User's flagging
-   * ability aligns with the moderation decicions made by moderators.
-   */
-  flag: KarmaThreshold;
-
-  /**
-   * comment represents the karma setting in relation to how well a User's comments are moderated.
-   */
-  comment: KarmaThreshold;
-}
-
-export interface Karma {
-  /**
-   * When true, checks will be completed to ensure that the Karma checks are
-   * completed.
-   */
-  enabled: boolean;
-
-  /**
-   * karmaThresholds contains the currently set thresholds for triggering Trust
-   * beheviour.
-   */
-  thresholds: KarmaThresholds;
-}
-
 export interface Settings extends ModerationSettings {
   customCssUrl?: string;
 
@@ -227,12 +159,12 @@ export interface Settings extends ModerationSettings {
    * karma is the set of settings related to how user Trust and Karma are
    * handled.
    */
-  karma: Karma;
+  karma: GQLKarma;
 
   /**
    * wordlist stores all the banned/suspect words.
    */
-  wordlist: Wordlist;
+  wordlist: GQLWordlist;
 
   /**
    * Set of configured authentication integrations.
@@ -242,5 +174,5 @@ export interface Settings extends ModerationSettings {
   /**
    * Various integrations with external services.
    */
-  integrations: ExternalIntegrations;
+  integrations: GQLExternalIntegrations;
 }

src/core/server/models/tenant.ts

@@ -91,6 +91,9 @@ export async function createTenant(db: Db, input: CreateTenantInput) {
       akismet: {
         enabled: false,
       },
+      perspective: {
+        enabled: false,
+      },
     },
   };
 

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

@@ -14,7 +14,7 @@ import { Request } from "talk-server/types/express";
 
 export type CreateComment = Omit<
   CreateCommentInput,
-  "status" | "action_counts"
+  "status" | "action_counts" | "metadata"
 >;
 
 export async function create(
@@ -44,7 +44,7 @@ export async function create(
   }
 
   // Run the comment through the moderation phases.
-  const { status } = await processForModeration({
+  const { status, metadata } = await processForModeration({
     asset,
     tenant,
     comment: input,
@@ -55,9 +55,10 @@ export async function create(
   // TODO: (wyattjoh) use the actions somehow.
 
   const comment = await createComment(mongo, tenant.id, {
+    ...input,
     status,
     action_counts: {},
-    ...input,
+    metadata,
   });
 
   if (input.parent_id) {

src/core/server/services/comments/moderation/index.spec.ts

@@ -0,0 +1,105 @@
+import {
+  GQLACTION_GROUP,
+  GQLACTION_TYPE,
+  GQLCOMMENT_STATUS,
+} from "talk-server/graph/tenant/schema/__generated__/types";
+import {
+  compose,
+  ModerationPhaseContext,
+} from "talk-server/services/comments/moderation";
+
+describe("compose", () => {
+  it("handles when a phase throws an error", async () => {
+    const err = new Error("this is an error");
+    const enhanced = compose([
+      () => {
+        throw err;
+      },
+    ]);
+
+    await expect(enhanced({} as ModerationPhaseContext)).rejects.toEqual(err);
+  });
+
+  it("handles when it returns a status", async () => {
+    const status = GQLCOMMENT_STATUS.ACCEPTED;
+    const enhanced = compose([() => ({ status })]);
+
+    await expect(enhanced({} as ModerationPhaseContext)).resolves.toEqual({
+      status,
+      metadata: {},
+      actions: [],
+    });
+  });
+
+  it("merges the metadata", async () => {
+    const status = GQLCOMMENT_STATUS.ACCEPTED;
+    const enhanced = compose([
+      () => ({ metadata: { first: true } }),
+      () => ({ status, metadata: { second: true } }),
+      () => ({ metadata: { third: true } }),
+    ]);
+
+    await expect(enhanced({} as ModerationPhaseContext)).resolves.toEqual({
+      status,
+      metadata: { first: true, second: true },
+      actions: [],
+    });
+  });
+
+  it("merges actions", async () => {
+    const status = GQLCOMMENT_STATUS.ACCEPTED;
+
+    const flags = [
+      {
+        action_type: GQLACTION_TYPE.FLAG,
+        group_id: GQLACTION_GROUP.TOXIC_COMMENT,
+      },
+      {
+        action_type: GQLACTION_TYPE.FLAG,
+        group_id: GQLACTION_GROUP.SPAM_COMMENT,
+      },
+    ];
+
+    const enhanced = compose([
+      () => ({
+        actions: [flags[0]],
+      }),
+      () => ({
+        status,
+        actions: [flags[1]],
+      }),
+      () => ({
+        actions: [
+          {
+            action_type: GQLACTION_TYPE.FLAG,
+            group_id: GQLACTION_GROUP.LINKS,
+          },
+        ],
+      }),
+    ]);
+
+    const final = await enhanced({} as ModerationPhaseContext);
+
+    for (const flag of flags) {
+      expect(final.actions).toContainEqual(flag);
+    }
+
+    expect(final.actions).not.toContainEqual({
+      action_type: GQLACTION_TYPE.FLAG,
+      group_id: GQLACTION_GROUP.LINKS,
+    });
+  });
+
+  it("handles when it does not return a status", async () => {
+    const enhanced = compose([
+      () => ({ metadata: { first: true } }),
+      () => ({ metadata: { second: true } }),
+    ]);
+
+    await expect(enhanced({} as ModerationPhaseContext)).resolves.toEqual({
+      status: GQLCOMMENT_STATUS.NONE,
+      metadata: { first: true, second: true },
+      actions: [],
+    });
+  });
+});

src/core/server/services/comments/moderation/index.ts

@@ -3,13 +3,13 @@ import { GQLCOMMENT_STATUS } from "talk-server/graph/tenant/schema/__generated__
 import { Action } from "talk-server/models/actions";
 import { Asset } from "talk-server/models/asset";
 import { Tenant } from "talk-server/models/tenant";
-import { CreateComment } from "talk-server/services/comments";
-
 import { User } from "talk-server/models/user";
+import { CreateComment } from "talk-server/services/comments";
 import { Request } from "talk-server/types/express";
+
 import { moderationPhases } from "./phases";
 
-// TODO: (wyattjoh) move into actions module.
+// TODO: (wyattjoh) move into actions module once we have action methods.
 export type CreateAction = Omit<
   Action,
   "id" | "item_type" | "item_id" | "created_at"
@@ -18,6 +18,7 @@ export type CreateAction = Omit<
 export interface PhaseResult {
   actions: CreateAction[];
   status: GQLCOMMENT_STATUS;
+  metadata: Record<string, any>;
 }
 
 export interface ModerationPhaseContext {
@@ -42,31 +43,47 @@ export type IntermediateModerationPhase = (
  * compose will create a moderation pipeline for which is executable with the
  * passed actions.
  */
-const compose = (
+export const compose = (
   phases: IntermediateModerationPhase[]
 ): ModerationPhase => async context => {
-  const actions: CreateAction[] = [];
+  const final: PhaseResult = {
+    status: GQLCOMMENT_STATUS.NONE,
+    actions: [],
+    metadata: {},
+  };
 
   // Loop over all the moderation phases and see if we've resolved the status.
   for (const phase of phases) {
     const result = await phase(context);
     if (result) {
-      if (result.actions) {
-        actions.push(...result.actions);
+      // If this result contained actions, then we should push it into the
+      // other actions.
+      const { actions } = result;
+      if (actions) {
+        final.actions.push(...actions);
+      }
+
+      // If this result contained metadata, then we should merge it into the
+      // other metadata.
+      const { metadata } = result;
+      if (metadata) {
+        final.metadata = {
+          ...final.metadata,
+          ...metadata,
+        };
       }
 
       // If this result contained a status, then we've finished resolving
       // phases!
       const { status } = result;
       if (status) {
-        return { status, actions };
+        final.status = status;
+        break;
       }
     }
   }
 
-  // If we didn't determine a different comment from a previous itteration, set
-  // it to 'NONE'.
-  return { status: GQLCOMMENT_STATUS.NONE, actions };
+  return final;
 };
 
 /**

src/core/server/services/comments/moderation/phases/assetClosed.ts

@@ -1,12 +1,15 @@
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 
 // This phase checks to see if the asset being processed is closed or not.
-export const assetClosed: IntermediateModerationPhase = ({ asset }) => {
+export const assetClosed: IntermediateModerationPhase = ({
+  asset,
+}): IntermediatePhaseResult | void => {
   // Check to see if the asset has closed commenting...
   if (asset.closedAt && asset.closedAt.valueOf() <= Date.now()) {
     // TODO: (wyattjoh) return better error.
     throw new Error("asset is currently closed for commenting");
   }
-
-  return;
 };

src/core/server/services/comments/moderation/phases/commentLength.ts

@@ -1,18 +1,22 @@
 import {
+  GQLACTION_GROUP,
   GQLACTION_TYPE,
   GQLCOMMENT_STATUS,
 } from "talk-server/graph/tenant/schema/__generated__/types";
 import { ModerationSettings } from "talk-server/models/settings";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 
 const testCharCount = (settings: Partial<ModerationSettings>, length: number) =>
   settings.charCountEnable && settings.charCount && length > settings.charCount;
 
-export const commentLength: IntermediateModerationPhase = async ({
+export const commentLength: IntermediateModerationPhase = ({
   asset,
   tenant,
   comment,
-}) => {
+}): IntermediatePhaseResult | void => {
   const length = comment.body.length;
 
   // Check to see if the body is too short, if it is, then complain about it!
@@ -32,7 +36,7 @@ export const commentLength: IntermediateModerationPhase = async ({
       actions: [
         {
           action_type: GQLACTION_TYPE.FLAG,
-          group_id: "BODY_COUNT",
+          group_id: GQLACTION_GROUP.BODY_COUNT,
           metadata: {
             count: length,
           },
@@ -40,6 +44,4 @@ export const commentLength: IntermediateModerationPhase = async ({
       ],
     };
   }
-
-  return;
 };

src/core/server/services/comments/moderation/phases/commentingDisabled.ts

@@ -1,5 +1,8 @@
 import { ModerationSettings } from "talk-server/models/settings";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 
 const testDisabledCommenting = (settings: Partial<ModerationSettings>) =>
   settings.disableCommenting;
@@ -7,7 +10,7 @@ const testDisabledCommenting = (settings: Partial<ModerationSettings>) =>
 export const commentingDisabled: IntermediateModerationPhase = ({
   asset,
   tenant,
-}) => {
+}): IntermediatePhaseResult | void => {
   // Check to see if the asset has closed commenting.
   if (
     testDisabledCommenting(tenant) ||

src/core/server/services/comments/moderation/phases/index.ts

@@ -1,6 +1,7 @@
 import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
 
 import { premod } from "talk-server/services/comments/moderation/phases/premod";
+import { toxic } from "talk-server/services/comments/moderation/phases/toxic";
 import { assetClosed } from "./assetClosed";
 import { commentingDisabled } from "./commentingDisabled";
 import { commentLength } from "./commentLength";
@@ -22,5 +23,6 @@ export const moderationPhases: IntermediateModerationPhase[] = [
   links,
   karma,
   spam,
+  toxic,
   premod,
 ];

src/core/server/services/comments/moderation/phases/karma.ts

@@ -1,8 +1,12 @@
 import {
+  GQLACTION_GROUP,
   GQLACTION_TYPE,
   GQLCOMMENT_STATUS,
 } from "talk-server/graph/tenant/schema/__generated__/types";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 import {
   getCommentTrustScore,
   isReliableCommenter,
@@ -10,7 +14,10 @@ import {
 
 // This phase checks to see if the user making the comment is allowed to do so
 // considering their reliability (Trust) status.
-export const karma: IntermediateModerationPhase = ({ tenant, author }) => {
+export const karma: IntermediateModerationPhase = ({
+  tenant,
+  author,
+}): IntermediatePhaseResult | void => {
   // If the user is not a reliable commenter (passed the unreliability
   // threshold by having too many rejected comments) then we can change the
   // status of the comment to `SYSTEM_WITHHELD`, therefore pushing the user's
@@ -27,7 +34,7 @@ export const karma: IntermediateModerationPhase = ({ tenant, author }) => {
       actions: [
         {
           action_type: GQLACTION_TYPE.FLAG,
-          group_id: "TRUST",
+          group_id: GQLACTION_GROUP.TRUST,
           metadata: {
             trust: getCommentTrustScore(author),
           },
@@ -35,6 +42,4 @@ export const karma: IntermediateModerationPhase = ({ tenant, author }) => {
       ],
     };
   }
-
-  return;
 };

src/core/server/services/comments/moderation/phases/links.ts

@@ -2,11 +2,15 @@ import linkify from "linkify-it";
 import tlds from "tlds";
 
 import {
+  GQLACTION_GROUP,
   GQLACTION_TYPE,
   GQLCOMMENT_STATUS,
 } from "talk-server/graph/tenant/schema/__generated__/types";
 import { ModerationSettings } from "talk-server/models/settings";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 
 /**
  * The preloaded linkify instance with common tlds.
@@ -24,8 +28,7 @@ export const links: IntermediateModerationPhase = ({
   asset,
   tenant,
   comment,
-  author,
-}) => {
+}): IntermediatePhaseResult | void => {
   if (
     testPremodLinksEnable(tenant, comment.body) ||
     (asset.settings && testPremodLinksEnable(asset.settings, comment.body))
@@ -36,7 +39,7 @@ export const links: IntermediateModerationPhase = ({
       actions: [
         {
           action_type: GQLACTION_TYPE.FLAG,
-          group_id: "LINKS",
+          group_id: GQLACTION_GROUP.LINKS,
           metadata: {
             links: comment.body,
           },
@@ -44,6 +47,4 @@ export const links: IntermediateModerationPhase = ({
       ],
     };
   }
-
-  return;
 };

src/core/server/services/comments/moderation/phases/premod.ts

@@ -3,14 +3,20 @@ import {
   GQLMODERATION_MODE,
 } from "talk-server/graph/tenant/schema/__generated__/types";
 import { ModerationSettings } from "talk-server/models/settings";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 
 const testModerationMode = (settings: Partial<ModerationSettings>) =>
   settings.moderation === GQLMODERATION_MODE.PRE;
 
 // This phase checks to see if the settings have premod enabled, if they do,
 // the comment is premod, otherwise, it's just none.
-export const premod: IntermediateModerationPhase = ({ asset, tenant }) => {
+export const premod: IntermediateModerationPhase = ({
+  asset,
+  tenant,
+}): IntermediatePhaseResult | void => {
   // If the settings say that we're in premod mode, then the comment is in
   // premod status.
 
@@ -23,6 +29,4 @@ export const premod: IntermediateModerationPhase = ({ asset, tenant }) => {
       status: GQLCOMMENT_STATUS.PREMOD,
     };
   }
-
-  return;
 };

src/core/server/services/comments/moderation/phases/spam.ts

@@ -1,10 +1,15 @@
 import { Client } from "akismet-api";
 
 import {
+  GQLACTION_GROUP,
   GQLACTION_TYPE,
   GQLCOMMENT_STATUS,
 } from "talk-server/graph/tenant/schema/__generated__/types";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import logger from "talk-server/logger";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 
 export const spam: IntermediateModerationPhase = async ({
   asset,
@@ -12,16 +17,26 @@ export const spam: IntermediateModerationPhase = async ({
   comment,
   author,
   req,
-}) => {
+}): Promise<IntermediatePhaseResult | void> => {
   const integration = tenant.integrations.akismet;
 
   // We can only check for spam if this comment originated from a graphql
   // request via an HTTP call.
-  if (!req || !integration.enabled) {
+  if (!req) {
+    logger.debug({ tenant_id: tenant.id }, "request was not available");
+    return;
+  }
+
+  if (!integration.enabled) {
+    logger.debug({ tenant_id: tenant.id }, "akismet integration was disabled");
     return;
   }
 
   if (!integration.key || !integration.site) {
+    logger.error(
+      { tenant_id: tenant.id },
+      "akismet integration was enabled but configuration was missing"
+    );
     return;
   }
 
@@ -34,41 +49,73 @@ export const spam: IntermediateModerationPhase = async ({
   // Grab the properties we need.
   const userIP = req.ip;
   if (!userIP) {
+    logger.debug(
+      { tenant_id: tenant.id },
+      "request did not contain ip address, aborting spam check"
+    );
     return;
   }
 
   const userAgent = req.get("User-Agent");
   if (!userAgent || userAgent.length === 0) {
+    logger.debug(
+      { tenant_id: tenant.id },
+      "request did not contain User-Agent header, aborting spam check"
+    );
     return;
   }
 
   const referrer = req.get("Referrer");
   if (!referrer || referrer.length === 0) {
+    logger.debug(
+      { tenant_id: tenant.id },
+      "request did not contain Referrer header, aborting spam check"
+    );
     return;
   }
 
-  // Check the comment for spam.
-  const isSpam = await client.checkSpam({
-    user_ip: userIP, // REQUIRED
-    referrer, // REQUIRED
-    user_agent: userAgent, // REQUIRED
-    comment_content: comment.body,
-    permalink: asset.url,
-    comment_author: author.displayName || author.username || "",
-    comment_type: "comment",
-    is_test: false,
-  });
-  if (isSpam) {
-    return {
-      status: GQLCOMMENT_STATUS.SYSTEM_WITHHELD,
-      actions: [
-        {
-          action_type: GQLACTION_TYPE.FLAG,
-          group_id: "SPAM_COMMENT",
-        },
-      ],
-    };
-  }
+  try {
+    logger.trace({ tenant_id: tenant.id }, "checking comment for spam");
 
-  return;
+    // Check the comment for spam.
+    const isSpam = await client.checkSpam({
+      user_ip: userIP, // REQUIRED
+      referrer, // REQUIRED
+      user_agent: userAgent, // REQUIRED
+      comment_content: comment.body,
+      permalink: asset.url,
+      comment_author: author.displayName || author.username || "",
+      comment_type: "comment",
+      is_test: false,
+    });
+    if (isSpam) {
+      logger.trace(
+        { tenant_id: tenant.id, is_spam: isSpam },
+        "comment contained spam"
+      );
+      return {
+        status: GQLCOMMENT_STATUS.SYSTEM_WITHHELD,
+        actions: [
+          {
+            action_type: GQLACTION_TYPE.FLAG,
+            group_id: GQLACTION_GROUP.SPAM_COMMENT,
+          },
+        ],
+        metadata: {
+          // Store the spam result from Akismet in the Comment metadata.
+          akismet: spam,
+        },
+      };
+    }
+
+    logger.trace(
+      { tenant_id: tenant.id, is_spam: isSpam },
+      "comment did not contain spam"
+    );
+  } catch (err) {
+    logger.error(
+      { tenant_id: tenant.id, err },
+      "could not determine if comment contained spam"
+    );
+  }
 };

src/core/server/services/comments/moderation/phases/staff.ts

@@ -2,7 +2,10 @@ import {
   GQLCOMMENT_STATUS,
   GQLUSER_ROLE,
 } from "talk-server/graph/tenant/schema/__generated__/types";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 
 // If a given user is a staff member, always approve their comment.
 export const staff: IntermediateModerationPhase = ({
@@ -10,12 +13,10 @@ export const staff: IntermediateModerationPhase = ({
   tenant,
   comment,
   author,
-}) => {
+}): IntermediatePhaseResult | void => {
   if (author.role !== GQLUSER_ROLE.COMMENTER) {
     return {
       status: GQLCOMMENT_STATUS.ACCEPTED,
     };
   }
-
-  return;
 };

src/core/server/services/comments/moderation/phases/toxic.ts

@@ -0,0 +1,183 @@
+import { isNil } from "lodash";
+import ms from "ms";
+import fetch from "node-fetch";
+
+import { Omit } from "talk-common/types";
+import {
+  GQLACTION_GROUP,
+  GQLACTION_TYPE,
+  GQLCOMMENT_STATUS,
+  GQLPerspectiveExternalIntegration,
+} from "talk-server/graph/tenant/schema/__generated__/types";
+import logger from "talk-server/logger";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
+
+export const toxic: IntermediateModerationPhase = async ({
+  tenant,
+  comment,
+}): Promise<IntermediatePhaseResult | void> => {
+  const integration = tenant.integrations.perspective;
+
+  if (!integration.enabled) {
+    // The Toxic comment plugin is not enabled.
+    logger.debug(
+      { tenant_id: tenant.id },
+      "perspective integration was disabled"
+    );
+    return;
+  }
+
+  if (!integration.key) {
+    // The Toxic comment requires a key in order to communicate with the API.
+    logger.error(
+      { tenant_id: tenant.id },
+      "perspective integration was enabled but configuration was missing"
+    );
+    return;
+  }
+
+  let endpoint = integration.endpoint;
+  if (isNil(endpoint)) {
+    // TODO: (wyattjoh) replace hardcoded default with config.
+    endpoint = "https://commentanalyzer.googleapis.com/v1alpha1";
+
+    logger.trace(
+      { tenant_id: tenant.id, endpoint },
+      "endpoint missing in integration settings, using defaults"
+    );
+  }
+
+  let threshold = integration.threshold;
+  if (isNil(threshold)) {
+    // TODO: (wyattjoh) replace hardcoded default with config.
+    threshold = 0.8;
+
+    logger.trace(
+      { tenant_id: tenant.id, threshold },
+      "threshold missing in integration settings, using defaults"
+    );
+  }
+
+  let doNotStore = integration.doNotStore;
+  if (isNil(doNotStore)) {
+    doNotStore = true;
+
+    logger.trace(
+      { tenant_id: tenant.id, do_not_store: doNotStore },
+      "doNotStore missing in integration settings, using defaults"
+    );
+  }
+
+  // TODO: (wyattjoh) replace hardcoded default with config.
+  const timeout = ms("300ms");
+
+  try {
+    logger.trace({ tenant_id: tenant.id }, "checking comment toxicity");
+
+    // Call into the Toxic comment API.
+    const scores = await getScores(
+      comment.body,
+      {
+        endpoint,
+        key: integration.key,
+        doNotStore,
+      },
+      timeout
+    );
+
+    const score = scores.SEVERE_TOXICITY.summaryScore;
+    const isToxic = score > threshold;
+    if (isToxic) {
+      logger.trace(
+        { tenant_id: tenant.id, score, is_toxic: isToxic, threshold },
+        "comment was toxic"
+      );
+      return {
+        status: GQLCOMMENT_STATUS.SYSTEM_WITHHELD,
+        actions: [
+          {
+            action_type: GQLACTION_TYPE.FLAG,
+            group_id: GQLACTION_GROUP.TOXIC_COMMENT,
+          },
+        ],
+        metadata: {
+          // Store the scores from perspective in the Comment metadata.
+          perspective: scores,
+        },
+      };
+    }
+
+    logger.trace(
+      { tenant_id: tenant.id, score, is_toxic: isToxic, threshold },
+      "comment was not toxic"
+    );
+  } catch (err) {
+    logger.error(
+      { tenant_id: tenant.id, err },
+      "could not determine comment toxicity"
+    );
+  }
+};
+
+/**
+ * getScores will return the toxicity scores for the comment text.
+ *
+ * @param text comment text to check for toxicity
+ * @param settings integration settings used to communicate with the perspective api
+ * @param timeout timeout for communicating with the perspective api
+ */
+async function getScores(
+  text: string,
+  {
+    key,
+    endpoint,
+    doNotStore,
+  }: Required<Omit<GQLPerspectiveExternalIntegration, "enabled" | "threshold">>,
+  timeout: number
+) {
+  try {
+    const response = await fetch(`${endpoint}/comments:analyze?key=${key}`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      timeout,
+      body: JSON.stringify({
+        comment: {
+          text,
+        },
+        // TODO: (wyattjoh) support other languages.
+        languages: ["en"],
+        doNotStore,
+        requestedAttributes: {
+          TOXICITY: {},
+          SEVERE_TOXICITY: {},
+        },
+      }),
+    });
+
+    // Grab the data out of the Perspective API.
+    const data = await response.json();
+
+    // Reformat the scores.
+    return {
+      TOXICITY: {
+        summaryScore: data.attributeScores.TOXICITY.summaryScore.value,
+      },
+      SEVERE_TOXICITY: {
+        summaryScore: data.attributeScores.SEVERE_TOXICITY.summaryScore.value,
+      },
+    };
+  } catch (err) {
+    // Ensure that the API key doesn't get leaked to the logs by accident.
+    if (err.message) {
+      err.message = err.message.replace(key, "***");
+    }
+
+    // Rethrow the error.
+    throw err;
+  }
+}

src/core/server/services/comments/moderation/phases/wordlist.ts

@@ -1,17 +1,19 @@
 import {
+  GQLACTION_GROUP,
   GQLACTION_TYPE,
   GQLCOMMENT_STATUS,
 } from "talk-server/graph/tenant/schema/__generated__/types";
-import { IntermediateModerationPhase } from "talk-server/services/comments/moderation";
+import {
+  IntermediateModerationPhase,
+  IntermediatePhaseResult,
+} from "talk-server/services/comments/moderation";
 import { containsMatchingPhrase } from "talk-server/services/comments/moderation/wordlist";
 
 // This phase checks the comment against the wordlist.
 export const wordlist: IntermediateModerationPhase = ({
-  asset,
   tenant,
   comment,
-  author,
-}) => {
+}): IntermediatePhaseResult | void => {
   // Decide the status based on whether or not the current asset/settings
   // has pre-mod enabled or not. If the comment was rejected based on the
   // wordlist, then reject it, otherwise if the moderation setting is
@@ -23,7 +25,7 @@ export const wordlist: IntermediateModerationPhase = ({
       actions: [
         {
           action_type: GQLACTION_TYPE.FLAG,
-          group_id: "BANNED_WORD",
+          group_id: GQLACTION_GROUP.BANNED_WORD,
         },
       ],
     };
@@ -40,11 +42,9 @@ export const wordlist: IntermediateModerationPhase = ({
       actions: [
         {
           action_type: GQLACTION_TYPE.FLAG,
-          group_id: "SUSPECT_WORD",
+          group_id: GQLACTION_GROUP.SUSPECT_WORD,
         },
       ],
     };
   }
-
-  return;
 };

src/core/server/services/tenant/cache/adapter.ts

@@ -0,0 +1,92 @@
+import { Config } from "talk-common/config";
+import TenantCache from "talk-server/services/tenant/cache";
+
+export type DeconstructionFn<T> = (tenantID: string, value: T) => Promise<void>;
+
+/**
+ * TenantCacheAdapter is designed to allow additional services to cache entries
+ * that are related to tenants that could possibly be cached. When caching of
+ * tenants are enabled, this acts as a map to store entries, and will
+ * automatically invalidate tenants that have been updated.
+ */
+export class TenantCacheAdapter<T> {
+  private cache = new Map<string, T>();
+  private tenantCache: TenantCache;
+  private isCachingEnabled: boolean;
+
+  private unsubscribeFn?: () => void;
+  private deconstructionFn?: DeconstructionFn<T>;
+
+  constructor(
+    tenantCache: TenantCache,
+    config: Config,
+    deconstructionFn?: DeconstructionFn<T>
+  ) {
+    this.tenantCache = tenantCache;
+    this.isCachingEnabled = !config.get("disable_tenant_caching");
+    this.deconstructionFn = deconstructionFn;
+  }
+
+  public subscribe() {
+    if (this.isCachingEnabled) {
+      // Unsubscribe from updates if we
+      this.unsubscribe();
+
+      this.unsubscribeFn = this.tenantCache.subscribe(async tenant => {
+        // Get the current set value for the item in the cache.
+        const value = this.get(tenant.id);
+
+        // Delete the tenant cache item when the tenant changes.
+        this.cache.delete(tenant.id);
+
+        if (this.deconstructionFn) {
+          // The deconstruction function is set. We will check that the value
+          // exists, and if it does, we will call the function with the given
+          // identifier, this allows the caller to attach deconstruction
+          // components to the tenant being removed. The only side affect to
+          // note is that by the time that the deconstruction function is
+          // called, the tenant has already been purged from the cache.
+          if (typeof value !== "undefined") {
+            await this.deconstructionFn(tenant.id, value);
+          }
+        }
+      });
+    }
+  }
+
+  /**
+   * This will disconnect the map/cache from getting updates.
+   */
+  public unsubscribe() {
+    if (this.isCachingEnabled && this.unsubscribeFn) {
+      this.unsubscribeFn();
+    }
+  }
+
+  /**
+   * get will return the cached entry keyed on the tenantID.
+   *
+   * @param tenantID the tenantID for the cached item
+   */
+  public get(tenantID: string): T | undefined {
+    if (this.isCachingEnabled) {
+      return this.cache.get(tenantID);
+    }
+
+    return;
+  }
+
+  /**
+   * set will set the cached entry into the map (if caching is enabled).
+   *
+   * @param tenantID the tenantID for the cached item
+   * @param value the value to set in the map (if caching is enabled)
+   */
+  public set(tenantID: string, value: T): TenantCacheAdapter<T> {
+    if (this.isCachingEnabled) {
+      this.cache.set(tenantID, value);
+    }
+
+    return this;
+  }
+}

src/core/server/services/tenant/cache/index.ts

@@ -4,6 +4,7 @@ import { Db } from "mongodb";
 import uuid from "uuid";
 
 import { EventEmitter } from "events";
+import { Config } from "talk-common/config";
 import logger from "talk-server/logger";
 import {
   retrieveAllTenants,
@@ -29,21 +30,13 @@ export default class TenantCache {
   /**
    * tenantsByID reference the tenants that have been cached/retrieved by ID.
    */
-  private tenantsByID = new DataLoader<string, Readonly<Tenant> | null>(ids => {
-    logger.debug({ ids: ids.length }, "now loading tenants");
-    return retrieveManyTenants(this.mongo, ids);
-  });
+  private tenantsByID: DataLoader<string, Readonly<Tenant> | null>;
 
   /**
    * tenantsByDomain reference the tenants that have been cached/retrieved by
    * Domain.
    */
-  private tenantsByDomain = new DataLoader<string, Readonly<Tenant> | null>(
-    domains => {
-      logger.debug({ domains: domains.length }, "now loading tenants");
-      return retrieveManyTenantsByDomain(this.mongo, domains);
-    }
-  );
+  private tenantsByDomain: DataLoader<string, Readonly<Tenant> | null>;
 
   /**
    * Create a new client application ID. This prevents duplicated messages
@@ -54,23 +47,70 @@ export default class TenantCache {
 
   private mongo: Db;
   private emitter = new EventEmitter();
+  private cachingEnabled: boolean;
+
+  constructor(mongo: Db, subscriber: Redis, config: Config) {
+    this.cachingEnabled = !config.get("disable_tenant_caching");
+    if (!this.cachingEnabled) {
+      logger.warn("tenant caching is disabled");
+    } else {
+      logger.debug("tenant caching is enabled");
+    }
 
-  constructor(mongo: Db, subscriber: Redis) {
     // Save the Db reference.
     this.mongo = mongo;
 
-    // Attach to messages on this connection so we can receive updates when
-    // the tenant are changed.
-    subscriber.on("message", this.onMessage);
+    // Configure the data loaders.
+    this.tenantsByID = new DataLoader(
+      async ids => {
+        logger.debug({ ids: ids.length }, "now loading tenants");
+        const tenants = await retrieveManyTenants(this.mongo, ids);
+        logger.debug(
+          { tenants: tenants.filter(t => t !== null).length },
+          "loaded tenants"
+        );
+        return tenants;
+      },
+      {
+        cache: this.cachingEnabled,
+      }
+    );
 
-    // Subscribe to tenant notifications.
-    subscriber.subscribe(TENANT_UPDATE_CHANNEL);
+    this.tenantsByDomain = new DataLoader(
+      async domains => {
+        logger.debug({ domains: domains.length }, "now loading tenants");
+        const tenants = await retrieveManyTenantsByDomain(this.mongo, domains);
+        logger.debug(
+          { tenants: tenants.filter(t => t !== null).length },
+          "loaded tenants"
+        );
+        return tenants;
+      },
+      {
+        cache: this.cachingEnabled,
+      }
+    );
+
+    // We don't need updates if we aren't synced to tenant updates.
+    if (this.cachingEnabled) {
+      // Attach to messages on this connection so we can receive updates when
+      // the tenant are changed.
+      subscriber.on("message", this.onMessage);
+
+      // Subscribe to tenant notifications.
+      subscriber.subscribe(TENANT_UPDATE_CHANNEL);
+    }
   }
 
   /**
    * primeAll will load all the tenants into the cache on startup.
    */
   public async primeAll() {
+    if (!this.cachingEnabled) {
+      logger.debug("tenants not primed, caching disabled");
+      return;
+    }
+
     // Grab all the tenants for this node.
     const tenants = await retrieveAllTenants(this.mongo);
 

src/core/server/services/users/karma.ts

@@ -1,13 +1,13 @@
 import { get } from "lodash";
 
-import { KarmaThresholds } from "talk-server/models/settings";
+import { GQLKarmaThresholds } from "talk-server/graph/tenant/schema/__generated__/types";
 import { User } from "talk-server/models/user";
 
 export const getCommentTrustScore = (user: User): number =>
   get(user, "metadata.trust.comment.karma", 0);
 
 export const isReliableCommenter = (
-  thresholds: KarmaThresholds,
+  thresholds: GQLKarmaThresholds,
   user: User
 ): boolean | null => {
   const score = getCommentTrustScore(user);