DEV-MERGE-LEGACY-BACKEND-2026/internal/academic/dualrun/fixture/run_marker.go

package fixture

import (
	"context"
	"database/sql"
	"fmt"
	"time"
)

// 本文件是**迁移运行标记**：把「这次映射是什么时刻跑的」落到目标库里，供另一个进程的
// verify 精确取用。
//
// 为什么必须落库：
//   create_time 这类**业务时间**列在补了 DEFAULT CURRENT_TIMESTAMP(3) 之后，
//   如果映射忘了传值，DB 会用「插入时刻」静默填上——不报错、不空、格式正确，
//   只有「与迁移执行时刻比对」这一条断言能抓住它。
//   而 map 与 verify 是两个独立进程，时间戳无法通过内存传递，所以写进 t_audit_log。
//
// 取值精度：毫秒（DATETIME(3)），与列精度一致，避免"差 1 毫秒"的伪差异。

// RunMarkerAction 是标记行的 action 值（t_audit_log.action）。
const RunMarkerAction = "ac15_fixture_run"

// WriteRunMarker 写入一条运行标记，返回该标记的时间戳（= 本次迁移的起点）。
//
// 用 t_audit_log 而不是新建表：它是 core DDL 里既有的审计表（14 列、0 行、已对齐默认值），
// 语义上也正是"谁在什么时候做了什么"。清理时按 action 删除即可。
func WriteRunMarker(ctx context.Context, db *sql.DB, phase string) (time.Time, error) {
	now := time.Now().Truncate(time.Millisecond)
	const q = `
INSERT INTO t_audit_log (actor_user_id, action, entity_type, entity_id, detail, trace_id, created_at)
VALUES (NULL, ?, 'fixture', ?, ?, 'ac15-fixture', ?)`
	detail := fmt.Sprintf(`{"phase":%q,"tool":"ac15fixture","note":"migration run marker"}`, phase)
	if _, err := db.ExecContext(ctx, q, RunMarkerAction, phase, detail,
		now.Format("2006-01-02 15:04:05.000")); err != nil {
		return time.Time{}, fmt.Errorf("写运行标记失败：%w", err)
	}
	return now, nil
}

// ReadRunMarker 读最近一条运行标记的时间戳。
func ReadRunMarker(ctx context.Context, db *sql.DB) (time.Time, bool, error) {
	var s sql.NullString
	err := db.QueryRowContext(ctx,
		`SELECT DATE_FORMAT(created_at,'%Y-%m-%d %H:%i:%s.%f') FROM t_audit_log
		  WHERE action = ? ORDER BY id DESC LIMIT 1`, RunMarkerAction).Scan(&s)
	if err == sql.ErrNoRows || (err == nil && !s.Valid) {
		return time.Time{}, false, nil
	}
	if err != nil {
		return time.Time{}, false, err
	}
	t, perr := time.ParseInLocation("2006-01-02 15:04:05.000000", s.String, time.Local)
	if perr != nil {
		return time.Time{}, false, fmt.Errorf("解析运行标记时间失败（%q）：%w", s.String, perr)
	}
	return t, true, nil
}

// ClearRunMarkers 删除全部运行标记（清理阶段调用）。
func ClearRunMarkers(ctx context.Context, db *sql.DB) (int64, error) {
	res, err := db.ExecContext(ctx, `DELETE FROM t_audit_log WHERE action = ?`, RunMarkerAction)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// countRowsAfter 统计某表 create_time >= marker 的行数。
//
// 这就是「漏传业务时间」的探针：DB 默认值只会给出「插入时刻」，必然 >= marker；
// 而 legacy 的历史业务时间一定 < marker（本次迁移发生在历史数据之后）。
func countRowsAfter(ctx context.Context, db *sql.DB, table, column string, marker time.Time) (int, error) {
	var n int
	q := fmt.Sprintf(`SELECT COUNT(*) FROM %s WHERE %s >= ?`, table, column)
	if err := db.QueryRowContext(ctx, q, marker.Format("2006-01-02 15:04:05.000")).Scan(&n); err != nil {
		return 0, fmt.Errorf("统计 %s.%s 失败：%w", table, column, err)
	}
	return n, nil
}

// createTimeAssertions 对一批「迁移过来的写类表」生成 create_time 断言。
//
// tables 里的列名统一叫 create_time（本项目的业务时间列名）；
// marker 取不到时不判 pass（视为失败并明确说明），避免"没标记就当通过"。
func createTimeAssertions(ctx context.Context, db *sql.DB, tables []string) []Assertion {
	marker, ok, err := ReadRunMarker(ctx, db)
	if err != nil {
		return []Assertion{{
			Name: "create_time 未落入 DB 默认值（业务时间必须来自 legacy）",
			Pass: false, Detail: fmt.Sprintf("读运行标记失败：%v", err),
		}}
	}
	if !ok {
		return []Assertion{{
			Name:   "create_time 未落入 DB 默认值（业务时间必须来自 legacy）",
			Pass:   false,
			Detail: "未找到运行标记（t_audit_log.action='ac15_fixture_run'）—— 无法判定，按失败处理（先跑 map 再跑 verify）",
		}}
	}
	out := make([]Assertion, 0, len(tables))
	for _, t := range tables {
		n, err := countRowsAfter(ctx, db, t, "create_time", marker)
		name := t + ".create_time 未落入 DB 默认值（>= 迁移执行时刻的行数 = 0）"
		if err != nil {
			out = append(out, Assertion{Name: name, Pass: false, Detail: err.Error()})
			continue
		}
		out = append(out, Assertion{
			Name: name,
			Pass: n == 0,
			Detail: fmt.Sprintf("迁移执行时刻=%s；create_time >= 该时刻的行数 = %d（期望 0）"+
				"—— 补了 DEFAULT CURRENT_TIMESTAMP(3) 之后，漏传业务时间会静默落成插入时刻，只有这条能抓",
				marker.Format("2006-01-02 15:04:05"), n),
		})
	}
	return out
}