Node.js ORM 深度对比：Prisma、TypeORM、Sequelize 与 Drizzle

// Active Record 风格（TypeORM 支持）
const user = await User.findOne({ where: { id: 1 } });
user.name = 'Updated Name';
await user.save(); // 模型实例自带持久化方法

// Data Mapper 风格（TypeORM Repository 模式 / Prisma）
const user = await userRepository.findOne({ where: { id: 1 } });
const updated = new User({ ...user, name: 'Updated Name' });
await userRepository.save(updated);

import { PrismaClient } from '@prisma/client'

const prisma = new PrismaClient({
  log: ['query', 'info', 'warn', 'error'],
})

// 类型完全推导：user 的类型为 User | null
const user = await prisma.user.findUnique({
  where: { email: 'alice@example.com' },
  include: {
    posts: { where: { published: true }, orderBy: { createdAt: 'desc' } },
    profile: true,
  },
})

// 类型安全的数据创建：data 字段的类型与 UserCreateInput 严格匹配
const newUser = await prisma.user.create({
  data: {
    email: 'bob@example.com',
    name: 'Bob',
    profile: {
      create: { bio: 'Software Engineer' }
    },
    posts: {
      create: [
        { title: 'First Post', published: true },
        { title: 'Draft', published: false },
      ]
    }
  },
  include: { profile: true, posts: true }
})

// 批量查询 + 分页
const posts = await prisma.post.findMany({
  where: {
    published: true,
    author: { role: 'ADMIN' },
  },
  orderBy: { createdAt: 'desc' },
  skip: 0,
  take: 20,
  include: {
    author: { select: { id: true, name: true } },
    tags: true,
  },
})

// 独立事务（全部成功或全部回滚）
const [alice, bobPost] = await prisma.$transaction([
  prisma.user.create({
    data: { email: 'alice@example.com', name: 'Alice' }
  }),
  prisma.post.create({
    data: { title: 'Hello Prisma', published: true, authorId: 2 }
  }),
])

// 交互式事务（支持业务逻辑判断）
const result = await prisma.$transaction(async (tx) => {
  const user = await tx.user.findUnique({ where: { id: 1 } })
  if (!user) throw new Error('User not found')

  await tx.post.updateMany({
    where: { authorId: user.id },
    data: { published: true },
  })

  return await tx.user.update({
    where: { id: user.id },
    data: { updatedAt: new Date() },
  })
})

// 批量写入（单条 INSERT ... VALUES 多行）
await prisma.user.createMany({
  data: [
    { email: 'user1@example.com', name: 'User 1' },
    { email: 'user2@example.com', name: 'User 2' },
    { email: 'user3@example.com', name: 'User 3' },
  ],
  skipDuplicates: true,
})

import {
  Entity, PrimaryGeneratedColumn, Column,
  CreateDateColumn, UpdateDateColumn,
  OneToMany, ManyToOne, JoinColumn,
  OneToOne, ManyToMany, JoinTable,
} from 'typeorm'

export enum UserRole {
  USER = 'USER',
  ADMIN = 'ADMIN',
  MODERATOR = 'MODERATOR',
}

@Entity('users')
export class User {
  @PrimaryGeneratedColumn()
  id: number

  @Column({ unique: true })
  email: string

  @Column({ nullable: true })
  name: string

  @Column({ type: 'enum', enum: UserRole, default: UserRole.USER })
  role: UserRole

  @OneToMany(() => Post, post => post.author)
  posts: Post[]

  @OneToOne(() => Profile, profile => profile.user)
  profile: Profile

  @CreateDateColumn()
  createdAt: Date

  @UpdateDateColumn()
  updatedAt: Date
}

@Entity('posts')
export class Post {
  @PrimaryGeneratedColumn()
  id: number

  @Column()
  title: string

  @Column({ type: 'text', nullable: true })
  content: string

  @Column({ default: false })
  published: boolean

  @ManyToOne(() => User, user => user.posts)
  @JoinColumn({ name: 'author_id' })
  author: User

  @Column()
  authorId: number

  @ManyToMany(() => Tag)
  @JoinTable({
    name: 'post_tags',
    joinColumn: { name: 'post_id', referencedColumnName: 'id' },
    inverseJoinColumn: { name: 'tag_id', referencedColumnName: 'id' },
  })
  tags: Tag[]

  @CreateDateColumn()
  createdAt: Date
}

@Entity('profiles')
export class Profile {
  @PrimaryGeneratedColumn()
  id: number

  @Column({ type: 'text', nullable: true })
  bio: string

  @Column({ nullable: true })
  avatar: string

  @OneToOne(() => User, user => user.profile)
  @JoinColumn()
  user: User
}

