package im

import (
	"context"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/require"
	"nhooyr.io/websocket"
	"nhooyr.io/websocket/wsjson"

	"github.com/tidbcloud/AgentTeam/internal/store/identity"
	storeim "github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
	"github.com/tidbcloud/AgentTeam/internal/store/workspace"
)

func TestRunWebSocketHeartbeatKeepsIdleConnectionAlive(t *testing.T) {
	heartbeatCtx, stopHeartbeat := context.WithCancel(context.Background())
	heartbeatErr := make(chan error, 1)
	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		conn, err := websocket.Accept(writer, request, nil)
		if err != nil {
			heartbeatErr <- fmt.Errorf("accept websocket: %w", err)
			return
		}
		defer conn.CloseNow()
		go func() {
			for {
				if _, _, err := conn.Read(heartbeatCtx); err != nil {
					return
				}
			}
		}()
		heartbeatErr <- runWebSocketHeartbeat(heartbeatCtx, conn, 5*time.Millisecond, 20*time.Millisecond, 500*time.Millisecond)
	}))
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http")
	conn, _, err := websocket.Dial(context.Background(), wsURL, nil)
	require.NoError(t, err)
	defer conn.CloseNow()

	readDone := make(chan struct{})
	go func() {
		defer close(readDone)
		_, _, _ = conn.Read(context.Background())
	}()

	select {
	case err := <-heartbeatErr:
		require.Failf(t, "heartbeat stopped", "idle connection closed before cancellation: %v", err)
	case <-time.After(650 * time.Millisecond):
	}

	stopHeartbeat()
	select {
	case err := <-heartbeatErr:
		require.ErrorIs(t, err, context.Canceled)
	case <-time.After(time.Second):
		require.Fail(t, "heartbeat did not stop after cancellation")
	}
	select {
	case <-readDone:
	case <-time.After(time.Second):
		require.Fail(t, "websocket reader did not stop after the server closed the connection")
	}
}

func TestWebSocketHeartbeatInitialDelayStaysWithinInterval(t *testing.T) {
	const interval = 25 * time.Second
	for range 100 {
		delay := webSocketHeartbeatInitialDelay(interval)
		require.GreaterOrEqual(t, delay, time.Duration(0))
		require.Less(t, delay, interval)
	}
}

func TestEventsWSHandlerNegotiatesNoContextTakeoverCompression(t *testing.T) {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:    fakeWSAuth{},
		WorkspaceStore: fakeWSWorkspaceResolver{},
		StreamEvents:   &fakeWSStreamStore{},
		StreamCursors:  &fakeWSCursorStore{},
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, resp, err := websocket.Dial(ctx, wsURL, &websocket.DialOptions{
		CompressionMode: websocket.CompressionNoContextTakeover,
	})
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")
	require.NotNil(t, resp)
	extensions := resp.Header.Get("Sec-WebSocket-Extensions")
	require.Contains(t, extensions, "permessage-deflate")
	require.Contains(t, extensions, "client_no_context_takeover")
	require.Contains(t, extensions, "server_no_context_takeover")
}

func TestEventsWSHandlerPushesLiveHubEvents(t *testing.T) {
	gin.SetMode(gin.TestMode)
	hub := NewStreamEventHub()
	streams := &fakeWSStreamStore{}
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:         fakeWSAuth{},
		WorkspaceStore:      fakeWSWorkspaceResolver{},
		StreamEvents:        streams,
		StreamCursors:       &fakeWSCursorStore{},
		StreamSubscriptions: hub,
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	require.Equal(t, "hello", hello.Kind)
	require.Equal(t, "owner", hello.ViewerRole)
	readBootstrapComplete(t, ctx, conn)

	hub.PublishStreamEvents([]storeim.StreamEvent{{
		WorkspaceID: "wsp_1",
		MemberID:    "mem_1",
		EventSeq:    1,
		EventID:     "evt_live_1",
		Type:        "message.created",
		ScopeID:     "chn_1",
		Payload:     map[string]any{"messageId": "msg_1"},
		CreatedAt:   time.Now().UTC(),
	}})

	var frame wsEventFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "event", frame.Kind)
	require.Equal(t, int64(1), frame.Event.EventSeq)
	require.Equal(t, "evt_live_1", frame.Event.EventID)
	require.Equal(t, "message.created", frame.Event.Type)
}

