Nuxt+Prisma+Dockerでデータベース（PostgreSql）の自動同期を実現

本稿では、Dockerにおけるデータベース同期のソリューションを紹介します。本稿の基本的な開発環境はNuxt4 + Prisma + Dockerであり、同期はコンテナ起動後、アプリケーションプロセスが立ち上がった後に発生します。核となるロジックは、データベース同期スクリプトとプロジェクト起動時に実行されるスクリプトです。Nuxtには`Server Start`タイプのプラグイン開発フックが提供されているため、非常に簡単に実現できます。その他の開発環境については、ご自身で実装を参考にしてください。

**Dockerイメージ構築段階では**データベースマイグレーションは実行されません。同期は**コンテナ起動後、アプリケーションプロセスが立ち上がった後**に発生し、Nitro起動プラグインがカスタムマイグレーションランナーを呼び出し、イメージ内の`prisma/migrations/*/migration.sql`を読み取り、Prisma互換の`_prisma_migrations`テーブルに書き込みます。

再現の要点は以下の3点です。

1.  **イメージに**`prisma/migrations`（および実行時Prisma Clientの成果物`prisma/generated`）**を含める**。
2.  **`server/lib/db-migrations.ts`を実装する**（以下の完全なコードを参照）。
3.  **アプリケーション起動の最も早い段階で**`ensureDatabaseMigrations()`を呼び出す（以下のプラグイン例を参照）。

`prisma/migrations`ディレクトリの内容は`prisma migrate dev`などのコマンドによって自動生成されます。各プロジェクトのスキーマは異なるため、本稿ではマイグレーションSQL自体については詳しく述べません。

## 全体フロー

```mermaid
sequenceDiagram
  participant Docker as Docker 構築
  participant CMD as node server/index.mjs
  participant Plugin as Nitro 起動プラグイン
  participant Mig as db-migrations.ts
  participant PG as PostgreSQL

  Docker->>Docker: pnpm build（prisma generate を含む）
  Docker->>Docker: COPY .output + prisma/migrations + prisma/generated
  Note over Docker: 構築期はDBに接続せず、マイグレーションも行わない

  CMD->>Plugin: プロセス起動
  Plugin->>Mig: ensureDatabaseMigrations()
  Mig->>PG: DBが存在しない場合は CREATE DATABASE
  Mig->>PG: advisory lock + _prisma_migrations の読み書き
  Mig->>PG: 未適用 migration.sql の実行
  Plugin->>CMD: マイグレーション完了後、シードデータなどの初期化を続行
```

「`entrypoint.sh`で`npx prisma migrate deploy`を実行する」方法と比較して、現在の**本番DockerfileのCMDは直接Nodeを起動**し、イメージ内にPrisma CLIをインストールすることに依存しません。マイグレーションロジックは、アプリケーション内で`pg`ドライバーを使用して完全に実行されます。

リポジトリには`docker/entrypoint.sh`（Prisma公式CLI方式）がまだ残っていますが、**現在のDockerfileのENTRYPOINT/CMDにはマウントされていません**。これはオプションの参考としてのみ提供されており、文末を参照してください。

---

## 環境変数

| 変数 | 必須 | 説明 |
|------|------|------|
| `DATABASE_URL` | はい | ターゲットデータベース接続文字列、データベース名を含む必要があります。`?schema=public`はPostgreSQLスキーマを指定します。 |
| `DATABASE_AUTO_MIGRATE` | いいえ | `false`に設定すると自動マイグレーションをスキップします。 |
| `DATABASE_BOOTSTRAP_URL` | いいえ | `CREATE DATABASE`に使用する管理接続。デフォルトでは`DATABASE_URL`のデータベース名を`postgres`に変更します。 |
| `DATABASE_SCHEMA` | いいえ | URLに`schema`パラメータがない場合のスキーマ名。デフォルトは`public`です。 |

Composeの例（データベース関連部分のみ表示）：

```yaml
environment:
  DATABASE_URL: "postgresql://postgres:パスワード@ホスト:5432/データベース名?schema=public"
  # DATABASE_AUTO_MIGRATE: "false"   # 外部運用でマイグレーションを実行する場合は無効にできます
  # DATABASE_BOOTSTRAP_URL: "postgresql://postgres:パスワード@ホスト:5432/postgres"
```

---

## Dockerイメージ：構築期に何をするか

構築段階では、アプリケーションのパッケージ化とマイグレーションファイルのコピーのみを行い、**データベースには接続しません**。

