package memory

import (
	"context"
	"errors"
	"strings"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
)

// 本文件锁定记忆整理「试算」的**加性异步**（契约 v0.76 §4.12.1；切片 MT-A）：
//
//	① **零副作用照旧**：异步跑完后 memory_items / memory_tidy_candidates /
//	   memory_tidy_runs **逐行不变**（与同步试算同一条硬约束，且用的是既有的
//	   逐行快照比对，不是另一套断言）；
//	② **结果 = 同步结果**：异步取回的候选与同一批数据上的同步试算逐字段一致；
//	③ **同时只允许 1 个在跑**：第二个请求复用既有 runId，不另起一轮；
//	④ **有界且诚实**：完成后有 TTL（30 分钟）、进程重启后旧 runId 如实失效
//	   （ErrTidyRunNotFound），未知 runId 同样如此 —— 绝不假装还在跑；
//	⑤ **失败带真实原因**：语义召回不可用 / 配置关闭 / 超时都回失败态 + 原始错误，
//	   绝不回一份空候选冒充「没有建议」。

// advClock 是可由测试推进的时钟（并发安全：后台 goroutine 也会读它）。
type advClock struct {
	mu sync.Mutex
	at time.Time
}

func newAdvClock(at time.Time) *advClock { return &advClock{at: at} }

func (c *advClock) now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.at
}

func (c *advClock) advance(d time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.at = c.at.Add(d)
}

// gateEmbedder 是「卡在召回里不放」的 embedder：它让一轮停在 embedding 调用上，
// 用来验「同时只允许 1 个在跑」。release 关闭后所有调用立即返回单位向量。
type gateEmbedder struct{ release chan struct{} }

func (g gateEmbedder) Embed(ctx context.Context, texts []string) ([]embedding.Vector, error) {
	select {
	case <-g.release:
	case <-ctx.Done():
		return nil, ctx.Err()
	}
	out := make([]embedding.Vector, 0, len(texts))
	for range texts {
		out = append(out, embedding.Vector{1, 0})
	}
	return out, nil
}

func (gateEmbedder) Dim() int { return 2 }

// hungEmbedder 模拟「embedding 服务半死不活」：一直等到 ctx 被取消。
type hungEmbedder struct{}

func (hungEmbedder) Embed(ctx context.Context, _ []string) ([]embedding.Vector, error) {
	<-ctx.Done()
	return nil, ctx.Err()
}

func (hungEmbedder) Dim() int { return 2 }

// waitTidyRun 轮询到想要的终态；超时或意外失败即让用例失败。
func waitTidyRun(t *testing.T, svc *TidyService, runID string, want TidyAsyncStatus) TidyAsyncDryRun {
	t.Helper()
	deadline := time.Now().Add(15 * time.Second)
	for time.Now().Before(deadline) {
		run, err := svc.DryRunResult(runID)
		if err != nil {
			t.Fatalf("DryRunResult(%s): %v", runID, err)
		}
		if run.Status == want {
			return run
		}
		if run.Status == TidyAsyncFailed && want != TidyAsyncFailed {
			t.Fatalf("运行提前失败：%v", run.Err)
		}
		time.Sleep(2 * time.Millisecond)
	}
	t.Fatalf("等待 %s 超时", want)
	return TidyAsyncDryRun{}
}

