Merge pull request #5 from PurdueRCAC/db-abstraction

Move database operations to a separate module

Author: FluxCapacitor2

backend/prisma/migrations/20251130070225_deployment_and_log_changes/migration.sql

@@ -0,0 +1,38 @@
+/*
+  Warnings:
+
+  - The values [STOPPED] on the enum `DeploymentStatus` will be removed. If these variants are still used in the database, this will fail.
+  - The values [SYSTEM] on the enum `LogType` will be removed. If these variants are still used in the database, this will fail.
+  - You are about to drop the column `builderJobId` on the `Deployment` table. All the data in the column will be lost.
+
+*/
+
+-- Set deployments with a status of STOPPED to COMPLETE
+UPDATE "Deployment" SET "status" = 'COMPLETE' WHERE "status" = 'STOPPED';
+
+-- AlterEnum
+BEGIN;
+CREATE TYPE "DeploymentStatus_new" AS ENUM ('QUEUED', 'PENDING', 'BUILDING', 'DEPLOYING', 'COMPLETE', 'ERROR', 'CANCELLED');
+ALTER TABLE "Deployment" ALTER COLUMN "status" DROP DEFAULT;
+ALTER TABLE "Deployment" ALTER COLUMN "status" TYPE "DeploymentStatus_new" USING ("status"::text::"DeploymentStatus_new");
+ALTER TYPE "DeploymentStatus" RENAME TO "DeploymentStatus_old";
+ALTER TYPE "DeploymentStatus_new" RENAME TO "DeploymentStatus";
+DROP TYPE "DeploymentStatus_old";
+ALTER TABLE "Deployment" ALTER COLUMN "status" SET DEFAULT 'QUEUED';
+COMMIT;
+
+-- Set logs with a status of SYSTEM to BUILD
+UPDATE "Log" SET "type" = 'BUILD' WHERE "type" = 'SYSTEM';
+
+-- AlterEnum
+BEGIN;
+CREATE TYPE "LogType_new" AS ENUM ('BUILD', 'RUNTIME');
+ALTER TABLE "Log" ALTER COLUMN "type" TYPE "LogType_new" USING ("type"::text::"LogType_new");
+ALTER TYPE "LogType" RENAME TO "LogType_old";
+ALTER TYPE "LogType_new" RENAME TO "LogType";
+DROP TYPE "LogType_old";
+COMMIT;
+
+-- AlterTable
+ALTER TABLE "Deployment" DROP COLUMN "builderJobId",
+ALTER COLUMN "status" SET DEFAULT 'QUEUED';

backend/prisma/schema.prisma

