08C Actor 与 Film：关联 CRUD、事务与并发实战

这一章把关系表做成真实 API。重点不是多写几个 Controller，而是明确：

操作的是 Actor、Film，还是 FilmActor 关系；
单条 SQL 与多步骤事务的边界；
ORM 异常怎样映射成稳定业务错误；
并发请求为什么不能只靠“先查再写”；
Sequelize API 大致会生成什么 SQL。

1. 本章接口

GET    /api/sakila/actors/:actor_id/films
PUT    /api/sakila/actors/:actor_id/films/:film_id
DELETE /api/sakila/actors/:actor_id/films/:film_id
PUT    /api/sakila/actors/:actor_id/films

语义：

接口	作用
GET collection	分页查询演员的电影
PUT member	确保一条关系存在
DELETE member	解除一条关系
PUT collection	用新列表整体替换全部关系

为什么单条建立关系使用 PUT？因为资源身份已经由 (actor_id, film_id) 唯一确定，重复请求可以保持相同最终状态。如果项目更希望重复创建返回 ACTOR_FILM_EXISTS，也可以使用 POST；必须先决定契约，不要让数据库异常随机决定 API 行为。

2. DTO 与 Service 类型

export interface ActorFilmDto {
  film_id: number;
  title: string;
  release_year: number | null;
  relation_last_update: string;
}

export interface ListActorFilmsOptions {
  actor_id: number;
  page: number;
  page_size: number;
  title?: string;
}

export interface ReplaceActorFilmsInput {
  film_ids: number[];
}

不要直接返回 FilmActor 或 Film Model。DTO 应明确：

哪些字段来自 Film；
哪些字段来自关系表；
Date 在 JSON 中应表示为 ISO string；
DECIMAL 若加入响应，通常应保留 string 精度。

3. Router 与验证顺序

router.use(authenticateAdmin);

router.get(
  "/actors/:actor_id/films",
  actorFilmListValidation,
  validateRequest,
  listActorFilmsController,
);

router.put(
  "/actors/:actor_id/films/:film_id",
  actorFilmIdValidation,
  validateRequest,
  attachActorFilmController,
);

router.delete(
  "/actors/:actor_id/films/:film_id",
  actorFilmIdValidation,
  validateRequest,
  detachActorFilmController,
);

router.put(
  "/actors/:actor_id/films",
  replaceActorFilmsValidation,
  validateRequest,
  replaceActorFilmsController,
);

固定 collection 路径与 member 路径要保持清晰。Express 按注册顺序匹配，复杂固定路径通常应放在宽泛参数路径之前。

校验示例：

import { body, param, query } from "express-validator";

function actorIdRule() {
  return param("actor_id")
    .isInt({ min: 1, max: 65_535 })
    .withMessage("actor_id 必须是有效正整数")
    .toInt();
}

function filmIdRule() {
  return param("film_id")
    .isInt({ min: 1, max: 65_535 })
    .withMessage("film_id 必须是有效正整数")
    .toInt();
}

export const actorFilmIdValidation = [actorIdRule(), filmIdRule()];

export const actorFilmListValidation = [
  actorIdRule(),
  query("page").optional().isInt({ min: 1 }).toInt(),
  query("page_size").optional().isInt({ min: 1, max: 100 }).toInt(),
  query("title").optional().isString().trim().isLength({ max: 128 }),
];

export const replaceActorFilmsValidation = [
  actorIdRule(),
  body("film_ids")
    .isArray({ min: 0, max: 500 })
    .withMessage("film_ids 必须是最多包含 500 项的数组"),
  body("film_ids.*")
    .isInt({ min: 1, max: 65_535 })
    .withMessage("每个 film_id 都必须是有效正整数")
    .toInt(),
];

3.1 不要复用后再继续修改同一条 ValidationChain

ValidationChain 是可变对象。如果把同一个 actorIdRule 对象放入多个数组后，又继续对它调用 .custom()，其他路由也可能受到影响。上面的示例使用规则工厂：每次调用 actorIdRule() 都返回一个独立 chain。简化形式如下：

function actorIdRule() {
  return param("actor_id").isInt({ min: 1 }).toInt();
}

在每个验证数组中调用一次 actorIdRule()，就不会在路由之间共享可变 chain。

4. Controller：只做 HTTP 适配

interface ActorFilmParams {
  actor_id: number;
  film_id: number;
}

export async function attachActorFilmController(
  req: Request<{ actor_id: string; film_id: string }>,
  res: Response<ApiResponse<null>>,
): Promise<void> {
  const input = matchedData<ActorFilmParams>(req, {
    locations: ["params"],
  });

  await attachFilmToActor(input.actor_id, input.film_id);

  res.json({ code: "OK", message: "success", data: null });
}

