package wire

import (
	"context"
	"encoding/json"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
)

// fakeAssetsRefresher captures asset refresh calls so tests can assert which
// kinds and how many times the dispatcher invoked each refresher.
type fakeAssetsRefresher struct {
	mu          sync.Mutex
	memoryCalls [][]string
	skillCalls  [][]string
	kbCalls     [][]string
}

type fakeAgentTeamAssetsRefresher struct {
	agentteamTestDispatcher
	fakeAssetsRefresher
}

type blockingAgentTeamAssetsRefresher struct {
	fakeAgentTeamAssetsRefresher
	skillRelease <-chan struct{}
}

func (f *blockingAgentTeamAssetsRefresher) RefreshSkillAccess(ctx context.Context, agentIDs []string) {
	if f.skillRelease != nil {
		select {
		case <-f.skillRelease:
		case <-ctx.Done():
		}
	}
	f.fakeAgentTeamAssetsRefresher.RefreshSkillAccess(ctx, agentIDs)
}

func (f *fakeAssetsRefresher) RefreshMemoryAccess(_ context.Context, agentIDs []string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.memoryCalls = append(f.memoryCalls, append([]string(nil), agentIDs...))
}

func (f *fakeAssetsRefresher) RefreshSkillAccess(_ context.Context, agentIDs []string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.skillCalls = append(f.skillCalls, append([]string(nil), agentIDs...))
}

func (f *fakeAssetsRefresher) RefreshKnowledgeBaseAccess(_ context.Context, agentIDs []string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.kbCalls = append(f.kbCalls, append([]string(nil), agentIDs...))
}

// newChannelForAssetsDedup builds a Protocol with just enough state for
// shouldApplyAssetsEvent — the dedup ring is lazily initialized inside the
// method, so we only need the mutex and the id.
func newChannelForAssetsDedup(t *testing.T) *Protocol {
	t.Helper()
	return &Protocol{id: "ch_test"}
}

// TestAssetsAccessChangedMsg_OccurredAtIsTime pins the wire-type contract:
// the server emits OccurredAt as an RFC3339 string (time.Time on the
// server). The Desktop struct MUST decode that without error — the prior
// int64 typing silently dropped every event because JSON decode failed.
func TestAssetsAccessChangedMsg_OccurredAtIsTime(t *testing.T) {
	raw := []byte(`{
		"type": "agent:assets:access-changed",
		"event_id": "evt_1",
		"agent_ids": ["agt_a"],
		"asset_types": ["skill"],
		"version": 42,
		"occurred_at": "2026-06-20T03:30:00Z",
		"source": "outbox"
	}`)
	var msg assetsAccessChangedMsg
	require.NoError(t, json.Unmarshal(raw, &msg))
	require.Equal(t, "evt_1", msg.EventID)
	require.Equal(t, []string{"agt_a"}, msg.AgentIDs)
	require.Equal(t, []string{"skill"}, msg.AssetTypes)
	require.Equal(t, int64(42), msg.Version)
	require.Equal(t, "outbox", msg.Source)
	require.False(t, msg.OccurredAt.IsZero(),
		"OccurredAt MUST decode from RFC3339 string; int64 typing silently wiped every event")
	want, err := time.Parse(time.RFC3339, "2026-06-20T03:30:00Z")
	require.NoError(t, err)
	require.True(t, msg.OccurredAt.Equal(want))
}

// TestDispatchAssetsAccessByKind_RoutesByKind pins the routing: memory-only
// events trigger only RefreshMemoryAccess; skill-only trigger only
// RefreshSkillAccess; mixed events trigger both.
func TestDispatchAssetsAccessByKind_RoutesByKind(t *testing.T) {
	f := &fakeAssetsRefresher{}
	ctx := context.Background()

	dispatchAssetsAccessByKind(ctx, f, normalizeAssetTypes([]string{"memory"}), []string{"agt_a"})
	dispatchAssetsAccessByKind(ctx, f, normalizeAssetTypes([]string{"skill"}), []string{"agt_a"})
	dispatchAssetsAccessByKind(ctx, f, normalizeAssetTypes([]string{"memory", "skill"}), []string{"agt_a"})
	dispatchAssetsAccessByKind(ctx, f, normalizeAssetTypes([]string{"kb"}), []string{"agt_a"})

	require.Len(t, f.memoryCalls, 2, "memory kind routed exactly twice")
	require.Len(t, f.skillCalls, 2, "skill kind routed exactly twice")
	require.Len(t, f.kbCalls, 1, "kb kind routed exactly once")
}

