migration-helper

Convex 迁移助手

在进行不兼容变更时，安全地迁移 Convex 架构和数据。

适用场景

• 为现有表添加新的必填字段
• 更改字段类型或结构
• 拆分或合并表
• 重命名字段
• 将嵌套数据迁移为关系型数据

迁移原则

1. 不自动迁移：Convex 不会自动迁移数据
2. 增量变更是安全的：添加可选字段或新表是安全的
3. 不兼容变更需要编写代码：必填字段、类型变更都需要迁移代码
4. 零停机：编写迁移代码时，要确保迁移期间应用仍可运行

安全变更（无需迁移）

添加可选字段

// Before
users: defineTable({
  name: v.string(),
})

// After - Safe! New field is optional
users: defineTable({
  name: v.string(),
  bio: v.optional(v.string()),
})

添加新表

// Safe to add completely new tables
posts: defineTable({
  userId: v.id("users"),
  title: v.string(),
}).index("by_user", ["userId"])

添加索引

// Safe to add indexes at any time
users: defineTable({
  name: v.string(),
  email: v.string(),
})
  .index("by_email", ["email"]) // New index

不兼容变更（需要迁移）

添加必填字段

问题：现有文档不会包含这个新字段。

解决方案：先将字段设为可选，回填数据，然后再设为必填。

// Step 1: Add as optional
users: defineTable({
  name: v.string(),
  email: v.optional(v.string()), // Start optional
})

// Step 2: Create migration
import { internalMutation } from "./_generated/server";
import { v } from "convex/values";

export const backfillEmails = internalMutation({
  args: {},
  handler: async (ctx) => {
    const users = await ctx.db.query("users").collect();

    for (const user of users) {
      if (!user.email) {
        await ctx.db.patch(user._id, {
          email: `user-${user._id}@example.com`, // Default value
        });
      }
    }
  },
});

// Step 3: Run migration via dashboard or CLI
// npx convex run migrations:backfillEmails

// Step 4: Make field required (after all data migrated)
users: defineTable({
  name: v.string(),
  email: v.string(), // Now required
})

更改字段类型

示例：将 tags: v.array(v.string()) 改为单独的表

// Step 1: Create new structure (additive)
tags: defineTable({
  name: v.string(),
}).index("by_name", ["name"]),

postTags: defineTable({
  postId: v.id("posts"),
  tagId: v.id("tags"),
})
  .index("by_post", ["postId"])
  .index("by_tag", ["tagId"]),

// Keep old field as optional during migration
posts: defineTable({
  title: v.string(),
  tags: v.optional(v.array(v.string())), // Keep temporarily
})

// Step 2: Write migration
export const migrateTags = internalMutation({
  args: { batchSize: v.optional(v.number()) },
  handler: async (ctx, args) => {
    const batchSize = args.batchSize ?? 100;

    const posts = await ctx.db
      .query("posts")
      .filter(q => q.neq(q.field("tags"), undefined))
      .take(batchSize);

    for (const post of posts) {
      if (!post.tags || post.tags.length === 0) {
        await ctx.db.patch(post._id, { tags: undefined });
        continue;
      }

      // Create tags and relationships
      for (const tagName of post.tags) {
        // Get or create tag
        let tag = await ctx.db
          .query("tags")
          .withIndex("by_name", q => q.eq("name", tagName))
          .unique();

        if (!tag) {
          const tagId = await ctx.db.insert("tags", { name: tagName });
          tag = { _id: tagId, name: tagName };
        }

        // Create relationship
        const existing = await ctx.db
          .query("postTags")
          .withIndex("by_post", q => q.eq("postId", post._id))
          .filter(q => q.eq(q.field("tagId"), tag._id))
          .unique();

        if (!existing) {
          await ctx.db.insert("postTags", {
            postId: post._id,
            tagId: tag._id,
          });
        }
      }

      // Remove old field
      await ctx.db.patch(post._id, { tags: undefined });
    }

    return { migrated: posts.length };
  },
});

// Step 3: Run in batches via cron or manually
// Run multiple times until all migrated

// Step 4: Remove old field from schema
posts: defineTable({
  title: v.string(),
  // tags field removed
})

重命名字段

// Step 1: Add new field (optional)
users: defineTable({
  name: v.string(),
  displayName: v.optional(v.string()), // New name
})