不要把 req、res 传给 Service。也不要因为 Request 泛型写了 number 就直接相信 URL 输入；运行时转换来自 validator，可信数据来自 matchedData()。

5. 查询演员的电影

对于“某演员的电影列表”，可以把 FilmActor 作为查询主体。actor_id 固定后，每条关系只对应一部 Film，分页和 count 比从 Actor 展开全部 Films 更容易理解。

先为 FilmActor 增加关联属性：

// FilmActor class 内
declare film?: NonAttribute<Film>;

并注册：

FilmActor.belongsTo(Film, {
  as: "film",
  foreignKey: "film_id",
});

Service 骨架：

export async function listActorFilms(
  options: ListActorFilmsOptions,
): Promise<PageResult<ActorFilmDto>> {
  const actor = await Actor.findByPk(options.actor_id, {
    attributes: ["actor_id"],
  });

  if (actor === null) {
    throw new AppError("ACTOR_NOT_FOUND", "演员不存在");
  }

  const offset = (options.page - 1) * options.page_size;

  const { count, rows } = await FilmActor.findAndCountAll({
    where: { actor_id: options.actor_id },
    attributes: ["actor_id", "film_id", "last_update"],
    include: [
      {
        model: Film,
        as: "film",
        required: true,
        attributes: ["film_id", "title", "release_year"],
        ...(options.title
          ? {
              where: {
                title: { [Op.like]: `%${escapeLike(options.title)}%` },
              },
            }
          : {}),
      },
    ],
    order: [
      [{ model: Film, as: "film" }, "title", "ASC"],
      ["film_id", "ASC"],
    ],
    limit: options.page_size,
    offset,
  });

  return {
    page: options.page,
    page_size: options.page_size,
    total: count,
    total_pages: Math.ceil(count / options.page_size),
    items: rows.map((relation) => {
      if (relation.film === undefined) {
        throw new Error("FilmActor 查询缺少 film include");
      }

      return {
        film_id: relation.film.film_id,
        title: relation.film.title,
        release_year: relation.film.release_year,
        relation_last_update: relation.last_update.toISOString(),
      };
    }),
  };
}

近似 SQL：

SELECT fa.actor_id, fa.film_id, fa.last_update,
       f.film_id, f.title, f.release_year
FROM film_actor AS fa
JOIN film AS f ON f.film_id = fa.film_id
WHERE fa.actor_id = ?
ORDER BY f.title ASC, fa.film_id ASC
LIMIT ? OFFSET ?;

由于查询基表已经限定一个 actor，且一条 FilmActor 只 include 一个 Film，这里不会发生一条关系被多个子表行放大的问题。如果继续 include Film 的 Category 等多值关联，则应重新检查 count 和分页。

6. 建立关系：显式 FilmActor.create

import {
  ForeignKeyConstraintError,
  UniqueConstraintError,
} from "sequelize";

export async function attachFilmToActor(
  actor_id: number,
  film_id: number,
): Promise<void> {
  await db.transaction(async (transaction) => {
    // 所有关联写操作统一先锁 Actor，才能让批量替换与单条增删互相串行化。
    const actor = await Actor.findByPk(actor_id, {
      attributes: ["actor_id"],
      transaction,
      lock: transaction.LOCK.UPDATE,
    });

    if (actor === null) {
      throw new AppError("ACTOR_NOT_FOUND", "演员不存在");
    }

    const film = await Film.findByPk(film_id, {
      attributes: ["film_id"],
      transaction,
    });

    if (film === null) {
      throw new AppError("FILM_NOT_FOUND", "电影不存在");
    }

    try {
      await FilmActor.create(
        { actor_id, film_id },
        { transaction },
      );
    } catch (error: unknown) {
      if (error instanceof UniqueConstraintError) {
        // PUT 的幂等语义：关系已经存在也视为目标状态已达成。
        return;
      }

      if (error instanceof ForeignKeyConstraintError) {
        // 处理“检查存在后、INSERT 前，目标又被其他事务删除”的竞态。
        throw new AppError(
          "ACTOR_OR_FILM_NOT_FOUND",
          "演员或电影不存在",
        );
      }

      throw error;
    }
  });
}

底层核心 SQL：

SELECT actor_id FROM actor WHERE actor_id = ? FOR UPDATE;
SELECT film_id FROM film WHERE film_id = ?;
INSERT INTO film_actor (actor_id, film_id) VALUES (?, ?);

如果不需要与“整体替换”操作互相串行，单条 FilmActor.create() 本身是原子的，可以不额外开启事务。但一旦项目采用统一的关系集合锁协议，所有关系写接口都必须遵守，不能只有批量接口加锁。

7. 删除关系

严格语义版本：

