feat(server): 报读表索引迁移与存量重复清洗

- AddIndexToStudentEnrollments：student_enrollments 补 (class_id, student_id) 普通索引，加速幂等查找（NON_UNIQUE=1 守卫 + 1061/1091 并发幂等） - AddActiveUniqueToStudentEnrollments：active 部分唯一函数索引（MySQL 8.0.13+ 版本守卫），闭合并发重复报读；up 前 GET_LOCK 串行化 + 清洗存量重复 active （保留最早、其余置 archived，keep_enrollment_id 备份可回滚），有界重试； down 三级恢复守卫（archived + 花名册 active + 无 up 后新建报读），未恢复行保留备份表供人工排查
2026-08-12 16:18:33 +08:00
parent 37b1ba45e1
commit 0ea31b8bc3
2 changed files with 275 additions and 0 deletions
@@ -0,0 +1,46 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+const INDEX_NAME = 'IDX_student_enrollments_class_student';
+
+/**
+ * student_enrollments 补 (class_id, student_id) 普通索引：
+ * 加速 ensureEnrollmentsForRoster 的幂等查找，以及唯一索引迁移里
+ * 重复 active 清洗的 JOIN；普通索引不引入共享锁/间隙锁。
+ * 非唯一索引：同一学生同一班允许 archived + active 等多条历史报读并存。
+ */
+export class AddIndexToStudentEnrollments1786515492285 implements MigrationInterface {
+  private async hasIndex(queryRunner: QueryRunner): Promise<boolean> {
+    const rows = (await queryRunner.query(
+      "SELECT COUNT(*) AS count FROM information_schema.statistics " +
+        "WHERE table_schema = DATABASE() AND table_name = 'student_enrollments' " +
+        "AND index_name = '" +
+        INDEX_NAME +
+        "' AND NON_UNIQUE = 1",
+    )) as Array<{ count: string | number }>;
+    return Number(rows[0]?.count ?? 0) > 0;
+  }
+
+  async up(queryRunner: QueryRunner): Promise<void> {
+    if (await this.hasIndex(queryRunner)) return;
+    try {
+      await queryRunner.query(
+        'CREATE INDEX `' + INDEX_NAME + '` ON `student_enrollments` (`class_id`, `student_id`)',
+      );
+    } catch (error) {
+      // migrationsRun 下多实例并发执行：败者会撞 ER_DUP_KEYNAME(1061)，
+      // 重查确认同名非唯一索引已存在则幂等跳过，否则原样抛出
+      if (!(await this.hasIndex(queryRunner))) throw error;
+    }
+  }
+
+  async down(queryRunner: QueryRunner): Promise<void> {
+    if (!(await this.hasIndex(queryRunner))) return;
+    try {
+      await queryRunner.query('DROP INDEX `' + INDEX_NAME + '` ON `student_enrollments`');
+    } catch (error) {
+      // 并发 revert / 手工已删除：DROP 撞 ER_CANT_DROP_FIELD_OR_KEY(1091)，
+      // 重查确认索引已不存在则幂等跳过，否则原样抛出（与 up 的 1061 处理对称）
+      if (await this.hasIndex(queryRunner)) throw error;
+    }
+  }
+}
@@ -0,0 +1,229 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+import { isDuplicateEntryError } from '../common/db-errors';
+
+const INDEX_NAME = 'IDX_student_enrollments_active_class_student';
+const BACKUP_TABLE = 'student_enrollments_active_cleanup';
+
+/**
+ * student_enrollments 增加 active 部分唯一索引：
+ * 仅对 status='active' 且 class_id 非空的行按 (class_id, student_id) 唯一，
+ * 闭合 ensureEnrollmentsForRoster 并发幂等（重复 active 插入直接 1062，
+ * 而非 FOR SHARE 间隙锁互等死锁）。archived 等多条历史报读、无班级报读不受影响。
+ * 函数索引（MySQL 8.0.13+），非唯一行在函数列上为 NULL，唯一索引允许多个 NULL。
+ *
+ * up 前先清洗存量重复 active（保留最早一条，其余置 archived），并把被清洗行
+ * 及其 keep_id（保留行 id）备份到 BACKUP_TABLE（累积追加，重试/重跑不丢首轮备份），
+ * down 时恢复，保证数据可回滚。清洗与 DDL 之间 MySQL 会隐式提交（DDL 无法事务化）：
+ * 建索引失败时再次清洗并重试（有界 3 次），覆盖「清洗提交后、DDL 前并发插入新重复行」的窗口。
+ */
+export class AddActiveUniqueToStudentEnrollments1786517000000 implements MigrationInterface {
+  /** 目标索引存在且是唯一函数索引（EXPRESSION 非空），才视为已应用。 */
+  private async hasIndex(queryRunner: QueryRunner): Promise<boolean> {
+    const rows = (await queryRunner.query(
+      "SELECT COUNT(*) AS count FROM information_schema.statistics " +
+        "WHERE table_schema = DATABASE() AND table_name = 'student_enrollments' " +
+        "AND index_name = '" +
+        INDEX_NAME +
+        "' AND NON_UNIQUE = 0 AND EXPRESSION IS NOT NULL",
+    )) as Array<{ count: string | number }>;
+    return Number(rows[0]?.count ?? 0) > 0;
+  }
+
+  private async hasBackupTable(queryRunner: QueryRunner): Promise<boolean> {
+    const rows = (await queryRunner.query(
+      "SELECT COUNT(*) AS count FROM information_schema.tables " +
+        "WHERE table_schema = DATABASE() AND table_name = '" +
+        BACKUP_TABLE +
+        "'",
+    )) as Array<{ count: string | number }>;
+    return Number(rows[0]?.count ?? 0) > 0;
+  }
+
+  /** 备份表存在性/结构保障：旧结构（无 keep_enrollment_id 列）就地升级，保留既有备份行。 */
+  private async ensureBackupTable(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      'CREATE TABLE IF NOT EXISTS `' +
+        BACKUP_TABLE +
+        '` (`enrollment_id` int NOT NULL, `old_status` varchar(20) NOT NULL, ' +
+        '`keep_enrollment_id` int NOT NULL, PRIMARY KEY (`enrollment_id`)) ENGINE=InnoDB',
+    );
+    const cols = (await queryRunner.query(
+      "SELECT COUNT(*) AS count FROM information_schema.columns " +
+        "WHERE table_schema = DATABASE() AND table_name = '" +
+        BACKUP_TABLE +
+        "' AND column_name = 'keep_enrollment_id'",
+    )) as Array<{ count: string | number }>;
+    if (Number(cols[0]?.count ?? 0) === 0) {
+      await queryRunner.query(
+        'ALTER TABLE `' + BACKUP_TABLE + '` ADD COLUMN `keep_enrollment_id` int NOT NULL DEFAULT 0',
+      );
+    }
+  }
+
+  /**
+   * 备份重复 active 行（含 keep_id，供 down 精确恢复）并置 archived。
+   * 累积追加（INSERT IGNORE + 保留旧行）：重试/失败重跑时首轮备份记录不丢
+   * （首轮行已置 archived，不再被 SELECT 匹配），保证 down 能完整恢复。
+   */
+  private async cleanupDuplicates(queryRunner: QueryRunner): Promise<void> {
+    await this.ensureBackupTable(queryRunner);
+    await queryRunner.query(
+      'INSERT IGNORE INTO `' +
+        BACKUP_TABLE +
+        '` (enrollment_id, old_status, keep_enrollment_id) ' +
+        'SELECT e.id, e.status, d.keep_id FROM student_enrollments e ' +
+        'JOIN (SELECT student_id, class_id, MIN(id) keep_id FROM student_enrollments ' +
+        "WHERE status = 'active' AND class_id IS NOT NULL " +
+        'GROUP BY student_id, class_id HAVING COUNT(*) > 1) d ' +
+        'ON e.student_id = d.student_id AND e.class_id = d.class_id ' +
+        "AND e.status = 'active' AND e.id <> d.keep_id",
+    );
+    await queryRunner.query(
+      'UPDATE student_enrollments e JOIN `' +
+        BACKUP_TABLE +
+        "` b ON e.id = b.enrollment_id AND e.status = 'active' " +
+        'JOIN (SELECT student_id, class_id, MIN(id) keep_id FROM student_enrollments ' +
+        "WHERE status = 'active' AND class_id IS NOT NULL " +
+        'GROUP BY student_id, class_id HAVING COUNT(*) > 1) d ' +
+        'ON e.student_id = d.student_id AND e.class_id = d.class_id ' +
+        "SET e.status = 'archived'",
+    );
+  }
+
+  private createIndexSql(): string {
+    return (
+      "CREATE UNIQUE INDEX `" +
+      INDEX_NAME +
+      "` ON `student_enrollments` " +
+      "((CASE WHEN status = 'active' AND class_id IS NOT NULL THEN class_id END), " +
+      "(CASE WHEN status = 'active' AND class_id IS NOT NULL THEN student_id END))"
+    );
+  }
+
+  /** 错误分类：1061 索引已存在 / 1062 重复行 / 其他原样抛出 */
+  private classifyCreateError(error: unknown): 'dup-keyname' | 'dup-entry' | null {
+    const driver =
+      (error as { driverError?: { code?: string; errno?: number } })?.driverError ?? error;
+    const code = (driver as { code?: string })?.code;
+    const errno = (driver as { errno?: number })?.errno;
+    if (code === 'ER_DUP_KEYNAME' || errno === 1061) return 'dup-keyname';
+    if (isDuplicateEntryError(error)) return 'dup-entry';
+    return null;
+  }
+
+  /**
+   * 函数索引需要 MySQL 8.0.13+；在清洗 DML 之前检查，避免数据已归档但建索引
+   * 因语法错误失败留下半状态（8.0.0–8.0.12、MySQL 5.7、MariaDB 均不支持函数索引）。
+   */
+  private async assertFunctionalIndexSupported(queryRunner: QueryRunner): Promise<void> {
+    const rows = (await queryRunner.query('SELECT VERSION() AS v')) as Array<{
+      v: string;
+    }>;
+    const version = String(rows[0]?.v ?? '');
+    const match = version.match(/^(\d+)\.(\d+)\.(\d+)/);
+    const major = Number(match?.[1] ?? 0);
+    const minor = Number(match?.[2] ?? 0);
+    const patch = Number(match?.[3] ?? 0);
+    // MariaDB VERSION() 形如 10.x / 5.5.5-10.x，不支持函数索引；MySQL 需 8.0.13+
+    if (/mariadb/i.test(version) || major < 8 || (major === 8 && minor === 0 && patch < 13)) {
+      throw new Error(
+        `student_enrollments active 部分唯一索引需要 MySQL 8.0.13+（函数索引），当前版本 ${version}`,
+      );
+    }
+  }
+
+  async up(queryRunner: QueryRunner): Promise<void> {
+    await this.assertFunctionalIndexSupported(queryRunner);
+    if (await this.hasIndex(queryRunner)) {
+      // 索引已存在（并发 runner / 手动 DDL）：确保备份表存在且结构最新，
+      // 保证 down 可安全执行（唯一索引下不存在重复 active，清洗无副作用）
+      await this.ensureBackupTable(queryRunner);
+      return;
+    }
+    // 命名锁串行化：migrationsRun 多实例并发启动 / 并发迁移时，避免两个 runner
+    // 竞争备份表与 DDL（GET_LOCK 是连接级锁，queryRunner 全程同连接）
+    const lockRows = (await queryRunner.query(
+      "SELECT GET_LOCK('gongxue_add_active_unique_index', 30) AS locked",
+    )) as Array<{ locked: number | string | null }>;
+    if (Number(lockRows[0]?.locked ?? 0) !== 1) {
+      throw new Error('获取迁移命名锁失败（gongxue_add_active_unique_index），请稍后重试');
+    }
+    try {
+      if (await this.hasIndex(queryRunner)) {
+        await this.ensureBackupTable(queryRunner);
+        return;
+      }
+      await this.cleanupDuplicates(queryRunner);
+      // 有界重试：持续写入下清洗与 DDL 之间仍可能插入新重复行，小上限避免无限循环；
+      // 耗尽重试后迁移失败，但备份表 + 已归档行构成安全可重跑状态（幂等）
+      for (let attempt = 1; attempt <= 3; attempt++) {
+        try {
+          await queryRunner.query(this.createIndexSql());
+          return;
+        } catch (error) {
+          const kind = this.classifyCreateError(error);
+          if (kind === 'dup-keyname') {
+            // 并发 runner 已建同名索引：确认后按已应用处理
+            if (!(await this.hasIndex(queryRunner))) throw error;
+            return;
+          }
+          if (kind !== 'dup-entry') throw error;
+          // 清洗提交后、DDL 前并发插入的新重复行导致建索引失败：再次清洗后重试
+          if (attempt === 3) throw error;
+          await this.cleanupDuplicates(queryRunner);
+        }
+      }
+    } finally {
+      // 连接断开时锁由 MySQL 自动释放；RELEASE_LOCK 失败不应掩盖主流程错误
+      try {
+        await queryRunner.query("SELECT RELEASE_LOCK('gongxue_add_active_unique_index')");
+      } catch {
+        // 忽略释放失败：连接异常断开会自动释放命名锁
+      }
+    }
+  }
+
+  async down(queryRunner: QueryRunner): Promise<void> {
+    if (await this.hasIndex(queryRunner)) {
+      try {
+        await queryRunner.query('DROP INDEX `' + INDEX_NAME + '` ON `student_enrollments`');
+      } catch (error) {
+        // 并发 revert / 手工已删除：DROP 撞 ER_CANT_DROP_FIELD_OR_KEY(1091)，
+        // 重查确认索引已不存在则幂等跳过，否则原样抛出
+        if (await this.hasIndex(queryRunner)) throw error;
+      }
+    }
+    if (await this.hasBackupTable(queryRunner)) {
+      // 恢复守卫：只恢复「仍为 archived」且「该生仍是该班 active 花名册成员」、
+      // 且「同班不存在 up 之后新建的 active 报读（keep_id 之外）」的行。
+      // keep_id 是 up 保留行（回滚时允许随 A 一并恢复原状）；A→A2 场景下 A2 的
+      // id ≠ keep_id，B 保持 archived 由下方计数查询保留备份表供人工排查
+      await queryRunner.query(
+        'UPDATE student_enrollments e JOIN `' +
+          BACKUP_TABLE +
+          "` b ON e.id = b.enrollment_id AND e.status = 'archived' " +
+          'JOIN class_student cs ON cs.class_id = e.class_id ' +
+          "AND cs.student_id = e.student_id AND cs.status = 'active' " +
+          'LEFT JOIN student_enrollments e2 ON e2.class_id = e.class_id ' +
+          "AND e2.student_id = e.student_id AND e2.status = 'active' " +
+          'AND e2.id <> e.id AND (b.keep_enrollment_id = 0 OR e2.id <> b.keep_enrollment_id) ' +
+          // 同批恢复的备份行彼此不阻塞（否则 >2 行重复组恢复顺序不定导致部分行永不恢复）
+          'LEFT JOIN `' +
+          BACKUP_TABLE +
+          '` b2 ON e2.id = b2.enrollment_id ' +
+          'SET e.status = b.old_status ' +
+          'WHERE e2.id IS NULL AND b2.enrollment_id IS NULL',
+      );
+      // 统计仍未恢复的备份行（被守卫跳过，如学生已离班/存在 up 后新建报读）：
+      // 有则保留备份表供人工排查，避免未恢复的数据被静默丢弃
+      const rows = (await queryRunner.query(
+        'SELECT COUNT(*) AS count FROM student_enrollments e JOIN `' +
+          BACKUP_TABLE +
+          "` b ON e.id = b.enrollment_id WHERE e.status = 'archived'",
+      )) as Array<{ count: string | number }>;
+      if (Number(rows[0]?.count ?? 0) === 0) {
+        await queryRunner.query('DROP TABLE `' + BACKUP_TABLE + '`');
+      }
+    }
+  }
+}