fix(database): make restores atomic and validate users schema before session migration

docker-entrypoint.sh

@@ -45,6 +45,7 @@ NODE
 if [ -d /app/prisma/migrations ] && [ "$(find /app/prisma/migrations -mindepth 1 -maxdepth 1 | wc -l)" -gt 0 ]; then
   db_action="$(node <<'NODE'
 const { PrismaClient } = require("@prisma/client");
+const { readdirSync } = require("fs");
 
 const prisma = new PrismaClient();
   async function main() {
@@ -75,6 +76,7 @@ const prisma = new PrismaClient();
 
     let failedMigrations = [];
     let migrationCount = 0;
+    let appliedMigrationNames = new Set();
 
     if (hasMigrationTable) {
       failedMigrations = await prisma.$queryRaw`
@@ -90,8 +92,27 @@ const prisma = new PrismaClient();
       `;
 
       migrationCount = migrationCountResult[0].count;
+
+      const appliedMigrationRows = await prisma.$queryRaw`
+        SELECT migration_name
+        FROM "_prisma_migrations"
+        WHERE finished_at IS NOT NULL
+          AND rolled_back_at IS NULL
+      `;
+      appliedMigrationNames = new Set(
+        appliedMigrationRows.map((row) => row.migration_name)
+      );
     }
 
+    const migrationNames = readdirSync("/app/prisma/migrations", {
+      withFileTypes: true,
+    })
+      .filter((entry) => entry.isDirectory())
+      .map((entry) => entry.name);
+    const pendingMigrationNames = migrationNames.filter(
+      (migrationName) => !appliedMigrationNames.has(migrationName)
+    );
+
     /**
     * STATE #1
     * Fresh database
@@ -172,6 +193,65 @@ const prisma = new PrismaClient();
         return;
       }
 
+      if (pendingMigrationNames.length > 0) {
+        const idColumnIntegrity = await prisma.$queryRaw`
+          SELECT
+            table_class.relname AS "tableName",
+            attribute.attnotnull AS "isNotNull",
+            EXISTS (
+              SELECT 1
+              FROM pg_index i
+              WHERE i.indrelid = table_class.oid
+                AND i.indisunique
+                AND i.indisvalid
+                AND i.indisready
+                AND i.indnkeyatts = 1
+                AND i.indkey[0] = attribute.attnum
+            ) AS "hasUniqueId"
+          FROM pg_class table_class
+          JOIN pg_namespace namespace
+            ON namespace.oid = table_class.relnamespace
+          JOIN pg_attribute attribute
+            ON attribute.attrelid = table_class.oid
+          WHERE namespace.nspname = 'public'
+            AND table_class.relkind IN ('r', 'p')
+            AND table_class.relname <> '_prisma_migrations'
+            AND attribute.attname = 'id'
+            AND NOT attribute.attisdropped
+          ORDER BY table_class.relname
+        `;
+
+        const invalidIdColumns = idColumnIntegrity.filter(
+          (row) => !row.hasUniqueId || !row.isNotNull
+        );
+
+        if (invalidIdColumns.length > 0) {
+          console.error(
+            `Database integrity check failed before applying pending migrations: ${pendingMigrationNames.join(", ")}.`
+          );
+
+          for (const invalidColumn of invalidIdColumns) {
+            const quotedTableName = `"${invalidColumn.tableName.replace(/"/g, '""')}"`;
+            const [counts] = await prisma.$queryRawUnsafe(`
+              SELECT
+                COUNT(*) FILTER (WHERE "id" IS NULL)::int AS "nullIds",
+                (COUNT(*) - COUNT(DISTINCT "id"))::int AS "duplicateIds"
+              FROM ${quotedTableName}
+            `);
+
+            console.error(
+              `${invalidColumn.tableName}.id must be NOT NULL and have a valid single-column PRIMARY KEY or UNIQUE index (duplicate rows: ${counts.duplicateIds}, null IDs: ${counts.nullIds}).`
+            );
+          }
+
+          console.error(
+            "Restore missing database constraints and resolve duplicate IDs before applying migrations."
+          );
+          console.log("blocked");
+          return;
+        }
+      }
+
       console.error(
         "Prisma migration history detected. Applying pending migrations."
       );

src/server/lib/database-restore.ts

@@ -0,0 +1,38 @@
+import { Transform } from "node:stream";
+
+export const CUSTOM_RESTORE_SCRIPT_ARGS = [
+  "--clean",
+  "--if-exists",
+  "--no-owner",
+  "--no-privileges",
+] as const;
+
+export const ATOMIC_PSQL_RESTORE_ARGS = [
+  "--single-transaction",
+  "--set",
+  "ON_ERROR_STOP=1",
+] as const;
+
+export function isUnsupportedRestoreSetting(line: string) {
+  return line.trim() === "SET transaction_timeout = 0;";
+}
+
+export function createRestoreSqlSanitizer() {
+  let pending = "";
+
+  return new Transform({
+    transform(chunk, _encoding, callback) {
+      pending += chunk.toString();
+      const lines = pending.split(/\r?\n/);
+      pending = lines.pop() ?? "";
+      const output = lines
+        .filter((line) => !isUnsupportedRestoreSetting(line))
+        .map((line) => `${line}\n`)
+        .join("");
+      callback(null, output);
+    },
+    flush(callback) {
+      callback(null, isUnsupportedRestoreSetting(pending) ? "" : pending);
+    },
+  });
+}

src/server/routes/settings.ts

@@ -7,6 +7,11 @@ import { promisify } from "node:util";
 import { pipeline } from "node:stream/promises";
 import { z } from "zod";
 import { resolvePublicAppOrigin } from "../lib/public-url";
+import {
+  ATOMIC_PSQL_RESTORE_ARGS,
+  createRestoreSqlSanitizer,
+  CUSTOM_RESTORE_SCRIPT_ARGS,
+} from "../lib/database-restore";
 import prisma from "../lib/prisma";
 import { decrypt, encrypt } from "../lib/crypto";
 import { authenticate, requireRole } from "../middleware/auth";
@@ -714,7 +719,7 @@ export async function settingsRoutes(app: FastifyInstance) {
     options: {
       inputPath?: string;
       outputPath?: string;
-      allowTransactionTimeoutCompatibilityError?: boolean;
+      connectToDatabase?: boolean;
     } = {},
   ) => {
     const databaseUrl = process.env.DATABASE_URL?.trim();
@@ -771,26 +776,7 @@ export async function settingsRoutes(app: FastifyInstance) {
               ? rejectedResult.reason.message
               : "";
 
-          const onlyTransactionTimeoutCompatibilityError =
-            options.allowTransactionTimeoutCompatibilityError &&
-            stderr.includes(
-              'unrecognized configuration parameter "transaction_timeout"',
-            ) &&
-            stderr
-              .replace(
-                /pg_restore: error: could not execute query: ERROR:\s+unrecognized configuration parameter "transaction_timeout"[\r\n]+Command was:\s*SET transaction_timeout = 0;\s*/g,
-                "",
-              )
-              .replace(
-                /pg_restore: warning: errors ignored on restore: 1\s*/g,
-                "",
-              )
-              .trim().length === 0;
-
-          if (
-            !rejectedResult &&
-            (exitCode === 0 || onlyTransactionTimeoutCompatibilityError)
-          ) {
+          if (!rejectedResult && exitCode === 0) {
             resolve();
             return;
           }
@@ -806,20 +792,29 @@ export async function settingsRoutes(app: FastifyInstance) {
     const databaseContainer =
       process.env.DATABASE_BACKUP_CONTAINER?.trim() ||
       `${process.env.COMPOSE_PROJECT_NAME?.trim() || "doktainer"}-postgres`;
+    const databaseArgs =
+      options.connectToDatabase === false
+        ? []
+        : ["--dbname", databaseCliUrl];
     const dockerArgs = [
       "exec",
       "-i",
       databaseContainer,
       command,
-      "--dbname",
-      databaseCliUrl,
+      ...databaseArgs,
       ...args,
     ];
     const localBinary = await findLocalPostgresBinary(command);
     const runLocal = async () => {
       if (!localBinary) return false;
-      const localArgs = ["--dbname", databaseCliUrl, ...args];
-      if (options.inputPath) localArgs.push(options.inputPath);
+      const localArgs = [...databaseArgs, ...args];
+      if (options.inputPath) {
+        if (command === "psql") {
+          localArgs.push("--file", options.inputPath);
+        } else {
+          localArgs.push(options.inputPath);
+        }
+      }
       await run(
         localBinary,
         localArgs,
@@ -898,6 +893,8 @@ export async function settingsRoutes(app: FastifyInstance) {
           .status(400)
           .send({ success: false, error: "Select a PostgreSQL dump file" });
       const filePath = join(tmpdir(), `doktainer-restore-${Date.now()}.dump`);
+      const restoredSqlPath = `${filePath}.sql`;
+      const sanitizedSqlPath = `${filePath}.sanitized.sql`;
       try {
         await pipeline(upload.file, createWriteStream(filePath));
         const header = Buffer.alloc(5);
@@ -909,24 +906,43 @@ export async function settingsRoutes(app: FastifyInstance) {
         }
 
         const isCustomArchive = header.toString("ascii") === "PGDMP";
+        if (isCustomArchive) {
+          await runDatabaseCommand(
+            "pg_restore",
+            [...CUSTOM_RESTORE_SCRIPT_ARGS],
+            {
+              inputPath: filePath,
+              outputPath: restoredSqlPath,
+              connectToDatabase: false,
+            },
+          );
+          await pipeline(
+            createReadStream(restoredSqlPath),
+            createRestoreSqlSanitizer(),
+            createWriteStream(sanitizedSqlPath),
+          );
+        }
+
         await runDatabaseCommand(
-          isCustomArchive ? "pg_restore" : "psql",
-          isCustomArchive
-            ? ["--clean", "--if-exists", "--no-owner"]
-            : ["--set", "ON_ERROR_STOP=1"],
-          {
-            inputPath: filePath,
-            allowTransactionTimeoutCompatibilityError: isCustomArchive,
-          },
+          "psql",
+          [...ATOMIC_PSQL_RESTORE_ARGS],
+          { inputPath: isCustomArchive ? sanitizedSqlPath : filePath },
         );
-        await auditLog({
-          userId: req.userId,
-          organizationId: req.organizationId,
-          action: "DATABASE_RESTORE",
-          category: "SYSTEM",
-          level: "WARNING",
-          message: "Doktainer database restored from uploaded dump",
-        });
+        try {
+          await auditLog({
+            userId: req.userId,
+            organizationId: req.organizationId,
+            action: "DATABASE_RESTORE",
+            category: "SYSTEM",
+            level: "WARNING",
+            message: "Doktainer database restored from uploaded dump",
+          });
+        } catch (error) {
+          app.log.warn(
+            { err: error },
+            "Database restore completed, but its audit event could not be recorded",
+          );
+        }
         return reply.send({
           success: true,
           message: "Database restore completed",
@@ -942,7 +958,11 @@ export async function settingsRoutes(app: FastifyInstance) {
                 : "Database restore failed",
           });
       } finally {
-        await fs.rm(filePath, { force: true });
+        await Promise.all([
+          fs.rm(filePath, { force: true }),
+          fs.rm(restoredSqlPath, { force: true }),
+          fs.rm(sanitizedSqlPath, { force: true }),
+        ]);
       }
     },
   );

tests/backend/database-restore-safety.test.ts

@@ -0,0 +1,59 @@
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+import { Readable } from "node:stream";
+import test from "node:test";
+
+import {
+  ATOMIC_PSQL_RESTORE_ARGS,
+  createRestoreSqlSanitizer,
+  CUSTOM_RESTORE_SCRIPT_ARGS,
+  isUnsupportedRestoreSetting,
+} from "../../src/server/lib/database-restore";
+
+async function sanitizeChunks(chunks: string[]) {
+  const output: Buffer[] = [];
+  const sanitizer = createRestoreSqlSanitizer();
+  sanitizer.on("data", (chunk: Buffer) => output.push(chunk));
+  await new Promise<void>((resolve, reject) => {
+    Readable.from(chunks).pipe(sanitizer).once("finish", resolve).once("error", reject);
+  });
+  return Buffer.concat(output).toString("utf8");
+}
+
+test("database restores execute atomically and stop on SQL errors", () => {
+  assert.deepEqual(ATOMIC_PSQL_RESTORE_ARGS, [
+    "--single-transaction",
+    "--set",
+    "ON_ERROR_STOP=1",
+  ]);
+  assert.deepEqual(CUSTOM_RESTORE_SCRIPT_ARGS, [
+    "--clean",
+    "--if-exists",
+    "--no-owner",
+    "--no-privileges",
+  ]);
+});
+
+test("restore sanitizer removes only the unsupported transaction timeout setting", async () => {
+  assert.equal(isUnsupportedRestoreSetting("SET transaction_timeout = 0;"), true);
+  assert.equal(isUnsupportedRestoreSetting("SET statement_timeout = 0;"), false);
+  assert.equal(
+    await sanitizeChunks([
+      "SET statement_timeout = 0;\nSET trans",
+      "action_timeout = 0;\nCREATE TABLE example (id text);\n",
+    ]),
+    "SET statement_timeout = 0;\nCREATE TABLE example (id text);\n",
+  );
+});
+
+test("container startup validates id invariants before any pending migration", () => {
+  const entrypoint = readFileSync("docker-entrypoint.sh", "utf8");
+
+  assert.doesNotMatch(entrypoint, /20260909_add_database_sessions/);
+  assert.match(entrypoint, /pendingMigrationNames\.length > 0/);
+  assert.match(entrypoint, /table_class\.relname <> '_prisma_migrations'/);
+  assert.match(entrypoint, /i\.indisunique/);
+  assert.match(entrypoint, /attribute\.attnotnull/);
+  assert.match(entrypoint, /Database integrity check failed/);
+  assert.match(entrypoint, /resolve duplicate IDs before applying migrations/);
+});