func TestEventsWSHandlerBootstrapAutoSubscribesJoinedChannelScopes(t *testing.T) {
	gin.SetMode(gin.TestMode)
	hub := NewStreamEventHub()
	streams := &fakeWSStreamStore{}
	channels := fakeBootstrapChannelStore{channels: []message.ChannelSummary{
		{ChannelID: "chn_joined", Joined: true},
		{ChannelID: "chn_not_joined", Joined: false},
	}}
	dms := fakeBootstrapDMStore{channels: []message.DMChannel{{ID: "dm_direct"}}}
	liveEmitter := NewLiveEventEmitter(hub, nil, nil)
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:         fakeWSAuth{},
		WorkspaceStore:      fakeWSWorkspaceResolver{},
		StreamEvents:        streams,
		StreamCursors:       &fakeWSCursorStore{},
		StreamSubscriptions: hub,
		VisibleScopes: fakeVisibleScopeAuthorizer{allowed: map[string]bool{
			"chn_joined": true,
		}},
		LiveEmitter: liveEmitter,
		Channels:    channels,
		DMs:         dms,
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	readBootstrapComplete(t, ctx, conn)

	require.Eventually(t, func() bool {
		return visibleScopeSubscriberCount(hub, "wsp_1", "chn_joined") == 1 &&
			visibleScopeSubscriberCount(hub, "wsp_1", "dm_direct") == 1
	}, time.Second, 10*time.Millisecond)
	require.Equal(t, 0, visibleScopeSubscriberCount(hub, "wsp_1", "chn_not_joined"),
		"bootstrap must not auto-subscribe scopes the user has not joined")

	// A route-level explicit lease can overlap the implicit bootstrap source.
	// Releasing that lease must not remove the joined-channel subscription.
	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{
		Kind: "subscribeVisibleScope", RequestID: "req_sub_joined", ScopeID: "chn_joined",
	}))
	readVisibleScopeAck(t, ctx, conn, "req_sub_joined", "chn_joined", "subscribe")
	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{
		Kind: "unsubscribeVisibleScope", RequestID: "req_unsub_joined", ScopeID: "chn_joined",
	}))
	readVisibleScopeAck(t, ctx, conn, "req_unsub_joined", "chn_joined", "unsubscribe")
	require.Equal(t, 1, visibleScopeSubscriberCount(hub, "wsp_1", "chn_joined"),
		"explicit release must preserve the implicit bootstrap source")

	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_joined", storeim.StreamEvent{
		Type:    "runtime.task_created",
		Payload: map[string]any{"taskId": "tsk_1"},
	})

	var frame wsEventFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "event", frame.Kind)
	require.Equal(t, "runtime.task_created", frame.Event.Type)
	require.Equal(t, "chn_joined", frame.Event.ScopeID)
	require.Equal(t, "mem_1", frame.Event.MemberID)
	require.Equal(t, int64(0), frame.Event.EventSeq)
}

func TestEventsWSHandlerExplicitVisibleTeamScopePushesTeamWorkflowUpdate(t *testing.T) {
	gin.SetMode(gin.TestMode)
	hub := NewStreamEventHub()
	streams := &fakeWSStreamStore{}
	liveEmitter := NewLiveEventEmitter(hub, nil, nil)
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:         fakeWSAuth{},
		WorkspaceStore:      fakeWSWorkspaceResolver{},
		StreamEvents:        streams,
		StreamCursors:       &fakeWSCursorStore{},
		StreamSubscriptions: hub,
		VisibleScopes: fakeVisibleScopeAuthorizer{allowed: map[string]bool{
			"chn_team": true,
		}},
		LiveEmitter: liveEmitter,
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	readBootstrapComplete(t, ctx, conn)
	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{
		Kind: "subscribeVisibleScope", RequestID: "req_team", ScopeID: "chn_team",
	}))
	readVisibleScopeAck(t, ctx, conn, "req_team", "chn_team", "subscribe")
	require.Eventually(t, func() bool {
		return visibleScopeSubscriberCount(hub, "wsp_1", "chn_team") == 1
	}, time.Second, 10*time.Millisecond)

	liveEmitter.EmitTeamWorkflowsUpdated(
		"wsp_1", "chn_team", "team_1", "twf_1", 3,
	)

	var frame wsEventFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "event", frame.Kind)
	require.Equal(t, liveEventTypeTeamWorkflowsUpdated, frame.Event.Type)
	require.Equal(t, "chn_team", frame.Event.ScopeID)
	require.Equal(t, "mem_1", frame.Event.MemberID)
	require.Equal(t, int64(0), frame.Event.EventSeq)
	require.Equal(t, "wsp_1", frame.Event.Payload["workspaceId"])
	require.Equal(t, "chn_team", frame.Event.Payload["sourceScopeId"])
	require.Equal(t, "team_1", frame.Event.Payload["teamId"])
	require.Equal(t, "twf_1", frame.Event.Payload["teamWorkflowId"])
	require.Equal(t, float64(3), frame.Event.Payload["revision"])
}

