-- S8-TENANT-ONLY-BATCH5：把 S8 规则执行主链的身份从 (Tenant, Factory) 收敛为 Tenant。
--
-- 本脚本与同批代码是**一个不可分割的单元**。单独跑任何一边都会出事：
--   · 只改代码不迁数据 → 旧 active exception 仍是 T{t}:F{f}:R... 而新 detection 产出 T{t}:R...，
--     调度器会认为旧异常"不再命中"从而开始 recovery 计数，同时按新键再建一条重复异常；
--   · 只迁数据不改代码 → 每次 tick 生成的仍是含 F 段的键，与库里刚改完的键对不上，同样重复建单。
--
-- 变更（DDL 部分在 MySQL 下会自动提交，脚本按"每步可独立重跑"设计，不依赖事务回滚）：
--   ① preflight：去掉 :F{factory} 段后是否产生键碰撞 —— 有则整脚本中止，不自动挑一条
--   ② 归档受影响行的最小恢复字段
--   ③ 重写 dedup_key：仅限严格匹配 ^T{数字}:F{数字}:R 的行
--   ④ 复验：重写后 collision 必须为 0
--   ⑤ ado_s8_rule_detection_state 唯一键 (tenant,factory,rule_code,dedup_key) → (tenant,rule_code,dedup_key)
--   ⑥ ado_s8_watch_rule 新增 UNIQUE(tenant_id, rule_code)
--
-- 明确**不做**：
--   · 不 DROP 任何 factory_id 列（本批做的是"移除隔离语义"，不是 schema 美化）
--   · 不动 ado_s8_exception 的 uk(tenant,factory,exception_code)（异常侧属 Batch 6）
--   · 不动 ado_s8_exception_bak_1_0_487（那是 1.0.487 的还原点，重写它等于毁掉备份）
--   · 不动 mdp_entity.dedup_key_path（数据中台的同名概念，与 S8 规则去重无关）
--
-- 迁移前实测（2026-09-06，本地隔离克隆 aidopdev_s8_local，非沿用历史数字）：
--   旧格式行数    : exception 33 · detection_log 155 · detection_state 28
--   非旧格式行数  : exception 68（不得触碰）· detection_log 0 · detection_state 0
--   去 F 段后碰撞: exception 活动单 0 · detection_state 0
--   watch_rule (tenant_id, rule_code) 重复组: 0

-- ============================================================================
-- ① PREFLIGHT：碰撞门禁
--
-- 去掉 F 段后若出现「同租户 + 同规则 + 同对象」的多个旧键映射到同一个新键，
-- 说明历史上确实靠工厂区分过两条不同的异常。那种情况下自动挑一条留下会静默丢数据，
-- 因此直接让脚本失败、交人工治理。
-- ============================================================================

SET @collision_exception := (
  SELECT COUNT(*) FROM (
    SELECT tenant_id, REGEXP_REPLACE(dedup_key, '^(T[0-9]+):F[0-9]+:', '$1:') nk, COUNT(*) c
    FROM ado_s8_exception
    WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R' AND is_deleted = 0 AND status <> 'CLOSED'
    GROUP BY 1, 2 HAVING c > 1) z);

SET @collision_state := (
  SELECT COUNT(*) FROM (
    SELECT tenant_id, rule_code, REGEXP_REPLACE(dedup_key, '^(T[0-9]+):F[0-9]+:', '$1:') nk, COUNT(*) c
    FROM ado_s8_rule_detection_state
    WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R'
    GROUP BY 1, 2, 3 HAVING c > 1) z);

SET @dup_watch_rule := (
  SELECT COUNT(*) FROM (
    SELECT tenant_id, rule_code, COUNT(*) c FROM ado_s8_watch_rule GROUP BY 1, 2 HAVING c > 1) z);

