29c - 迁移文件生成与审查

┌─────────────────────────────────────────────────────────────┐
│                    数据库迁移方式分类                          │
├─────────────────────┬───────────────────────────────────────┤
│  声明式（Declarative）│  命令式（Imperative）                  │
│                     │                                       │
│  定义"目标状态"       │  定义"变更步骤"                        │
│  工具自动计算 diff    │  开发者手写 up/down                    │
│                     │                                       │
│  ✅ Prisma          │  ✅ Alembic（手动模式）                 │
│  ✅ Drizzle         │  ✅ TypeORM                            │
│  ✅ Atlas           │  ✅ Diesel                             │
│  ✅ Django          │  ✅ Flyway                             │
│                     │  ✅ Liquibase                          │
├─────────────────────┴───────────────────────────────────────┤
│  混合模式：Alembic autogenerate（声明式生成 + 命令式微调）      │
└─────────────────────────────────────────────────────────────┘

# 1. 修改 schema.prisma（添加新字段）
# 2. 生成并应用迁移
npx prisma migrate dev --name add_user_avatar
 
# 生成的文件结构：
# prisma/migrations/
# ├── 20250615120000_add_user_avatar/
# │   └── migration.sql
# └── migration_lock.toml

-- prisma/migrations/20250615120000_add_user_avatar/migration.sql
-- CreateTable 或 AlterTable
ALTER TABLE "User" ADD COLUMN "avatar_url" TEXT;
ALTER TABLE "User" ADD COLUMN "avatar_updated_at" TIMESTAMP(3);
 
-- CreateIndex
CREATE INDEX "User_avatar_updated_at_idx" ON "User"("avatar_updated_at");

# 在 CI/CD 中使用 migrate deploy（不会生成新迁移）
npx prisma migrate deploy
 
# 检查迁移状态
npx prisma migrate status

┌─────────────────────────────────────────────────────────┐
│              Prisma AI Safety Guardrails                 │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  AI Agent 请求执行:                                      │
│  prisma migrate reset                                   │
│  prisma db push --force-reset                           │
│                                                         │
│  ┌─────────────────────────────────────────────┐        │
│  │  🛡️ Safety Guardrail 拦截                    │        │
│  │                                             │        │
│  │  检测到破坏性操作！                            │        │
│  │  此操作将删除所有数据并重建数据库。              │        │
│  │                                             │        │
│  │  ❌ 自动执行被阻止                            │        │
│  │  ✅ 需要人工确认才能继续                       │        │
│  └─────────────────────────────────────────────┘        │
│                                                         │
│  被保护的命令：                                           │
│  • prisma migrate reset                                 │
│  • prisma db push --force-reset                         │
│  • prisma db push --accept-data-loss                    │
│  • prisma migrate dev（当需要重置时）                     │
│                                                         │
└─────────────────────────────────────────────────────────┘

// schema.prisma
generator client {
  provider = "prisma-client-js"
  // AI Safety Guardrails 在 v6.15+ 默认启用
  // 无需额外配置
}

# AI 尝试执行：
$ npx prisma migrate reset

# Prisma 检测到非交互式环境（AI 工具），自动阻止：
Error: This command will reset your database and delete all data.
This operation was blocked because it was detected as being run
by an AI tool. Please run this command manually to confirm.

你是一位资深数据库工程师。请审查以下 Prisma 生成的迁移 SQL 文件，
检查安全风险并提供改进建议。

## 迁移文件内容
[粘贴 migration.sql 内容]

## 当前数据库规模
- 表 [表名] 约 [行数] 行
- 数据库类型：[PostgreSQL/MySQL/SQLite]
- 是否有读副本：[是/否]

## 审查清单
请逐项检查：
1. **破坏性操作**：是否包含 DROP TABLE、DROP COLUMN、TRUNCATE？
2. **数据丢失风险**：ALTER COLUMN 是否可能导致数据截断？
3. **锁表风险**：ALTER TABLE 是否会在大表上获取排他锁？
4. **索引创建**：CREATE INDEX 是否使用了 CONCURRENTLY（PostgreSQL）？
5. **默认值**：新增 NOT NULL 列是否提供了默认值？
6. **回滚可行性**：此迁移是否可以安全回滚？如何回滚？
7. **向后兼容性**：旧版本代码是否能与新 Schema 兼容？

## 输出格式
- 🔴 高风险：必须修复
- 🟡 中风险：建议修复
- 🟢 低风险：可接受
- 对每个风险项提供具体的修复建议

你是一位 Prisma 专家。请帮我安全地修改 Schema 以实现以下需求：

## 当前 Schema
[粘贴当前 schema.prisma 相关部分]

## 需求变更
[描述需要的变更]

## 安全要求
1. 不能有数据丢失风险
2. 必须支持零停机部署（旧代码和新代码同时运行）
3. 如果需要多步迁移，请拆分为多个步骤并说明执行顺序
4. 对于大表（>100万行），避免锁表操作

## 输出
1. 修改后的 schema.prisma 片段
2. 预期生成的 SQL（用于审查）
3. 部署步骤和注意事项
4. 回滚方案

// src/db/schema.ts
import { pgTable, uuid, varchar, timestamp, text, index, boolean } from 'drizzle-orm/pg-core';
 
export const users = pgTable('users', {
  id: uuid('id').primaryKey().defaultRandom(),
  email: varchar('email', { length: 255 }).notNull().unique(),
  name: varchar('name', { length: 100 }).notNull(),
  avatarUrl: text('avatar_url'),
  isActive: boolean('is_active').notNull().default(true),
  createdAt: timestamp('created_at').notNull().defaultNow(),
  updatedAt: timestamp('updated_at').notNull().defaultNow(),
}, (table) => [
  index('users_email_idx').on(table.email),
  index('users_created_at_idx').on(table.createdAt),
]);
 
export const posts = pgTable('posts', {
  id: uuid('id').primaryKey().defaultRandom(),
  title: varchar('title', { length: 255 }).notNull(),
  content: text('content'),
  authorId: uuid('author_id').notNull().references(() => users.id, { onDelete: 'cascade' }),
  publishedAt: timestamp('published_at'),
  createdAt: timestamp('created_at').notNull().defaultNow(),
}, (table) => [
  index('posts_author_id_idx').on(table.authorId),
  index('posts_published_at_idx').on(table.publishedAt),
]);

# drizzle.config.ts 配置
# export default defineConfig({
#   schema: './src/db/schema.ts',
#   out: './drizzle',
#   dialect: 'postgresql',
# });
 
# 生成迁移
npx drizzle-kit generate
 
# 生成的文件结构：
# drizzle/
# ├── 0000_initial_schema.sql
# ├── 0001_add_posts_table.sql
# └── meta/
#     ├── 0000_snapshot.json
#     ├── 0001_snapshot.json
#     └── _journal.json

-- drizzle/0001_add_posts_table.sql
CREATE TABLE IF NOT EXISTS "posts" (
  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
  "title" varchar(255) NOT NULL,
  "content" text,
  "author_id" uuid NOT NULL,
  "published_at" timestamp,
  "created_at" timestamp DEFAULT now() NOT NULL
);
 
-- 外键约束
ALTER TABLE "posts"
  ADD CONSTRAINT "posts_author_id_users_id_fk"
  FOREIGN KEY ("author_id") REFERENCES "users"("id")
  ON DELETE cascade ON UPDATE no action;
 
-- 索引
CREATE INDEX IF NOT EXISTS "posts_author_id_idx" ON "posts" USING btree ("author_id");
CREATE INDEX IF NOT EXISTS "posts_published_at_idx" ON "posts" USING btree ("published_at");

// ❌ 危险：直接删除列（Drizzle 会生成 ALTER TABLE DROP COLUMN）
// 修改前的 Schema
export const users = pgTable('users', {
  id: uuid('id').primaryKey().defaultRandom(),
  email: varchar('email', { length: 255 }).notNull(),
  oldField: text('old_field'), // 删除这个字段
});
 