func TestEventsWSHandlerSubscribeVisibleScopePushesThreadEventForParentReadableNonMember(t *testing.T) {
	gin.SetMode(gin.TestMode)
	hub := NewStreamEventHub()
	streams := &fakeWSStreamStore{}
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:         fakeWSAuth{},
		WorkspaceStore:      fakeWSWorkspaceResolver{},
		StreamEvents:        streams,
		StreamCursors:       &fakeWSCursorStore{},
		StreamSubscriptions: hub,
		VisibleScopes:       fakeVisibleScopeAuthorizer{allowed: map[string]bool{"thr_1": true}},
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	require.Equal(t, "hello", hello.Kind)
	readBootstrapComplete(t, ctx, conn)

	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{Kind: "subscribeVisibleScope", RequestID: "req_thread", ScopeID: "thr_1"}))
	readVisibleScopeAck(t, ctx, conn, "req_thread", "thr_1", "subscribe")
	require.Eventually(t, func() bool {
		return visibleScopeSubscriberCount(hub, "wsp_1", "thr_1") == 1
	}, time.Second, 10*time.Millisecond)

	events := []storeim.StreamEvent{{
		WorkspaceID: "wsp_1",
		MemberID:    "mem_thread_member",
		EventSeq:    12,
		EventID:     "evt_thread_1",
		Type:        "message.created",
		ScopeID:     "thr_1",
		EntityID:    "msg_1",
		Payload:     map[string]any{"messageId": "msg_1"},
		CreatedAt:   time.Now().UTC(),
	}}
	hub.PublishStreamEvents(events)
	hub.PublishVisibleScopeEvents(events)

	var frame wsEventFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "event", frame.Kind)
	require.Equal(t, "message.created", frame.Event.Type)
	require.Equal(t, "thr_1", frame.Event.ScopeID)
	require.Equal(t, "mem_1", frame.Event.MemberID)
	require.Equal(t, int64(0), frame.Event.EventSeq)
	require.Empty(t, frame.Event.EventID)
	require.Equal(t, map[string]any{"messageId": "msg_1"}, frame.Event.Payload)
}

func TestEventsWSHandlerSubscribeVisibleScopeReplaysWakeStatusSnapshot(t *testing.T) {
	gin.SetMode(gin.TestMode)
	hub := NewStreamEventHub()
	streams := &fakeWSStreamStore{}
	liveEmitter := NewLiveEventEmitter(hub, nil, nil)
	liveEmitter.storeWakeStatusSnapshot("wsp_1", liveWakeStatusSnapshot{
		agentID:        "agt_1",
		agentName:      "technical-agent",
		displayName:    "Visible Agent",
		channelID:      "thr_1",
		messageID:      "msg_1",
		wakeToken:      "wake_1",
		target:         "#quick-start:msg_1",
		status:         "received",
		replyMessageID: "msg_reply_1",
		timestamp:      time.Now().UTC().Format(time.RFC3339Nano),
	})
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:         fakeWSAuth{},
		WorkspaceStore:      fakeWSWorkspaceResolver{},
		StreamEvents:        streams,
		StreamCursors:       &fakeWSCursorStore{},
		StreamSubscriptions: hub,
		VisibleScopes:       fakeVisibleScopeAuthorizer{allowed: map[string]bool{"thr_1": true}},
		LiveEmitter:         liveEmitter,
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	require.Equal(t, "hello", hello.Kind)
	readBootstrapComplete(t, ctx, conn)

	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{Kind: "subscribeVisibleScope", RequestID: "req_wake", ScopeID: "thr_1"}))
	readVisibleScopeAck(t, ctx, conn, "req_wake", "thr_1", "subscribe")

	var frame wsEventFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "event", frame.Kind)
	require.Equal(t, "runtime.agent_wake_status", frame.Event.Type)
	require.Equal(t, "thr_1", frame.Event.ScopeID)
	require.Equal(t, "agt_1", frame.Event.Payload["agentId"])
	require.Equal(t, "technical-agent", frame.Event.Payload["agentName"])
	require.Equal(t, "Visible Agent", frame.Event.Payload["agentDisplayName"])
	require.Equal(t, "msg_1", frame.Event.Payload["messageId"])
	require.Equal(t, "received", frame.Event.Payload["status"])
}