-- 门禁实现说明：
--   AutoVersionUpdate 明确拒绝含自定义语句分隔符指令的脚本（"当前执行器不支持"），
--   而且它是对全文做子串匹配 —— 连注释里出现那个关键字都会被拦，
--   因此本文件通篇不出现该词。既然用不了存储过程 + SIGNAL，
--   改用**具名 CHECK 约束**：约束名本身携带失败原因，
--   MySQL 的报错文案会带上它（"Check constraint 'xxx' is violated"），
--   而该文案会被 AutoVersionUpdate 写进 sys_db_migration_log.error_message，
--   于是运维在日志里直接看得到"为什么被挡"，不必回头翻脚本。
--   本版本随即被记为 Failed 并阻止应用启动 —— 迁移没过这件事不可能被忽略。

-- 先把碰撞明细落一张诊断表：被挡住时运维需要知道**具体哪几个键**撞了，
-- 只给一个计数会让人无从下手。
DROP TABLE IF EXISTS ado_s8_tenant_only_collision_1_0_493;
CREATE TABLE ado_s8_tenant_only_collision_1_0_493 (
  source_table VARCHAR(64) NOT NULL,
  tenant_id BIGINT NULL,
  rule_code VARCHAR(64) NULL,
  new_dedup_key VARCHAR(256) NULL,
  row_count INT NULL
) COMMENT 'S8-TENANT-ONLY-BATCH5 去 F 段后的 dedup_key 碰撞明细（为空表示无碰撞）';

INSERT INTO ado_s8_tenant_only_collision_1_0_493 (source_table, tenant_id, rule_code, new_dedup_key, row_count)
SELECT 'ado_s8_exception', tenant_id, source_rule_code,
       REGEXP_REPLACE(dedup_key, '^(T[0-9]+):F[0-9]+:', '$1:'), COUNT(*)
FROM ado_s8_exception
WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R' AND is_deleted = 0 AND status <> 'CLOSED'
GROUP BY 1, 2, 3, 4 HAVING COUNT(*) > 1;

INSERT INTO ado_s8_tenant_only_collision_1_0_493 (source_table, tenant_id, rule_code, new_dedup_key, row_count)
SELECT 'ado_s8_rule_detection_state', tenant_id, rule_code,
       REGEXP_REPLACE(dedup_key, '^(T[0-9]+):F[0-9]+:', '$1:'), COUNT(*)
FROM ado_s8_rule_detection_state
WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R'
GROUP BY 1, 2, 3, 4 HAVING COUNT(*) > 1;

INSERT INTO ado_s8_tenant_only_collision_1_0_493 (source_table, tenant_id, rule_code, new_dedup_key, row_count)
SELECT 'ado_s8_watch_rule_duplicate_identity', tenant_id, rule_code, NULL, COUNT(*)
FROM ado_s8_watch_rule GROUP BY 1, 2, 3, 4 HAVING COUNT(*) > 1;

-- 门禁本体：probe 非 0 即违反 CHECK，脚本在**任何写入之前**失败。
DROP TABLE IF EXISTS tmp_s8_493_preflight_guard;
CREATE TABLE tmp_s8_493_preflight_guard (
  probe INT NOT NULL,
  CONSTRAINT s8_tenant_only_BLOCKED_see_collision_table_1_0_493 CHECK (probe = 0)
) COMMENT 'S8-TENANT-ONLY-BATCH5 preflight 门禁；成功后自动删除';

INSERT INTO tmp_s8_493_preflight_guard (probe)
VALUES (@collision_exception + @collision_state + @dup_watch_rule);

DROP TABLE IF EXISTS tmp_s8_493_preflight_guard;

-- ============================================================================
-- ② 归档
--
-- 只存恢复所需的最小字段，不整表复制：受影响的是 dedup_key 这一列，
-- 备份整张 exception 表既慢又没有额外价值。
-- INSERT IGNORE + 主键：脚本重跑时不会把已改写后的值再存一遍覆盖原值。
-- ============================================================================

CREATE TABLE IF NOT EXISTS ado_s8_exception_dedup_bak_1_0_493 (
  id BIGINT NOT NULL PRIMARY KEY,
  tenant_id BIGINT NULL,
  factory_id BIGINT NULL,
  source_rule_code VARCHAR(64) NULL,
  dedup_key VARCHAR(128) NULL
) COMMENT 'S8-TENANT-ONLY-BATCH5 dedup_key 迁移前快照';

