package gateway

import (
	"context"
	"errors"
	"strings"
	"testing"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)

// 本文件锁定切片 P0 的**落库**那一半：主动消息无论走活跃客户端路由
// （gateway hub）还是 IM 桥路由（channelbridge），都必须成为主会话
// （`history.MainSessionID`）里**恰好一条** assistant 消息。
//
// 为什么需要它：修复前活跃客户端路由只广播 `proactive_message` 帧、一行都不写，
// 于是「帧被 2 个连接收下」被读成「已推送」，刷新后什么都不剩（主人报的正是
// 「后台思考的主动推送一个都没推出来」）。另一半风险是**双份** —— IM 适配器
// 出站消息本来就会被镜像回主会话（channelbridge/proactive.go 的 persistToMain），
// 若活跃路由也写一次、而一条推送同时走了两条路，主会话里就会出现两条。
// 因此这里两个路由各自跑一遍，逐个断言「恰好一条」，并额外断言
// 一条推送只会触发**一路**写入。

// proactiveStoreMsg 构造一条最小化的自主主动消息，等同于 backthink 的
// buildProactiveMessage 在无渠道决策（通用哨兵值）时的产物。
func proactiveStoreMsg(content string) adapter.UnifiedMessage {
	return adapter.UnifiedMessage{
		Platform:    "cyrene-backthink",
		ChannelType: adapter.ChannelDirect,
		Content:     content,
		ContentType: adapter.ContentText,
		Sender:      adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
	}
}

// mainSessionRows 读回主会话的落库行（最新优先）。
func mainSessionRows(t *testing.T, store *history.Store) []history.Message {
	t.Helper()
	rows, err := store.List(context.Background(), history.MainSessionID, 0, 50, "")
	if err != nil {
		t.Fatalf("list main session: %v", err)
	}
	return rows
}

// TestHub_SendProactive_PersistsAssistantMessageToMainSession 验证活跃客户端
// 路由把主动消息写成主会话里的一条 assistant 消息（「气泡恒落」的根：刷新后
// 仍在），且发出去的帧带上落库行的权威 id/时间戳 —— 客户端据此把实时气泡
// 对齐到历史行，而不是让它刷新即丢。
func TestHub_SendProactive_PersistsAssistantMessageToMainSession(t *testing.T) {
	t.Parallel()
	store := newHistoryStoreForGateway(t)
	hb := newTestHub(nil, nil)
	hb.history = store
	owner := newOwnerConn(hb, 8)
	hb.register(owner)

	delivery, err := hb.SendProactive(context.Background(), proactiveStoreMsg("记得喝水哦"))
	if err != nil {
		t.Fatalf("SendProactive err = %v", err)
	}
	if delivery.Connections != 1 {
		t.Fatalf("connections = %d, want 1", delivery.Connections)
	}
	if !delivery.Persisted {
		t.Fatalf("Persisted = false, want true (主动消息必须落进主会话，否则刷新就没了)")
	}

	rows := mainSessionRows(t, store)
	if len(rows) != 1 {
		t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条（>1 = 双份）", len(rows))
	}
	row := rows[0]
	if row.Role != history.RoleAssistant {
		t.Fatalf("role = %q, want %q（她单方面开口的话，绝不能冒充主人的发言）", row.Role, history.RoleAssistant)
	}
	if row.SessionID != history.MainSessionID {
		t.Fatalf("session = %q, want %q", row.SessionID, history.MainSessionID)
	}
	// 正文经共享规整器（turnreply.NormalizeParagraphs）：纯散文被包成一个 `<p>`
	// ——与 Web 回合 / IM 出站 / IM 侧主动推送同一份口径（M25：本路由曾绕过它）。
	if row.Content != "<p>记得喝水哦</p>" {
		t.Fatalf("content = %q, want the normalized proactive text", row.Content)
	}
	if row.TurnID != "" {
		t.Fatalf("turn_id = %q, want empty（主动推送不属于任何一路回合）", row.TurnID)
	}

	ev := readFakeFrame(t, owner)
	if ev.Type != wsProactiveMessage {
		t.Fatalf("frame type = %q, want %q", ev.Type, wsProactiveMessage)
	}
	data, ok := ev.Data.(map[string]any)
	if !ok {
		t.Fatalf("frame data type = %T, want map[string]any", ev.Data)
	}
	if data["id"] != row.ID {
		t.Fatalf("frame id = %v, want the persisted row id %q", data["id"], row.ID)
	}
	if got, _ := data["createdAt"].(float64); int64(got) != row.CreatedAt {
		t.Fatalf("frame createdAt = %v, want %d", data["createdAt"], row.CreatedAt)
	}
}

