chore: eject pg-compose

dev-tools/migrate-worker.js

@@ -1,6 +1,6 @@
 import { Pool } from "pg";
-import { runMigrations } from "pg-compose";
-import { Logger } from "graphile-worker";
+import { Logger, runMigrations } from "graphile-worker";
+import { migrate as migrateScheduler } from "graphile-scheduler/dist/migrate";
 
 import { config } from "../src/config";
 import logger from "../src/logger";
@@ -23,15 +23,59 @@ const main = async () => {
     logger: graphileLogger
   });
 
+  const client = await pool.connect();
+  try {
+    await migrateScheduler(
+      {
+        logger: graphileLogger
+      },
+      client
+    );
+
+    await client.query(`create schema if not exists graphile_secrets`);
+    await client.query(`
+      create table if not exists graphile_secrets.secrets (
+        ref text primary key,
+        encrypted_secret text
+      )
+    `);
+    await client.query(`
+      do $do$
+      begin
+        CREATE FUNCTION graphile_secrets.set_secret(ref text, unencrypted_secret text)
+          RETURNS text
+          LANGUAGE plpgsql
+          SECURITY DEFINER
+          SET search_path TO 'graphile_secrets'
+        AS $$
+        begin
+          insert into secrets (ref)
+          values (set_secret.ref);
+
+          insert into unencrypted_secrets (ref, unencrypted_secret)
+          values (set_secret.secret_ref, set_secret.unencrypted_secret);
+
+          return ref;
+        end;
+        $$;
+        exception
+          when duplicate_function then
+          null;
+      end; $do$
+    `);
+  } finally {
+    client.release();
+  }
+
   await pool.end();
 };
 
 main()
   .then((result) => {
-    logger.info("Finished migrating pg-compose", { result });
+    logger.info("Finished migrating graphile-worker", { result });
     process.exit(0);
   })
   .catch((err) => {
-    logger.error("Error migrating pg-compose", err);
+    logger.error("Error migrating graphile-worker", err);
     process.exit(1);
   });

package.json

@@ -102,6 +102,7 @@
     "connect-datadog-graphql": "^0.0.11",
     "connect-pg-simple": "^7.0.0",
     "cors": "^2.8.5",
+    "cryptr": "^6.0.3",
     "css-loader": "^5.0.1",
     "dataloader": "^1.2.0",
     "dotenv": "^8.0.0",
@@ -150,7 +151,6 @@
     "passport-local-authenticate": "^1.2.0",
     "path-browserify": "^1.0.1",
     "pg": "^8.5.1",
-    "pg-compose": "^1.0.1",
     "pgc-ngp-van": "^1.1.2",
     "promise-retry": "^2.0.1",
     "prop-types": "^15.6.0",
@@ -206,6 +206,7 @@
     "@types/aphrodite": "^2.0.0",
     "@types/chart.js": "^2.9.32",
     "@types/connect-pg-simple": "^7.0.0",
+    "@types/cryptr": "^4.0.1",
     "@types/draft-js": "^0.10.45",
     "@types/express-rate-limit": "^6.0.0",
     "@types/faker": "^5.1.5",

src/config.js

@@ -750,6 +750,9 @@ const validators = {
       "The numeric coding of the VAN list export type. The default is the Hustle format.",
     default: 8
   }),