// 修改后的 Schema（直接删除 oldField）
export const users = pgTable('users', {
  id: uuid('id').primaryKey().defaultRandom(),
  email: varchar('email', { length: 255 }).notNull(),
  // oldField 被删除
});
 
// Drizzle 会生成：
// ALTER TABLE "users" DROP COLUMN "old_field";
// ⚠️ 这是不可逆操作！数据将永久丢失！

// ✅ 安全：分步迁移
// 步骤 1：标记字段为废弃（添加注释，但不删除）
// 步骤 2：确认所有代码不再读取该字段
// 步骤 3：在下一个版本中删除字段
// 步骤 4：在确认回滚窗口过后，清理迁移

你是一位 Drizzle ORM 专家。请审查以下 Drizzle 生成的迁移 SQL。

## 迁移 SQL
[粘贴 drizzle 生成的 .sql 文件内容]

## Schema 变更描述
[描述从什么变到什么]

## 审查要点
1. 是否有 DROP COLUMN/DROP TABLE 操作？如果有，是否可以改为分步迁移？
2. ALTER TABLE 操作是否会在大表上造成锁表？
3. 新增的 NOT NULL 列是否有 DEFAULT 值？
4. 外键约束是否正确设置了 ON DELETE 行为？
5. 索引创建是否合理？是否需要 CONCURRENTLY？
6. 此迁移是否支持回滚？请提供回滚 SQL。

## 输出
- 风险评估（🔴/🟡/🟢）
- 具体问题和修复建议
- 回滚 SQL
- 部署注意事项

project/
├── alembic/
│   ├── env.py              # 迁移环境配置
│   ├── script.py.mako      # 迁移脚本模板
│   └── versions/           # 迁移文件目录
│       ├── 001_abc123_initial.py
│       ├── 002_def456_add_posts.py
│       └── 003_ghi789_add_avatar.py
├── alembic.ini             # Alembic 配置文件
├── app/
│   └── models.py           # SQLAlchemy Models
└── requirements.txt

# app/models.py - SQLAlchemy Model
from sqlalchemy import Column, String, Text, Boolean, DateTime, ForeignKey, Index
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase, relationship
from datetime import datetime
import uuid
 
class Base(DeclarativeBase):
    pass
 
class User(Base):
    __tablename__ = 'users'
 
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String(255), nullable=False, unique=True)
    name = Column(String(100), nullable=False)
    avatar_url = Column(Text, nullable=True)  # 新增字段
    is_active = Column(Boolean, nullable=False, default=True)
    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
 
    posts = relationship('Post', back_populates='author')
 
    __table_args__ = (
        Index('ix_users_email', 'email'),
        Index('ix_users_created_at', 'created_at'),
    )

# 生成迁移
alembic revision --autogenerate -m "add user avatar"
 
# 生成的迁移文件：alembic/versions/003_ghi789_add_avatar.py

"""add user avatar
 
Revision ID: ghi789
Revises: def456
Create Date: 2025-06-15 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
 
# revision identifiers
revision: str = 'ghi789'
down_revision: Union[str, None] = 'def456'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
 
 
def upgrade() -> None:
    # ### commands auto generated by Alembic ###
    op.add_column('users', sa.Column('avatar_url', sa.Text(), nullable=True))
    # ### end Alembic commands ###
 
 
def downgrade() -> None:
    # ### commands auto generated by Alembic ###
    op.drop_column('users', 'avatar_url')
    # ### end Alembic commands ###

"""migrate user status from string to enum
 
Revision ID: jkl012
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import ENUM
 
revision = 'jkl012'
down_revision = 'ghi789'
 
# 定义新的 Enum 类型
user_status = ENUM('active', 'inactive', 'suspended', name='user_status', create_type=False)
 
 
def upgrade() -> None:
    # 步骤 1：创建 Enum 类型
    user_status.create(op.get_bind(), checkfirst=True)
 
    # 步骤 2：添加新列
    op.add_column('users', sa.Column('status', user_status, nullable=True))
 
    # 步骤 3：数据迁移（将旧数据转换到新列）
    op.execute("""
        UPDATE users
        SET status = CASE
            WHEN is_active = true THEN 'active'::user_status
            ELSE 'inactive'::user_status
        END
    """)
 
    # 步骤 4：设置 NOT NULL
    op.alter_column('users', 'status', nullable=False)
 
    # 步骤 5：删除旧列（在确认安全后的下一次迁移中执行）
    # op.drop_column('users', 'is_active')  # ⚠️ 不要在同一迁移中删除！
 
 
def downgrade() -> None:
    op.drop_column('users', 'status')
    user_status.drop(op.get_bind(), checkfirst=True)

你是一位 SQLAlchemy/Alembic 专家。请帮我编写安全的 Alembic 迁移脚本。

## 当前 Model
[粘贴当前 SQLAlchemy Model 代码]

## 需求变更
[描述需要的变更]

## 要求
1. 使用 Alembic op 操作（不要直接写 SQL，除非必要）
2. upgrade() 和 downgrade() 必须完整且可逆
3. 如果涉及数据迁移，使用 op.execute() 并确保幂等性
4. 对于大表操作，考虑分批处理
5. 新增 NOT NULL 列必须先添加为 nullable=True，填充数据后再改为 NOT NULL
6. 如果 autogenerate 无法检测（如列重命名），请手动编写

## 输出
1. 完整的迁移脚本（upgrade + downgrade）
2. 执行前的安全检查清单
3. 预估执行时间和锁表影响
4. 回滚验证步骤

// src/data-source.ts
import { DataSource } from 'typeorm';
import { User } from './entities/User';
import { Post } from './entities/Post';
 
export const AppDataSource = new DataSource({
  type: 'postgres',
  host: process.env.DB_HOST || 'localhost',
  port: parseInt(process.env.DB_PORT || '5432'),
  username: process.env.DB_USER || 'postgres',
  password: process.env.DB_PASSWORD || 'postgres',
  database: process.env.DB_NAME || 'myapp',
  entities: [User, Post],
  migrations: ['src/migrations/*.ts'],
  synchronize: false,  // ❌ 生产环境必须为 false
  logging: process.env.NODE_ENV === 'development',
});

// src/entities/User.ts
import {
  Entity, PrimaryGeneratedColumn, Column,
  CreateDateColumn, UpdateDateColumn, Index, OneToMany
} from 'typeorm';
import { Post } from './Post';
 
@Entity('users')
export class User {
  @PrimaryGeneratedColumn('uuid')
  id: string;
 
  @Column({ type: 'varchar', length: 255, unique: true })
  @Index('idx_users_email')
  email: string;
 
  @Column({ type: 'varchar', length: 100 })
  name: string;
 
  @Column({ type: 'text', nullable: true })
  avatarUrl: string | null;
 
  @Column({ type: 'boolean', default: true })
  isActive: boolean;
 
  @CreateDateColumn()
  createdAt: Date;
 
  @UpdateDateColumn()
  updatedAt: Date;
 
  @OneToMany(() => Post, (post) => post.author)
  posts: Post[];
}

# 使用 ts-node 运行（需要 package.json 中配置脚本）
# package.json:
# "scripts": {
#   "typeorm": "typeorm-ts-node-commonjs",
#   "migration:generate": "npm run typeorm -- migration:generate -d src/data-source.ts",
#   "migration:run": "npm run typeorm -- migration:run -d src/data-source.ts",
#   "migration:revert": "npm run typeorm -- migration:revert -d src/data-source.ts"
# }
 
npm run migration:generate -- src/migrations/AddUserAvatar

// src/migrations/1718452800000-AddUserAvatar.ts
import { MigrationInterface, QueryRunner } from 'typeorm';
 
export class AddUserAvatar1718452800000 implements MigrationInterface {
  name = 'AddUserAvatar1718452800000';
 
  public async up(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query(
      `ALTER TABLE "users" ADD "avatar_url" text`
    );
  }
 