@Entity('tags')
export class Tag {
  @PrimaryGeneratedColumn()
  id: number

  @Column({ unique: true })
  name: string
}

import { DataSource } from 'typeorm'

const dataSource = new DataSource({
  type: 'postgres',
  host: 'localhost',
  port: 5432,
  username: 'user',
  password: 'pass',
  database: 'mydb',
  entities: [User, Post, Profile, Tag],
  synchronize: false,
  logging: true,
})

await dataSource.initialize()

const userRepo = dataSource.getRepository(User)
const postRepo = dataSource.getRepository(Post)

// 基础 CRUD
const user = await userRepo.findOne({
  where: { email: 'alice@example.com' },
  relations: ['posts', 'profile'],
})

// QueryBuilder：复杂查询的首选
const posts = await postRepo
  .createQueryBuilder('post')
  .leftJoinAndSelect('post.author', 'author')
  .leftJoinAndSelect('post.tags', 'tag')
  .where('post.published = :published', { published: true })
  .andWhere('author.role = :role', { role: UserRole.ADMIN })
  .orderBy('post.created_at', 'DESC')
  .skip(0)
  .take(20)
  .getMany()

// 聚合查询
const stats = await postRepo
  .createQueryBuilder('post')
  .select('author.name', 'authorName')
  .addSelect('COUNT(post.id)', 'postCount')
  .addSelect('AVG(LENGTH(post.content))', 'avgContentLength')
  .leftJoin('post.author', 'author')
  .groupBy('author.id')
  .having('COUNT(post.id) > :minCount', { minCount: 5 })
  .getRawMany()

// 声明式事务（使用装饰器可以在应用中自动管理）
await dataSource.transaction(async (manager) => {
  const userRepoTx = manager.getRepository(User)
  const postRepoTx = manager.getRepository(Post)

  const user = await userRepoTx.save({
    email: 'alice@example.com',
    name: 'Alice',
    profile: { bio: 'Hello TypeORM' },
  })

  await postRepoTx.save({
    title: 'First Post',
    published: true,
    author: user,
    tags: [{ name: 'orm' }, { name: 'typescript' }],
  })
})

// Active Record 风格的实例方法（可选）
user.name = 'Updated Name'
await user.save()

// 级联删除配置在实体装饰器中：@OneToMany(() => Post, post => post.author, { cascade: true, onDelete: 'CASCADE' })

import { Sequelize, DataTypes, Model } from 'sequelize'

const sequelize = new Sequelize('postgres://user:pass@localhost:5432/mydb')

class User extends Model {
  declare id: number
  declare email: string
  declare name: string | null
  declare role: 'USER' | 'ADMIN' | 'MODERATOR'
  declare createdAt: Date
  declare updatedAt: Date
}

User.init(
  {
    id: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
    email: { type: DataTypes.STRING, allowNull: false, unique: true },
    name: { type: DataTypes.STRING },
    role: { type: DataTypes.ENUM('USER', 'ADMIN', 'MODERATOR'), defaultValue: 'USER' },
  },
  { sequelize, tableName: 'users', timestamps: true, modelName: 'User' }
)

class Post extends Model {
  declare id: number
  declare title: string
  declare content: string | null
  declare published: boolean
  declare authorId: number
}

Post.init(
  {
    id: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
    title: { type: DataTypes.STRING, allowNull: false },
    content: { type: DataTypes.TEXT },
    published: { type: DataTypes.BOOLEAN, defaultValue: false },
  },
  { sequelize, tableName: 'posts', timestamps: true, modelName: 'Post' }
)

class Tag extends Model {
  declare id: number
  declare name: string
}

Tag.init(
  {
    id: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
    name: { type: DataTypes.STRING, allowNull: false, unique: true },
  },
  { sequelize, tableName: 'tags', timestamps: false, modelName: 'Tag' }
)

// 关联定义
User.hasMany(Post, { foreignKey: 'authorId', as: 'posts' })
Post.belongsTo(User, { foreignKey: 'authorId', as: 'author' })

Post.belongsToMany(Tag, { through: 'PostTags', as: 'tags' })
Tag.belongsToMany(Post, { through: 'PostTags', as: 'posts' })

await sequelize.sync({ alter: true })

// 查询 + 关联加载
const user = await User.findOne({
  where: { email: 'alice@example.com' },
  include: [
    { model: Post, as: 'posts', where: { published: true }, required: false },
  ],
})

// 分页查询
const posts = await Post.findAll({
  where: { published: true },
  include: [
    { model: User, as: 'author', attributes: ['id', 'name'] },
    { model: Tag, as: 'tags' },
  ],
  order: [['createdAt', 'DESC']],
  offset: 0,
  limit: 20,
})