export async function detachFilmFromActor(
  actor_id: number,
  film_id: number,
): Promise<void> {
  await db.transaction(async (transaction) => {
    const actor = await Actor.findByPk(actor_id, {
      attributes: ["actor_id"],
      transaction,
      lock: transaction.LOCK.UPDATE,
    });

    if (actor === null) {
      throw new AppError("ACTOR_NOT_FOUND", "演员不存在");
    }

    const deleted_count = await FilmActor.destroy({
      where: { actor_id, film_id },
      transaction,
    });

    if (deleted_count === 0) {
      throw new AppError(
        "ACTOR_FILM_NOT_FOUND",
        "演员与电影的关联不存在",
      );
    }
  });
}

幂等 DELETE 版本可以把 deleted_count === 0 也视为成功。两种设计都不会删除 Actor 或 Film。

近似 SQL：

DELETE FROM film_actor
WHERE actor_id = ? AND film_id = ?;

8. 整体替换关系集合

请求：

PUT /api/sakila/actors/1/films
Content-Type: application/json

{
  "film_ids": [10, 20, 30, 30]
}

Service 应先去重，再执行完整替换：

export async function replaceActorFilms(
  actor_id: number,
  input: ReplaceActorFilmsInput,
): Promise<void> {
  const film_ids = [...new Set(input.film_ids)];

  await db.transaction(async (transaction) => {
    const actor = await Actor.findByPk(actor_id, {
      attributes: ["actor_id"],
      transaction,
      lock: transaction.LOCK.UPDATE,
    });

    if (actor === null) {
      throw new AppError("ACTOR_NOT_FOUND", "演员不存在");
    }

    const films = await Film.findAll({
      where: { film_id: film_ids },
      attributes: ["film_id"],
      transaction,
    });

    if (films.length !== film_ids.length) {
      const existing_ids = new Set(films.map((film) => film.film_id));
      const missing_ids = film_ids.filter((id) => !existing_ids.has(id));

      throw new AppError(
        "FILM_NOT_FOUND",
        `电影不存在：${missing_ids.join(", ")}`,
      );
    }

    await FilmActor.destroy({
      where: { actor_id },
      transaction,
    });

    if (film_ids.length > 0) {
      await FilmActor.bulkCreate(
        film_ids.map((film_id) => ({ actor_id, film_id })),
        { transaction },
      );
    }
  });
}

空数组必须明确处理。不要想当然地把空数组传给所有 IN、NOT IN 或 bulk API，不同版本和方言可能产生不同 SQL。

这个实现采用“全部删除，再全部插入”，好处是语义直观。更高性能的实现可以计算集合差异，只删多余、只加缺失，但复杂度更高，第一版没有必要过早优化。

8.1 为什么不能漏传 transaction

错误示例：

await db.transaction(async (transaction) => {
  await FilmActor.destroy({
    where: { actor_id },
    transaction,
  });

  await FilmActor.bulkCreate(rows); // 错：这条语句在事务外
});

如果 bulkCreate 失败，事务只能恢复 DELETE，无法撤销已经在其他步骤中独立提交的写入。事务中的每一条查询都要显式传递同一个 transaction。

9. `setFilms()` 的等价实现

完成显式版本后，可以尝试：

await actor.setFilms(film_ids, { transaction });

但必须知道它可能执行多条 SQL：

SELECT 当前 FilmActor
DELETE 新列表里不存在的关系
INSERT 缺失关系

它仍然需要事务，也仍然需要验证目标 Film 是否存在、处理外键异常和并发。Mixin 只是减少样板代码，不会自动替你完成业务契约。

10. 中间表有业务字段时怎样更新

假设以后新增：

ALTER TABLE film_actor
  ADD COLUMN character_name VARCHAR(100) NULL,
  ADD COLUMN billing_order SMALLINT UNSIGNED NULL;

此时关系本身有了可更新内容：

const [affected_count] = await FilmActor.update(
  {
    character_name: input.character_name,
    billing_order: input.billing_order,
  },
  {
    where: { actor_id, film_id },
    transaction,
  },
);

if (affected_count === 0) {
  throw new AppError("ACTOR_FILM_NOT_FOUND", "关联不存在");
}

近似 SQL：

UPDATE film_actor
SET character_name = ?, billing_order = ?
WHERE actor_id = ? AND film_id = ?;

这和 Film.update({ title }) 完全不同：前者更新关系属性，后者更新电影实体。

11. 错误映射建议

ACTOR_NOT_FOUND
FILM_NOT_FOUND
ACTOR_FILM_EXISTS
ACTOR_FILM_NOT_FOUND
ACTOR_OR_FILM_NOT_FOUND
ACTOR_IN_USE
FILM_IN_USE

数据库异常到业务语义：