@@ -186,12 +186,11 @@ model Deployment {
   createdAt     DateTime @default(now())
   updatedAt     DateTime @updatedAt
   commitMessage String?
-  builderJobId  String?
   checkRunId    Int?
 
   // Used to update a deployment when the correct workflow run finishes
   workflowRunId Int?             @unique
-  status        DeploymentStatus @default(PENDING)
+  status        DeploymentStatus @default(QUEUED)
   // A random value used by the builder (in place of the deployment ID) to make changes to this Deployment.
   // API clients should only be able to update deployments if they have the deployment's secret or some other form of authorization.
   secret        String?          @unique
@@ -290,8 +289,6 @@ enum LogStream {
 }
 
 enum LogType {
-  // System logs created by the AnvilOps backend
-  SYSTEM
   // Logs from users' image builds
   BUILD
   // Logs from users' pods
@@ -311,8 +308,8 @@ enum DeploymentStatus {
   COMPLETE
   // There was a problem somewhere during the process
   ERROR
-  // Deployment is no longer active
-  STOPPED
+  // The deployment was building or deploying, but it was cancelled in favor of another deployment
+  CANCELLED
 }
 
 model Session {

backend/prisma/types.ts

@@ -1,11 +1,8 @@
-import { Prisma } from "../src/generated/prisma/client.ts";
-import { db } from "../src/lib/db.ts";
-
-const Resources = ["cpu", "memory"] as const;
 type BaseResources = {
   cpu?: string;
   memory?: string;
 };
+
 declare global {
   namespace PrismaJson {
     type EnvVar = {
@@ -25,11 +22,4 @@ declare global {
       isPreviewing: boolean;
     };
   }
-  type ExtendedDeploymentConfig = Prisma.Result<
-    typeof db.deploymentConfig,
-    {},
-    "findFirst"
-  >;
 }
-
-export {};

backend/src/db/crypto.ts

@@ -0,0 +1,60 @@
+import crypto, { createCipheriv, createDecipheriv } from "node:crypto";
+import { env } from "../lib/env.ts";
+
+const masterKey = Buffer.from(env.FIELD_ENCRYPTION_KEY, "base64");
+const separator = "|";
+
+const unwrapKey = (wrapped: string): Buffer => {
+  const iv = Buffer.alloc(8, 0xa6); // Recommended default initial value
+  const decipher = createDecipheriv("aes256-wrap", masterKey, iv);
+  return Buffer.concat([decipher.update(wrapped, "base64"), decipher.final()]);
+};
+
+const wrapKey = (key: Buffer): string => {
+  const iv = Buffer.alloc(8, 0xa6);
+  const cipher = createCipheriv("aes256-wrap", masterKey, iv);
+  return cipher.update(key, undefined, "base64") + cipher.final("base64");
+};
+
+const encrypt = (secret: string, key: Buffer): string => {
+  const iv = crypto.randomBytes(12);
+  const cipher = createCipheriv("aes-256-gcm", key, iv);
+  const res = Buffer.concat([cipher.update(secret), cipher.final()]);
+  const authTag = cipher.getAuthTag();
+  return [authTag, iv, res]
+    .map((buf) => buf.toString("base64"))
+    .join(separator);
+};
+
+const decrypt = (ctxtFull: string, key: Buffer): string => {
+  const [authTagEncoded, ivEncoded, ctxtEncoded] = ctxtFull.split(separator);
+  const iv = Buffer.from(ivEncoded, "base64");
+  const ctxt = Buffer.from(ctxtEncoded, "base64");
+  const authTag = Buffer.from(authTagEncoded, "base64");
+  const decipher = createDecipheriv("aes-256-gcm", key, iv);
+  decipher.setAuthTag(authTag);
+
+  return decipher.update(ctxt, undefined, "utf8") + decipher.final("utf8");
+};
+
+const genKey = (): Buffer => {
+  return crypto.randomBytes(32);
+};
+
+export const encryptEnv = (plaintext: PrismaJson.EnvVar[], key: string) => {
+  const unwrapped = unwrapKey(key);
+  return plaintext.map((envVar) => ({
+    ...envVar,
+    value: encrypt(envVar.value, unwrapped),
+  }));
+};
+
+export const decryptEnv = (ciphertext: PrismaJson.EnvVar[], key: string) => {
+  const unwrapped = unwrapKey(key);
+  return ciphertext.map((envVar) => ({
+    ...envVar,
+    value: decrypt(envVar.value, unwrapped),
+  }));
+};
+
+export const generateKey = () => wrapKey(genKey());

backend/src/db/index.ts

@@ -0,0 +1,117 @@
+import { PrismaPg } from "@prisma/adapter-pg";
+import type { DefaultArgs } from "@prisma/client/runtime/client";
+import connectPgSimple from "connect-pg-simple";
+import session from "express-session";
+import { Pool, type Notification } from "pg";
+import "../../prisma/types.ts";
+import { PrismaClient } from "../generated/prisma/client.ts";
+import { env } from "../lib/env.ts";
+import { AppRepo } from "./repo/app.ts";
+import { AppGroupRepo } from "./repo/appGroup.ts";
+import { CacheRepo } from "./repo/cache.ts";
+import { DeploymentRepo } from "./repo/deployment.ts";
+import { InvitationRepo } from "./repo/invitation.ts";
+import { OrganizationRepo } from "./repo/organization.ts";
+import { RepoImportStateRepo } from "./repo/repoImportState.ts";
+import { UserRepo } from "./repo/user.ts";
+
+export class NotFoundError extends Error {}
+export class ConflictError extends Error {}
+
+export type PrismaClientType = PrismaClient<
+  { adapter: PrismaPg; omit: { deployment: { secret: true } } },
+  never,
+  DefaultArgs
+>;
+
+/**
+ * A Postgres database implementation
+ */
+export class Database {
+  private DATABASE_URL =
+    env.DATABASE_URL ??
+    `postgresql://${env.POSTGRES_USER}:${env.POSTGRES_PASSWORD}@${env.POSTGRES_HOSTNAME}/${env.POSTGRES_DB}`;
+
+  private pool = new Pool({
+    connectionString: this.DATABASE_URL,
+    connectionTimeoutMillis: 5000,
+  });
+
+  private prismaPostgresAdapter = new PrismaPg({
+    connectionString: this.DATABASE_URL,
+  });
+
+  private client: PrismaClientType = new PrismaClient({
+    adapter: this.prismaPostgresAdapter,
+    omit: {
+      deployment: {
+        secret: true,
+      },
+    },
+  });
+
+  app = new AppRepo(this.client);
+
+  appGroup = new AppGroupRepo(this.client);
+
+  cache = new CacheRepo(this.client);
+
+  deployment = new DeploymentRepo(this.client, this.publish.bind(this));
+
+  invitation = new InvitationRepo(this.client);
+
+  org = new OrganizationRepo(this.client);
+
+  repoImportState = new RepoImportStateRepo(this.client);
+
+  user = new UserRepo(this.client);
+
+  sessionStore = new (connectPgSimple(session))({
+    conString: this.DATABASE_URL,
+  });
+
+  /**
+   * Subscribes to the given channel and runs the callback when a message is received on that channel.
+   *
+   * @returns A cleanup function to remove the listener
+   */
+  async subscribe(channel: string, callback: (msg: Notification) => void) {
+    const conn = await this.pool.connect();
+    if (!channel.match(/^[a-zA-Z0-9_]+$/g)) {
+      // Sanitize against potential SQL injection. Postgres unfortunately doesn't provide a way to parameterize the
+      // channel name for LISTEN and UNLISTEN, so we validate that the channel name is a valid SQL identifier here.
+      throw new Error(
+        "Invalid channel name: '" +
+          channel +
+          "'. Expected only letters, numbers, and underscores.",
+      );
+    }
+
+    const listener = (msg: Notification) => {
+      if (msg.channel === channel) {
+        callback(msg);
+      }
+    };
+    conn.on("notification", listener);
+
+    await conn.query(`LISTEN "${channel}"`);
+
+    return async () => {
+      await conn.query(`UNLISTEN "${channel}"`);
+      conn.off("notification", listener);
+      conn.release();
+    };
+  }
+
+  /**
+   * Publishes a message on the given channel.
+   * @see {subscribe}
+   * @param channel The channel to publish on
+   * @param payload The message to publish
+   */
+  async publish(channel: string, payload: string) {
+    await this.pool.query("SELECT pg_notify($1, $2);", [channel, payload]);
+  }
+}
+
+export const db = new Database();