// Step 2: Copy data
export const renameField = internalMutation({
  handler: async (ctx) => {
    const users = await ctx.db.query("users").collect();

    for (const user of users) {
      await ctx.db.patch(user._id, {
        displayName: user.name,
      });
    }
  },
});

// Step 3: Update schema (remove old field)
users: defineTable({
  displayName: v.string(),
})

// Step 4: Update all code to use new field name

迁移模式

批量处理

对于大型表，分批处理：

export const migrateBatch = internalMutation({
  args: {
    cursor: v.optional(v.string()),
    batchSize: v.number(),
  },
  handler: async (ctx, args) => {
    const batchSize = args.batchSize;
    let query = ctx.db.query("largeTable");

    // Use cursor for pagination if needed
    const items = await query.take(batchSize);

    for (const item of items) {
      await ctx.db.patch(item._id, {
        // migration logic
      });
    }

    return {
      processed: items.length,
      hasMore: items.length === batchSize,
    };
  },
});

定时迁移

使用 cron 作业逐步迁移：

// convex/crons.ts
import { cronJobs } from "convex/server";
import { internal } from "./_generated/api";

const crons = cronJobs();

crons.interval(
  "migrate-batch",
  { minutes: 5 }, // Every 5 minutes
  internal.migrations.migrateBatch,
  { batchSize: 100 }
);

export default crons;

双写模式

用于零停机迁移：

// Write to both old and new structure during transition
export const createPost = mutation({
  args: { title: v.string(), tags: v.array(v.string()) },
  handler: async (ctx, args) => {
    const user = await getCurrentUser(ctx);

    // Create post
    const postId = await ctx.db.insert("posts", {
      userId: user._id,
      title: args.title,
      // Keep writing old field during migration
      tags: args.tags,
    });

    // ALSO write to new structure
    for (const tagName of args.tags) {
      let tag = await ctx.db
        .query("tags")
        .withIndex("by_name", q => q.eq("name", tagName))
        .unique();

      if (!tag) {
        const tagId = await ctx.db.insert("tags", { name: tagName });
        tag = { _id: tagId };
      }

      await ctx.db.insert("postTags", {
        postId,
        tagId: tag._id,
      });
    }

    return postId;
  },
});

// After migration complete, remove old writes

测试迁移

验证迁移是否成功

export const verifyMigration = query({
  args: {},
  handler: async (ctx) => {
    const total = (await ctx.db.query("users").collect()).length;
    const migrated = (await ctx.db
      .query("users")
      .filter(q => q.neq(q.field("newField"), undefined))
      .collect()
    ).length;

    return {
      total,
      migrated,
      remaining: total - migrated,
      percentComplete: (migrated / total) * 100,
    };
  },
});

迁移检查清单

• 确定不兼容变更
• 将新结构添加为可选或增量变更
• 编写迁移函数（内部 mutation）
• 使用样例数据测试迁移
• 如果数据集较大，分批运行迁移
• 验证迁移已完成（所有记录均已更新）
• 更新应用代码以使用新结构
• 部署新代码
• 从架构中移除旧字段
• 清理迁移代码

常见问题

1. 不要立即将字段设为必填：始终先将其设为可选
2. 不要在单个事务中迁移：大型迁移应分批进行
3. 不要忘记更新查询：更新所有使用旧字段的代码
4. 不要过早删除旧字段：等待所有数据迁移完成
5. 彻底测试：先在开发环境中验证迁移

示例：完整迁移流程

// 1. Current schema
export default defineSchema({
  users: defineTable({
    name: v.string(),
  }),
});

// 2. Add optional field
export default defineSchema({
  users: defineTable({
    name: v.string(),
    role: v.optional(v.union(
      v.literal("user"),
      v.literal("admin")
    )),
  }),
});

// 3. Migration function
export const addDefaultRoles = internalMutation({
  handler: async (ctx) => {
    const users = await ctx.db.query("users").collect();
    for (const user of users) {
      if (!user.role) {
        await ctx.db.patch(user._id, { role: "user" });
      }
    }
  },
});

// 4. Run migration: npx convex run migrations:addDefaultRoles

// 5. Verify: Check all users have role

// 6. Make required
export default defineSchema({
  users: defineTable({
    name: v.string(),
    role: v.union(
      v.literal("user"),
      v.literal("admin")
    ),
  }),
});