package gateway

// 本文件是链路图**落盘持久化 + 多轮历史回放**的跨层闭环（切片 F2；design/19 §4 P3）。
// 它用真的写入路径（`loop.FlowPublisher.EndRound` → `flowgraph.SQLStore.RecordRound`
// → 真 SQLite + 迁移 0022），再经**真的** HTTP handler 读回来，钉住三件只有端到端
// 才成立、也正是主人要的事：
//
//   - **后端重启后仍能查回历史轮次的图**（本切片的内核）：换一套「进程」（新连接、
//     新 store、空内存的发布器），图仍从库里读得回来；
//   - **最近一轮优先内存、miss 才落盘**：内存里有更新的那轮时，省略 turnId 不回落盘旧轮；
//   - **白名单外会话既不落盘也查不到**：隔离渠道会话连图都不建，`turns` 恒空、`graph` 恒 null。
//
// 参数翻译与语义在 `admin_flow_graph_test.go`（v0.71 那批）与本文件互补。

import (
	"context"
	"database/sql"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/flowgraph"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)

// flowGraphResponseDTO 是 v0.73 响应体的线上投影（含新增三键）。
type flowGraphResponseDTO struct {
	SessionID string `json:"sessionId"`
	TurnID    string `json:"turnId"`
	Source    string `json:"source"`
	Graph     *struct {
		ID        string `json:"id"`
		SessionID string `json:"sessionId"`
		Nodes     []struct {
			Kind string `json:"kind"`
		} `json:"nodes"`
	} `json:"graph"`
	Turns []struct {
		TurnID    string `json:"turnId"`
		CreatedAt string `json:"createdAt"`
	} `json:"turns"`
}

func decodeFlowGraphResp(t *testing.T, body []byte) flowGraphResponseDTO {
	t.Helper()
	var out flowGraphResponseDTO
	if err := json.Unmarshal(decodeBody(t, body)["data"], &out); err != nil {
		t.Fatalf("decode flow graph data: %v (body=%s)", err, body)
	}
	return out
}

// openFlowGraphStore 打开一个真实的 SQLite 文件库、跑迁移 0022，返回连接与存储句柄。
// caller 可显式 Close 连接以模拟进程退出（`t.Cleanup` 会兜底再关一次，无害）。
func openFlowGraphStore(t *testing.T, path string) (*sql.DB, *flowgraph.SQLStore) {
	t.Helper()
	ctx := context.Background()
	conn, err := db.Open(ctx, path, gwTestLogger())
	if err != nil {
		t.Fatalf("open db: %v", err)
	}
	if err := db.MigrateUp(ctx, conn, gwTestLogger()); err != nil {
		_ = conn.Close()
		t.Fatalf("migrate: %v", err)
	}
	store, err := flowgraph.NewStore(flowgraph.Config{DB: conn, Log: gwTestLogger()})
	if err != nil {
		_ = conn.Close()
		t.Fatalf("new flowgraph store: %v", err)
	}
	t.Cleanup(func() { _ = conn.Close() })
	return conn, store
}

// flowPublisherWithStore 造一个把每轮收尾落进 store 的发布器（生产那条路径）。
func flowPublisherWithStore(store *flowgraph.SQLStore, allow func(string) bool) *loop.FlowPublisher {
	return loop.NewFlowPublisher(func(string, any) {}, allow,
		loop.WithRoundRecorder(store), loop.WithFlowLogger(gwTestLogger()))
}

// runFlowRound 让发布器跑完一整轮（起轮 → 一个 llm span 起止 → 收轮）。
func runFlowRound(p *loop.FlowPublisher, session, input, output string) {
	p.StartRound(session, input)
	sp := &trace.Span{ID: "span-llm", Name: "llm.call", Kind: trace.SpanKindLLM, StartedAt: time.Now()}
	p.SpanStarted(session, sp)
	sp.Finish(trace.SpanStatusOK)
	p.SpanFinished(session, sp)
	p.EndRound(session, output)
}

// mkFlowGraph 造一轮最小图，供直接写库（不经发布器）的用例使用。
func mkFlowGraph(session string, round int) *trace.FlowGraph {
	return &trace.FlowGraph{
		ID:        session + "-" + itoa(round),
		SessionID: session,
		Nodes: []trace.FlowNode{
			{ID: "input", Label: "你好", Kind: "input", Status: trace.FlowStatusDone},
			{ID: "output", Label: "回复", Kind: "output", Status: trace.FlowStatusDone},
		},
		Edges: []trace.FlowEdge{{From: "input", To: "output"}},
	}
}

func itoa(n int) string {
	if n == 0 {
		return "0"
	}
	digits := make([]byte, 0, 4)
	for n > 0 {
		digits = append([]byte{byte('0' + n%10)}, digits...)
		n /= 10
	}
	return string(digits)
}

// flowURL 拼出该端点的 URL（带可选 turnId）。
func flowURL(ts *httptest.Server, session, turn string) string {
	u := ts.URL + "/api/v1/admin/flow/graph?sessionId=" + session
	if turn != "" {
		u += "&turnId=" + turn
	}
	return u
}

// TestAdminFlowGraph_PersistedHistoryReadBack_正常：省略 turnId 回内存最近一轮
// （source live），带 turnId 回该历史轮（source persisted），且 turns 列全三轮、最新在前。
func TestAdminFlowGraph_PersistedHistoryReadBack_正常(t *testing.T) {
	t.Parallel()
	path := filepath.Join(t.TempDir(), "cyrene.db")
	_, store := openFlowGraphStore(t, path)
	pub := flowPublisherWithStore(store, func(string) bool { return true })
	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	for _, io := range []struct{ in, out string }{
		{"第一轮", "答一"}, {"第二轮", "答二"}, {"第三轮", "答三"},
	} {
		runFlowRound(pub, "web", io.in, io.out)
	}

	resp, body := doReq(t, http.MethodGet, flowURL(ts, "web", ""), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("latest status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	latest := decodeFlowGraphResp(t, body)
	if latest.Source != "live" {
		t.Fatalf("最新一轮 source = %q, want live（先查内存）", latest.Source)
	}
	if latest.TurnID != "web-3" {
		t.Fatalf("最新一轮 turnId = %q, want web-3", latest.TurnID)
	}
	if len(latest.Turns) != 3 {
		t.Fatalf("turns = %d, want 3（每轮一行，不是每 span 一行）", len(latest.Turns))
	}
	for i, want := range []string{"web-3", "web-2", "web-1"} {
		if latest.Turns[i].TurnID != want {
			t.Fatalf("turns[%d] = %q, want %q（最新在前）", i, latest.Turns[i].TurnID, want)
		}
	}

	resp, body = doReq(t, http.MethodGet, flowURL(ts, "web", "web-1"), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("history status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	hist := decodeFlowGraphResp(t, body)
	if hist.Source != "persisted" {
		t.Fatalf("历史轮 source = %q, want persisted（内存里没有 web-1）", hist.Source)
	}
	if hist.Graph == nil || hist.Graph.ID != "web-1" {
		t.Fatalf("历史轮 graph = %+v, want web-1", hist.Graph)
	}
	kinds := map[string]bool{}
	for _, n := range hist.Graph.Nodes {
		kinds[n.Kind] = true
	}
	if !kinds["input"] || !kinds["output"] {
		t.Fatalf("历史轮图缺主干端点: %+v", hist.Graph.Nodes)
	}
}

// TestAdminFlowGraph_SurvivesRestart_正常 是本切片的**内核**（主人要求）：后端重启后
// 仍能查回历史轮次。换一套「进程」（新连接 + 新 store + **空内存**的新发布器），
// 唯一能拿到图的方式就是从库读回。
func TestAdminFlowGraph_SurvivesRestart_正常(t *testing.T) {
	t.Parallel()
	path := filepath.Join(t.TempDir(), "cyrene.db")

	// 第一个「进程」：跑两轮，图进库。
	conn1, store1 := openFlowGraphStore(t, path)
	pub1 := flowPublisherWithStore(store1, func(string) bool { return true })
	_, ts1, _ := buildEnv(t, cfgDefault(), Deps{Flow: pub1, FlowGraphs: store1})
	runFlowRound(pub1, "web", "第一轮", "答一")
	runFlowRound(pub1, "web", "第二轮", "答二")
	ts1.Close()
	_ = conn1.Close() // 模拟进程退出

	// 第二个「进程」：空内存的新发布器 + 同一个文件库的新连接 / 新 store。
	_, store2 := openFlowGraphStore(t, path)
	pub2 := flowPublisherWithStore(store2, func(string) bool { return true })
	_, ts2, jwt2 := buildEnv(t, cfgDefault(), Deps{Flow: pub2, FlowGraphs: store2})
	tok := signAny(t, jwt2, "acct", auth.RoleOwner)

	resp, body := doReq(t, http.MethodGet, flowURL(ts2, "web", ""), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("after restart status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	got := decodeFlowGraphResp(t, body)
	if got.Graph == nil {
		t.Fatalf("重启后 graph = null，但该会话刚跑完两轮并已落盘（这正是主人要的「本地数据」）: %s", body)
	}
	if got.Source != "persisted" || got.TurnID != "web-2" {
		t.Fatalf("重启后 source/turnId = %q/%q, want persisted/web-2", got.Source, got.TurnID)
	}

	// 更早那一轮也能按 turnId 取回。
	resp, body = doReq(t, http.MethodGet, flowURL(ts2, "web", "web-1"), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("restart history status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	hist := decodeFlowGraphResp(t, body)
	if hist.Graph == nil || hist.Graph.ID != "web-1" {
		t.Fatalf("重启后 web-1 = %+v, want web-1", hist.Graph)
	}
}

// TestAdminFlowGraph_LatestPrefersMemory_正常 钉住「最新一轮永远最新」：内存里有更新的
// 那轮（web-2）而库里的最新只有 web-1 时，省略 turnId **必须**回内存那轮。
func TestAdminFlowGraph_LatestPrefersMemory_正常(t *testing.T) {
	t.Parallel()
	path := filepath.Join(t.TempDir(), "cyrene.db")
	_, store := openFlowGraphStore(t, path)
	if err := store.RecordRound(mkFlowGraph("web", 1)); err != nil {
		t.Fatalf("seed persisted round: %v", err)
	}
	// 发布器**不挂 recorder**：内存跑两轮（web-2），但不再落盘 ⇒ 库里仍旧只有 web-1。
	pub := loop.NewFlowPublisher(func(string, any) {}, func(string) bool { return true })
	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	runFlowRound(pub, "web", "第一轮", "答一")
	runFlowRound(pub, "web", "第二轮", "答二")

	resp, body := doReq(t, http.MethodGet, flowURL(ts, "web", ""), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	got := decodeFlowGraphResp(t, body)
	if got.Source != "live" || got.TurnID != "web-2" {
		t.Fatalf("source/turnId = %q/%q, want live/web-2（省略 turnId 时先查内存，绝不回落盘的旧轮）",
			got.Source, got.TurnID)
	}
}

// TestAdminFlowGraph_OutsideWhitelistNotPersisted_正常 钉住白名单复用：发布会话范围外
// 的会话（隔离渠道会话）连图都不建 ⇒ 既不落盘、也查不到，`turns` 恒空。
func TestAdminFlowGraph_OutsideWhitelistNotPersisted_正常(t *testing.T) {
	t.Parallel()
	path := filepath.Join(t.TempDir(), "cyrene.db")
	_, store := openFlowGraphStore(t, path)
	pub := flowPublisherWithStore(store, func(session string) bool { return session == "web" })
	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	runFlowRound(pub, "sess-im-stranger", "陌生人说话", "回应")
	runFlowRound(pub, "web", "主人说话", "回应")

	resp, body := doReq(t, http.MethodGet, flowURL(ts, "sess-im-stranger", ""), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("stranger status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	stranger := decodeFlowGraphResp(t, body)
	if stranger.Graph != nil {
		t.Fatalf("白名单外会话不该有图，却拿到 %+v", stranger.Graph)
	}
	if stranger.Source != "none" {
		t.Fatalf("source = %q, want none", stranger.Source)
	}
	if len(stranger.Turns) != 0 {
		t.Fatalf("白名单外会话不该落盘（turns=%+v）", stranger.Turns)
	}
	// 反向：白名单内的会话照常可查、有落盘轮（保证上面的空不是端点整体坏了）。
	_, body = doReq(t, http.MethodGet, flowURL(ts, "web", ""), nil, tok)
	if inScope := decodeFlowGraphResp(t, body); inScope.Graph == nil || len(inScope.Turns) != 1 {
		t.Fatalf("白名单内会话应有图且有一轮落盘: %s", body)
	}
}

// TestAdminFlowGraph_UnknownTurnIsHonestNull_边界 钉住「该 turnId 不存在」是诚实空态：
// 200 + `graph:null` + `source:"none"`，绝不用别的轮次顶替。
func TestAdminFlowGraph_UnknownTurnIsHonestNull_边界(t *testing.T) {
	t.Parallel()
	path := filepath.Join(t.TempDir(), "cyrene.db")
	_, store := openFlowGraphStore(t, path)
	pub := flowPublisherWithStore(store, func(string) bool { return true })
	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	runFlowRound(pub, "web", "第一轮", "答一")

	resp, body := doReq(t, http.MethodGet, flowURL(ts, "web", "web-999"), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	got := decodeFlowGraphResp(t, body)
	if got.Graph != nil {
		t.Fatalf("不存在的轮次 graph = %+v, want null", got.Graph)
	}
	if got.Source != "none" || got.TurnID != "" {
		t.Fatalf("source/turnId = %q/%q, want none/\"\"", got.Source, got.TurnID)
	}
	// `graph` 键必须恒在（客户端靠它区分「没有图」与「字段缺失」）。
	var envelope map[string]json.RawMessage
	if err := json.Unmarshal(decodeBody(t, body)["data"], &envelope); err != nil {
		t.Fatalf("decode envelope: %v", err)
	}
	if _, present := envelope["graph"]; !present {
		t.Fatalf("`graph` 键必须恒在（空时是 null）: %s", body)
	}
}

// TestAdminFlowGraph_EmptyTurnsSerialiseAsArray_边界 钉住线上空形状：没有落盘轮次时
// `turns` 是 `[]`（客户端不必在 null 与空数组之间分叉）。
func TestAdminFlowGraph_EmptyTurnsSerialiseAsArray_边界(t *testing.T) {
	t.Parallel()
	path := filepath.Join(t.TempDir(), "cyrene.db")
	_, store := openFlowGraphStore(t, path)
	pub := flowPublisherWithStore(store, func(string) bool { return true })
	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	resp, body := doReq(t, http.MethodGet, flowURL(ts, "web", ""), nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	if !strings.Contains(string(body), `"turns":[]`) {
		t.Fatalf("空 turns 必须序列化为 [] 而非 null: %s", body)
	}
}