Cyrene_v2/internal/db/migrate_flow_graphs_test.go

package db

import (
	"context"
	"testing"
)

// flowGraphsMigrationID 是链路图落盘表的迁移 id（design/19 §4 P3；切片 F2）。
const flowGraphsMigrationID = "0022_flow_graphs"

// TestMigrateFlowGraphsReversible 钉住 0022（flow_graphs，切片 F2）：up 建表与索引、
// 重复 up 幂等、down 整表消失、再 up 恢复端到端可用。它用自己的内存库，绝不碰
// data/cyrene.db。
//
// 断言的是**库里的结构事实**（表/列/索引在不在、默认值读出来是什么），不按日志文本
// 断言 —— 与 migrate_channel_events_test.go 同一口径。
func TestMigrateFlowGraphsReversible(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	conn := openMemory(t)

	if err := MigrateUp(ctx, conn, testLogger()); err != nil {
		t.Fatalf("MigrateUp: %v", err)
	}

	if !schemaObjectExists(t, ctx, conn, "table", "flow_graphs") {
		t.Fatal("flow_graphs should exist after MigrateUp")
	}
	// 列名错一个，写入面就会静默写错地方。
	for _, col := range []string{"id", "session_id", "turn_id", "graph_json", "created_at", "updated_at"} {
		if !columnExists(t, ctx, conn, "flow_graphs", col) {
			t.Fatalf("flow_graphs.%s should exist after MigrateUp", col)
		}
	}
	// 查询/GC 与 upsert 依赖的两个索引。
	for _, idx := range []string{"idx_flow_graphs_session_id_turn_id", "idx_flow_graphs_created_at"} {
		if !schemaObjectExists(t, ctx, conn, "index", idx) {
			t.Fatalf("%s should exist after MigrateUp", idx)
		}
	}

	// 默认值：只写必填列也能落行，时间戳由 SQLite 默认填。
	if _, err := conn.ExecContext(ctx,
		`INSERT INTO flow_graphs (session_id, turn_id, graph_json) VALUES (?, ?, ?)`,
		"web", "web-1", `{"id":"web-1"}`); err != nil {
		t.Fatalf("insert minimal flow graph: %v", err)
	}
	var createdAt string
	if err := conn.QueryRowContext(ctx,
		`SELECT created_at FROM flow_graphs WHERE session_id = ?`, "web").Scan(&createdAt); err != nil {
		t.Fatalf("read minimal flow graph: %v", err)
	}
	if createdAt == "" {
		t.Fatal("created_at should be filled by the migration default")
	}

	// 重复 up：已记录的迁移被跳过，结构仍在且不报错（幂等）。
	if err := MigrateUp(ctx, conn, testLogger()); err != nil {
		t.Fatalf("MigrateUp second call: %v", err)
	}
	if !schemaObjectExists(t, ctx, conn, "table", "flow_graphs") {
		t.Fatal("flow_graphs should still exist after idempotent re-run")
	}

	// 恰好回滚到 0022（含它之后新加的迁移），日后再加迁移也不会让本测试失效。
	ids := upMigrationIDs(t)
	idx := -1
	for i, id := range ids {
		if id == flowGraphsMigrationID {
			idx = i
		}
	}
	if idx < 0 {
		t.Fatalf("%s not found in %v", flowGraphsMigrationID, ids)
	}
	if err := MigrateDown(ctx, conn, len(ids)-idx, testLogger()); err != nil {
		t.Fatalf("MigrateDown(%d): %v", len(ids)-idx, err)
	}
	if schemaObjectExists(t, ctx, conn, "table", "flow_graphs") {
		t.Fatal("flow_graphs should be dropped by the 0022 down")
	}

	// 重新 up 恢复端到端可用。
	if err := MigrateUp(ctx, conn, testLogger()); err != nil {
		t.Fatalf("re-MigrateUp: %v", err)
	}
	if !schemaObjectExists(t, ctx, conn, "table", "flow_graphs") {
		t.Fatal("flow_graphs should exist again after re-MigrateUp")
	}
}

// TestMigrate0022UpsertSameTurn_正常 钉住唯一键的**行为**（不只是「索引存在」）：
// 同一 `(session_id, turn_id)` 二次写入必须更新**一行**，而不是插出两行「同一轮」
// —— 否则多轮列表会出现重复，且「第 N 轮」到底存的是哪份图就说不清了。
func TestMigrate0022UpsertSameTurn_正常(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	conn := openMemory(t)

	if err := MigrateUp(ctx, conn, testLogger()); err != nil {
		t.Fatalf("MigrateUp: %v", err)
	}

	const upsert = `INSERT INTO flow_graphs (session_id, turn_id, graph_json, created_at, updated_at)
	VALUES (?, ?, ?, ?, ?)
	ON CONFLICT(session_id, turn_id) DO UPDATE SET
	    graph_json = excluded.graph_json,
	    updated_at = excluded.updated_at`
	write := func(payload string) {
		t.Helper()
		if _, err := conn.ExecContext(ctx, upsert, "web", "web-1", payload,
			"2026-10-05T00:00:00.000Z", "2026-10-05T00:00:00.000Z"); err != nil {
			t.Fatalf("upsert flow graph: %v", err)
		}
	}
	write(`{"id":"web-1","nodes":[]}`)
	write(`{"id":"web-1","nodes":[{"id":"input"}]}`)

	var rows int
	if err := conn.QueryRowContext(ctx,
		`SELECT count(*) FROM flow_graphs WHERE session_id = ? AND turn_id = ?`, "web", "web-1").Scan(&rows); err != nil {
		t.Fatalf("count flow graphs: %v", err)
	}
	if rows != 1 {
		t.Fatalf("flow_graphs rows = %d, want 1 (unique key must merge, not duplicate)", rows)
	}
	var payload string
	if err := conn.QueryRowContext(ctx,
		`SELECT graph_json FROM flow_graphs WHERE session_id = ? AND turn_id = ?`, "web", "web-1").Scan(&payload); err != nil {
		t.Fatalf("read graph_json: %v", err)
	}
	if payload != `{"id":"web-1","nodes":[{"id":"input"}]}` {
		t.Fatalf("graph_json = %q, want the second (latest) payload", payload)
	}
}