// TestTidyAsyncDryRun_ZeroSideEffectsAndMatchesSync 是判据 ①②：异步路径**复用**
// 同步试算的单测口径 —— 跑完后三张表逐行不变，且取回的候选与同步试算逐字段一致。
func TestTidyAsyncDryRun_ZeroSideEffectsAndMatchesSync(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	seedDryRunFixture(t, store)
	before := snapshotTidyTables(t, store)

	svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder(), time.Now, nil)
	handle, err := svc.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun: %v", err)
	}
	if handle.RunID == "" || handle.StartedAt.IsZero() {
		t.Fatalf("句柄必须带 runId 与 startedAt：%+v", handle)
	}

	run := waitTidyRun(t, svc, handle.RunID, TidyAsyncDone)
	if run.Result == nil {
		t.Fatal("done 必须带 result（形状 = 同步试算的结果）")
	}
	if len(run.Result.Candidates) == 0 {
		t.Fatalf("夹具应当产出候选（否则这条断言是空转）：%+v", run.Result.Stats)
	}

	sync, err := svc.DryRun(ctx)
	if err != nil {
		t.Fatalf("DryRun: %v", err)
	}
	got := sortedSignatures(run.Result.Candidates)
	want := sortedSignatures(sync.Candidates)
	if strings.Join(got, "\n") != strings.Join(want, "\n") {
		t.Fatalf("异步候选与同步试算不一致：\n async=%v\n sync =%v", got, want)
	}
	if run.Result.RecallMode != sync.RecallMode || run.Result.Stats.Compared != sync.Stats.Compared {
		t.Fatalf("异步与同步的观测口径不一致：%+v vs %+v", run.Result, sync)
	}

	after := snapshotTidyTables(t, store)
	assertRowsUnchanged(t, "memory_items", before.items, after.items)
	assertRowsUnchanged(t, "memory_tidy_candidates", before.candidates, after.candidates)
	assertRowsUnchanged(t, "memory_tidy_runs", before.runs, after.runs)
}

// TestTidyAsyncDryRun_OnlyOneRunAtATimeReusesRunID 是判据 ③：同时只允许 1 个在跑，
// 第二个请求**复用**既有 runId（契约 §4.12.1 给的「如实拒绝或复用」中选了复用）。
func TestTidyAsyncDryRun_OnlyOneRunAtATimeReusesRunID(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	seedDryRunFixture(t, store)
	gate := gateEmbedder{release: make(chan struct{})}
	svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), gate, time.Now, nil)

	first, err := svc.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun: %v", err)
	}
	second, err := svc.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun(第二次): %v", err)
	}
	if second.RunID != first.RunID {
		t.Fatalf("在跑期间第二个请求必须复用 runId（同时只允许 1 个）：%q vs %q", second.RunID, first.RunID)
	}
	if !second.StartedAt.Equal(first.StartedAt) {
		t.Fatalf("复用必须是**同一次**运行：startedAt %v vs %v", second.StartedAt, first.StartedAt)
	}
	run, err := svc.DryRunResult(first.RunID)
	if err != nil {
		t.Fatalf("DryRunResult: %v", err)
	}
	if run.Status != TidyAsyncRunning {
		t.Fatalf("还没释放闸门就报 %q，want running", run.Status)
	}

	close(gate.release)
	done := waitTidyRun(t, svc, first.RunID, TidyAsyncDone)
	if done.Result == nil {
		t.Fatal("完成后必须有结果")
	}

	// 上一轮跑完之后，新请求才是**新的一次**运行（runId 不重复用）。
	third, err := svc.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun(第三次): %v", err)
	}
	if third.RunID == first.RunID {
		t.Fatal("上一轮已结束，新请求必须是一次新运行（runId 必须不同）")
	}
	waitTidyRun(t, svc, third.RunID, TidyAsyncDone)
}

// TestTidyAsyncDryRun_TTLExpiryIsHonest 是判据 ④（TTL）：TTL 内可查，超时后如实
// 查不到，绝不假装还在跑。
func TestTidyAsyncDryRun_TTLExpiryIsHonest(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	seedDryRunFixture(t, store)
	clock := newAdvClock(time.Now())
	svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder(), clock.now, nil)

	handle, err := svc.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun: %v", err)
	}
	waitTidyRun(t, svc, handle.RunID, TidyAsyncDone)

	clock.advance(defaultTidyRunTTL - time.Minute)
	if _, err := svc.DryRunResult(handle.RunID); err != nil {
		t.Fatalf("TTL 内必须还能查到：%v", err)
	}
	clock.advance(2 * time.Minute)
	if _, err := svc.DryRunResult(handle.RunID); !errors.Is(err, ErrTidyRunNotFound) {
		t.Fatalf("超 TTL 后必须如实报 ErrTidyRunNotFound，got %v", err)
	}
}

