MikroORM 进阶实战：JWT 认证、QueryBuilder、虚拟实体与软删除完整指南

导读

本文以 MikroORM 官方《Getting Started》系列教程的 Chapter 4（Advanced） 为骨架，在 Fastify + SQLite 的博客 API 项目背景下，系统讲解从路由注册优化到生产部署的完整进阶链路：自定义仓储（Custom Repository）、基于 @fastify/jwt 的 JWT 认证、Embeddables 嵌入对象、Zod 输入校验、QueryBuilder 聚合查询、虚拟实体与视图实体、onFlush 事件实现软删除、结果缓存、脱离 Web 容器的脚本与 CRON 场景，以及 TypeScript 编译与 Vite 打包两种部署方式。读完本文，你将掌握 MikroORM 在真实 API 项目中从"能跑"到"工程化"的全部关键手段，并能对照 源码实现 理解每个特性的底层机制。

路由注册的工程化重构

在实现剩余端点之前，教程先把散落在 bootstrap() 中的路由内联注册收敛为模块化的路由工厂函数，这也是后续所有端点能够清晰维护的前提。

在 src/modules/article 下新建 routes.ts，导出工厂函数，内部通过 initORM() 拿到全局 ORM 服务，并立即使用 findAndCount 实现分页列表：

import { FastifyInstance } from 'fastify';
import { initORM } from '../../db.js';

export async function registerArticleRoutes(app: FastifyInstance) {
  const db = initORM();

  app.get('/', async request => {
    const { limit, offset } = request.query as { limit?: number; offset?: number };
    const [items, total] = await db.article.findAndCount({}, {
      limit, offset,
    });

    return { items, total };
  });
}

findAndCount 是仓储层对 em.findAndCount 的封装（见 EntityRepository.ts），一次调用同时返回"数据数组"与"满足条件的总数"，天然适配分页接口。User 模块同样先建占位工厂：

import { FastifyInstance } from 'fastify';
import { initORM } from '../../db.js';

export async function registerUserRoutes(app: FastifyInstance) {
  // no routes yet
}

最后在 bootstrap() 中用 Fastify 的 app.register() 挂载，并通过 prefix 统一路由前缀：

 // register routes here
-app.get('/article', async request => {
-  ...
-});
+app.register(registerArticleRoutes, { prefix: 'article' });
+app.register(registerUserRoutes, { prefix: 'user' });

注册接口与自定义仓储

第一个 User 端点：sign-up

POST /user/sign-up 接收 email、fullName、password 三个必填字段，逻辑分三步：字段存在性检查 → 邮箱唯一性检查 → 创建实体并 flush：

export async function registerUserRoutes(app: FastifyInstance) {
  const db = initORM();

  // register new user
  app.post('/sign-up', async request => {
    const body = request.body as EntityData<User>;

    if (!body.email || !body.fullName || !body.password) {
      throw new Error('One of required fields is missing: email, fullName, password');
    }

    if ((await db.user.count({ email: body.email })) > 0) {
      throw new Error('This email is already registered, maybe you want to sign in?');
    }

    const user = db.user.create({
      fullName: body.fullName,
      email: body.email,
      password: body.password,
      bio: body.bio ?? '',
    });
    await db.em.flush();

    // after flush, we have the `user.id` set
    console.log(`User ${user.id} created`);

    return user;
  });
}

这里有两个值得注意的细节：

• db.user.create() 是 em.create() 的仓储快捷方式，它只做实体实例化与元数据填充，不写数据库；真正落库发生在 db.em.flush()。因此 flush 之后 user.id 才可用（自增主键由数据库回填）。
• bio 使用空串兜底 body.bio ?? ''，对应实体中 bio: p.text().default('') 的默认值设计。

自定义仓储：把业务查询收敛到仓储

"邮箱是否已注册"的检查散落在路由里既不易读也难复用。教程将其下沉为仓储方法。在 src/modules/user 下新建 user.repository.ts：

import { EntityRepository } from '@mikro-orm/sqlite';
import { User } from './user.entity.js';

export class UserRepository extends EntityRepository<User> {

  async exists(email: string) {
    const count = await this.count({ email });
    return count > 0;
  }

}

在 defineEntity 的配置中通过 repository 选项把实体绑定到自定义仓储类：

import { defineEntity, type InferEntity, p } from '@mikro-orm/core';
import { BaseSchema } from '../common/base.entity.js';
import { UserRepository } from './user.repository.js';

export const UserSchema = defineEntity({
  name: 'User',
  extends: BaseSchema,
  repository: () => UserRepository,
  properties: {
    fullName: p.string(),
    email: p.string(),
    password: p.string().hidden().lazy(),
    bio: p.text().default(''),
    articles: () => p.oneToMany(ArticleSchema).mappedBy('author'),
  },
  // hooks remain the same
});

// setClass from Chapter 2 remains the same

注意两点：

• password 同时调用了 .hidden()（序列化时隐藏）与 .lazy()（懒加载属性，查询时不自动读取，需要显式 populate，后文 login 中会用到）。
• repository 选项接收一个返回类的工厂函数 () => UserRepository，避免循环依赖问题。

Services 类型同步收紧：

export interface Services {
  orm: MikroORM;
  em: EntityManager;
- user: EntityRepository<User>;
+ user: UserRepository;
  article: EntityRepository<IArticle>;
  tag: EntityRepository<ITag>;
}