  public async down(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query(
      `ALTER TABLE "users" DROP COLUMN "avatar_url"`
    );
  }
}

// ❌ 常见错误：在生产环境使用 synchronize: true
const dataSource = new DataSource({
  synchronize: true,  // 这会自动修改数据库 Schema，可能导致数据丢失！
});
 
// ✅ 正确做法：生产环境只使用迁移
const dataSource = new DataSource({
  synchronize: false,
  migrationsRun: true,  // 应用启动时自动运行迁移（可选）
});

// ❌ 常见错误：migration:generate 生成了意外的变更
// 原因：Entity 与数据库不同步（可能有人手动修改了数据库）
// 解决：先用 schema:log 查看差异，确认后再生成
 
// ✅ 安全流程
// 1. 检查当前迁移状态
// npm run typeorm -- migration:show -d src/data-source.ts
// 2. 生成迁移（仔细审查生成的 SQL）
// npm run migration:generate -- src/migrations/MyChange
// 3. 审查生成的文件（重点检查 up 和 down 方法）
// 4. 在测试环境运行
// npm run migration:run
// 5. 验证数据完整性
// 6. 部署到生产

你是一位 TypeORM 专家。请帮我编写安全的 TypeORM 迁移。

## 当前 Entity
[粘贴当前 TypeORM Entity 代码]

## 需求变更
[描述需要的变更]

## 要求
1. 使用 QueryRunner API 编写 up() 和 down() 方法
2. down() 必须完整可逆
3. 对于数据迁移，使用事务包裹
4. 大表操作需要分批处理
5. 考虑 PostgreSQL/MySQL 的差异

## 输出
1. 完整的迁移类（implements MigrationInterface）
2. 对应的 Entity 变更
3. 部署注意事项

project/
├── migrations/
│   ├── 00000000000000_diesel_initial_setup/
│   │   ├── up.sql
│   │   └── down.sql
│   ├── 2025-06-15-120000_create_users/
│   │   ├── up.sql
│   │   └── down.sql
│   └── 2025-06-15-130000_create_posts/
│       ├── up.sql
│       └── down.sql
├── src/
│   ├── schema.rs          # 自动生成（diesel print-schema）
│   └── models.rs          # 手动编写的 Rust 结构体
├── diesel.toml
└── Cargo.toml

-- migrations/2025-06-15-120000_create_users/up.sql
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email VARCHAR(255) NOT NULL UNIQUE,
    name VARCHAR(100) NOT NULL,
    avatar_url TEXT,
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);
 
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_created_at ON users(created_at);

-- migrations/2025-06-15-120000_create_users/down.sql
DROP TABLE IF EXISTS users;

-- migrations/2025-06-16-100000_add_user_avatar/up.sql
ALTER TABLE users ADD COLUMN avatar_url TEXT;
ALTER TABLE users ADD COLUMN avatar_updated_at TIMESTAMP;
 
CREATE INDEX idx_users_avatar_updated ON users(avatar_updated_at)
    WHERE avatar_url IS NOT NULL;  -- 部分索引：只索引有头像的用户

-- migrations/2025-06-16-100000_add_user_avatar/down.sql
DROP INDEX IF EXISTS idx_users_avatar_updated;
ALTER TABLE users DROP COLUMN IF EXISTS avatar_updated_at;
ALTER TABLE users DROP COLUMN IF EXISTS avatar_url;

# 运行迁移后，自动更新 schema.rs
diesel migration run
diesel print-schema > src/schema.rs

// src/schema.rs（自动生成）
diesel::table! {
    users (id) {
        id -> Uuid,
        email -> Varchar,
        name -> Varchar,
        avatar_url -> Nullable<Text>,
        is_active -> Bool,
        created_at -> Timestamp,
        updated_at -> Timestamp,
    }
}
 
diesel::table! {
    posts (id) {
        id -> Uuid,
        title -> Varchar,
        content -> Nullable<Text>,
        author_id -> Uuid,
        published_at -> Nullable<Timestamp>,
        created_at -> Timestamp,
    }
}
 
diesel::joinable!(posts -> users (author_id));
diesel::allow_tables_to_appear_in_same_query!(users, posts);

// src/models.rs
use diesel::prelude::*;
use uuid::Uuid;
use chrono::NaiveDateTime;
 
#[derive(Queryable, Selectable, Debug)]
#[diesel(table_name = crate::schema::users)]
pub struct User {
    pub id: Uuid,
    pub email: String,
    pub name: String,
    pub avatar_url: Option<String>,
    pub is_active: bool,
    pub created_at: NaiveDateTime,
    pub updated_at: NaiveDateTime,
}
 
#[derive(Insertable)]
#[diesel(table_name = crate::schema::users)]
pub struct NewUser<'a> {
    pub email: &'a str,
    pub name: &'a str,
    pub avatar_url: Option<&'a str>,
}
 
// 如果 schema.rs 中没有 avatar_url 字段，
// 这里会编译报错：no field `avatar_url` on type `users::table`
// ✅ 编译时就能发现迁移遗漏！

你是一位 Rust/Diesel 专家。请帮我编写安全的 Diesel 迁移 SQL。

## 当前 schema.rs
[粘贴当前 schema.rs 内容]

## 需求变更
[描述需要的变更]

## 数据库类型
[PostgreSQL/MySQL/SQLite]

## 要求
1. 编写 up.sql 和 down.sql
2. down.sql 必须完整可逆
3. 使用 IF EXISTS / IF NOT EXISTS 防止重复执行
4. 对于 PostgreSQL，索引创建考虑使用 CONCURRENTLY
5. 新增 NOT NULL 列必须提供 DEFAULT 值或分步迁移
6. 提供对应的 Rust Model 结构体更新

## 输出
1. up.sql 内容
2. down.sql 内容
3. 更新后的 schema.rs 片段
4. 更新后的 models.rs 片段
5. 部署注意事项

# myapp/models.py
from django.db import models
import uuid
 