// TestTidyAsyncDryRun_RestartDropsRunIDs 是判据 ④（重启语义）：结果只活在进程
// 内存里，**同一份存储上新建的实例**（= 进程重启）查不到旧 runId。
func TestTidyAsyncDryRun_RestartDropsRunIDs(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	seedDryRunFixture(t, store)
	first := newTidyServiceForTest(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder(), time.Now, nil)
	handle, err := first.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun: %v", err)
	}
	waitTidyRun(t, first, handle.RunID, TidyAsyncDone)

	restarted := newTidyServiceForTest(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder(), time.Now, nil)
	if _, err := restarted.DryRunResult(handle.RunID); !errors.Is(err, ErrTidyRunNotFound) {
		t.Fatalf("重启后旧 runId 必须如实失效（注册表是进程内存态），got %v", err)
	}
}

// TestTidyAsyncDryRun_UnknownRunIDIsHonest 钉住「未知 runId」的诚实语义。
func TestTidyAsyncDryRun_UnknownRunIDIsHonest(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), nil, time.Now, nil)
	for _, id := range []string{"", "run-nonexistent", "../../etc/passwd"} {
		if _, err := svc.DryRunResult(id); !errors.Is(err, ErrTidyRunNotFound) {
			t.Fatalf("DryRunResult(%q) err = %v, want ErrTidyRunNotFound", id, err)
		}
	}
}

// TestTidyAsyncDryRun_FailureCarriesRealCause 是判据 ⑤：语义召回不可用时失败态带
// **真实原因**（ErrTidyRecallUnavailable），且没有 result —— 面板因此不会把
// 「没算出来」读成「算过了，没有建议」。
func TestTidyAsyncDryRun_FailureCarriesRealCause(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	seedDryRunFixture(t, store)
	before := snapshotTidyTables(t, store)
	svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80),
		failingEmbedder{err: errors.New("dial tcp 127.0.0.1:11434: 目标计算机积极拒绝")}, time.Now, nil)

	handle, err := svc.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun: %v", err)
	}
	run := waitTidyRun(t, svc, handle.RunID, TidyAsyncFailed)
	if !errors.Is(run.Err, ErrTidyRecallUnavailable) {
		t.Fatalf("失败原因 = %v, want ErrTidyRecallUnavailable", run.Err)
	}
	if run.Result != nil {
		t.Fatal("失败态不得带 result（那会把「没算出来」伪装成结果）")
	}
	after := snapshotTidyTables(t, store)
	assertRowsUnchanged(t, "memory_items", before.items, after.items)
	assertRowsUnchanged(t, "memory_tidy_candidates", before.candidates, after.candidates)
	assertRowsUnchanged(t, "memory_tidy_runs", before.runs, after.runs)
}

// TestTidyAsyncDryRun_DisabledFailsHonestly 钉住「配置关闭」在异步路径上的口径：
// 仍然如实回 runId（不假装排队成功），失败原因就是 ErrTidyDisabled —— 与同步
// 路径同一个错误，网关据此投影成同一张失败码表。
func TestTidyAsyncDryRun_DisabledFailsHonestly(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	seedDryRunFixture(t, store)
	cfg := tidyTestCfg(0.80)
	cfg.Enabled = false
	svc := newTidyServiceForTest(t, store, cfg, dryRunFixtureEmbedder(), time.Now, nil)

	handle, err := svc.StartDryRun(0)
	if err != nil {
		t.Fatalf("StartDryRun: %v", err)
	}
	run := waitTidyRun(t, svc, handle.RunID, TidyAsyncFailed)
	if !errors.Is(run.Err, ErrTidyDisabled) {
		t.Fatalf("关闭时失败原因 = %v, want ErrTidyDisabled", run.Err)
	}
}

// TestTidyAsyncDryRun_TimeoutFailsHonestly 钉住墙钟上限：embedding 卡死到超时 ⇒
// 失败态里的错误链含 context.DeadlineExceeded（网关据此投影成 tidy_timeout / 召回
// 不可用，见 gateway 的同一张映射表）。
func TestTidyAsyncDryRun_TimeoutFailsHonestly(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	seedDryRunFixture(t, store)
	svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), hungEmbedder{}, time.Now, nil)

	handle, err := svc.StartDryRun(300 * time.Millisecond)
	if err != nil {
		t.Fatalf("StartDryRun: %v", err)
	}
	run := waitTidyRun(t, svc, handle.RunID, TidyAsyncFailed)
	if !errors.Is(run.Err, context.DeadlineExceeded) {
		t.Fatalf("超时的错误链必须含 context.DeadlineExceeded，got %v", run.Err)
	}
}