CREATE TABLE IF NOT EXISTS ado_s8_detection_log_dedup_bak_1_0_493 (
  id BIGINT NOT NULL PRIMARY KEY,
  tenant_id BIGINT NULL,
  factory_id BIGINT NULL,
  rule_code VARCHAR(64) NULL,
  dedup_key VARCHAR(256) NULL
) COMMENT 'S8-TENANT-ONLY-BATCH5 dedup_key 迁移前快照';

CREATE TABLE IF NOT EXISTS ado_s8_rule_detection_state_dedup_bak_1_0_493 (
  id BIGINT NOT NULL PRIMARY KEY,
  tenant_id BIGINT NULL,
  factory_id BIGINT NULL,
  rule_code VARCHAR(64) NULL,
  dedup_key VARCHAR(128) NULL
) COMMENT 'S8-TENANT-ONLY-BATCH5 dedup_key 迁移前快照';

INSERT IGNORE INTO ado_s8_exception_dedup_bak_1_0_493 (id, tenant_id, factory_id, source_rule_code, dedup_key)
SELECT id, tenant_id, factory_id, source_rule_code, dedup_key
FROM ado_s8_exception WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R';

INSERT IGNORE INTO ado_s8_detection_log_dedup_bak_1_0_493 (id, tenant_id, factory_id, rule_code, dedup_key)
SELECT id, tenant_id, factory_id, rule_code, dedup_key
FROM ado_s8_detection_log WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R';

INSERT IGNORE INTO ado_s8_rule_detection_state_dedup_bak_1_0_493 (id, tenant_id, factory_id, rule_code, dedup_key)
SELECT id, tenant_id, factory_id, rule_code, dedup_key
FROM ado_s8_rule_detection_state WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R';

-- ============================================================================
-- ③ 重写 dedup_key
--
-- 只改严格匹配 ^T{数字}:F{数字}:R 的行，且只去掉 :F{数字} 这一段 ——
-- ⚠️ 反向引用必须写 $1：MySQL 的 REGEXP_REPLACE 用 ICU 语法，
--    写成 \1 不会被当作分组引用，而是被吞成字面量 "1" —— 租户前缀会整段丢失，
--    得到 "1:RRULE_..." 这种既不是旧格式也不是新格式的坏键。
--    本地隔离库首次执行时实测踩到过，已由归档表还原后修正。
-- 不做全表 REPLACE(':F', '')：那会误伤任何恰好含 ":F" 的对象标识（例如 PO 号里带 :F）。
-- 正则锚定行首 + 数字类，把改写面收敛到确实由旧 BuildDedupKey 生成的键。
-- 幂等：改写后的行不再匹配该正则，重跑即 0 行受影响。
-- ============================================================================

UPDATE ado_s8_exception
SET dedup_key = REGEXP_REPLACE(dedup_key, '^(T[0-9]+):F[0-9]+:', '$1:')
WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R';

UPDATE ado_s8_detection_log
SET dedup_key = REGEXP_REPLACE(dedup_key, '^(T[0-9]+):F[0-9]+:', '$1:')
WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R';

UPDATE ado_s8_rule_detection_state
SET dedup_key = REGEXP_REPLACE(dedup_key, '^(T[0-9]+):F[0-9]+:', '$1:')
WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R';

-- ============================================================================
-- ④ 复验：重写后不得残留旧格式、不得出现碰撞
-- ============================================================================