func TestEventsWSHandlerRejectsUnauthorizedVisibleScopeSubscription(t *testing.T) {
	gin.SetMode(gin.TestMode)
	hub := NewStreamEventHub()
	streams := &fakeWSStreamStore{}
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:         fakeWSAuth{},
		WorkspaceStore:      fakeWSWorkspaceResolver{},
		StreamEvents:        streams,
		StreamCursors:       &fakeWSCursorStore{},
		StreamSubscriptions: hub,
		VisibleScopes:       fakeVisibleScopeAuthorizer{allowed: map[string]bool{"thr_allowed": true}},
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	require.Equal(t, "hello", hello.Kind)
	readBootstrapComplete(t, ctx, conn)

	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{Kind: "subscribeVisibleScope", RequestID: "req_denied", ScopeID: "thr_denied"}))
	var frame wsErrorFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "error", frame.Kind)
	require.Equal(t, "unauthorized_visible_scope", frame.Code)
	require.Equal(t, "req_denied", frame.RequestID)
	require.Equal(t, "thr_denied", frame.ScopeID)
	require.Equal(t, 0, visibleScopeSubscriberCount(hub, "wsp_1", "thr_denied"))
}

func TestEventsWSHandlerRejectsFutureAck(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streams := &fakeWSStreamStore{events: []storeim.StreamEvent{{
		WorkspaceID: "wsp_1",
		MemberID:    "mem_1",
		EventSeq:    1,
		EventID:     "evt_1",
		Type:        "message.created",
		CreatedAt:   time.Now().UTC(),
	}}}
	cursors := &fakeWSCursorStore{cursor: storeim.StreamCursor{
		WorkspaceID:      "wsp_1",
		MemberID:         "mem_1",
		ClientID:         "web",
		AckedEventSeq:    1,
		AckedEventID:     "evt_1",
		CursorGeneration: "gen_1",
	}}
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:    fakeWSAuth{},
		WorkspaceStore: fakeWSWorkspaceResolver{},
		StreamEvents:   streams,
		StreamCursors:  cursors,
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web&lastAckedEventSeq=999"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	require.Equal(t, int64(1), hello.ResumeFrom)
	readBootstrapComplete(t, ctx, conn)

	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{Kind: "ack", ClientID: "web", AckedEventSeq: 2, AckedEventID: "evt_2"}))
	var frame wsErrorFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "error", frame.Kind)
	require.Equal(t, "invalid_ack", frame.Code)
	require.Equal(t, 0, cursors.updateCount())
}

func TestEventsWSHandlerRejectsAckEventIDMismatch(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streams := &fakeWSStreamStore{events: []storeim.StreamEvent{{
		WorkspaceID: "wsp_1",
		MemberID:    "mem_1",
		EventSeq:    1,
		EventID:     "evt_1",
		Type:        "message.created",
		CreatedAt:   time.Now().UTC(),
	}}}
	cursors := &fakeWSCursorStore{}
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:    fakeWSAuth{},
		WorkspaceStore: fakeWSWorkspaceResolver{},
		StreamEvents:   streams,
		StreamCursors:  cursors,
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	require.Equal(t, int64(0), hello.ResumeFrom)
	var eventFrame wsEventFrame
	require.NoError(t, wsjson.Read(ctx, conn, &eventFrame))
	require.Equal(t, int64(1), eventFrame.Event.EventSeq)
	readBootstrapComplete(t, ctx, conn)

	require.NoError(t, wsjson.Write(ctx, conn, wsControlFrame{Kind: "ack", ClientID: "web", AckedEventSeq: 1, AckedEventID: "evt_wrong"}))
	var frame wsErrorFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "error", frame.Kind)
	require.Equal(t, "invalid_ack", frame.Code)
	require.Equal(t, 0, cursors.updateCount())
}