func TestRefreshSkillAccessForSyncedAgentsFiltersEmptyIDs(t *testing.T) {
	f := &fakeAgentTeamAssetsRefresher{}
	c := &Protocol{id: "ch_skills", disp: f}

	c.refreshSkillAccessForSyncedAgents([]types.MachineAgentInfo{
		{ID: "agt_a"},
		{ID: " "},
		{ID: "agt_b"},
	})

	require.Equal(t, [][]string{{"agt_a", "agt_b"}}, f.skillCalls)
	require.Empty(t, f.memoryCalls)
	require.Empty(t, f.kbCalls)
}

func TestRefreshKnowledgeBaseAccessForSyncedAgentsFiltersEmptyIDs(t *testing.T) {
	f := &fakeAgentTeamAssetsRefresher{}
	c := &Protocol{id: "ch_kb", disp: f}

	c.refreshKnowledgeBaseAccessForSyncedAgents([]types.MachineAgentInfo{
		{ID: "agt_a"},
		{ID: ""},
		{ID: "agt_b"},
	})

	require.Equal(t, [][]string{{"agt_a", "agt_b"}}, f.kbCalls)
	require.Empty(t, f.memoryCalls)
	require.Empty(t, f.skillCalls)
}

func TestRefreshMemoryAccessForSyncedAgentsFiltersEmptyIDs(t *testing.T) {
	f := &fakeAgentTeamAssetsRefresher{}
	c := &Protocol{id: "ch_memory", disp: f}

	c.refreshMemoryAccessForSyncedAgents([]types.MachineAgentInfo{
		{ID: "agt_a"},
		{ID: ""},
		{ID: "agt_b"},
	})

	require.Equal(t, [][]string{{"agt_a", "agt_b"}}, f.memoryCalls)
	require.Empty(t, f.skillCalls)
	require.Empty(t, f.kbCalls)
}

func TestHandleAgentListResponseUnblocksWaitersBeforeSkillPrefetchCompletes(t *testing.T) {
	const syncTimeout = 5 * time.Second

	releaseSkillRefresh := make(chan struct{})
	refresher := &blockingAgentTeamAssetsRefresher{skillRelease: releaseSkillRefresh}
	c := &Protocol{
		id:     "agentteam-test",
		disp:   refresher,
		sendCh: make(chan []byte, 1),
	}

	done := make(chan bool, 1)
	go func() {
		done <- c.requestAgentListSync(syncTimeout)
	}()

	require.Eventually(t, func() bool {
		c.mu.Lock()
		defer c.mu.Unlock()
		return c.agentListSyncInFlight && len(c.agentListSyncWaiters) == 1
	}, syncTimeout, 5*time.Millisecond)

	require.Eventually(t, func() bool { return len(c.sendCh) == 1 }, syncTimeout, 5*time.Millisecond)
	raw := <-c.sendCh
	require.JSONEq(t, `{"type":"machine:agent:list"}`, string(raw))

	c.handleAgentListResponse([]byte(`{
		"type":"machine:agent:list:response",
		"agents":[{"id":"agt_a","name":"agent-a","displayName":"Agent A","runtime":"codex","status":"idle","activity":""}]
	}`))

	select {
	case ok := <-done:
		require.True(t, ok, "requestAgentListSync should complete after catalog sync even if skill prefetch is still blocked")
	case <-time.After(syncTimeout):
		t.Fatal("requestAgentListSync remained blocked on skill prefetch")
	}

	close(releaseSkillRefresh)
	require.Eventually(t, func() bool {
		refresher.mu.Lock()
		defer refresher.mu.Unlock()
		return len(refresher.skillCalls) == 1
	}, syncTimeout, 5*time.Millisecond)
}