class User(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    email = models.EmailField(max_length=255, unique=True, db_index=True)
    name = models.CharField(max_length=100)
    avatar_url = models.URLField(blank=True, null=True)
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
 
    class Meta:
        db_table = 'users'
        indexes = [
            models.Index(fields=['created_at'], name='idx_users_created_at'),
            models.Index(fields=['email'], name='idx_users_email'),
        ]
 
class Post(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    title = models.CharField(max_length=255)
    content = models.TextField(blank=True)
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')
    published_at = models.DateTimeField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
 
    class Meta:
        db_table = 'posts'
        indexes = [
            models.Index(fields=['author'], name='idx_posts_author'),
            models.Index(fields=['published_at'], name='idx_posts_published'),
        ]

# 生成迁移文件
python manage.py makemigrations myapp
 
# 查看生成的 SQL（审查用）
python manage.py sqlmigrate myapp 0002
 
# 执行迁移
python manage.py migrate

# myapp/migrations/0002_user_avatar_url.py
from django.db import migrations, models
 
class Migration(migrations.Migration):
 
    dependencies = [
        ('myapp', '0001_initial'),
    ]
 
    operations = [
        migrations.AddField(
            model_name='user',
            name='avatar_url',
            field=models.URLField(blank=True, null=True),
        ),
    ]

# ❌ 危险：一步删除字段
class Migration(migrations.Migration):
    operations = [
        migrations.RemoveField(
            model_name='user',
            name='old_field',
        ),
    ]
# 如果旧版本代码还在运行，会导致 500 错误！
 
# ✅ 安全：两阶段删除
# 阶段 1（版本 N）：从代码中移除对 old_field 的所有引用
# 阶段 2（版本 N+1）：删除数据库字段
class Migration(migrations.Migration):
    operations = [
        migrations.RemoveField(
            model_name='user',
            name='old_field',
        ),
    ]

# myapp/migrations/0003_migrate_status_data.py
from django.db import migrations
 
def migrate_status_forward(apps, schema_editor):
    """将 is_active 布尔值迁移到 status 枚举"""
    User = apps.get_model('myapp', 'User')
    # 分批处理，避免内存溢出
    batch_size = 1000
    users = User.objects.all()
    for i in range(0, users.count(), batch_size):
        batch = users[i:i + batch_size]
        for user in batch:
            user.status = 'active' if user.is_active else 'inactive'
        User.objects.bulk_update(batch, ['status'], batch_size=batch_size)
 
def migrate_status_backward(apps, schema_editor):
    """回滚：将 status 枚举迁移回 is_active 布尔值"""
    User = apps.get_model('myapp', 'User')
    batch_size = 1000
    users = User.objects.all()
    for i in range(0, users.count(), batch_size):
        batch = users[i:i + batch_size]
        for user in batch:
            user.is_active = user.status == 'active'
        User.objects.bulk_update(batch, ['is_active'], batch_size=batch_size)
 
class Migration(migrations.Migration):
    dependencies = [
        ('myapp', '0002_user_avatar_url'),
    ]
 
    operations = [
        migrations.RunPython(
            migrate_status_forward,
            migrate_status_backward,  # ✅ 必须提供回滚函数
        ),
    ]

你是一位 Django 专家。请帮我编写安全的 Django 迁移。

## 当前 Model
[粘贴当前 Django Model 代码]

## 需求变更
[描述需要的变更]

## 部署方式
[滚动部署/蓝绿部署/单机部署]

## 要求
1. 遵循两阶段删除原则（先移除代码引用，再删除字段）
2. 数据迁移使用 RunPython，必须提供 reverse 函数
3. 大表数据迁移使用分批处理（batch_size=1000）
4. 新增 NOT NULL 字段必须提供 default 或分步迁移
5. 考虑滚动部署时新旧代码并存的兼容性
6. 使用 sqlmigrate 预览 SQL 并评估锁表风险

## 输出
1. 完整的迁移文件（operations 列表）
2. 对应的 Model 变更
3. 部署步骤（先迁移还是先部署代码？）
4. 回滚步骤

┌─────────────────────────────────────────────────────────────────┐
│                    迁移安全审查四维模型                            │
├────────────────┬────────────────────────────────────────────────┤
│                │                                                │
│  🔴 破坏性操作  │  DROP TABLE / DROP COLUMN / TRUNCATE           │
│  (Destructive) │  ALTER COLUMN TYPE（缩小范围）                   │
│                │  DROP INDEX / DROP CONSTRAINT                   │
│                │                                                │
├────────────────┼────────────────────────────────────────────────┤
│                │                                                │
│  🟡 锁表风险    │  ALTER TABLE ADD COLUMN ... NOT NULL            │
│  (Lock Risk)   │  ALTER TABLE ADD COLUMN ... DEFAULT（MySQL<8.0）│
│                │  CREATE INDEX（非 CONCURRENTLY）                 │
│                │  ALTER TABLE ... ADD CONSTRAINT                 │
│                │  RENAME TABLE / RENAME COLUMN                   │
│                │                                                │
├────────────────┼────────────────────────────────────────────────┤
│                │                                                │
│  🟠 数据丢失    │  ALTER COLUMN TYPE（精度降低）                   │
│  (Data Loss)   │  ALTER COLUMN SET NOT NULL（含 NULL 数据）       │
│                │  UPDATE ... SET（不可逆数据转换）                 │
│                │  DELETE FROM（数据清理）                          │
│                │                                                │
├────────────────┼────────────────────────────────────────────────┤
│                │                                                │
│  🔵 兼容性      │  RENAME COLUMN（旧代码引用旧名）                 │
│  (Compat)      │  DROP DEFAULT（旧代码依赖默认值）                 │
│                │  ADD NOT NULL（旧代码不写入该字段）                │
│                │  CHANGE ENUM VALUES（旧代码使用旧值）             │
│                │                                                │
└────────────────┴────────────────────────────────────────────────┘

你是一位资深数据库工程师和迁移安全专家。请对以下迁移文件进行全面安全审查。

## 迁移文件
[粘贴迁移文件内容（SQL 或 ORM 迁移脚本）]

## 环境信息
- 数据库类型：[PostgreSQL 16 / MySQL 8.4 / SQLite]
- 受影响表的行数：[表名: 约 X 行]
- 部署方式：[滚动部署 / 蓝绿部署 / 停机部署]
- 是否有读副本：[是/否]
- 当前 QPS：[约 X 请求/秒]

## 审查维度

### 1. 破坏性操作检测
- 是否包含 DROP TABLE / DROP COLUMN / TRUNCATE？
- 是否有不可逆的数据类型变更？
- 是否删除了索引或约束？

### 2. 锁表风险评估
- ALTER TABLE 操作是否会获取排他锁？
- 锁持续时间预估（基于表大小）？
- 是否可以使用 CONCURRENTLY / INSTANT / INPLACE 替代？

### 3. 数据完整性
- 新增 NOT NULL 列是否有默认值？
- 数据类型变更是否可能导致数据截断？
- 外键约束是否正确设置了级联行为？

### 4. 向后兼容性
- 旧版本代码是否能与新 Schema 兼容？
- 是否需要多步迁移（expand-contract 模式）？
- 回滚后旧代码是否能正常工作？

### 5. 性能影响
- 迁移执行时间预估？
- 是否需要分批执行？
- 对在线查询的影响？

## 输出格式
对每个发现的问题：
- 🔴 高风险 / 🟡 中风险 / 🟢 低风险
- 问题描述
- 影响范围
- 修复建议（提供具体的替代 SQL）
- 回滚方案

最后提供：
- 总体风险评级
- 推荐的执行顺序
- 执行前检查清单
- 执行后验证清单

你是一位数据库性能专家。我需要在一个大表上执行 Schema 变更，
请帮我设计安全的迁移方案。

## 表信息
- 表名：[表名]
- 行数：[约 X 百万行]
- 表大小：[约 X GB]
- 数据库：[PostgreSQL / MySQL]
- 当前 QPS：[约 X 请求/秒]
- 维护窗口：[有/无，如有则为 X 小时]

## 需要执行的变更
[描述 Schema 变更]

## 约束条件
- 最大允许停机时间：[0 / X 分钟]
- 最大允许性能降级：[X%]
- 是否可以分批执行：[是/否]

## 请提供
1. 分步迁移方案（每步的 SQL 和预估时间）
2. 每步的锁表分析
3. 监控指标（执行期间需要关注什么）
4. 回滚方案（每步的回滚 SQL）
5. 执行脚本（可直接运行的 SQL 脚本，包含进度输出）

┌─────────────────────────────────────────────────────────────────┐
│              Expand-Contract 迁移模式                             │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  阶段 1：Expand（扩展）                                          │
│  ┌─────────────────────────────────────────────────────┐        │
│  │  • 添加新列/新表                                      │        │
│  │  • 新列设为 nullable 或有 default                     │        │
│  │  • 不删除任何旧结构                                    │        │
│  │  • 旧代码继续正常工作                                  │        │
│  └─────────────────────────────────────────────────────┘        │
│                          ↓                                      │
│  阶段 2：Migrate（迁移数据 + 代码）                               │
│  ┌─────────────────────────────────────────────────────┐        │
│  │  • 后台任务迁移历史数据到新结构                         │        │
│  │  • 部署新代码（同时写入新旧结构）                       │        │
│  │  • 使用触发器或应用层双写保持同步                       │        │
│  │  • 验证新旧数据一致性                                  │        │
│  └─────────────────────────────────────────────────────┘        │
│                          ↓                                      │
│  阶段 3：Contract（收缩）                                        │
│  ┌─────────────────────────────────────────────────────┐        │
│  │  • 部署只读新结构的代码                                │        │
│  │  • 确认所有服务已切换到新结构                           │        │
│  │  • 等待回滚窗口过期（通常 7-14 天）                    │        │
│  │  • 删除旧列/旧表                                      │        │
│  └─────────────────────────────────────────────────────┘        │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

-- ❌ 危险：直接重命名（会导致旧代码崩溃）
ALTER TABLE users RENAME COLUMN email TO email_address;
 
-- ✅ 安全：Expand-Contract 三阶段
 
-- 阶段 1：Expand（添加新列）
ALTER TABLE users ADD COLUMN email_address VARCHAR(255);
UPDATE users SET email_address = email;  -- 迁移历史数据
-- 添加触发器保持同步
CREATE OR REPLACE FUNCTION sync_email_columns()
RETURNS TRIGGER AS $$
BEGIN
    IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
        IF NEW.email IS DISTINCT FROM NEW.email_address THEN
            IF NEW.email IS NOT NULL AND NEW.email_address IS NULL THEN
                NEW.email_address := NEW.email;
            ELSIF NEW.email_address IS NOT NULL THEN
                NEW.email := NEW.email_address;
            END IF;
        END IF;
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
 
CREATE TRIGGER trg_sync_email
    BEFORE INSERT OR UPDATE ON users
    FOR EACH ROW EXECUTE FUNCTION sync_email_columns();
 
-- 阶段 2：Migrate（部署新代码，读写 email_address）
-- 新代码同时写入 email 和 email_address
-- 读取优先使用 email_address
 
-- 阶段 3：Contract（确认所有服务已切换后）
DROP TRIGGER trg_sync_email ON users;
DROP FUNCTION sync_email_columns();
ALTER TABLE users DROP COLUMN email;
-- 可选：添加唯一约束
ALTER TABLE users ADD CONSTRAINT uq_users_email_address UNIQUE (email_address);
ALTER TABLE users ALTER COLUMN email_address SET NOT NULL;

┌─────────────────────────────────────────────────────────────────┐
│              蓝绿部署 + 数据库迁移                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  步骤 1：当前状态                                                │
│  ┌──────────┐     ┌──────────┐                                  │
│  │ 蓝（活跃） │────→│  数据库   │                                  │
│  │ App v1   │     │ Schema v1│                                  │
│  └──────────┘     └──────────┘                                  │
│                                                                 │
│  步骤 2：执行向前兼容的迁移                                       │
│  ┌──────────┐     ┌──────────┐                                  │
│  │ 蓝（活跃） │────→│  数据库   │  ← Schema v2（兼容 v1 代码）     │
│  │ App v1   │     │ Schema v2│                                  │
│  └──────────┘     └──────────┘                                  │
│                                                                 │
│  步骤 3：部署新代码到绿环境                                       │
│  ┌──────────┐     ┌──────────┐     ┌──────────┐                │
│  │ 蓝（活跃） │────→│  数据库   │←────│ 绿（待命） │                │
│  │ App v1   │     │ Schema v2│     │ App v2   │                │
│  └──────────┘     └──────────┘     └──────────┘                │
│                                                                 │
│  步骤 4：切换流量到绿环境                                         │
│  ┌──────────┐     ┌──────────┐     ┌──────────┐                │
│  │ 蓝（待命） │     │  数据库   │←────│ 绿（活跃） │                │
│  │ App v1   │     │ Schema v2│     │ App v2   │                │
│  └──────────┘     └──────────┘     └──────────┘                │
│                                                                 │
│  关键约束：Schema v2 必须同时兼容 App v1 和 App v2               │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

✅ 允许的迁移操作（向前兼容）：
  • ADD COLUMN（nullable 或有 default）
  • ADD TABLE
  • ADD INDEX（CONCURRENTLY）
  • ADD CONSTRAINT ... NOT VALID

❌ 禁止的迁移操作（破坏向前兼容）：
  • DROP COLUMN（旧代码还在读）
  • DROP TABLE（旧代码还在用）
  • RENAME COLUMN/TABLE
  • ALTER COLUMN TYPE（可能不兼容）
  • ALTER COLUMN SET NOT NULL（旧代码可能写入 NULL）

迁移出现问题？
│
├── 数据未被修改？
│   ├── 是 → 直接执行 down migration
│   └── 否 → 继续评估 ↓
│
├── down migration 是否安全？
│   ├── 是（仅删除新增的列/表）→ 执行 down migration
│   └── 否（涉及数据恢复）→ 继续评估 ↓
│
├── 有数据库备份？
│   ├── 是 → 评估从备份恢复的时间
│   │   ├── 可接受 → 从备份恢复
│   │   └── 不可接受 → 手动修复
│   └── 否 → 手动修复
│
└── 手动修复
    ├── 编写修复 SQL
    ├── 在测试环境验证
    └── 在生产环境执行

#!/bin/bash
# verify_migration_rollback.sh
# 在 CI 中验证每个迁移的 up + down 是否可逆
 
set -e
 
echo "=== 迁移回滚验证 ==="
 
# 1. 创建测试数据库
createdb migration_test_db
 
# 2. 应用所有迁移
echo ">>> 应用所有迁移..."
DATABASE_URL="postgresql://localhost/migration_test_db" \
  npx prisma migrate deploy  # 或对应的 ORM 命令
 
# 3. 记录当前 Schema
pg_dump --schema-only migration_test_db > schema_after_up.sql
 
# 4. 插入测试数据
echo ">>> 插入测试数据..."
psql migration_test_db -f test_data.sql
 
# 5. 回滚最近一次迁移
echo ">>> 回滚最近一次迁移..."
# Alembic: alembic downgrade -1
# Django: python manage.py migrate app 000X
# Diesel: diesel migration revert
 
# 6. 重新应用迁移
echo ">>> 重新应用迁移..."
# 对应的 migrate 命令
 
# 7. 比较 Schema
pg_dump --schema-only migration_test_db > schema_after_redo.sql
diff schema_after_up.sql schema_after_redo.sql
 
if [ $? -eq 0 ]; then
    echo "✅ 迁移回滚验证通过：up → down → up 结果一致"
else
    echo "❌ 迁移回滚验证失败：Schema 不一致！"
    exit 1
fi
 
# 8. 验证数据完整性
echo ">>> 验证数据完整性..."
psql migration_test_db -c "SELECT count(*) FROM users;" | grep -q "0" && {
    echo "❌ 数据丢失！"
    exit 1
}
 
echo "✅ 数据完整性验证通过"
 
# 9. 清理
dropdb migration_test_db
rm -f schema_after_up.sql schema_after_redo.sql
 
echo "=== 验证完成 ==="

# Alembic 数据迁移最佳实践示例
"""migrate user addresses to separate table
 
Revision ID: mno345
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy import text
 
revision = 'mno345'
down_revision = 'jkl012'
 
 
def upgrade() -> None:
    # 步骤 1：创建新表
    op.create_table(
        'addresses',
        sa.Column('id', sa.UUID(), primary_key=True,
                  server_default=sa.text('gen_random_uuid()')),
        sa.Column('user_id', sa.UUID(), sa.ForeignKey('users.id'),
                  nullable=False),
        sa.Column('street', sa.String(255), nullable=False),
        sa.Column('city', sa.String(100), nullable=False),
        sa.Column('country', sa.String(100), nullable=False),
        sa.Column('created_at', sa.DateTime(),
                  server_default=sa.text('NOW()'), nullable=False),
    )
 
    # 步骤 2：分批迁移数据（避免长事务和内存溢出）
    conn = op.get_bind()
    batch_size = 5000
    offset = 0
 
    while True:
        result = conn.execute(text("""
            INSERT INTO addresses (user_id, street, city, country)
            SELECT id, address_street, address_city, address_country
            FROM users
            WHERE address_street IS NOT NULL
            ORDER BY id
            LIMIT :batch_size OFFSET :offset
            RETURNING id
        """), {"batch_size": batch_size, "offset": offset})
 
        rows = result.fetchall()
        if not rows:
            break
 
        offset += batch_size
        # 可选：输出进度
        print(f"  已迁移 {offset} 条地址记录...")
 
    # 步骤 3：创建索引（在数据迁移完成后）
    op.create_index('idx_addresses_user_id', 'addresses', ['user_id'])
 
    # 注意：不要在这里删除旧列！留到下一个迁移版本
 
 
def downgrade() -> None:
    # 回滚：将数据迁移回 users 表
    conn = op.get_bind()
    conn.execute(text("""
        UPDATE users u
        SET address_street = a.street,
            address_city = a.city,
            address_country = a.country
        FROM addresses a
        WHERE a.user_id = u.id
    """))
 
    op.drop_table('addresses')

# .github/workflows/migration-safety.yml
name: Migration Safety Check
 
on:
  pull_request:
    paths:
      - 'prisma/migrations/**'
      - 'drizzle/**'
      - 'alembic/versions/**'
      - 'migrations/**'
      - 'src/migrations/**'
 
jobs:
  migration-review:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_DB: test_db
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: postgres
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
 
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0  # 需要完整历史来比较变更
 
      - name: Setup Node.js
        uses: actions/setup-node@v4
        with:
          node-version: '22'
 
      - name: Install dependencies
        run: npm ci
 
      # 步骤 1：检测迁移文件变更
      - name: Detect migration changes
        id: detect
        run: |
          CHANGED_FILES=$(git diff --name-only origin/main...HEAD)
          echo "changed_files<<EOF" >> $GITHUB_OUTPUT
          echo "$CHANGED_FILES" >> $GITHUB_OUTPUT
          echo "EOF" >> $GITHUB_OUTPUT
 
          # 检查是否有迁移文件变更
          if echo "$CHANGED_FILES" | grep -qE '(migrations/|drizzle/)'; then
            echo "has_migrations=true" >> $GITHUB_OUTPUT
          else
            echo "has_migrations=false" >> $GITHUB_OUTPUT
          fi
 
      # 步骤 2：破坏性操作检测
      - name: Check for destructive operations
        if: steps.detect.outputs.has_migrations == 'true'
        run: |
          echo "=== 检查破坏性操作 ==="
          MIGRATION_FILES=$(git diff --name-only origin/main...HEAD | grep -E '\.(sql|py|ts)$' | grep -iE '(migration|alembic|drizzle)')
 
          FOUND_ISSUES=false
          for file in $MIGRATION_FILES; do
            if [ -f "$file" ]; then
              echo "检查文件: $file"
 
              # 检测 DROP 操作
              if grep -inE 'DROP\s+(TABLE|COLUMN|INDEX|CONSTRAINT)' "$file"; then
                echo "🔴 发现 DROP 操作: $file"
                FOUND_ISSUES=true
              fi
 
              # 检测 TRUNCATE 操作
              if grep -inE 'TRUNCATE\s+TABLE' "$file"; then
                echo "🔴 发现 TRUNCATE 操作: $file"
                FOUND_ISSUES=true
              fi
 
              # 检测 ALTER COLUMN TYPE
              if grep -inE 'ALTER\s+COLUMN.*TYPE' "$file"; then
                echo "🟡 发现 ALTER COLUMN TYPE: $file"
                FOUND_ISSUES=true
              fi
 
              # 检测非 CONCURRENTLY 的 CREATE INDEX
              if grep -inE 'CREATE\s+INDEX(?!\s+CONCURRENTLY)' "$file"; then
                echo "🟡 发现非 CONCURRENTLY 的 CREATE INDEX: $file"
              fi
 
              # 检测 NOT NULL 无 DEFAULT
              if grep -inE 'NOT\s+NULL' "$file" | grep -ivE 'DEFAULT'; then
                echo "🟡 发现 NOT NULL 无 DEFAULT: $file"
              fi
            fi
          done
 
          if [ "$FOUND_ISSUES" = true ]; then
            echo ""
            echo "⚠️ 发现潜在危险操作，请确认是否安全。"
            echo "如果确认安全，请在 PR 描述中添加 [migration-approved] 标签。"
            # 不直接失败，而是添加 PR 评论
          fi
 
      # 步骤 3：迁移可逆性验证
      - name: Verify migration reversibility
        if: steps.detect.outputs.has_migrations == 'true'
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
        run: |
          echo "=== 验证迁移可逆性 ==="
 
          # 应用所有迁移
          npx prisma migrate deploy  # 或对应的 ORM 命令
 
          # 记录 Schema
          pg_dump --schema-only -h localhost -U postgres test_db > schema_v1.sql
 
          # 注意：回滚验证取决于 ORM
          # Prisma 不原生支持 down migration
          # Alembic: alembic downgrade -1 && alembic upgrade head
          # Django: python manage.py migrate app 000X && python manage.py migrate
 
          echo "✅ 迁移应用成功"
 
      # 步骤 4：生成迁移审查报告
      - name: Generate migration review report
        if: steps.detect.outputs.has_migrations == 'true'
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const { execSync } = require('child_process');
 
            // 获取变更的迁移文件
            const changedFiles = execSync(
              'git diff --name-only origin/main...HEAD'
            ).toString().split('\n').filter(f =>
              f.match(/(migration|alembic|drizzle)/i) && f.match(/\.(sql|py|ts)$/)
            );
 
            let report = '## 🗄️ 数据库迁移审查报告\n\n';
            report += `发现 ${changedFiles.length} 个迁移文件变更：\n\n`;
 
            for (const file of changedFiles) {
              if (fs.existsSync(file)) {
                const content = fs.readFileSync(file, 'utf8');
                report += `### \`${file}\`\n\n`;
 
                // 检测风险
                const risks = [];
                if (/DROP\s+(TABLE|COLUMN)/i.test(content))
                  risks.push('🔴 包含 DROP 操作');
                if (/TRUNCATE/i.test(content))
                  risks.push('🔴 包含 TRUNCATE 操作');
                if (/ALTER\s+COLUMN.*TYPE/i.test(content))
                  risks.push('🟡 包含列类型变更');
                if (/CREATE\s+INDEX(?!\s+CONCURRENTLY)/i.test(content))
                  risks.push('🟡 CREATE INDEX 未使用 CONCURRENTLY');
                if (/NOT\s+NULL/i.test(content) && !/DEFAULT/i.test(content))
                  risks.push('🟡 NOT NULL 列可能缺少 DEFAULT');
 
                if (risks.length > 0) {
                  report += '**风险项：**\n';
                  risks.forEach(r => report += `- ${r}\n`);
                } else {
                  report += '✅ 未发现明显风险\n';
                }
                report += '\n';
              }
            }
 
            report += '\n---\n';
            report += '> 🤖 此报告由 CI 自动生成。';
            report += '请数据库负责人审查后批准。\n';
 
            // 发布 PR 评论
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: report
            });
 
      # 步骤 5：需要审批的迁移
      - name: Require approval for dangerous migrations
        if: steps.detect.outputs.has_migrations == 'true'
        run: |
          MIGRATION_FILES=$(git diff --name-only origin/main...HEAD | grep -iE '(migration|alembic|drizzle)')
 
          HAS_DANGEROUS=false
          for file in $MIGRATION_FILES; do
            if [ -f "$file" ] && grep -qiE 'DROP\s+(TABLE|COLUMN)|TRUNCATE' "$file"; then
              HAS_DANGEROUS=true
              break
            fi
          done
 
          if [ "$HAS_DANGEROUS" = true ]; then
            # 检查 PR 描述是否包含审批标签
            PR_BODY="${{ github.event.pull_request.body }}"
            if echo "$PR_BODY" | grep -q '\[migration-approved\]'; then
              echo "✅ 迁移已获批准"
            else
              echo "❌ 发现破坏性迁移操作，需要在 PR 描述中添加 [migration-approved] 标签"
              exit 1
            fi
          fi

# .github/workflows/sql-review.yml
name: SQL Review
 
on:
  pull_request:
    paths:
      - '**/*.sql'
 
jobs:
  sql-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Bytebase SQL Review
        uses: bytebase/sql-review-action@v1
        with:
          # SQL Review 规则配置
          override-file-path: '.github/sql-review-rules.yml'
          # 数据库类型
          database-type: 'POSTGRES'

# .github/sql-review-rules.yml
# Bytebase SQL Review 规则配置
rules:
  # 禁止 DROP TABLE
  - type: statement.disallow-drop-table
    level: ERROR
 
  # 禁止 DROP COLUMN（需要审批）
  - type: statement.disallow-drop-column
    level: WARNING
 
  # 要求 CREATE INDEX CONCURRENTLY
  - type: index.create-concurrently
    level: ERROR
 
  # NOT NULL 列必须有 DEFAULT
  - type: column.require-default-for-not-null
    level: ERROR
 
  # 表必须有主键
  - type: table.require-pk
    level: ERROR
 
  # 列命名规范（snake_case）
  - type: naming.column
    level: WARNING
    payload:
      format: "^[a-z][a-z0-9_]*$"
 
  # 索引命名规范
  - type: naming.index
    level: WARNING
    payload:
      format: "^idx_[a-z][a-z0-9_]*$"
 
  # 外键命名规范
  - type: naming.foreign-key
    level: WARNING
    payload:
      format: "^fk_[a-z][a-z0-9_]*$"

# .github/workflows/atlas-ci.yml
name: Atlas Migration CI
 
on:
  pull_request:
    paths:
      - 'schema.hcl'
      - 'migrations/**'
 
jobs:
  atlas-lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Atlas Lint
        uses: ariga/atlas-action/migrate/lint@v1
        with:
          dir: 'file://migrations'
          dev-url: 'docker://postgres/16/dev'
          # 检测破坏性变更
          # 检测数据依赖问题
          # 验证迁移文件完整性

// .kiro/hooks/migration-review.json
{
  "name": "Migration Safety Review",
  "description": "当迁移文件被创建或修改时，自动进行安全审查",
  "event": "fileCreated",
  "filePatterns": [
    "prisma/migrations/**/*.sql",
    "drizzle/**/*.sql",
    "alembic/versions/**/*.py",
    "migrations/**/*.sql",
    "migrations/**/*.py",
    "src/migrations/**/*.ts"
  ],
  "action": "askAgent",
  "prompt": "请对这个新创建的迁移文件进行安全审查。检查以下风险：\n1. 破坏性操作（DROP TABLE/COLUMN/INDEX）\n2. 锁表风险（ALTER TABLE 在大表上）\n3. 数据丢失风险（NOT NULL 无 DEFAULT、类型缩小）\n4. 向后兼容性（旧代码是否能与新 Schema 兼容）\n5. 回滚可行性（down migration 是否完整）\n\n对每个发现的问题标注风险级别（🔴高/🟡中/🟢低）并提供修复建议。"
}

// .kiro/hooks/schema-change-reminder.json
{
  "name": "Schema Change Migration Reminder",
  "description": "当 Schema 文件被修改时，提醒生成迁移文件",
  "event": "fileEdited",
  "filePatterns": [
    "prisma/schema.prisma",
    "src/db/schema.ts",
    "src/db/schema/*.ts",
    "app/models.py",
    "*/models.py"
  ],
  "action": "askAgent",
  "prompt": "检测到 Schema/Model 文件被修改。请：\n1. 分析变更内容\n2. 评估是否需要生成迁移文件\n3. 如果需要，提供生成迁移的命令\n4. 预估迁移的安全风险\n5. 如果涉及大表变更，建议使用 Expand-Contract 模式"
}

// .kiro/hooks/pre-deploy-migration-check.json
{
  "name": "Pre-Deploy Migration Check",
  "description": "部署前检查迁移文件的安全性和完整性",
  "event": "userTriggered",
  "action": "askAgent",
  "prompt": "请执行部署前迁移检查：\n1. 列出所有待执行的迁移文件\n2. 检查每个迁移的安全性（破坏性操作、锁表风险）\n3. 验证迁移顺序是否正确\n4. 确认所有迁移都有对应的回滚方案\n5. 生成部署清单和执行顺序建议"
}

<!-- .kiro/steering/database-migration-safety.md -->
# 数据库迁移安全规范
 
## 迁移文件生成规则
 
1. **永远不要直接修改已应用的迁移文件**
   - 已应用的迁移是不可变的历史记录
   - 需要修改时，创建新的迁移文件
 
2. **每个迁移文件只做一件事**
   - 不要在一个迁移中混合 Schema 变更和数据迁移
   - Schema 变更和数据迁移应该是独立的迁移文件
 
3. **所有迁移必须可回滚**
   - 每个 up migration 必须有对应的 down migration
   - down migration 必须经过测试验证
 
4. **大表操作必须使用安全模式**
   - PostgreSQL：CREATE INDEX CONCURRENTLY
   - PostgreSQL：ADD CONSTRAINT ... NOT VALID + VALIDATE CONSTRAINT
   - MySQL：使用 pt-online-schema-change 或 gh-ost
   - 数据迁移：分批处理（batch_size ≤ 5000）
 
5. **破坏性操作必须使用 Expand-Contract 模式**
   - DROP COLUMN：先停止写入 → 下个版本删除
   - RENAME COLUMN：新增列 → 数据同步 → 迁移代码 → 删除旧列
   - ALTER COLUMN TYPE：新增列 → 数据转换 → 迁移代码 → 删除旧列
 
6. **新增 NOT NULL 列的安全步骤**
   - 步骤 1：ADD COLUMN ... DEFAULT value（nullable）
   - 步骤 2：UPDATE 填充历史数据（分批）
   - 步骤 3：ALTER COLUMN SET NOT NULL
 
7. **迁移文件命名规范**
   - 使用描述性名称：add_user_avatar、create_posts_table
   - 不要使用通用名称：update、fix、change
 
## AI 生成迁移的审查清单
 
当 AI 生成迁移文件时，必须检查：
- [ ] 是否包含破坏性操作？
- [ ] 大表操作是否使用了安全模式？
- [ ] NOT NULL 列是否有 DEFAULT？
- [ ] 索引创建是否使用了 CONCURRENTLY？
- [ ] down migration 是否完整且可逆？
- [ ] 是否考虑了滚动部署的兼容性？
- [ ] 数据迁移是否使用了分批处理？
- [ ] 外键约束是否正确设置了级联行为？

┌──────────────────────────────────────────────────────────────────────────┐
│                        迁移命令速查表                                      │
├──────────┬───────────────────────────────────────────────────────────────┤
│ 操作      │ 各 ORM 对应命令                                               │
├──────────┼───────────────────────────────────────────────────────────────┤
│          │ Prisma:    npx prisma migrate dev --name <name>               │
│ 生成迁移  │ Drizzle:   npx drizzle-kit generate                          │
│          │ Alembic:   alembic revision --autogenerate -m "<msg>"         │
│          │ TypeORM:   npx typeorm migration:generate src/migrations/Name │
│          │ Diesel:    diesel migration generate <name>                   │
│          │ Django:    python manage.py makemigrations                    │
├──────────┼───────────────────────────────────────────────────────────────┤
│          │ Prisma:    npx prisma migrate deploy                          │
│ 应用迁移  │ Drizzle:   npx drizzle-kit migrate                           │
│          │ Alembic:   alembic upgrade head                               │
│          │ TypeORM:   npx typeorm migration:run -d src/data-source.ts    │
│          │ Diesel:    diesel migration run                               │
│          │ Django:    python manage.py migrate                           │
├──────────┼───────────────────────────────────────────────────────────────┤
│          │ Prisma:    npx prisma migrate resolve（手动标记）               │
│ 回滚迁移  │ Drizzle:   手动执行回滚 SQL                                    │
│          │ Alembic:   alembic downgrade -1                               │
│          │ TypeORM:   npx typeorm migration:revert -d src/data-source.ts │
│          │ Diesel:    diesel migration revert                            │
│          │ Django:    python manage.py migrate <app> <migration_number>  │
├──────────┼───────────────────────────────────────────────────────────────┤
│          │ Prisma:    npx prisma migrate status                          │
│ 查看状态  │ Drizzle:   npx drizzle-kit check                             │
│          │ Alembic:   alembic current / alembic history                  │
│          │ TypeORM:   npx typeorm migration:show -d src/data-source.ts   │
│          │ Diesel:    diesel migration list                              │
│          │ Django:    python manage.py showmigrations                    │
├──────────┼───────────────────────────────────────────────────────────────┤
│          │ Prisma:    npx prisma migrate diff                            │
│ 预览 SQL  │ Drizzle:   查看生成的 .sql 文件                                │
│          │ Alembic:   alembic upgrade head --sql                         │
│          │ TypeORM:   查看生成的 .ts 文件中的 SQL                          │
│          │ Diesel:    查看 up.sql / down.sql                             │
│          │ Django:    python manage.py sqlmigrate <app> <number>         │
└──────────┴───────────────────────────────────────────────────────────────┘

-- Migration 001: expand_user_table.sql
-- 安全级别：🟢 低风险（仅添加 nullable 列和新表）
 
-- 1. 添加新列（nullable，不影响现有代码）
ALTER TABLE users ADD COLUMN first_name VARCHAR(50);
ALTER TABLE users ADD COLUMN last_name VARCHAR(50);
ALTER TABLE users ADD COLUMN phone_verified BOOLEAN DEFAULT FALSE;
 
-- 2. 创建地址表
CREATE TABLE addresses (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    type VARCHAR(20) NOT NULL DEFAULT 'shipping',  -- shipping, billing
    street VARCHAR(255) NOT NULL,
    city VARCHAR(100) NOT NULL,
    state VARCHAR(100),
    postal_code VARCHAR(20),
    country VARCHAR(100) NOT NULL DEFAULT 'CN',
    is_default BOOLEAN NOT NULL DEFAULT FALSE,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);
 
-- 3. 创建索引（CONCURRENTLY 避免锁表）
CREATE INDEX CONCURRENTLY idx_addresses_user_id ON addresses(user_id);
CREATE INDEX CONCURRENTLY idx_addresses_user_default
    ON addresses(user_id) WHERE is_default = TRUE;
CREATE INDEX CONCURRENTLY idx_users_phone_verified
    ON users(phone_verified) WHERE phone_verified = TRUE;

-- Migration 001: expand_user_table_down.sql（回滚）
DROP INDEX IF EXISTS idx_users_phone_verified;
DROP INDEX IF EXISTS idx_addresses_user_default;
DROP INDEX IF EXISTS idx_addresses_user_id;
DROP TABLE IF EXISTS addresses;
ALTER TABLE users DROP COLUMN IF EXISTS phone_verified;
ALTER TABLE users DROP COLUMN IF EXISTS last_name;
ALTER TABLE users DROP COLUMN IF EXISTS first_name;

-- Migration 002: migrate_user_data.sql
-- 安全级别：🟡 中风险（数据迁移，需分批执行）
 
-- 1. 拆分 full_name 到 first_name + last_name（分批处理）
DO $$
DECLARE
    batch_size INT := 5000;
    total_rows INT;
    processed INT := 0;
BEGIN
    SELECT count(*) INTO total_rows FROM users WHERE first_name IS NULL;
    RAISE NOTICE '需要迁移 % 行数据', total_rows;
 
    LOOP
        UPDATE users
        SET
            first_name = COALESCE(
                split_part(full_name, ' ', 1),
                full_name
            ),
            last_name = COALESCE(
                NULLIF(split_part(full_name, ' ', 2), ''),
                ''
            )
        WHERE id IN (
            SELECT id FROM users
            WHERE first_name IS NULL
            ORDER BY id
            LIMIT batch_size
        );
 
        GET DIAGNOSTICS processed = ROW_COUNT;
        EXIT WHEN processed = 0;
 
        RAISE NOTICE '已处理 % 行', processed;
        -- 短暂暂停，减少对在线查询的影响
        PERFORM pg_sleep(0.1);
    END LOOP;
 
    RAISE NOTICE '名字拆分完成';
END $$;
 
-- 2. 迁移地址数据到 addresses 表（分批处理）
DO $$
DECLARE
    batch_size INT := 5000;
    processed INT := 0;
BEGIN
    LOOP
        INSERT INTO addresses (user_id, type, street, city, state, postal_code, country, is_default)
        SELECT
            u.id,
            'shipping',
            u.address_street,
            u.address_city,
            u.address_state,
            u.address_postal_code,
            COALESCE(u.address_country, 'CN'),
            TRUE
        FROM users u
        WHERE u.address_street IS NOT NULL
          AND NOT EXISTS (
              SELECT 1 FROM addresses a WHERE a.user_id = u.id
          )
        ORDER BY u.id
        LIMIT batch_size;
 
        GET DIAGNOSTICS processed = ROW_COUNT;
        EXIT WHEN processed = 0;
 
        RAISE NOTICE '已迁移 % 条地址', processed;
        PERFORM pg_sleep(0.1);
    END LOOP;
 
    RAISE NOTICE '地址迁移完成';
END $$;
 
-- 3. 验证数据一致性
DO $$
DECLARE
    null_names INT;
    missing_addresses INT;
BEGIN
    SELECT count(*) INTO null_names
    FROM users WHERE first_name IS NULL AND full_name IS NOT NULL;
 
    SELECT count(*) INTO missing_addresses
    FROM users u
    WHERE u.address_street IS NOT NULL
      AND NOT EXISTS (SELECT 1 FROM addresses a WHERE a.user_id = u.id);
 
    IF null_names > 0 THEN
        RAISE EXCEPTION '发现 % 个用户名字未迁移！', null_names;
    END IF;
 
    IF missing_addresses > 0 THEN
        RAISE EXCEPTION '发现 % 个用户地址未迁移！', missing_addresses;
    END IF;
 
    RAISE NOTICE '✅ 数据一致性验证通过';
END $$;

// 新代码同时读写新旧字段（过渡期）
class UserService {
  async getUser(id: string) {
    const user = await prisma.user.findUnique({
      where: { id },
      include: { addresses: true },
    });
 
    return {
      ...user,
      // 优先使用新字段，回退到旧字段
      firstName: user.firstName || user.fullName?.split(' ')[0] || '',
      lastName: user.lastName || user.fullName?.split(' ')[1] || '',
      // 地址优先从 addresses 表读取
      address: user.addresses?.[0] || {
        street: user.addressStreet,
        city: user.addressCity,
      },
    };
  }
 
  async updateUser(id: string, data: UpdateUserDto) {
    // 双写：同时更新新旧字段
    await prisma.user.update({
      where: { id },
      data: {
        firstName: data.firstName,
        lastName: data.lastName,
        fullName: `${data.firstName} ${data.lastName}`,  // 保持旧字段同步
      },
    });
  }
}

-- Migration 003: contract_user_table.sql
-- 安全级别：🔴 高风险（删除列，需确认所有代码已迁移）
-- ⚠️ 执行前检查清单：
-- [ ] 所有服务已部署使用新字段的代码
-- [ ] 监控确认无代码访问旧字段
-- [ ] 数据一致性验证通过
-- [ ] 已创建数据库备份
 
-- 1. 设置新列为 NOT NULL（数据已填充）
ALTER TABLE users ALTER COLUMN first_name SET NOT NULL;
ALTER TABLE users ALTER COLUMN last_name SET NOT NULL;
 
-- 2. 删除旧列（不可逆！确认后执行）
ALTER TABLE users DROP COLUMN full_name;
ALTER TABLE users DROP COLUMN address_street;
ALTER TABLE users DROP COLUMN address_city;
ALTER TABLE users DROP COLUMN address_state;
ALTER TABLE users DROP COLUMN address_postal_code;
ALTER TABLE users DROP COLUMN address_country;
 
-- 3. 添加新索引
CREATE INDEX CONCURRENTLY idx_users_last_name ON users(last_name);