func TestEventsWSHandlerReplaysBeyondSinglePage(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streams := &fakeWSStreamStore{events: make([]storeim.StreamEvent, 0, streamReplayLimit+5)}
	for seq := int64(1); seq <= streamReplayLimit+5; seq++ {
		streams.events = append(streams.events, storeim.StreamEvent{
			WorkspaceID: "wsp_1",
			MemberID:    "mem_1",
			EventSeq:    seq,
			EventID:     "evt_replay_" + time.Unix(seq, 0).UTC().Format("150405"),
			Type:        "message.created",
			CreatedAt:   time.Now().UTC(),
		})
	}
	router := gin.New()
	RegisterWebSocketRoutes(router, Dependencies{
		SessionAuth:    fakeWSAuth{},
		WorkspaceStore: fakeWSWorkspaceResolver{},
		StreamEvents:   streams,
	})
	server := httptest.NewServer(router)
	defer server.Close()

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/wsp_1/events/stream?accessToken=token&clientId=web"
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, wsURL, nil)
	require.NoError(t, err)
	defer conn.Close(websocket.StatusNormalClosure, "")

	var hello wsHelloFrame
	require.NoError(t, wsjson.Read(ctx, conn, &hello))
	require.Equal(t, int64(streamReplayLimit+5), hello.LatestEventSeq)

	var lastSeq int64
	for i := 0; i < streamReplayLimit+5; i++ {
		var frame wsEventFrame
		require.NoError(t, wsjson.Read(ctx, conn, &frame))
		require.Equal(t, "event", frame.Kind)
		require.Greater(t, frame.Event.EventSeq, lastSeq)
		lastSeq = frame.Event.EventSeq
	}
	require.Equal(t, int64(streamReplayLimit+5), lastSeq)
	readBootstrapComplete(t, ctx, conn)
}

type fakeWSAuth struct{}

func (fakeWSAuth) AuthenticateSession(context.Context, string) (identity.AuthenticatedUser, error) {
	now := time.Now().UTC()
	return identity.AuthenticatedUser{User: model.User{ID: "usr_1", Email: "user@example.com", EmailVerifiedAt: &now}}, nil
}

type fakeWSWorkspaceResolver struct{}

func (fakeWSWorkspaceResolver) GetWorkspaceForUser(context.Context, string, string) (workspace.Summary, error) {
	return workspace.Summary{
		WorkspaceID:   "wsp_1",
		MemberID:      "mem_1",
		MemberUserID:  "usr_1",
		WorkspaceRole: "owner",
	}, nil
}

type fakeWSStreamStore struct {
	mu     sync.Mutex
	events []storeim.StreamEvent
}

func (store *fakeWSStreamStore) AppendEvents(context.Context, string, string, []storeim.StreamEvent) error {
	return nil
}

func (store *fakeWSStreamStore) ListEventsAfter(_ context.Context, workspaceID string, memberID string, afterSeq int64, limit int) ([]storeim.StreamEvent, error) {
	store.mu.Lock()
	defer store.mu.Unlock()
	var events []storeim.StreamEvent
	for _, event := range store.events {
		if event.WorkspaceID == workspaceID && event.MemberID == memberID && event.EventSeq > afterSeq {
			events = append(events, event)
			if limit > 0 && len(events) >= limit {
				break
			}
		}
	}
	return events, nil
}

func (store *fakeWSStreamStore) GetLatestSeq(_ context.Context, workspaceID string, memberID string) (int64, error) {
	store.mu.Lock()
	defer store.mu.Unlock()
	var latest int64
	for _, event := range store.events {
		if event.WorkspaceID == workspaceID && event.MemberID == memberID && event.EventSeq > latest {
			latest = event.EventSeq
		}
	}
	return latest, nil
}

type fakeWSCursorStore struct {
	mu      sync.Mutex
	cursor  storeim.StreamCursor
	cursors map[string]storeim.StreamCursor
	updates []storeim.StreamCursor
}

func (store *fakeWSCursorStore) UpdateCursor(_ context.Context, workspaceID string, memberID string, clientID string, ackedSeq int64) error {
	return store.UpdateCursorWithEvent(context.Background(), workspaceID, memberID, clientID, ackedSeq, "", "")
}