// 定义默认作用域与命名作用域
User.addScope('active', { where: { role: { [Op.ne]: 'BANNED' } } })
User.addScope('admins', { where: { role: 'ADMIN' } })

const admins = await User.scope('admins').findAll()

// 生命周期钩子
User.addHook('beforeCreate', (user) => {
  if (!user.role) user.role = 'USER'
})

Post.addHook('afterDestroy', async (post) => {
  await AuditLog.create({ action: 'post_deleted', postId: post.id })
})

// 托管事务
const result = await sequelize.transaction(async (t) => {
  const user = await User.create(
    { email: 'alice@example.com', name: 'Alice' },
    { transaction: t }
  )
  const post = await Post.create(
    { title: 'Hello Sequelize', published: true, authorId: user.id },
    { transaction: t }
  )
  return { user, post }
})

// 手动事务
const t = await sequelize.transaction()
try {
  await User.create({ email: 'bob@example.com' }, { transaction: t })
  await t.commit()
} catch (error) {
  await t.rollback()
}

import { pgTable, serial, varchar, text, boolean, timestamp, integer, pgEnum } from 'drizzle-orm/pg-core'

export const roleEnum = pgEnum('role', ['USER', 'ADMIN', 'MODERATOR'])

export const users = pgTable('users', {
  id: serial('id').primaryKey(),
  email: varchar('email', { length: 255 }).notNull().unique(),
  name: varchar('name', { length: 255 }),
  role: roleEnum('role').default('USER').notNull(),
  createdAt: timestamp('created_at').defaultNow(),
  updatedAt: timestamp('updated_at').defaultNow().$onUpdate(() => new Date()),
})

export const posts = pgTable('posts', {
  id: serial('id').primaryKey(),
  title: varchar('title', { length: 255 }).notNull(),
  content: text('content'),
  published: boolean('published').default(false),
  authorId: integer('author_id').notNull().references(() => users.id),
  createdAt: timestamp('created_at').defaultNow(),
})

export const tags = pgTable('tags', {
  id: serial('id').primaryKey(),
  name: varchar('name', { length: 255 }).notNull().unique(),
})

// 多对多关联需要手动定义联结表
export const postTags = pgTable('post_tags', {
  postId: integer('post_id').notNull().references(() => posts.id),
  tagId: integer('tag_id').notNull().references(() => tags.id),
})

import { relations } from 'drizzle-orm'

export const usersRelations = relations(users, ({ many, one }) => ({
  posts: many(posts),
  profile: one(profiles),
}))

export const postsRelations = relations(posts, ({ one, many }) => ({
  author: one(users, { fields: [posts.authorId], references: [users.id] }),
  tags: many(postTags),
}))

// 查询 API
import { eq, desc, and, count, avg, sql } from 'drizzle-orm'
import { drizzle } from 'drizzle-orm/node-postgres'
import { Pool } from 'pg'

const pool = new Pool({ connectionString: process.env.DATABASE_URL })
const db = drizzle(pool, { schema: { users, posts, tags, postTags, usersRelations, postsRelations } })

// 基础查询（类型安全）
const user = await db.query.users.findFirst({
  where: eq(users.email, 'alice@example.com'),
  with: {
    posts: { where: eq(posts.published, true), orderBy: [desc(posts.createdAt)] },
    profile: true,
  },
})

// SQL-like 的 select 查询
const result = await db
  .select({
    id: users.id,
    name: users.name,
    postCount: count(posts.id),
  })
  .from(users)
  .leftJoin(posts, eq(users.id, posts.authorId))
  .where(eq(users.role, 'ADMIN'))
  .groupBy(users.id)
  .having(sql`COUNT(${posts.id}) > 5`)

// 聚合
const stats = await db
  .select({
    totalUsers: count(users.id),
    avgPostsPerUser: avg(count(posts.id)),
  })
  .from(users)
  .leftJoin(posts, eq(users.id, posts.authorId))

const posts = await prisma.post.findMany({
  where: {
    published: true,
    author: { role: 'ADMIN' },
  },
  orderBy: { createdAt: 'desc' },
  take: 20,
  include: {
    author: { select: { id: true, name: true } },
    tags: true,
  },
})

const posts = await postRepo
  .createQueryBuilder('post')
  .leftJoinAndSelect('post.author', 'author')
  .leftJoinAndSelect('post.tags', 'tag')
  .where('post.published = :published', { published: true })
  .andWhere('author.role = :role', { role: UserRole.ADMIN })
  .orderBy('post.createdAt', 'DESC')
  .take(20)
  .getMany()