路由中的调用随之简化：

-if ((await db.user.count({ email: body.email })) > 0) {
+if (await db.user.exists(body.email)) {
  throw new Error('This email is already registered, maybe you want to sign in?');
}

登录与 JWT 认证

login 仓储方法与懒加载密码

第二个 User 端点是 POST /user/sign-in。登录逻辑同样放进自定义仓储。注意这里故意抛出通用错误信息"Invalid combination of email and password"，避免泄露"该邮箱已注册"这类可用于枚举用户的信息：

export class UserRepository extends EntityRepository<User> {

  // ...

  async login(email: string, password: string) {
    // use a more generic error so you don't leak that such email is registered
    const err = new Error('Invalid combination of email and password');
    const user = await this.findOneOrFail({ email }, {
      populate: ['password'], // password is a lazy property, we need to populate it
      failHandler: () => err,
    });

    if (await user.verifyPassword(password)) {
      return user;
    }

    throw err;
  }

}

populate: ['password'] 是本节的关键：因为 password 被定义为 lazy 属性，默认查询不会加载它，登录时必须显式 populate 才能拿到明文密码与 verifyPassword() 比对。failHandler 用于自定义"找不到实体"时抛出的错误——这里直接复用 err，保证"邮箱不存在"与"密码错误"返回完全一致的报错。

测试先行：先看到 500，再实现认证

教程先写测试再实现认证，用 app.inject() 直接向 Fastify 实例发请求（无需真实端口），并刻意使用不同端口支持并行测试：

import { FastifyInstance } from 'fastify';
import { afterAll, beforeAll, expect, test } from 'vitest';
import { initTestApp } from './utils.js';

let app: FastifyInstance;

beforeAll(async () => {
  // using different ports to allow parallel testing
  app = await initTestApp(30002);
});

afterAll(async () => {
  // closing only the fastify app - it will close the database connection via onClose hook automatically
  await app.close();
});

test('login', async () => {
  const res1 = await app.inject({
    method: 'post',
    url: '/user/sign-in',
    payload: {
      email: '[e-mail]',
      password: 'password123',
    },
  });

  expect(res1.statusCode).toBe(200);
  expect(res1.json()).toMatchObject({
    fullName: 'Foo Bar',
  });

  const res2 = await app.inject({
    method: 'post',
    url: '/user/sign-in',
    payload: {
      email: '[e-mail]',
      password: 'password456',
    },
  });

  expect(res2.statusCode).toBe(401);
  expect(res2.json()).toMatchObject({ error: 'Invalid combination of email and password' });
});

此时运行 npm test，密码错误的用例必然失败——因为路由直接 throw 了错误，Fastify 默认返回 500：

 FAIL  test/user.test.ts > login
AssertionError: expected 500 to be 401 // Object.is equality

- Expected
+ Received

- 401
+ 500

这正是教程的意图：先用测试暴露"认证缺失"这一真实问题，再引入 JWT 方案把它修好。

接入 @fastify/jwt

安装插件：

npm install @fastify/jwt

在 bootstrap() 中注册，secret 优先读环境变量，测试环境回退到固定值：

import fastifyJWT from '@fastify/jwt';

// ...

// register JWT plugin
app.register(fastifyJWT, {
  secret: process.env.JWT_SECRET ?? '12345678', // fallback for testing
});

注册后 request 对象获得 user 属性，app 对象获得两个方法：

• app.jwt.sign()：由 payload 生成 token；
• request.jwtVerify()：校验并解码 token，还原 payload。

token 中只存 user.id。为此在 UserSchema 上新增一个虚拟属性：

token: p.string().persist(false).nullable(),

.persist(false) 意味着该属性不映射为数据库列（虚拟属性），但可以被赋值、参与序列化——正好用来把 JWT 挂在返回的 user 对象上。这与 defineEntity 的属性链式构建器一一对应（见 defineEntity.ts 中 persist() 的定义）。

AuthError 与全局错误处理器

先定义可被精确识别的业务错误类型：

export class AuthError extends Error {}

并在 UserRepository 中替换原来的普通 Error：

import { AuthError } from '../common/utils.js';

export class UserRepository extends EntityRepository<User> {

  // ...

  async login(email: string, password: string) {
    const err = new AuthError('Invalid combination of email and password');
    const user = await this.findOneOrFail({ email }, {
      populate: ['password'], // password is a lazy property, we need to populate it
      failHandler: () => err,
    });

    if (await user.verifyPassword(password)) {
      return user;
    }

    throw err;
  }

}

在 sign-up 与 sign-in 处理器中签发 token：

// register new user
app.post('/sign-up', async request => {
  // ...

  const user = db.user.create({
    fullName: body.fullName,
    email: body.email,
    password: body.password,
    bio: body.bio ?? '',
  });
  await db.em.flush();

  user.token = app.jwt.sign({ id: user.id })

  return user;
});

// login existing user
app.post('/sign-in', async request => {
  const { email, password } = request.body as { email: string; password: string };
  const user = await db.user.login(email, password);
  user.token = app.jwt.sign({ id: user.id })

  return user;
});

认证中间件与全局错误处理

在 bootstrap() 中注册两个关键钩子。第一个 onRequest 钩子放在 ORM 的 RequestContext 钩子之后（这样才能在钩子内使用 db.user），负责解析 token 并加载当前用户；token 无效时静默忽略——只在该端点确实需要登录时（通过 getUserFromToken）才强制校验：

// register auth hook after the ORM one to use the context
app.addHook('onRequest', async request => {
  try {
    const ret = await request.jwtVerify<{ id: number }>();
    request.user = await db.user.findOneOrFail(ret.id);
  } catch (e) {
    app.log.error(e);
    // ignore token errors, we validate the request.user exists only where needed
  }
});

第二个是全局错误处理器，把 AuthError 映射为 401、把 ORM 的 NotFoundError（findOneOrFail 找不到实体时抛出）映射为 404，其余错误保持 500：

// register global error handler to process 404 errors from `findOneOrFail` calls
app.setErrorHandler((error, request, reply) => {
  if (error instanceof AuthError) {
    return reply.status(401).send({ error: error.message });
  }

  // we also handle not found errors automatically
  // `NotFoundError` is an error thrown by the ORM via `em.findOneOrFail()` method
  if (error instanceof NotFoundError) {
    return reply.status(404).send({ error: error.message });
  }

  app.log.error(error);
  reply.status(500).send({ error: error.message });
});

至此测试重新通过：客户端携带有效 token 时，request.user 会被自动加载；token 缺失或过期时，只有需要登录的端点会拒绝。

当前用户：profile 端点与 wrap().assign()

先加一个工具函数，从请求中取出当前用户：

import { FastifyRequest } from 'fastify';
import { User } from '../user/user.entity.js';

export function getUserFromToken(req: FastifyRequest): User {
  if (!req.user) {
    throw new Error('Please provide your token via Authorization header');
  }

  return req.user as User;
}

再实现查询与修改两个端点。修改时使用 wrap(user).assign()：该辅助方法会把普通数据正确映射到实体图，例如自动把外键值转换为实体引用，最后 flush 统一落库：

app.get('/profile', async request => {
  const user = getUserFromToken(request);
  return user;
});

app.patch('/profile', async request => {
  const user = getUserFromToken(request);
  wrap(user).assign(request.body as EntityData<User>);
  await db.em.flush();
  return user;
});

:::info Exercise 教程留给你一个练习：为这两个端点补齐测试。 :::

Embeddables：把一组列映射为一个对象

回到 User 实体，为它增加可选的社交账号（twitter / facebook / linkedin）。Embeddables 允许把多列映射到单个对象属性，非常适合"值对象"场景（详见 embeddables.md）。

用 defineEntity 定义一个 embeddable schema，再嵌入实体：

import { defineEntity, InferEntity, p } from '@mikro-orm/core';

// Define the embeddable schema
export const SocialSchema = defineEntity({
  name: 'Social',
  embeddable: true,
  properties: {
    twitter: p.string().nullable(),
    facebook: p.string().nullable(),
    linkedin: p.string().nullable(),
  },
});

export type ISocial = InferEntity<typeof SocialSchema>;

export const UserSchema = defineEntity({
  name: 'User',
  extends: BaseSchema,
  repository: () => UserRepository,
  properties: {
    fullName: p.string(),
    email: p.string(),
    password: p.string().hidden().lazy(),
    bio: p.text().default(''),
    articles: () => p.oneToMany(ArticleSchema).mappedBy('author'),
    social: () => p.embedded(SocialSchema).nullable(),
  },
  // hooks and setClass remain the same
});

默认情况下，嵌入对象会按前缀展开为多个独立列（前缀默认取属性名 social）。用 CLI 的 schema:update --dump 预览 DDL（只输出 SQL，不真正执行）：

$ npx mikro-orm schema:update --dump

alter table `user` add column `social_twitter` text null;
alter table `user` add column `social_facebook` text null;
alter table `user` add column `social_linkedin` text null;

如果希望整个对象存进一个 JSON 列，只需给嵌入属性追加 .object()：

social: () => p.embedded(SocialSchema).object().nullable(),

再次预览：

$ npx mikro-orm schema:update --dump

alter table `user` add column `social` json null;

确认方案后，生成并应用迁移：

$ npx mikro-orm migration:create

Migration20231105213316.ts successfully created

$ npx mikro-orm migration:up

Processing 'Migration20231105213316'
Applied 'Migration20231105213316'
Successfully migrated up to the latest version

从源码看，Embeddables 在元数据发现阶段会被特殊处理：MetadataDiscovery 会先排序发现实体（embeddable 优先），再把嵌入属性展开、合并进宿主实体（见 MetadataDiscovery.ts），prefixMode 等细节均可通过配置调整。

Zod 校验：让输入在进入 ORM 前先通过类型闸门

sign-up 现在要接收 social 属性。由于 em.create() 是构建实体的统一入口，直接传入即可：

const user = db.em.create(User, {   // `User` is the class from setClass
  fullName: body.fullName,
  email: body.email,
  password: body.password,
  bio: body.bio ?? '',
  social: body.social as ISocial,
});
await db.em.flush();

但教程更推荐显式校验输入：MikroORM 自身会校验必填属性与类型（详见 property-validation.md），但业务校验（如字段格式）应由应用层负责。Zod 在这里还有一个附加收益——parse 返回的 dto 是完全类型推导的，可以直接通过 em.create() 的类型检查，省去类型断言。

安装并编写 schema：

npm install zod

const socialSchema = z.object({
  twitter: z.string().optional(),
  facebook: z.string().optional(),
  linkedin: z.string().optional(),
});

const userSchema = z.object({
  email: z.string(),
  fullName: z.string(),
  password: z.string(),
  bio: z.string().optional(),
  social: socialSchema.optional(),
});

app.post('/sign-up', async request => {
  const dto = userSchema.parse(request.body);

  if (await db.user.exists(dto.email)) {
    throw new Error('This email is already registered, maybe you want to sign in?');
  }

  // thanks to zod, our `dto` is fully typed and passes the `em.create()` checks
  const user = db.user.create(dto);
  await db.em.flush(); // no need for explicit `em.persist()` when we use `em.create()`

  // after flush, we have the `user.id` set
  user.token = app.jwt.sign({ id: user.id });

  return user;
});

:::info 本示例展示的 Zod 校验刻意保持基础（与 MikroORM 自带的必填/类型校验能力对齐），更深入的用法请参考 property-validation.md。 :::

Article 剩余端点

详情、评论、创建

文章详情：根据 slug 用 findOneOrFail 查询，一次性 populate 作者、评论作者与正文（text 同样为 lazy 属性）：

app.get('/:slug', async request => {
  const { slug } = request.params as { slug: string };
  return db.article.findOneOrFail({ slug }, {
    populate: ['author', 'comments.author', 'text'],
  });
});

:::warning 教程刻意省略了请求参数的校验（超出本指南范围），但在真实项目中务必先校验再使用。 :::

发表评论：用 getUserFromToken 取当前用户，按 slug 找到文章，然后 create 评论。注意两点：em.create() 会自动把新实体加入工作单元，无需显式 em.persist()；向 article.comments.add(comment) 本质上是"顺带"操作，因为设置了 Comment.author 后关联会自动传播（见 propagation.md）：

app.post('/:slug/comment', async request => {
  const { slug, text } = request.params as { slug: string; text: string };
  const author = getUserFromToken(request);
  const article = await db.article.findOneOrFail({ slug });
  const comment = db.comment.create({ author, article, text });

  // We can add the comment to `article.comments` collection,
  // but in fact it is a no-op, as it will be automatically
  // propagated by setting Comment.author property.
  article.comments.add(comment);

  // mention we don't need to persist anything explicitly
  await db.em.flush();

  return comment;
});

创建文章：逻辑高度相似，只是换成了 article 实体：

app.post('/', async request => {
  const { title, description, text } = request.body as { title: string; description: string; text: string };
  const author = getUserFromToken(request);
  const article = db.article.create({
    title,
    description,
    text,
    author,
  });

  await db.em.flush();

  return article;
});

更新：wrap().assign() 与权限校验

更新接口用 wrap(article).assign() 把请求体映射到实体图（自动把外键转换为实体引用），并先做作者权限校验：

app.patch('/:id', async request => {
  const user = getUserFromToken(request);
  const params = request.params as { id: string };
  const article = await db.article.findOneOrFail(+params.id);
  verifyArticlePermissions(user, article);
  wrap(article).assign(request.body as EntityData<IArticle>);
  await db.em.flush();

  return article;
});

权限校验工具函数：

export function verifyArticlePermissions(user: User, article: IArticle): void {
  if (article.author.id !== user.id) {
    throw new Error('You are not the author of this article!');
  }
}

提示：wrap(entity).assign() 与 em.assign() 等价，且 em.assign() 对非受管实体同样有效。

一步到位：em.upsert() / upsertMany()

如果"创建或更新"是同一语义（例如幂等的客户端提交），可以改用 em.upsert()——它底层生成 INSERT ... ON CONFLICT 语句，一步完成：

-const article = await db.article.findOneOrFail(+params.id);
-wrap(article).assign(request.body as EntityData<IArticle>);
-await db.em.flush();
+const article = await db.article.upsert(request.body as IArticle);

批量场景使用 em.upsertMany()，全部数据在单条查询内完成。从源码看，upsert 会先尝试命中 Identity Map（unitOfWork.getById），命中则直接 assign 合并；实体已是受管状态且未开启 upsertManaged 时也会直接合并返回（见 EntityManager.ts）。详见 entity-manager.md。

删除：先查再删，与 nativeDelete

删除流程：先 findOne（注意不是 findOneOrFail，找不到时返回 { notFound: true }），校验权限后 em.remove(article).flush() 删除——remove 只是标记待删除，真正执行在 flush()：

app.delete('/:id', async request => {
  const user = getUserFromToken(request);
  const params = request.params as { id: string };
  const article = await db.article.findOne(+params.id);

  if (!article) {
    return { notFound: true };
  }

  verifyArticlePermissions(user, article);
  // mention `nativeDelete` alternative if we don't care about validations much
  await db.em.remove(article).flush();

  return { success: true };
});

如果不需要加载实体、不关心级联与生命周期钩子，可以直接执行原生删除（nativeDelete 是仓储层对 em.nativeDelete 的透传，见 EntityRepository.ts）：

await db.article.nativeDelete(+params.id);

Unit of Work 的自动批量能力

本指南虽未用到，但值得强调：基于 Unit of Work 的 EntityManager 会把同类型实体的多次变更自动批量合并为每条实体一条 SQL。

批量插入

for (let i = 1; i <= 5; i++) {
  em.create(User, {
    fullName: `Peter ${i}`,
    email: `peter+${i}@foo.bar`,
    password: '...',
  });
}

await em.flush();

insert into `user` (`name`, `email`) values
  ('Peter 1', '[e-mail]'),
  ('Peter 2', '[e-mail]'),
  ('Peter 3', '[e-mail]'),
  ('Peter 4', '[e-mail]'),
  ('Peter 5', '[e-mail]');

批量更新（CASE WHEN）

const users = await em.find(User, {});

for (const user of users) {
  user.name += ' changed!';
}

await em.flush();

update `user` set
  `name` = case
    when (`id` = 1) then 'Peter 1 changed!'
    when (`id` = 2) then 'Peter 2 changed!'
    when (`id` = 3) then 'Peter 3 changed!'
    when (`id` = 4) then 'Peter 4 changed!'
    when (`id` = 5) then 'Peter 5 changed!'
    else `priority` end
  where `id` in (1, 2, 3, 4, 5);

批量删除

const users = await em.find(User, {});

em.remove(users);

await em.flush();

delete from `user` where `id` in (1, 2, 3, 4, 5);

这正是 Unit of Work 模式的价值：调用方无需关心批处理细节，flush() 会负责聚合与排序所有变更。

关闭变更追踪：disableIdentityMap

某些场景下你希望某次查询不进入 Identity Map、不参与变更集追踪。用 disableIdentityMap 选项即可：从源码实现看（EntityManager.ts），它会 fork 一个保留事务上下文的新 EntityManager，在 fork 中执行查询后立即 fork.clear()，从而主 Identity Map 保持干净，但单次 find 返回的实体之间仍然互相连通。

const user = await db.user.findOneOrFail({ email: '[e-mail]' }, {
  disableIdentityMap: true,
});
user.name = 'changed';
await db.em.flush(); // calling flush have no effect, as the entity is not managed

这样返回的实体是游离（detached）实体，对它的修改不会被追踪。要重新纳入管理，需先 em.merge()。此外，disableIdentityMap 也可在 ORM 配置层全局开启（Configuration.ts）。

虚拟实体：把任意 SQL 查询映射为实体

最初的列表接口用 findAndCount 直接返回分页数据。若要自定义响应形状（例如附带作者名、标签数组、评论数），虚拟实体是首选方案：它不映射任何数据库表，而是在查询时动态求值一段 SQL（或 QueryBuilder），把结果映射为实体。虚拟实体仅用于读取，没有主键，因此不参与变更追踪（想要真正的数据库视图请看后文"视图实体"）。

用 defineEntity 定义时提供 expression 选项——可以是 SQL 字符串，也可以是返回 SQL/QueryBuilder 的回调；属性只支持标量：

import { defineEntity, InferEntity, EntityManager, p } from '@mikro-orm/core';
import { ArticleSchema } from './article.entity.js';

export const ArticleListingSchema = defineEntity({
  name: 'ArticleListing',
  expression: (em: EntityManager) => {
    return em.getRepository(ArticleSchema).listArticlesQuery();
  },
  properties: {
    slug: p.string(),
    title: p.string(),
    description: p.string(),
    tags: p.array(),
    author: p.integer(),
    authorName: p.string(),
    totalComments: p.integer(),
  },
});

export type IArticleListing = InferEntity<typeof ArticleListingSchema>;

同时为 Article 建立自定义仓储，把 listArticles() 封装进去（ArticleListingSchema 是虚拟实体，没有实体类，因此传给 findAndCount 的是 schema 本身）：

import { FindOptions, sql, EntityRepository } from '@mikro-orm/sqlite';
import { type IArticle, ArticleSchema } from './article.entity.js';
import { type IArticleListing, ArticleListingSchema } from './article-listing.entity.js';

// extending the EntityRepository exported from driver package, so we can access things like the QB factory
export class ArticleRepository extends EntityRepository<IArticle> {

  listArticlesQuery() {
    // just a placeholder for now
    return this.createQueryBuilder('a');
  }

  async listArticles(options: FindOptions<IArticleListing>) {
    const [items, total] = await this.em.findAndCount(ArticleListingSchema, {}, options);
    return { items, total };
  }

}

端点换用新方法：

// list articles
app.get('/', async request => {
  const { limit, offset } = request.query as { limit?: number; offset?: number };

  const { items, total } = await db.article.listArticles({
    limit, offset,
  });

  return { items, total };
});

QueryBuilder：聚合、JOIN 与子查询

listArticlesQuery() 要完成三件事：统计每篇文章的评论数、聚合标签名、联表取作者名。全部用 QueryBuilder 实现。

第一步：选择基础列：

return this.createQueryBuilder('a')
  .select(['slug', 'title', 'description', 'author']);

第二步：JOIN User 取作者名，用 sql.ref() 为列设置自定义别名（u.full_name 来自 User 实体隐含的下划线字段名）：

return this.createQueryBuilder('a')
  .select(['slug', 'title', 'description', 'author'])
  .addSelect(sql.ref('u.full_name').as('authorName'))
  .join('author', 'u')

第三步：两个子查询。评论数用 count() 子查询按 a.id 关联；标签用 group_concat 聚合，注意 sql\...`` 标签模板标记原始 SQL 片段以防被转义：

import { FindOptions, sql, EntityRepository } from '@mikro-orm/sqlite';
import { type IArticle, ArticleSchema } from './article.entity.js';
import { type IArticleListing, ArticleListingSchema } from './article-listing.entity.js';
import { CommentSchema } from './comment.entity.js';

export class ArticleRepository extends EntityRepository<IArticle> {

  // ...

  listArticlesQuery() {
    // sub-query for total number of comments
    const totalComments = this.em.createQueryBuilder(CommentSchema)
      .count()
      .where({ article: sql.ref('a.id') })
      // by calling `qb.as()` we alias the sub-query
      .as('totalComments');

    // sub-query for all used tags
    const usedTags = this.em.createQueryBuilder(ArticleSchema, 'aa')
      // we need to mark raw query fragment with `sql` helper
      // otherwise it would be escaped
      .select(sql`group_concat(distinct t.name)`)
      .join('aa.tags', 't')
      .where({ 'aa.id': sql.ref('a.id') })
      .groupBy('aa.author')
      .as('tags');

    // build final query
    return this.createQueryBuilder('a')
      .select(['slug', 'title', 'description', 'author'])
      .addSelect(sql.ref('u.full_name').as('authorName'))
      .join('author', 'u')
      .addSelect([totalComments, usedTags]);
  }

}

sql.ref() 与 sql\`` 都是原始 SQL 工具（详见 raw-queries.md）：前者引用列/标识符，后者标记整段原始表达式，二者都绕过默认的标识符转义。

手动执行 QueryBuilder

上面的例子把 QueryBuilder 交给虚拟实体执行。手动执行有三种模式（execute() 的第一参数）：

const res1 = await qb.execute('all'); // returns array of objects, default behavior
const res2 = await qb.execute('get'); // returns single object
const res3 = await qb.execute('run'); // returns object like `{ affectedRows: number, insertId: number, row: any }`

第二参数控制列名到属性名的映射。例如 Article 的 createdAt 属性隐式映射到 created_at 列：

const res1 = await em.createQueryBuilder(ArticleSchema).select('*').execute('get', true);
console.log(res1); // `createdAt` will be defined, while `created_at` will be missing

const res2 = await em.createQueryBuilder(ArticleSchema).select('*').execute('get', false);
console.log(res2); // `created_at` will be defined, while `createdAt` will be missing

若希望结果直接是实体实例（而非纯对象），用 getResult() / getSingleResult()：

const article = await em.createQueryBuilder(ArticleSchema)
  .select('*')
  .where({ id: 1 })
  .getSingleResult();
console.log(article); // Article { id: 1, ... }

const articles = await em.createQueryBuilder(ArticleSchema)
  .select('*')
  .getResult();
console.log(articles[0] instanceof Article); // true

qb.getResultList() 是 qb.getResult() 的别名。

更新测试：让断言跟上新的响应形状

接口响应结构变了，既有测试需要同步更新。先在 TestSeeder 里为文章补上测试评论（注意 em.assign 可以直接对已创建实体的集合属性赋值）：

export class TestSeeder extends Seeder {
  async run(em: EntityManager): Promise<void> {
-   em.create(UserSchema, {
+   const author = em.create(UserSchema, {
      fullName: "Foo Bar",
      email: "[e-mail]",
      // ...
    });

+   em.assign(author.articles[0], {
+     comments: [
+       { author, text: `random comment ${Math.random()}` },
+       { author, text: `random comment ${Math.random()}` },
+     ],
+   });
+
+   em.assign(author.articles[1], {
+     comments: [{ author, text: `random comment ${Math.random()}` }],
+   });
+
+   em.assign(author.articles[2], {
+     comments: [
+       { author, text: `random comment ${Math.random()}` },
+       { author, text: `random comment ${Math.random()}` },
+       { author, text: `random comment ${Math.random()}` },
+     ],
+   });
  }
}

测试断言随之调整：

expect(res.json()).toMatchObject({
  items: [
-   { author: 1, slug: "title-13", title: "title 1/3" },
-   { author: 1, slug: "title-23", title: "title 2/3" },
-   { author: 1, slug: "title-33", title: "title 3/3" },
+   {
+     slug: expect.any(String),
+     title: 'title 1/3',
+     description: 'desc 1/3',
+     tags: ['foo1', 'foo2'],
+     authorName: 'Foo Bar',
+     totalComments: 2,
+   },
+   {
+     slug: expect.any(String),
+     title: 'title 2/3',
+     description: 'desc 2/3',
+     tags: ['foo2'],
+     authorName: 'Foo Bar',
+     totalComments: 1,
+   },
+   {
+     slug: expect.any(String),
+     title: 'title 3/3',
+     description: 'desc 3/3',
+     tags: ['foo2', 'foo3'],
+     authorName: 'Foo Bar',
+     totalComments: 3,
+   },
  ],
  total: 3,
});

结果缓存

MikroORM 内置简单的结果缓存机制，只需在 em.find() 的 options 里加 cache。取值有三种形式：

• true：使用默认过期时间（全局可配，默认 1 秒，见 Configuration.ts 中 resultCache.expiration: 1000 的默认值）；
• 数字：显式指定过期毫秒数；
• 元组 <a href="https://link.gitcode.com/i/b12d1c53c4099d5f82a05f757d2eee21" target="_blank">cacheKey, expiration]：第一个元素是缓存键（string），第二个是过期毫秒数；可用该键通过 em.clearCache(cacheKey) 主动失效缓存（[EntityManager.ts）。

给文章列表接口加 5 秒缓存：

// list articles
app.get('/', async request => {
  const { limit, offset } = request.query as { limit?: number; offset?: number };

  const { items, total } = await db.article.listArticles({
    limit, offset,
    cache: 5_000, // 5 seconds
  });

  return { items, total };
});

开启 调试模式 后，5 秒内重复访问该端点，只会看到第一次请求真正产生 SQL。缓存适配器可通过 resultCache.adapter 替换为 Redis 等外部实现。

视图实体：把子查询固化为数据库视图

虚拟实体每次查询都会把 expression 内联为子查询。如果希望物化为真实数据库视图，给 schema 加 view: true 即可——expression 随即成为视图定义，数据库创建一次视图，后续查询直接读它：

import { defineEntity, InferEntity, p } from '@mikro-orm/core';

export const ArticleListingViewSchema = defineEntity({
  name: 'ArticleListingView',
  view: true,
  expression: `
    select a.slug, a.title, a.description, a.author_id as author,
           u.full_name as author_name,
           (select count(*) from comment c where c.article_id = a.id) as total_comments,
           (select group_concat(distinct t.name) from article_tags at2
              join tag t on t.id = at2.tag_id
              where at2.article_id = a.id) as tags
    from article a
    join user u on u.id = a.author_id
  `,
  properties: {
    slug: p.string().primary(),
    title: p.string(),
    description: p.string(),
    tags: p.array(),
    author: p.integer(),
    authorName: p.string(),
    totalComments: p.integer(),
  },
});

export type IArticleListingView = InferEntity<typeof ArticleListingViewSchema>;

虚拟实体与视图实体的核心差异：

• 虚拟实体：查询时把 expression 内联为子查询，无主键，不参与变更追踪；
• 视图实体：生成 CREATE VIEW 语句，有主键、可进入 Identity Map，但仍默认只读。

源码层面，view: true（以及 PostgreSQL 专属的 view: { materialized: true }）会在元数据同步时归一化为视图标记（见 typings.ts）。

由于视图实体创建了真实数据库对象，需要生成并应用迁移：

npx mikro-orm migration:create
npx mikro-orm migration:up

如果在测试中直接使用 orm.schema.create() / orm.schema.update()，视图会被自动创建，无需额外步骤。

:::info 视图实体是只读的——ORM 不会为它生成 INSERT / UPDATE / DELETE 语句；expression 是作为视图定义的纯 SQL 字符串。 :::

用 onFlush 事件实现软删除

为评论增加软删除：不物理删除，而是打上 deletedAt 时间戳，并用过滤器让查询默认排除已删除记录。

先在 Comment 实体上添加 deletedAt 属性与 softDelete 过滤器（default: true 表示所有查询自动追加 WHERE deleted_at IS NULL）：

export const CommentSchema = defineEntity({
  name: 'Comment',
  extends: BaseSchema,
  properties: {
    text: p.string(),
    article: () => p.manyToOne(ArticleSchema).ref(),
    author: () => p.manyToOne(UserSchema).ref(),
    deletedAt: p.datetime().nullable(),
  },
  filters: {
    softDelete: { cond: { deletedAt: null }, default: true },
  },
});

接着实现事件订阅器：onFlush 在变更集计算完成之后、真实 SQL 执行之前触发（FlushEventArgs 携带 uow，即 Unit of Work），是"把 DELETE 改写成 UPDATE"的理想时机：

import type { EventSubscriber, FlushEventArgs } from '@mikro-orm/core';
import { ChangeSetType } from '@mikro-orm/core';

export class SoftDeleteSubscriber implements EventSubscriber {

  async onFlush(args: FlushEventArgs): Promise<void> {
    const changeSets = args.uow.getChangeSets();

    for (const cs of changeSets) {
      if (cs.type !== ChangeSetType.DELETE) {
        continue;
      }

      // only soft-delete entities that have a `deletedAt` property
      if (!cs.meta.properties.deletedAt) {
        continue;
      }

      // convert the DELETE to an UPDATE that sets `deletedAt`
      cs.entity.deletedAt = new Date();
      args.uow.computeChangeSet(cs.entity, ChangeSetType.UPDATE);
    }
  }

}

在 ORM 配置中注册订阅器：

import { SoftDeleteSubscriber } from './modules/common/soft-delete.subscriber.js';

export default defineConfig({
  // ...
  subscribers: [new SoftDeleteSubscriber()],
});

此后调用 em.remove(comment) + em.flush()，评论不会被物理删除，而是写入 deletedAt；配合过滤器，常规查询自动排除已软删评论。

需要查询已删除记录（如管理后台、撤销功能）时，显式关闭过滤器：

// include soft-deleted comments
const allComments = await em.find(CommentSchema, {}, {
  filters: { softDelete: false },
});

订阅器对 deletedAt 的检查是通用的：任何实体只要加上 deletedAt 属性与同名过滤器，就能复用这套软删除机制。事件与过滤器的完整语义参见 events.md 与 filters.md。

独立脚本与 CRON 任务

Web 场景中 RequestContext 会为每个请求创建独立的 EntityManager fork。但独立脚本、数据迁移、CRON 任务在 Web 容器之外运行，需要不同的上下文管理策略。

一次性脚本

最稳妥的方式是显式 fork（与第 1 章的做法一致）：

import { initORM } from '../src/db.js';

const db = initORM();
const em = db.em.fork();

// work with the forked EntityManager
const oldArticles = await em.find(ArticleSchema, {
  createdAt: { $lt: new Date('2020-01-01') },
});
em.remove(oldArticles);
await em.flush();

await db.orm.close();

如果确信脚本没有并发访问，也可以在配置层面放开全局上下文限制：

const db = initORM({ allowGlobalContext: true });

// now you can use db.em directly
const users = await db.em.find(User, {});

:::warning 切勿把 allowGlobalContext 当作生产环境缺失 RequestContext 的补丁：它只是压制了校验错误，并未解决共享 Identity Map 带来的内存增长与响应不稳定问题。仅适合无并发的简单脚本与测试；其余场景应使用 RequestContext、@CreateRequestContext() 装饰器或 em.fork()。原理详见 identity-map.md。 :::

CRON 任务

与 Web 服务并行运行的周期任务，用 RequestContext.create() 让每次执行拥有隔离上下文：

import { RequestContext } from '@mikro-orm/core';
import { initORM } from './db.js';

export async function setupCronJobs() {
  const db = initORM();

  // run every hour
  setInterval(async () => {
    await RequestContext.create(db.em, async () => {
      // this runs in its own context, safe from other concurrent operations
      const expiredArticles = await db.article.find({
        createdAt: { $lt: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000) },
      });
      // ... process expired articles
      await db.em.flush();
    });
  }, 60 * 60 * 1000);
}

核心原则只有一条：绝不在并发操作间共享同一个 EntityManager——要么 fork，要么用 RequestContext 隔离每次操作。

生产部署

由于 defineEntity 使用显式实体引用（而非基于文件系统的目录发现），部署有两条路线。

方案一：TypeScript 直接编译

最简单的做法是 tsc 编译后运行产物：

"scripts": {
  "build": "tsc",
  "start": "tsx src/server.ts",
  "start:prod": "node dist/server.js",
  "test": "vitest"
},

npm run build
npm run start:prod

方案二：Vite 打包为单文件

defineEntity 在实体发现阶段不需要运行时文件系统访问，因此应用完全兼容打包器——可以把全部依赖打进单个文件，非常适合容器化与 Serverless 部署。

安装 Vite：

npm install vite --save-dev

配置 SSR 构建（noExternal 让 MikroORM 相关包进入打包产物）：

import { defineConfig } from 'vite';

export default defineConfig({
  build: {
    ssr: 'src/server.ts',
    outDir: 'dist',
    sourcemap: true,
    target: 'node22',
  },
  ssr: {
    // bundle MikroORM packages into the output
    noExternal: ['@mikro-orm/sqlite', '@mikro-orm/sql', '@mikro-orm/core'],
  },
});

加入 bundle 脚本：

"scripts": {
  "build": "tsc",
  "bundle": "vite build",
  "start": "tsx src/server.ts",
  "start:prod": "node dist/server.js",
  "test": "vitest"
},

执行打包并启动：

npm run bundle
npm run start:prod

汇总：bootstrap 全貌（Checkpoint 4）

至此所有端点、认证、错误处理与钩子齐备。完整 bootstrap() 如下，可作为对照清单：

import { NotFoundError, RequestContext } from "@mikro-orm/core";
import { fastify } from "fastify";
import fastifyJWT from "@fastify/jwt";
import { initORM } from "./db.js";
import { registerArticleRoutes } from "./modules/article/routes.js";
import { registerUserRoutes } from "./modules/user/routes.js";
import { AuthError } from "./modules/common/utils.js";

export async function bootstrap(port = 3001, migrate = true) {
  const db = initORM();

  if (migrate) {
    // sync the schema
    await db.orm.migrator.up();
  }

  const app = fastify();

  // register JWT plugin
  app.register(fastifyJWT, {
    secret: process.env.JWT_SECRET ?? "12345678", // fallback for testing
  });

  // register request context hook
  app.addHook("onRequest", (request, reply, done) => {
    RequestContext.create(db.em, done);
  });

  // register auth hook after the ORM one to use the context
  app.addHook("onRequest", async (request) => {
    try {
      const ret = await request.jwtVerify<{ id: number }>();
      request.user = await db.user.findOneOrFail(ret.id);
    } catch (e) {
      app.log.error(e);
      // ignore token errors, we validate the request.user exists only where needed
    }
  });

  // register global error handler to process 404 errors from `findOneOrFail` calls
  app.setErrorHandler((error, request, reply) => {
    if (error instanceof AuthError) {
      return reply.status(401).send({ error: error.message });
    }

    if (error instanceof NotFoundError) {
      return reply.status(404).send({ error: error.message });
    }

    app.log.error(error);
    reply.status(500).send({ error: error.message });
  });

  // shut down the connection when closing the app
  app.addHook("onClose", async () => {
    await db.orm.close();
  });

  // register routes here
  app.register(registerArticleRoutes, { prefix: "article" });
  app.register(registerUserRoutes, { prefix: "user" });

  const url = await app.listen({ port });

  return { app, url, db };
}

本章覆盖的进阶主题——模块化路由、自定义仓储、JWT 认证、Embeddables、Zod 校验、QueryBuilder 聚合、虚拟/视图实体、软删除订阅器、结果缓存、非 Web 场景上下文管理以及双路部署——基本涵盖了 MikroORM 在真实 API 服务中会遇到的绝大多数工程问题，后续可在此基础上继续深入 type-safety.md 的类型安全实践。