// TestShouldApplyAssetsEvent_DedupByEventID pins the per-event_id
// idempotency contract: replaying the same event_id MUST NOT be applied a
// second time. This protects against server retries and reconnect replays.
//
// The previous channel-level version-counter scheme failed this: two events
// with the same version but different event_ids for different agents would
// collapse to one, and an older replay could suppress a newer unrelated
// event.
func TestShouldApplyAssetsEvent_DedupByEventID(t *testing.T) {
	c := newChannelForAssetsDedup(t)

	require.True(t, c.shouldApplyAssetsEvent("evt_X"))
	require.False(t, c.shouldApplyAssetsEvent("evt_X"))
	require.False(t, c.shouldApplyAssetsEvent("evt_X"))

	// A different event_id MUST still go through.
	require.True(t, c.shouldApplyAssetsEvent("evt_Y"))
	require.False(t, c.shouldApplyAssetsEvent("evt_Y"))
}

// TestShouldApplyAssetsEvent_EmptyIDAlwaysApplied pins the legacy compat
// path: events without event_id cannot be deduped, so they always apply.
func TestShouldApplyAssetsEvent_EmptyIDAlwaysApplied(t *testing.T) {
	c := newChannelForAssetsDedup(t)

	require.True(t, c.shouldApplyAssetsEvent(""))
	require.True(t, c.shouldApplyAssetsEvent(""))
	require.True(t, c.shouldApplyAssetsEvent("   "))
}

// TestShouldApplyAssetsEvent_RingBounded confirms the seen-set does not
// grow unboundedly. Once the ring is full, the oldest entry is evicted and
// a subsequent distinct event is still applied. Replaying the evicted id
// also re-applies (the in-memory ring is best-effort, not durable).
func TestShouldApplyAssetsEvent_RingBounded(t *testing.T) {
	c := newChannelForAssetsDedup(t)

	for i := 0; i < assetsEventDedupCapacity+1; i++ {
		require.True(t, c.shouldApplyAssetsEvent("evt_"+strconv.Itoa(i)),
			"distinct event_id %d must be applied", i)
	}
	require.LessOrEqual(t, len(c.seenAssetsEventIDs), assetsEventDedupCapacity,
		"seen-set MUST be bounded by assetsEventDedupCapacity")

	// The very first id was evicted by the ring; replaying it now applies
	// again because the in-memory ring is best-effort dedup, not durable.
	require.True(t, c.shouldApplyAssetsEvent("evt_0"),
		"evicted id may reapply — ring dedup is best-effort against replays, not a durable guarantee")
}

// TestNormalizeAssetTypes pins the backward-compat fallback: empty slice
// means "both known kinds" so legacy pushes (which omit asset_types) still
// drive memory + skill refresh. Unknown kinds are silently dropped.
func TestNormalizeAssetTypes(t *testing.T) {
	both := normalizeAssetTypes(nil)
	require.True(t, both["memory"])
	require.True(t, both["skill"])
	require.False(t, both["kb"], "legacy events without asset_types refresh only pre-existing memory+skill kinds")

	both = normalizeAssetTypes([]string{})
	require.True(t, both["memory"])
	require.True(t, both["skill"])
	require.False(t, both["kb"])

	onlySkill := normalizeAssetTypes([]string{"skill"})
	require.False(t, onlySkill["memory"])
	require.True(t, onlySkill["skill"])

	mixed := normalizeAssetTypes([]string{"MEMORY", " skill ", "kb", "unknown"})
	require.True(t, mixed["memory"], "kind lookup is case-insensitive and trimmed")
	require.True(t, mixed["skill"])
	require.True(t, mixed["kb"])
	require.False(t, mixed["unknown"], "unknown kinds are dropped")
}