func (store *fakeWSCursorStore) UpdateCursorWithEvent(_ context.Context, workspaceID string, memberID string, clientID string, ackedSeq int64, ackedEventID string, cursorGeneration string) error {
	store.mu.Lock()
	defer store.mu.Unlock()
	existing, ok := store.cursors[clientID]
	if !ok && store.cursor.ClientID == clientID {
		existing = store.cursor
		ok = true
	}
	if ok && ackedSeq < existing.AckedEventSeq {
		return nil
	}
	cursor := storeim.StreamCursor{
		WorkspaceID:      workspaceID,
		MemberID:         memberID,
		ClientID:         clientID,
		AckedEventSeq:    ackedSeq,
		AckedEventID:     ackedEventID,
		CursorGeneration: cursorGeneration,
	}
	store.cursor = cursor
	if store.cursors == nil {
		store.cursors = make(map[string]storeim.StreamCursor)
	}
	store.cursors[clientID] = cursor
	store.updates = append(store.updates, cursor)
	return nil
}

func (store *fakeWSCursorStore) GetCursor(_ context.Context, workspaceID string, memberID string, clientID string) (storeim.StreamCursor, error) {
	store.mu.Lock()
	defer store.mu.Unlock()
	if cursor, ok := store.cursors[clientID]; ok {
		return cursor, nil
	}
	if store.cursor.ClientID == clientID {
		return store.cursor, nil
	}
	return storeim.StreamCursor{WorkspaceID: workspaceID, MemberID: memberID, ClientID: ""}, nil
}

func (store *fakeWSCursorStore) updateCount() int {
	store.mu.Lock()
	defer store.mu.Unlock()
	return len(store.updates)
}

type fakeVisibleScopeAuthorizer struct {
	allowed map[string]bool
}

func (authorizer fakeVisibleScopeAuthorizer) CanReadVisibleScope(_ context.Context, _ string, _ string, scopeID string) (bool, error) {
	return authorizer.allowed[strings.TrimSpace(scopeID)], nil
}

func visibleScopeSubscriberCount(hub *StreamEventHub, workspaceID string, scopeID string) int {
	hub.mu.Lock()
	defer hub.mu.Unlock()
	return len(hub.scopeSubscribers[streamHubScopeKey{workspaceID: strings.TrimSpace(workspaceID), scopeID: strings.TrimSpace(scopeID)}])
}

// readBootstrapComplete consumes the bootstrap_complete frame the handler emits
// after hello + runtime/unread/wake-status bootstrap. Tests that do not wire
// the bootstrap collaborators still receive the frame because the handler
// always finalizes bootstrap with it.
func readBootstrapComplete(t *testing.T, ctx context.Context, conn *websocket.Conn) {
	t.Helper()
	var frame wsBootstrapCompleteFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "bootstrap_complete", frame.Kind)
}

func readVisibleScopeAck(t *testing.T, ctx context.Context, conn *websocket.Conn, requestID string, scopeID string, action string) {
	t.Helper()
	var frame wsVisibleScopeAckFrame
	require.NoError(t, wsjson.Read(ctx, conn, &frame))
	require.Equal(t, "visible_scope.ok", frame.Kind)
	require.Equal(t, requestID, frame.RequestID)
	require.Equal(t, scopeID, frame.ScopeID)
	require.Equal(t, action, frame.Action)
}

type fakeBootstrapChannelStore struct {
	channels []message.ChannelSummary
}

func (store fakeBootstrapChannelStore) ListChannelsForUser(context.Context, string, string) ([]message.ChannelSummary, error) {
	return store.channels, nil
}

type fakeBootstrapDMStore struct {
	channels []message.DMChannel
}

func (store fakeBootstrapDMStore) ListDMChannelsForUserWithLimit(context.Context, string, string, int) ([]message.DMChannel, error) {
	return store.channels, nil
}

func (store fakeBootstrapDMStore) GetDMChannelForUser(context.Context, string, string, string) (message.DMChannel, error) {
	return message.DMChannel{}, nil
}

func (store fakeBootstrapDMStore) OpenDMChannelWithAgent(context.Context, string, string, string) (message.DMChannel, error) {
	return message.DMChannel{}, nil
}

func (store fakeBootstrapDMStore) OpenDMChannelWithUser(context.Context, string, string, string) (message.DMChannel, error) {
	return message.DMChannel{}, nil
}