const posts = await Post.findAll({
  where: { published: true },
  include: [
    {
      model: User,
      as: 'author',
      where: { role: 'ADMIN' },
      attributes: ['id', 'name'],
    },
    { model: Tag, as: 'tags' },
  ],
  order: [['createdAt', 'DESC']],
  limit: 20,
})

const posts = await db.query.posts.findMany({
  where: and(eq(posts.published, true), eq(users.role, 'ADMIN')),
  orderBy: [desc(posts.createdAt)],
  limit: 20,
  with: {
    author: { columns: { id: true, name: true } },
    tags: true,
  },
})

// Prisma 的原始 SQL 查询
const result = await prisma.$queryRaw`
  SELECT p.id, p.title, u.name as author_name,
         COUNT(DISTINCT l.user_id) as like_count,
         ROW_NUMBER() OVER (PARTITION BY u.id ORDER BY p.created_at DESC) as rn
  FROM posts p
  JOIN users u ON p.author_id = u.id
  LEFT JOIN post_likes l ON p.id = l.post_id
  WHERE p.published = true
  GROUP BY p.id, u.name
  HAVING COUNT(DISTINCT l.user_id) > 10
`

// TypeORM 原生查询
const results = await dataSource.query(`
  SELECT p.id, p.title, u.name as author_name
  FROM posts p
  JOIN users u ON p.author_id = u.id
  WHERE p.created_at > $1
`, [new Date(Date.now() - 7 * 24 * 60 * 60 * 1000)])

import { sql } from 'drizzle-orm'

// Drizzle 中直接拼接 SQL 表达式
const result = await db.execute(sql`
  WITH recent_posts AS (
    SELECT * FROM ${posts} WHERE created_at > NOW() - INTERVAL '7 days'
  )
  SELECT * FROM recent_posts WHERE published = true
`)

# 开发阶段：自动检测 Schema 变更并生成迁移文件
npx prisma migrate dev --name add_user_role

# 代码审查：团队成员审查迁移 SQL 后再合并
# CI/CD 部署阶段：只应用已审查的迁移
npx prisma migrate deploy

# 生产排障：标记迁移状态
npx prisma migrate resolve --applied "20240101000000_init"

# 生成迁移文件（基于当前 Schema 和数据库的差异）
npx typeorm migration:generate -n AddUserRole

# 编写手动迁移（复杂变更）
npx typeorm migration:create -n RefactorPostsTable

# 执行迁移
npx typeorm migration:run
npx typeorm migration:revert

npx sequelize-cli migration:generate --name add-user-role
npx sequelize-cli db:migrate
npx sequelize-cli db:migrate:undo

npx drizzle-kit generate:pg   # 生成 SQL 迁移
npx drizzle-kit push:pg       # 快速推送到开发数据库
npx drizzle-kit check         # 检查迁移状态

// Prisma 连接池配置
const prisma = new PrismaClient({
  datasources: {
    db: {
      url: process.env.DATABASE_URL,
    },
  },
  // Prisma 内部使用 pgbouncer 兼容模式或直连模式
  // 连接池通过 DATABASE_URL 中的 connection_limit 参数控制
})

// TypeORM 连接池
const dataSource = new DataSource({
  type: 'postgres',
  url: process.env.DATABASE_URL,
  extra: {
    max: 20,              // 最大连接数
    idleTimeoutMillis: 30000,
    connectionTimeoutMillis: 2000,
  },
})

// Sequelize 连接池
const sequelize = new Sequelize(process.env.DATABASE_URL!, {
  pool: {
    max: 20,
    min: 5,
    acquire: 30000,
    idle: 10000,
  },
})

// Drizzle（连接池由底层驱动管理）
import { Pool } from 'pg'
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 20,
  idleTimeoutMillis: 30000,
})
const db = drizzle(pool)

// 使用独立的数据库运行测试
const DATABASE_URL = 'postgresql://test:test@localhost:5433/test_db'

beforeAll(async () => {
  // 应用迁移到测试数据库
  execSync('npx prisma migrate deploy', {
    env: { ...process.env, DATABASE_URL },
  })
  prisma = new PrismaClient({ datasources: { db: { url: DATABASE_URL } } })
})

afterEach(async () => {
  // 清理测试数据（按依赖顺序）
  await prisma.post.deleteMany()
  await prisma.user.deleteMany()
})

import { PostgreSqlContainer } from '@testcontainers/postgresql'

let container: StartedPostgreSqlContainer
let testDataSource: DataSource | typeof prisma

beforeAll(async () => {
  container = await new PostgreSqlContainer().start()
  const url = container.getConnectionUri()
  // 使用 url 初始化 TypeORM / Prisma / Drizzle
}, 30000)

afterAll(async () => {
  await container.stop()
})