// failingHistory 让 AppendFull 失败，用于验证「落库失败**可见**」：帧照发
// （实时气泡不能丢）、错误如实返回、Persisted=false（绝不谎报已推送）。
type failingHistory struct {
	HistoryStore
	err error
}

func (f failingHistory) AppendFull(context.Context, history.Message) (history.Message, error) {
	return history.Message{}, f.err
}

// TestHub_SendProactive_PersistFailureIsVisible 验证落库失败时不做假成功：
// 错误被返回、Persisted=false、Detail 的措辞由调用方如实反映，而帧仍然发出。
func TestHub_SendProactive_PersistFailureIsVisible(t *testing.T) {
	t.Parallel()
	hb := newTestHub(nil, nil)
	hb.history = failingHistory{err: errors.New("db down")}
	owner := newOwnerConn(hb, 8)
	hb.register(owner)

	delivery, err := hb.SendProactive(context.Background(), proactiveStoreMsg("在吗"))
	if err == nil {
		t.Fatalf("err = nil, want the persist failure to be reported")
	}
	if delivery.Persisted {
		t.Fatalf("Persisted = true, want false（没落库就是没落库）")
	}
	if delivery.Connections != 1 {
		t.Fatalf("connections = %d, want 1（落库失败不该吞掉实时帧）", delivery.Connections)
	}
	var frameData any
	if ev := readFakeFrame(t, owner); ev.Type != wsProactiveMessage {
		t.Fatalf("frame type = %q, want %q", ev.Type, wsProactiveMessage)
	} else {
		frameData = ev.Data
	}
	data, ok := frameData.(map[string]any)
	if !ok {
		t.Fatalf("frame data type = %T, want map[string]any", frameData)
	}
	if _, hasID := data["id"]; hasID {
		t.Fatalf("frame carries id %v, want the key omitted（没有落库行就没有权威 id）", data["id"])
	}
}

// fakeIMAdapter 是 IM 路由测试所需的最小适配器（只用到连接态与出站）。
type fakeIMAdapter struct {
	platform string
	sends    []adapter.UnifiedMessage
}

func (f *fakeIMAdapter) Name() string                      { return f.platform }
func (f *fakeIMAdapter) Config() map[string]any            { return map[string]any{} }
func (f *fakeIMAdapter) Connect(context.Context) error     { return nil }
func (f *fakeIMAdapter) Disconnect(context.Context) error  { return nil }
func (f *fakeIMAdapter) IsConnected() bool                 { return true }
func (f *fakeIMAdapter) HealthCheck(context.Context) error { return nil }
func (f *fakeIMAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) {
	<-ctx.Done()
}
func (f *fakeIMAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error {
	f.sends = append(f.sends, resp.Messages...)
	return nil
}
func (f *fakeIMAdapter) SendProactive(_ context.Context, msg adapter.UnifiedMessage) error {
	f.sends = append(f.sends, msg)
	return nil
}
func (f *fakeIMAdapter) SetMessageHandler(adapter.MessageHandler) {}
func (f *fakeIMAdapter) Capabilities() adapter.Capabilities       { return adapter.DefaultCapabilities() }
func (f *fakeIMAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) {
	return nil, nil
}

// stubBridgeChatter 满足 channelbridge.Chatter（本测试不会跑回合）。
type stubBridgeChatter struct{}

func (stubBridgeChatter) Run(context.Context, string, []model.Message) (*loop.Result, error) {
	return &loop.Result{Text: "hi"}, nil
}

// stubBridgeSessions 满足 channelbridge.SessionStore（本测试不会走入站路径）。
type stubBridgeSessions struct{}

func (stubBridgeSessions) GetOrCreate(context.Context, adapter.UnifiedMessage, *user.OwnerProfile) (session.Session, error) {
	return session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}, nil
}