Sequelize 异常	常见原因	建议处理
`UniqueConstraintError`	复合主键重复	幂等成功或 ACTOR_FILM_EXISTS
`ForeignKeyConstraintError` on INSERT	父记录不存在	重新确认后映射 not found
`ForeignKeyConstraintError` on parent DELETE	仍有关联	ACTOR_IN_USE / FILM_IN_USE
其他 DatabaseError	SQL、连接或结构问题	记录 cause，返回安全的内部错误

不要把 error.message、SQL、路径或数据库名称直接返回客户端。未知错误保留 cause 交给统一错误日志。

12. 并发和锁需要统一协议

只在 replaceActorFilms() 锁 Actor，并不能阻止另一个不加锁的 attachFilmToActor() 同时插入关系。锁只有在所有相关写入口遵循同一协议时才有效：

所有关系写操作
  → 开事务
  → 先锁 actor 行
  → 再读取/修改 film_actor
  → 提交

如果一次锁多个实体，保持固定顺序，例如始终先 Actor 后 Film，减少死锁概率。死锁仍然可能发生；只有整个事务是幂等的情况下，才适合做有限次数重试。

不要用行锁替代：

复合主键；
外键；
唯一约束；
Service 权限和状态检查。

13. 测试矩阵

13.1 建立关系

[ ] Actor、Film 均存在，新增一行。
[ ] 重复 PUT 符合幂等约定，数据库仍只有一行。
[ ] Actor 不存在，返回 ACTOR_NOT_FOUND。
[ ] Film 不存在，返回 FILM_NOT_FOUND。
[ ] 两个并发请求创建相同关系，数据库最终只有一行。

13.2 删除关系

[ ] 删除后 Actor、Film 仍存在。
[ ] 第二次删除符合约定的严格或幂等语义。
[ ] 只删除目标 pair，没有误删演员的全部电影。

13.3 查询

[ ] 没有电影的 Actor 返回空 items，而不是 ACTOR_NOT_FOUND。
[ ] Actor 不存在与空集合能够区分。
[ ] 不返回完整 FilmActor Model。
[ ] 同名电影分页稳定。
[ ] title 中含 %、_ 时按普通字符搜索。

13.4 整体替换

[ ] 输入重复 ID 会先去重。
[ ] 空数组会清空关系。
[ ] 任意一个 Film 不存在时，旧关系完整保留。
[ ] 故意让 bulkCreate 失败，确认 DELETE 被回滚。
[ ] 两个并发替换请求不会混合出第三种集合。

14. SQL 日志验收

为每个 Service 记录：

请求参数
实际 SQL 条数
是否使用事务
是否发生 SELECT N+1
使用了哪个索引
错误最终来自业务检查还是数据库约束

重点观察：

EXPLAIN SELECT ... FROM film_actor WHERE actor_id = ?;
EXPLAIN SELECT ... FROM film_actor WHERE film_id = ?;

15. 与第 09 章 Customer 的对应关系

Actor 与 Film：

Actor.belongsToMany(Film, through FilmActor)
Film.belongsToMany(Actor, through FilmActor)

Customer 与 Address：

Customer.belongsTo(Address, foreignKey address_id)
Address.hasMany(Customer, foreignKey address_id)

创建已有地址的客户只需给 customer.address_id 赋值，数据库外键负责兜底。只有“同时新建 Address 和 Customer”包含两次写入，才需要事务保证失败时不留下孤立 Address。

不要因为学会了 belongsToMany，就给所有有关联的表都增加中间表。先从业务基数和外键所在位置判断关系类型。

16. 复盘问题

为什么 addFilm() 不等于 createFilm()？
为什么 removeFilm() 不会删除 Film？
为什么 setFilms() 比 addFilm() 更危险？
为什么先查询关系不存在仍要捕获 UniqueConstraintError？
为什么批量替换必须把 DELETE 和 INSERT 放在同一事务？
为什么只给批量接口加行锁还不够？
为什么关系表没有业务字段时通常不设计 PATCH？
为什么 Customer 与 Address 不需要中间表？

08C Actor 与 Film：关联 CRUD、事务与并发实战

1. 本章接口

2. DTO 与 Service 类型

3. Router 与验证顺序

3.1 不要复用后再继续修改同一条 ValidationChain

4. Controller：只做 HTTP 适配

5. 查询演员的电影

6. 建立关系：显式 FilmActor.create

7. 删除关系

8. 整体替换关系集合

8.1 为什么不能漏传 transaction

9. setFilms() 的等价实现

10. 中间表有业务字段时怎样更新

11. 错误映射建议

12. 并发和锁需要统一协议

13. 测试矩阵

13.1 建立关系

13.2 删除关系

13.3 查询

13.4 整体替换

14. SQL 日志验收

15. 与第 09 章 Customer 的对应关系

16. 复盘问题

9. `setFilms()` 的等价实现