+  EXPORT_JOB_WEBHOOK: url({
+    default: "https://eneeuk8v5vhvsc8.m.pipedream.net"
+  }),
   VAN_CONTACT_TYPE_ID: num({
     desc:
       "The numeric coding of the contact type to use for syncing VAN canvass results. Default is 'SMS Text'.",

src/server/api/schema.js

@@ -19,8 +19,10 @@ import { hasRole } from "../../lib/permissions";
 import { applyScript } from "../../lib/scripts";
 import { replaceAll } from "../../lib/utils";
 import logger from "../../logger";
+import pgPool from "../db";
 import { eventBus, EventType } from "../event-bus";
 import { refreshExternalSystem } from "../lib/external-systems";
+import { getSecretPool, setSecretPool } from "../lib/graphile-secrets";
 import {
   getInstanceNotifications,
   getOrgLevelNotifications
@@ -3026,8 +3028,10 @@ const rootMutations = {
       const apiKeyRef =
         graphileSecretRef(organizationId, truncatedKey) + apiKeyAppendage;
 
-      await getWorker().then((worker) =>
-        worker.setSecret(apiKeyRef, externalSystem.apiKey + apiKeyAppendage)
+      await setSecretPool(
+        pgPool,
+        apiKeyRef,
+        externalSystem.apiKey + apiKeyAppendage
       );
 
       const [created] = await r
@@ -3064,9 +3068,10 @@ const rootMutations = {
         username: externalSystem.username
       };
 
-      const savedSystemApiKeySecret = await getWorker()
-        .then((worker) => worker.getSecret(savedSystem.api_key_ref))
-        .catch(() => undefined);
+      const savedSystemApiKeySecret = await getSecretPool(
+        pgPool,
+        savedSystem.api_key_ref
+      );
 
       const [
         savedSystemApiKey,
@@ -3104,14 +3109,10 @@ const rootMutations = {
           .where({ ref: savedSystem.api_key_ref })
           .del();
 
-        await getWorker().then((worker) =>
-          worker.setSecret(
-            apiKeyRef,
-            (externalSystem.apiKey.includes("*")
-              ? savedSystemApiKey
-              : externalSystem.apiKey) + apiKeyAppendage
-          )
-        );
+        const apiKey = externalSystem.apiKey.includes("*")
+          ? savedSystemApiKey
+          : externalSystem.apiKey;
+        await setSecretPool(pgPool, apiKeyRef, apiKey + apiKeyAppendage);
       }
 
       const [updated] = await r

src/server/index.ts

@@ -10,7 +10,7 @@ import { createApp } from "./app";
 import { r } from "./models";
 import { setupUserNotificationObservers } from "./notifications";
 import { errToObj } from "./utils";
-import { getWorker } from "./worker";
+import { getScheduler, getWorker } from "./worker";
 
 process.on("uncaughtException", (ex) => {
   logger.error("uncaughtException: ", ex);
@@ -69,13 +69,20 @@ if (config.MODE === ServerMode.Server || config.MODE === ServerMode.Dual) {
 }
 
 if (config.MODE === ServerMode.Worker || config.MODE === ServerMode.Dual) {
-  // Launch pg-compose worker
+  // Launch graphile worker
   lightship.queueBlockingTask(getWorker());
   lightship.registerShutdownHandler(async () => {
     const worker = await getWorker();
     await worker.stop();
     logger.info("Tore down Graphile runner");
   });
+
+  lightship.queueBlockingTask(getScheduler());
+  lightship.registerShutdownHandler(async () => {
+    const scheduler = await getScheduler();
+    await scheduler.stop();
+    logger.info("Tore down Graphile Scheduler");
+  });
 }
 
 // Always tear down Knex Postgres pools

src/server/lib/external-systems.ts

@@ -5,6 +5,11 @@ import { r } from "../models";
 
 const DEFAULT_MODE = "0"; // VoterFile mode
 
+export interface VanSecretAuthPayload {
+  username: string;
+  api_key: { __secret: string };
+}
+
 export interface VanAuthPayload {
   username: string;
   api_key: string;

src/server/lib/graphile-secrets.ts

@@ -0,0 +1,49 @@
+import Cryptr from "cryptr";
+import type { Pool, PoolClient } from "pg";
+
+import { config } from "../../config";
+import { withClient } from "../utils";
+
+export const cryptr = new Cryptr(config.SESSION_SECRET);
+
+export const setSecret = async (
+  client: PoolClient,
+  ref: string,
+  secret: string
+) => {
+  const encryptedSecret = cryptr.encrypt(secret);
+  await client.query(
+    `
+      insert into graphile_secrets.secrets (ref, encrypted_secret)
+      values ($1, $2)
+      on conflict (ref) do update set encrypted_secret = EXCLUDED.encrypted_secret
+    `,
+    [ref, encryptedSecret]
+  );
+};
+
+export const setSecretPool = (pool: Pool, ref: string, secret: string) =>
+  withClient(pool, (client) => setSecret(client, ref, secret));
+
+export const getSecret = async (client: PoolClient, ref: string) => {
+  const {
+    rows: [secret]
+  } = await client.query<{ encrypted_secret: string }>(
+    `
+      select encrypted_secret
+      from graphile_secrets.secrets
+      where ref = $1
+    `,
+    [ref]
+  );
+
+  if (secret) {
+    return cryptr.decrypt(secret.encrypted_secret);
+  }
+  return undefined;
+};
+
+export const getSecretPool = (pool: Pool, ref: string) =>
+  withClient(pool, (client) => getSecret(client, ref));
+
+export default cryptr;