// newProactiveTestBridge 装配一条接在**真历史存储**上的 IM 桥（一个已连接、
// 在白名单里的主人渠道），使 IM 路由的落库能被真库断言。
func newProactiveTestBridge(t *testing.T, store *history.Store, onPersisted *[]history.Message) (*channelbridge.Bridge, *fakeIMAdapter) {
	t.Helper()
	owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{{Platform: "telegram", AccountID: "100"}}}
	var hook func(history.Message)
	if onPersisted != nil {
		hook = func(m history.Message) { *onPersisted = append(*onPersisted, m) }
	}
	b, err := channelbridge.New(channelbridge.Options{
		Log:         gwTestLogger(),
		Loop:        stubBridgeChatter{},
		Sessions:    stubBridgeSessions{},
		Owner:       owner,
		History:     store,
		OnPersisted: hook,
	})
	if err != nil {
		t.Fatalf("new bridge: %v", err)
	}
	ad := &fakeIMAdapter{platform: "telegram"}
	b.Add(ad)
	return b, ad
}

// TestChannelBridge_SendProactive_MirrorsExactlyOnce 是**现状核查**的可执行
// 证据：IM 路由本来就会把主动消息镜像回主会话（且只镜像一次）。因此在这条
// 路由上再加一次落库就是双份 —— 活跃路由的落库正是靠「两条路由互斥」避开它的。
func TestChannelBridge_SendProactive_MirrorsExactlyOnce(t *testing.T) {
	t.Parallel()
	store := newHistoryStoreForGateway(t)
	var persisted []history.Message
	bridge, ad := newProactiveTestBridge(t, store, &persisted)

	if err := bridge.SendProactive(context.Background(), proactiveStoreMsg("记得喝水哦")); err != nil {
		t.Fatalf("SendProactive err = %v", err)
	}
	if len(ad.sends) != 1 {
		t.Fatalf("IM sends = %d, want 1", len(ad.sends))
	}
	rows := mainSessionRows(t, store)
	if len(rows) != 1 {
		t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条（>1 = 双份）", len(rows))
	}
	if rows[0].Role != history.RoleAssistant || !strings.Contains(rows[0].Content, "记得喝水哦") {
		t.Fatalf("row = %+v, want an assistant message carrying the proactive text", rows[0])
	}
	if len(persisted) != 1 || persisted[0].ID != rows[0].ID {
		t.Fatalf("OnPersisted 应恰好收到刚落地的那一条（Web 端据此实时镜像），得到 %+v", persisted)
	}
}

// TestProactiveRoutes_EachRouteAddsExactlyOneRow 把两条路由对**同一条**主动
// 消息各跑一次：主会话里应恰好两条（每条路由各一条）。它同时钉住两件事 ——
// ① 两条路由都真的落库（不再有「只发帧不落库」的那条）；② 没有一条路由写两份
// （若活跃路由也走 IM 镜像那一次，这里会是 3 条）。
func TestProactiveRoutes_EachRouteAddsExactlyOneRow(t *testing.T) {
	t.Parallel()
	store := newHistoryStoreForGateway(t)
	hb := newTestHub(nil, nil)
	hb.history = store
	hb.register(newOwnerConn(hb, 8))
	bridge, _ := newProactiveTestBridge(t, store, nil)

	msg := proactiveStoreMsg("两条路各跑一次")
	if _, err := hb.SendProactive(context.Background(), msg); err != nil {
		t.Fatalf("active route err = %v", err)
	}
	if got := len(mainSessionRows(t, store)); got != 1 {
		t.Fatalf("活跃路由之后主会话有 %d 条，want 1", got)
	}
	if err := bridge.SendProactive(context.Background(), msg); err != nil {
		t.Fatalf("im route err = %v", err)
	}
	rows := mainSessionRows(t, store)
	if len(rows) != 2 {
		t.Fatalf("两条路由各跑一次后主会话有 %d 条，want 2（3 条 = 有一条路由写了双份）", len(rows))
	}
	for _, row := range rows {
		if row.Role != history.RoleAssistant {
			t.Fatalf("row role = %q, want assistant", row.Role)
		}
	}
}