```dockerfile
FROM node:22.21.1-alpine3.22 AS builder
WORKDIR /app
RUN corepack enable
COPY package.json pnpm-lock.yaml ./
RUN pnpm install --frozen-lockfile
COPY . .
RUN pnpm build

FROM node:22.21.1-alpine3.22 AS runner
WORKDIR /app

COPY --from=builder /app/.output ./
COPY --from=builder /app/prisma/generated ./prisma/generated
COPY --from=builder /app/prisma/migrations ./prisma/migrations

ENV DATABASE_URL="postgresql://user:pass@host:5432/dbname?schema=public"
EXPOSE 9090
CMD ["node", "/app/server/index.mjs"]
```

説明：

-   `pnpm build`の前に、開発機/CIで`prisma generate`が実行され、`prisma/generated`が存在し、イメージにCOPYされる必要があります。
-   実行時の作業ディレクトリは`/app`であり、マイグレーションランナーは`process.cwd()/prisma/migrations`からSQLを読み取ります。これは上記のCOPYパスと一致します。
-   最終的なイメージには`prisma` CLIパッケージを**インストールする必要はありません**。

---

## 起動プラグイン：ビジネス初期化の前にマイグレーションを実行

マイグレーションは、Prismaクエリの前に完了する必要があります。Nitroプラグインを介して起動時に`await`します。

```typescript
import { ensureDatabaseMigrations } from "~~/server/lib/db-migrations";

async function runStartupInitialization() {
  await ensureDatabaseMigrations();
}

export default defineNitroPlugin(async () => {
  await runStartupInitialization();
});
```

同じプラグインファイル内の後続の`prisma.systemConfig.create`などは**データシード**に属し、構造同期ではありません。再現時には`ensureDatabaseMigrations()`の呼び出しを保持し、シードロジックはプロジェクトに応じて削除または削減してください。

---

## コア：カスタムマイグレーションランナー（完全なコード）

パスの規約：`server/lib/db-migrations.ts`。

動作の概要：

1.  ディレクトリ名でソートして`prisma/migrations/<name>/migration.sql`を読み取ります。
2.  ターゲットデータベースが存在しない場合（PostgreSQL `3D000`）、bootstrap接続を使用して`CREATE DATABASE`を実行します。
3.  スキーマを作成/使用し、`pg_advisory_lock`を取得して複数インスタンスの同時マイグレーションを回避します。
4.  `_prisma_migrations`テーブル（フィールドはPrismaと一致）を管理し、チェックサムはSQLのSHA-256です。
5.  **ベースライン**：既存のビジネステーブルがあるがマイグレーション履歴がない場合、`validateExistingSchemaIsLatest`で検証が成功すれば、マイグレーション記録のみを書き込み、SQLを重複して実行しません（次節の適応説明を参照）。
6.  残りの未適用マイグレーションをトランザクション内でSQLを実行し、記録します。

依存関係：`pg`（`package.json`の`dependencies`に既に含まれています）。

```typescript
import { createHash, randomUUID } from "node:crypto";
import { access, readdir, readFile } from "node:fs/promises";
import path from "node:path";
import { Client } from "pg";

const MIGRATIONS_DIR = path.resolve(process.cwd(), "prisma", "migrations");
const MIGRATIONS_TABLE = "_prisma_migrations";
const LOCK_KEY_1 = 20260328;
const LOCK_KEY_2 = 1;

type MigrationFile = {
  name: string;
  sql: string;
  checksum: string;
};

type DatabaseTarget = {
  databaseUrl: string;
  adminUrl: string;
  databaseName: string;
  schemaName: string;
};

// テーブル名リスト、ここでは例として挙げるのみ
const EXPECTED_TABLES = [
  "users",
  "settings",
];

let migrationTask: Promise<void> | null = null;

function getChecksum(sql: string) {
  return createHash("sha256").update(sql).digest("hex");
}

function quoteIdentifier(value: string) {
  return `"${value.replaceAll(`"`, `""`)}"`;
}