SET @left_over := (
  (SELECT COUNT(*) FROM ado_s8_exception WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R')
  + (SELECT COUNT(*) FROM ado_s8_detection_log WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R')
  + (SELECT COUNT(*) FROM ado_s8_rule_detection_state WHERE dedup_key REGEXP '^T[0-9]+:F[0-9]+:R'));

SET @post_collision := (
  SELECT COUNT(*) FROM (
    SELECT tenant_id, rule_code, dedup_key, COUNT(*) c
    FROM ado_s8_rule_detection_state GROUP BY 1, 2, 3 HAVING c > 1) z);

-- 与 preflight 同一机制：约束名即失败原因。
DROP TABLE IF EXISTS tmp_s8_493_postcheck_guard;
CREATE TABLE tmp_s8_493_postcheck_guard (
  probe INT NOT NULL,
  CONSTRAINT s8_tenant_only_POSTCHECK_leftover_or_collision_1_0_493 CHECK (probe = 0)
) COMMENT 'S8-TENANT-ONLY-BATCH5 改写后复验门禁；成功后自动删除';

INSERT INTO tmp_s8_493_postcheck_guard (probe) VALUES (@left_over + @post_collision);

DROP TABLE IF EXISTS tmp_s8_493_postcheck_guard;

-- ============================================================================
-- ⑤ ado_s8_rule_detection_state 唯一键去掉 factory_id
--
-- 必须与代码同批：upsert 的 WHERE 已改为 (tenant, rule_code, dedup_key)，
-- 而旧唯一键仍允许同键在不同 factory 下存两行 —— 抗抖计数会分裂，
-- trigger/recover 阈值静默失效。
-- 用 information_schema 判存在性：MySQL 无 DROP INDEX IF EXISTS。
-- ============================================================================

SET @has_old_uk := (SELECT COUNT(*) FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'ado_s8_rule_detection_state'
    AND INDEX_NAME = 'uk_s8_rule_detection_state_dedup');
SET @sql := IF(@has_old_uk > 0,
  'ALTER TABLE ado_s8_rule_detection_state DROP INDEX uk_s8_rule_detection_state_dedup', 'DO 0');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;

SET @has_new_uk := (SELECT COUNT(*) FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'ado_s8_rule_detection_state'
    AND INDEX_NAME = 'uk_s8_rule_detection_state_dedup');
SET @sql := IF(@has_new_uk = 0,
  'ALTER TABLE ado_s8_rule_detection_state ADD UNIQUE INDEX uk_s8_rule_detection_state_dedup (tenant_id, rule_code, dedup_key)', 'DO 0');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;

-- 辅助索引同步去掉 factory_id（非唯一，纯查询效率）。
SET @has_idx := (SELECT COUNT(*) FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'ado_s8_rule_detection_state'
    AND INDEX_NAME = 'idx_s8_rule_detection_state_rule' AND COLUMN_NAME = 'factory_id');
SET @sql := IF(@has_idx > 0,
  'ALTER TABLE ado_s8_rule_detection_state DROP INDEX idx_s8_rule_detection_state_rule', 'DO 0');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;

SET @has_idx2 := (SELECT COUNT(*) FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'ado_s8_rule_detection_state'
    AND INDEX_NAME = 'idx_s8_rule_detection_state_rule');
SET @sql := IF(@has_idx2 = 0,
  'ALTER TABLE ado_s8_rule_detection_state ADD INDEX idx_s8_rule_detection_state_rule (tenant_id, rule_code)', 'DO 0');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;

-- ============================================================================
-- ⑥ ado_s8_watch_rule 新增 UNIQUE(tenant_id, rule_code)
--
-- Batch 2 把运行策略身份定为 (TenantId, RuleCode)，但当时只有应用层 AnyAsync 兜底，
-- 因此只保证"顺序幂等"，两个实例同时供给理论上仍可能建出重复行。
-- 本索引把那个缺口在 DB 层封死。preflight 已确认无重复行。
-- ============================================================================

SET @has_rule_uk := (SELECT COUNT(*) FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'ado_s8_watch_rule'
    AND INDEX_NAME = 'uk_s8_watch_rule_tenant_code');
SET @sql := IF(@has_rule_uk = 0,
  'ALTER TABLE ado_s8_watch_rule ADD UNIQUE INDEX uk_s8_watch_rule_tenant_code (tenant_id, rule_code)', 'DO 0');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;