function getDatabaseTarget(): DatabaseTarget {
  const databaseUrl = process.env.DATABASE_URL;
  if (!databaseUrl) {
    throw new Error("DATABASE_URL is required before applying migrations.");
  }

  const parsed = new URL(databaseUrl);
  const databaseName = decodeURIComponent(parsed.pathname.replace(/^\//, ""));
  if (!databaseName) {
    throw new Error("DATABASE_URL must include a database name.");
  }

  const schemaName =
    parsed.searchParams.get("schema") ||
    process.env.DATABASE_SCHEMA ||
    "public";

  const adminUrl =
    process.env.DATABASE_BOOTSTRAP_URL ||
    (() => {
      const bootstrap = new URL(databaseUrl);
      bootstrap.pathname = "/postgres";
      return bootstrap.toString();
    })();

  return {
    databaseUrl,
    adminUrl,
    databaseName,
    schemaName,
  };
}

async function readMigrationFiles(): Promise<MigrationFile[]> {
  try {
    await access(MIGRATIONS_DIR);
  } catch {
    return [];
  }

  const entries = await readdir(MIGRATIONS_DIR, { withFileTypes: true });
  const migrations = await Promise.all(
    entries
      .filter((entry) => entry.isDirectory())
      .sort((a, b) => a.name.localeCompare(b.name))
      .map(async (entry) => {
        const sql = await readFile(
          path.join(MIGRATIONS_DIR, entry.name, "migration.sql"),
          "utf8"
        );

        return {
          name: entry.name,
          sql,
          checksum: getChecksum(sql),
        };
      })
  );

  return migrations;
}

async function ensureMigrationsTable(client: Client) {
  await client.query(`
    CREATE TABLE IF NOT EXISTS "${MIGRATIONS_TABLE}" (
      "id" TEXT PRIMARY KEY,
      "checksum" TEXT NOT NULL,
      "finished_at" TIMESTAMPTZ,
      "migration_name" TEXT NOT NULL UNIQUE,
      "logs" TEXT,
      "rolled_back_at" TIMESTAMPTZ,
      "started_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
      "applied_steps_count" INTEGER NOT NULL DEFAULT 0
    );
  `);
}

async function ensureSchema(client: Client, schemaName: string) {
  await client.query(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(schemaName)}`);
  await client.query(`SET search_path TO ${quoteIdentifier(schemaName)}`);
}

async function loadAppliedMigrations(client: Client) {
  const result = await client.query<{
    migration_name: string;
    checksum: string;
  }>(
    `SELECT "migration_name", "checksum"
     FROM "${MIGRATIONS_TABLE}"
     WHERE "rolled_back_at" IS NULL`
  );

  return new Map(
    result.rows.map((row) => [row.migration_name, row.checksum] as const)
  );
}

async function hasUserTables(client: Client) {
  const result = await client.query<{ exists: boolean }>(
    `
      SELECT EXISTS (
        SELECT 1
        FROM information_schema.tables
        WHERE table_schema = current_schema()
          AND table_name = ANY($1::text[])
      ) AS "exists"
    `,
    [EXPECTED_TABLES]
  );

  return Boolean(result.rows[0]?.exists);
}

async function validateExistingSchemaIsLatest(client: Client) {
  const tableCountResult = await client.query<{ count: string }>(
    `
      SELECT COUNT(*)::text AS "count"
      FROM information_schema.tables
      WHERE table_schema = current_schema()
        AND table_name = ANY($1::text[])
    `,
    [EXPECTED_TABLES]
  );

  const existingTableCount = Number(tableCountResult.rows[0]?.count ?? 0);
  if (existingTableCount !== EXPECTED_TABLES.length) {
    return false;
  }

  const metaColumnResult = await client.query<{ exists: boolean }>(
    `
      SELECT EXISTS (
        SELECT 1
        FROM information_schema.columns
        WHERE table_schema = current_schema()
          AND table_name = 'user_chat_messages'
          AND column_name = 'meta'
      ) AS "exists"
    `
  );

  const removedColumnsResult = await client.query<{
    fund_account_type_exists: boolean;
    flow_pay_type_exists: boolean;
    fixed_flow_pay_type_exists: boolean;
  }>(`
    SELECT
      EXISTS (
        SELECT 1
        FROM information_schema.columns
        WHERE table_schema = current_schema()
          AND table_name = 'user_fund_accounts'
          AND column_name = 'accountType'
      ) AS "fund_account_type_exists",
      EXISTS (
        SELECT 1
        FROM information_schema.columns
        WHERE table_schema = current_schema()
          AND table_name = 'user_flows'
          AND column_name = 'payType'
      ) AS "flow_pay_type_exists",
      EXISTS (
        SELECT 1
        FROM information_schema.columns
        WHERE table_schema = current_schema()
          AND table_name = 'user_fixed_flows'
          AND column_name = 'payType'
      ) AS "fixed_flow_pay_type_exists"
  `);

  const removed = removedColumnsResult.rows[0];

  return (
    Boolean(metaColumnResult.rows[0]?.exists) &&
    !removed?.fund_account_type_exists &&
    !removed?.flow_pay_type_exists &&
    !removed?.fixed_flow_pay_type_exists
  );
}

async function ensureDatabaseExists(target: DatabaseTarget) {
  const probeClient = new Client({
    connectionString: target.databaseUrl,
  });

  try {
    await probeClient.connect();
    await probeClient.end();
    return;
  } catch (error: any) {
    try {
      await probeClient.end();
    } catch {
      // 接続失敗後のクローズエラーは無視
    }

    if (error?.code !== "3D000") {
      throw error;
    }
  }

  const adminClient = new Client({
    connectionString: target.adminUrl,
  });

  await adminClient.connect();
  try {
    const existsResult = await adminClient.query<{ exists: boolean }>(
      "SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = $1) AS exists",
      [target.databaseName]
    );

    if (!existsResult.rows[0]?.exists) {
      await adminClient.query(
        `CREATE DATABASE ${quoteIdentifier(target.databaseName)}`
      );
      console.log(`[db:migrate] created database ${target.databaseName}`);
    }
  } finally {
    await adminClient.end();
  }
}

async function baselineMigrations(
  client: Client,
  migrations: MigrationFile[],
  applied: Map<string, string>
) {
  if (applied.size > 0) {
    return;
  }

  const hasTables = await hasUserTables(client);
  if (!hasTables) {
    return;
  }

  const isLatestSchema = await validateExistingSchemaIsLatest(client);
  if (!isLatestSchema) {
    throw new Error(
      "Existing database schema detected without _prisma_migrations history, and the schema does not match the current expected version. Refusing automatic baseline to avoid duplicate execution or destructive drift."
    );
  }

  await client.query("BEGIN");
  try {
    for (const migration of migrations) {
      await client.query(
        `INSERT INTO "${MIGRATIONS_TABLE}" ("id", "checksum", "finished_at", "migration_name", "started_at", "applied_steps_count") VALUES ($1, $2, NOW(), $3, NOW(), 1)`,
        [randomUUID(), migration.checksum, migration.name]
      );
      applied.set(migration.name, migration.checksum);
    }
    await client.query("COMMIT");
    console.log("[db:migrate] baselined existing schema");
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  }
}

async function applyPendingMigrations() {
  const target = getDatabaseTarget();

  const migrations = await readMigrationFiles();
  if (migrations.length < 1) {
    return;
  }

  await ensureDatabaseExists(target);

  const client = new Client({
    connectionString: target.databaseUrl,
  });

  await client.connect();

  try {
    await ensureSchema(client, target.schemaName);
    await client.query("SELECT pg_advisory_lock($1, $2)", [
      LOCK_KEY_1,
      LOCK_KEY_2,
    ]);
    await ensureMigrationsTable(client);

    const applied = await loadAppliedMigrations(client);
    await baselineMigrations(client, migrations, applied);

    for (const migration of migrations) {
      const appliedChecksum = applied.get(migration.name);

      if (appliedChecksum) {
        if (appliedChecksum !== migration.checksum) {
          throw new Error(
            `Migration checksum mismatch: ${migration.name}. Existing databases cannot safely apply a modified migration file.`
          );
        }
        continue;
      }

      await client.query("BEGIN");
      try {
        await client.query(migration.sql);
        await client.query(
          `INSERT INTO "${MIGRATIONS_TABLE}" ("id", "checksum", "finished_at", "migration_name", "started_at", "applied_steps_count") VALUES ($1, $2, NOW(), $3, NOW(), 1)`,
          [randomUUID(), migration.checksum, migration.name]
        );
        await client.query("COMMIT");
        console.log(`[db:migrate] applied ${migration.name}`);
      } catch (error) {
        await client.query("ROLLBACK");
        throw error;
      }
    }
  } finally {
    try {
      await client.query("SELECT pg_advisory_unlock($1, $2)", [
        LOCK_KEY_1,
        LOCK_KEY_2,
      ]);
    } catch {
      // シャットダウン/エラー処理中のアンロック失敗は無視
    }
    await client.end();
  }
}

export async function ensureDatabaseMigrations() {
  if (process.env.DATABASE_AUTO_MIGRATE === "false") {
    return;
  }

  migrationTask ??= applyPendingMigrations();
  return migrationTask;
}
```

## オプションのソリューション：entrypointでPrisma CLIを呼び出す

Prisma CLIが提供する`npx prisma migrate deploy`機能を利用してデータベース同期を実現しますが、このソリューションではコンテナ内にPrisma CLIをインストールする必要があり、イメージサイズが大幅に増加するため、推奨されません。

## 運用とトラブルシューティング

| 現象 | 考えられる原因 |
|------|----------|
| 起動エラー `DATABASE_URL is required` | 環境変数が注入されていません |
| `Migration checksum mismatch` | 適用済みのマイグレーションファイルが変更されています。履歴SQLを変更するのではなく、新しいマイグレーションを作成する必要があります。 |
| ベースラインが実行を拒否 | 既存のデータベーススキーマが`validateExistingSchemaIsLatest`に適合していません。手動で調整するか、`prisma migrate resolve`のような処理を先に行う必要があります。 |
| コンテナ内でマイグレーションが見つからない | `prisma/migrations`がCOPYされていないか、`cwd`が`/app`ではありません。 |
| 複数のレプリカが同時に起動 | advisory lockがマイグレーションを直列化します。それでも、デプロイ時には単一インスタンスで起動するか、`DATABASE_AUTO_MIGRATE=false`を設定してJobで実行することをお勧めします。 |

ログプレフィックス：`[db:migrate]`。例：`created database`、`baselined existing schema`、`applied <migration_name>`。