package im

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

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

	"github.com/tidbcloud/AgentTeam/internal/api"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"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/workspace"
	"github.com/tidbcloud/AgentTeam/internal/workspace/owner"
)

const (
	streamReplayLimit   = 500
	wsWriteTimeout      = 5 * time.Second
	wsHeartbeatInterval = 25 * time.Second
	wsHeartbeatTimeout  = 20 * time.Second

	bootstrapChannelsLimit = 100
	bootstrapListTimeout   = 5 * time.Second
)

// EventsWSHandler owns the WebSocket-only Workspace IM event stream route.
type EventsWSHandler struct {
	sessionAuth               api.SessionAuthenticator
	workspaceStore            api.WorkspaceResolver
	emailVerificationRequired bool
	streamEvents              StreamEventStore
	streamCursors             StreamCursorStore
	streamSubscriptions       StreamEventSubscriber
	visibleScopes             VisibleScopeAuthorizer
	logger                    *slog.Logger

	// Runtime/control-plane bootstrap collaborators. When any of these is nil,
	// the corresponding bootstrap block is skipped, allowing partial wiring in
	// tests and route-contract binaries that do not exercise the runtime path.
	runtimeBootstrap RuntimeBootstrapSource
	unreadRuntime    UnreadProjectionReader
	liveEmitter      *LiveEventEmitter
	channelLister    bootstrapChannelLister
}

// bootstrapChannelLister enumerates the channels and DMs a user can see, so the
// streaming bootstrap can replay wake-statuses for each channel scope.
type bootstrapChannelLister interface {
	ListChannelsForUser(ctx context.Context, userID string, workspaceID string) ([]message.ChannelSummary, error)
	ListDMChannelsForUserWithLimit(ctx context.Context, userID string, workspaceID string, limit int) ([]message.DMChannel, error)
}

type wsIdentity struct {
	user      identity.AuthenticatedUser
	workspace workspace.Summary
}

type wsControlFrame struct {
	Kind             string `json:"kind"`
	Type             string `json:"type,omitempty"`
	RequestID        string `json:"requestId,omitempty"`
	ClientID         string `json:"clientId,omitempty"`
	AckedEventSeq    int64  `json:"ackedEventSeq,omitempty"`
	AckedEventID     string `json:"ackedEventId,omitempty"`
	AfterEventSeq    int64  `json:"afterEventSeq,omitempty"`
	CursorGeneration string `json:"cursorGeneration,omitempty"`
	ScopeID          string `json:"scopeId,omitempty"`
}

type wsHelloFrame struct {
	Kind             string    `json:"kind"`
	Type             string    `json:"type"`
	WorkspaceID      string    `json:"workspaceId"`
	MemberID         string    `json:"memberId"`
	ViewerRole       string    `json:"viewerRole"`
	ClientID         string    `json:"clientId"`
	LatestEventSeq   int64     `json:"latestEventSeq"`
	ResumeFrom       int64     `json:"resumeFrom"`
	SnapshotRequired bool      `json:"snapshotRequired"`
	ServerTime       time.Time `json:"serverTime"`
}

type wsEventFrame struct {
	Kind  string              `json:"kind"`
	Event streamEventResponse `json:"event"`
}

type wsAckFrame struct {
	Kind           string `json:"kind"`
	ClientID       string `json:"clientId"`
	AckedEventSeq  int64  `json:"ackedEventSeq"`
	LatestEventSeq int64  `json:"latestEventSeq"`
}

type wsBootstrapCompleteFrame struct {
	Kind string `json:"kind"`
}

type wsVisibleScopeAckFrame struct {
	Kind      string `json:"kind"`
	RequestID string `json:"requestId,omitempty"`
	ScopeID   string `json:"scopeId"`
	Action    string `json:"action"`
}

type wsErrorFrame struct {
	Kind      string `json:"kind"`
	Code      string `json:"code"`
	Message   string `json:"message"`
	RequestID string `json:"requestId,omitempty"`
	ScopeID   string `json:"scopeId,omitempty"`
}

type streamEventResponse struct {
	WorkspaceID      string         `json:"workspaceId"`
	MemberID         string         `json:"memberId,omitempty"`
	EventSeq         int64          `json:"eventSeq"`
	EventID          string         `json:"eventId"`
	Type             string         `json:"type"`
	ScopeID          string         `json:"scopeId,omitempty"`
	ConversationID   string         `json:"conversationId,omitempty"`
	EntityID         string         `json:"entityId,omitempty"`
	EntitySeq        *int64         `json:"entitySeq,omitempty"`
	Payload          map[string]any `json:"payload,omitempty"`
	Patches          map[string]any `json:"patches,omitempty"`
	CursorGeneration string         `json:"cursorGeneration,omitempty"`
	CreatedAt        time.Time      `json:"createdAt"`
	ExpiresAt        *time.Time     `json:"expiresAt,omitempty"`
}

// Handle upgrades GET /api/v1/workspaces/:workspaceId/events/stream.
func (h *EventsWSHandler) Handle(ctx *gin.Context) {
	identity, ok := h.authenticate(ctx)
	if !ok {
		return
	}
	memberID := strings.TrimSpace(identity.workspace.MemberID)
	if memberID == "" {
		api.Error(ctx, http.StatusForbidden, "workspace member is required")
		return
	}
	clientID := strings.TrimSpace(ctx.Query("clientId"))
	if clientID == "" {
		api.Error(ctx, http.StatusBadRequest, "clientId is required")
		return
	}
	resumeFrom := parseNonNegativeInt64(ctx.Query("lastAckedEventSeq"))
	if cursorParam := strings.TrimSpace(ctx.Query("cursor")); cursorParam != "" {
		if parsed := parseNonNegativeInt64(cursorParam); parsed > resumeFrom {
			resumeFrom = parsed
		}
	}
	if h.streamCursors != nil {
		cursor, err := h.streamCursors.GetCursor(ctx.Request.Context(), identity.workspace.WorkspaceID, memberID, clientID)
		if err != nil {
			api.StoreError(ctx, err)
			return
		}
		// The durable member-local cursor is the source of truth when cursor
		// storage is available. Legacy browser-local seq values without a
		// matching server checkpoint may belong to another logged-in user and
		// must not skip replay. Workspace IM client IDs are scoped by the
		// frontend to the active server/workspace/user before they are allowed
		// to resume from local storage on first connect.
		if cursor.ClientID != "" {
			resumeFrom = cursor.AckedEventSeq
		} else if isWorkspaceIMStreamClientID(clientID, identity.workspace.WorkspaceID, identity.user.User.ID) {
			if resumeFrom < 0 {
				resumeFrom = 0
			}
		} else {
			resumeFrom = 0
		}
	}
	latestSeq, err := h.latestSeq(ctx.Request.Context(), identity.workspace.WorkspaceID, memberID)
	if err != nil {
		api.StoreError(ctx, err)
		return
	}
	if resumeFrom > latestSeq {
		resumeFrom = latestSeq
	}

	conn, err := websocket.Accept(ctx.Writer, ctx.Request, &websocket.AcceptOptions{
		InsecureSkipVerify: true,
		CompressionMode:    websocket.CompressionNoContextTakeover,
	})
	if err != nil {
		h.log().Warn("accept workspace im websocket", slog.Any("error", err))
		return
	}
	defer conn.Close(websocket.StatusNormalClosure, "")

	writer := &lockedWSWriter{conn: conn}
	var subscription *StreamEventSubscription
	if h.streamSubscriptions != nil {
		subscription = h.streamSubscriptions.SubscribeStreamEvents(ctx.Request.Context(), identity.workspace.WorkspaceID, memberID, identity.user.User.ID)
		defer subscription.Close()
	}

	if err := writer.write(ctx.Request.Context(), wsHelloFrame{
		Kind:           "hello",
		Type:           "hello",
		WorkspaceID:    identity.workspace.WorkspaceID,
		MemberID:       memberID,
		ViewerRole:     strings.TrimSpace(identity.workspace.WorkspaceRole),
		ClientID:       clientID,
		LatestEventSeq: latestSeq,
		ResumeFrom:     resumeFrom,
		ServerTime:     time.Now().UTC(),
	}); err != nil {
		return
	}
	replayedThrough, err := h.replay(ctx.Request.Context(), writer, identity.workspace.WorkspaceID, memberID, resumeFrom)
	if err != nil {
		h.log().Warn("replay workspace im websocket events", slog.Any("error", err))
		return
	}
	if err := h.bootstrap(ctx.Request.Context(), writer, subscription, identity.workspace.WorkspaceID, memberID, identity.user.User.ID); err != nil {
		h.log().Warn("bootstrap workspace im websocket runtime state", slog.Any("error", err))
		return
	}
	streamErr := h.runStream(ctx.Request.Context(), writer, subscription, identity.workspace.WorkspaceID, memberID, clientID, replayedThrough)
	if streamErr != nil && !isNormalWebSocketClose(streamErr) && !errors.Is(streamErr, context.Canceled) {
		h.log().Debug("workspace im websocket closed", slog.Any("error", streamErr))
	}
}

func isWorkspaceIMStreamClientID(clientID string, workspaceID string, userID string) bool {
	parts := strings.Split(clientID, ":")
	if len(parts) != 5 || parts[0] != "workspace-im" || parts[1] != "v1" {
		return false
	}
	if parts[2] != strings.TrimSpace(workspaceID) || parts[3] != strings.TrimSpace(userID) {
		return false
	}
	instanceID := parts[4]
	if len(instanceID) < 20 || len(instanceID) > 100 || !strings.HasPrefix(instanceID, "cid_") {
		return false
	}
	for _, r := range instanceID {
		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' || r == '-' {
			continue
		}
		return false
	}
	return true
}

func (h *EventsWSHandler) runStream(ctx context.Context, writer *lockedWSWriter, subscription *StreamEventSubscription, workspaceID string, memberID string, clientID string, replayedThrough int64) error {
	streamCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	errCh := make(chan error, 3)
	logging.Go(h.log(), func() {
		errCh <- h.readLoop(streamCtx, writer, subscription, workspaceID, memberID, clientID)
	})
	logging.Go(h.log(), func() {
		errCh <- runWebSocketHeartbeat(
			streamCtx,
			writer.conn,
			webSocketHeartbeatInitialDelay(wsHeartbeatInterval),
			wsHeartbeatInterval,
			wsHeartbeatTimeout,
		)
	})
	if subscription != nil {
		logging.Go(h.log(), func() {
			errCh <- h.liveLoop(streamCtx, writer, subscription, replayedThrough)
		})
	}
	err := <-errCh
	cancel()
	return err
}

// webSocketHeartbeatInitialDelay spreads newly established connections across
// one interval so a fleet-wide reconnect does not synchronize control frames.
func webSocketHeartbeatInitialDelay(interval time.Duration) time.Duration {
	return time.Duration(rand.Int64N(int64(interval)))
}

func runWebSocketHeartbeat(ctx context.Context, conn *websocket.Conn, initialDelay time.Duration, interval time.Duration, timeout time.Duration) error {
	timer := time.NewTimer(initialDelay)
	defer timer.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-timer.C:
			pingCtx, cancel := context.WithTimeout(ctx, timeout)
			err := conn.Ping(pingCtx)
			cancel()
			if err != nil {
				if ctxErr := ctx.Err(); ctxErr != nil {
					return ctxErr
				}
				return fmt.Errorf("workspace im websocket heartbeat: %w", err)
			}
			timer.Reset(interval)
		}
	}
}

func (h *EventsWSHandler) liveLoop(ctx context.Context, writer *lockedWSWriter, subscription *StreamEventSubscription, replayedThrough int64) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case event, ok := <-subscription.Events():
			if !ok {
				if dropped, backpressure := subscription.backpressureEvent(); backpressure {
					h.log().Warn("workspace im stream subscriber closed by backpressure",
						slog.String("workspace_id", dropped.WorkspaceID),
						slog.String("member_id", dropped.MemberID),
						slog.Int64("event_seq", dropped.EventSeq),
						slog.String("event_id", dropped.EventID),
					)
				}
				if err := ctx.Err(); err != nil {
					return err
				}
				return fmt.Errorf("workspace im stream subscription closed")
			}
			if event.EventSeq > 0 && event.EventSeq <= replayedThrough {
				continue
			}
			if err := writer.write(ctx, wsEventFrame{Kind: "event", Event: toStreamEventResponse(event)}); err != nil {
				return err
			}
		}
	}
}

func (h *EventsWSHandler) authenticate(ctx *gin.Context) (wsIdentity, bool) {
	if h.sessionAuth == nil || h.workspaceStore == nil {
		api.Error(ctx, http.StatusUnauthorized, "Invalid access token")
		return wsIdentity{}, false
	}
	token := strings.TrimSpace(ctx.Query("accessToken"))
	if token == "" {
		if bearer, ok := api.BearerToken(ctx); ok {
			token = bearer
		}
	}
	if token == "" {
		api.Error(ctx, http.StatusUnauthorized, "Missing access token")
		return wsIdentity{}, false
	}
	authenticated, err := h.sessionAuth.AuthenticateSession(ctx.Request.Context(), token)
	if err != nil {
		api.Error(ctx, http.StatusUnauthorized, "Invalid access token")
		return wsIdentity{}, false
	}
	if h.emailVerificationRequired && !authenticated.User.HasVerifiedIdentity() {
		api.ErrorWithCode(ctx, http.StatusForbidden, "Email or mobile not verified", "email_not_verified")
		return wsIdentity{}, false
	}
	workspaceID := strings.TrimSpace(ctx.Param("workspaceId"))
	if workspaceID == "" {
		api.Error(ctx, http.StatusBadRequest, "workspaceId is required")
		return wsIdentity{}, false
	}
	workspaceSummary, err := h.workspaceStore.GetWorkspaceForUser(ctx.Request.Context(), authenticated.User.ID, workspaceID)
	if err != nil {
		if errors.Is(err, owner.ErrOwnerMismatch) {
			api.ErrorWorkspaceOwnerMismatch(ctx)
			return wsIdentity{}, false
		}
		api.Error(ctx, http.StatusForbidden, "Forbidden")
		return wsIdentity{}, false
	}
	return wsIdentity{user: authenticated, workspace: workspaceSummary}, true
}

func (h *EventsWSHandler) latestSeq(ctx context.Context, workspaceID string, memberID string) (int64, error) {
	if h.streamEvents == nil {
		return 0, nil
	}
	return h.streamEvents.GetLatestSeq(ctx, workspaceID, memberID)
}

func (h *EventsWSHandler) replay(ctx context.Context, writer *lockedWSWriter, workspaceID string, memberID string, afterSeq int64) (int64, error) {
	if h.streamEvents == nil {
		return afterSeq, nil
	}
	replayedThrough := afterSeq
	for {
		events, err := h.streamEvents.ListEventsAfter(ctx, workspaceID, memberID, replayedThrough, streamReplayLimit)
		if err != nil {
			return replayedThrough, err
		}
		if len(events) == 0 {
			return replayedThrough, nil
		}
		for _, event := range events {
			if event.EventSeq <= replayedThrough {
				continue
			}
			replayedThrough = event.EventSeq
			if err := writer.write(ctx, wsEventFrame{Kind: "event", Event: toStreamEventResponse(event)}); err != nil {
				return replayedThrough, err
			}
		}
		if len(events) < streamReplayLimit {
			return replayedThrough, nil
		}
	}
}

// bootstrap emits live-only runtime.* events (machines/daemons/agents, unread
// projection, per-channel wake-status snapshots) directly to the new
// connection, then sends a bootstrap_complete control frame. Live events that
// arrive concurrently are buffered on the subscription and processed by the
// live agentteam after bootstrap_complete.
//
// For each joined channel and DM, bootstrap also auto-subscribes the
// connection to that scope's live-only fanout (runtime.task_*, wake-status,
// etc.) via SubscribeVisibleScopeEvents. This restores the Socket.IO
// "auto-join my channels on connect" semantic so channel-scoped runtime
// events reach the connection regardless of which page the user is currently
// viewing; per-scope subscribeVisibleScope frames from the client remain
// available for opt-in non-member scopes such as reply threads.
func (h *EventsWSHandler) bootstrap(ctx context.Context, writer *lockedWSWriter, subscription *StreamEventSubscription, workspaceID string, memberID string, userID string) error {
	emit := func(event storeim.StreamEvent) {
		if err := writer.write(ctx, wsEventFrame{Kind: "event", Event: toStreamEventResponse(event)}); err != nil {
			h.log().Debug("bootstrap write failed", slog.Any("error", err))
		}
	}

	if h.channelLister != nil && h.liveEmitter != nil {
		listCtx, cancel := context.WithTimeout(ctx, bootstrapListTimeout)
		channels, err := h.channelLister.ListChannelsForUser(listCtx, userID, workspaceID)
		if err != nil {
			cancel()
			return fmt.Errorf("list channels for bootstrap: %w", err)
		}
		dms, err := h.channelLister.ListDMChannelsForUserWithLimit(listCtx, userID, workspaceID, bootstrapChannelsLimit)
		cancel()
		if err != nil {
			return fmt.Errorf("list dm channels for bootstrap: %w", err)
		}
		seen := make(map[string]struct{}, len(channels)+len(dms))
		for _, channel := range channels {
			if !channel.Joined {
				continue
			}
			channelID := strings.TrimSpace(channel.ChannelID)
			if channelID == "" {
				continue
			}
			if _, dup := seen[channelID]; dup {
				continue
			}
			seen[channelID] = struct{}{}
			h.subscribeBootstrapScope(subscription, workspaceID, memberID, channelID)
			h.liveEmitter.ReplayWakeStatusesForChannel(workspaceID, channelID, emit)
		}
		for _, dm := range dms {
			dmID := strings.TrimSpace(dm.ID)
			if dmID == "" {
				continue
			}
			if _, dup := seen[dmID]; dup {
				continue
			}
			seen[dmID] = struct{}{}
			h.subscribeBootstrapScope(subscription, workspaceID, memberID, dmID)
			h.liveEmitter.ReplayWakeStatusesForChannel(workspaceID, dmID, emit)
		}
	}

	if h.runtimeBootstrap != nil && h.liveEmitter != nil {
		snapshot := h.runtimeBootstrap.WorkspaceRuntimeSnapshot(workspaceID)
		h.liveEmitter.ReplayWorkspaceRuntimeBootstrap(workspaceID, userID, snapshot, emit)
	}

	if h.unreadRuntime != nil && h.liveEmitter != nil {
		h.liveEmitter.ReplayUnreadProjectionState(workspaceID, memberID, h.unreadRuntime, emit)
	}

	return writer.write(ctx, wsBootstrapCompleteFrame{Kind: "bootstrap_complete"})
}

// subscribeBootstrapScope auto-joins the connection to a channel or DM scope
// during bootstrap so channel-scoped live-only events (runtime.task_*,
// agent_wake_status, etc.) reach the connection even when the user is viewing
// a non-channel surface such as /tasks/views/*. This mirrors the original
// Socket.IO behavior of joining every joined-channel room on connect. The
// scope membership is connection-local and cleaned up when the subscription
// closes; it does not create a durable scope_membership row.
func (h *EventsWSHandler) subscribeBootstrapScope(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string) {
	if subscription == nil {
		return
	}
	scopeSubscriber, ok := h.streamSubscriptions.(ImplicitVisibleScopeEventSubscriber)
	if !ok || scopeSubscriber == nil {
		return
	}
	scopeSubscriber.SubscribeImplicitVisibleScopeEvents(subscription, workspaceID, memberID, scopeID)
}

func (h *EventsWSHandler) readLoop(ctx context.Context, writer *lockedWSWriter, subscription *StreamEventSubscription, workspaceID string, memberID string, defaultClientID string) error {
	for {
		var frame wsControlFrame
		if err := wsjson.Read(ctx, writer.conn, &frame); err != nil {
			return err
		}
		kind := strings.TrimSpace(firstNonEmpty(frame.Kind, frame.Type))
		switch kind {
		case "ack":
			clientID := strings.TrimSpace(firstNonEmpty(frame.ClientID, defaultClientID))
			if clientID == "" || frame.AckedEventSeq < 0 {
				_ = writer.write(ctx, wsErrorFrame{Kind: "error", Code: "invalid_ack", Message: "invalid ack frame"})
				continue
			}
			latest, err := h.latestSeq(ctx, workspaceID, memberID)
			if err != nil {
				return err
			}
			if frame.AckedEventSeq > latest {
				_ = writer.write(ctx, wsErrorFrame{Kind: "error", Code: "invalid_ack", Message: "acked event seq is ahead of the stream"})
				continue
			}
			if err := h.validateAckedEvent(ctx, workspaceID, memberID, frame.AckedEventSeq, frame.AckedEventID); err != nil {
				_ = writer.write(ctx, wsErrorFrame{Kind: "error", Code: "invalid_ack", Message: err.Error()})
				continue
			}
			if updater, ok := h.streamCursors.(StreamCursorEventUpdater); ok {
				if err := updater.UpdateCursorWithEvent(ctx, workspaceID, memberID, clientID, frame.AckedEventSeq, frame.AckedEventID, frame.CursorGeneration); err != nil {
					return err
				}
			} else if h.streamCursors != nil {
				if err := h.streamCursors.UpdateCursor(ctx, workspaceID, memberID, clientID, frame.AckedEventSeq); err != nil {
					return err
				}
			}
			latest, err = h.latestSeq(ctx, workspaceID, memberID)
			if err != nil {
				return err
			}
			if err := writer.write(ctx, wsAckFrame{Kind: "ack.ok", ClientID: clientID, AckedEventSeq: frame.AckedEventSeq, LatestEventSeq: latest}); err != nil {
				return err
			}
		case "resume":
			if _, err := h.replay(ctx, writer, workspaceID, memberID, frame.AfterEventSeq); err != nil {
				return err
			}
		case "subscribeVisibleScope":
			if err := h.subscribeVisibleScope(ctx, writer, subscription, workspaceID, memberID, frame.RequestID, frame.ScopeID); err != nil {
				return err
			}
		case "unsubscribeVisibleScope":
			if err := h.unsubscribeVisibleScope(ctx, writer, subscription, workspaceID, memberID, frame.RequestID, frame.ScopeID); err != nil {
				return err
			}
		case "pong":
		default:
			_ = writer.write(ctx, wsErrorFrame{Kind: "error", Code: "unsupported_frame", Message: "unsupported control frame"})
		}
	}
}

func (h *EventsWSHandler) subscribeVisibleScope(ctx context.Context, writer *lockedWSWriter, subscription *StreamEventSubscription, workspaceID string, memberID string, requestID string, scopeID string) error {
	requestID = strings.TrimSpace(requestID)
	scopeID = strings.TrimSpace(scopeID)
	if scopeID == "" {
		_ = writer.write(ctx, visibleScopeErrorFrame("invalid_visible_scope", "scopeId is required", requestID, scopeID))
		return nil
	}
	visibleSubscriber, ok := h.streamSubscriptions.(VisibleScopeEventSubscriber)
	if subscription == nil || !ok {
		_ = writer.write(ctx, visibleScopeErrorFrame("visible_scope_unavailable", "visible-scope live fanout is unavailable", requestID, scopeID))
		return nil
	}
	if h.visibleScopes == nil {
		_ = writer.write(ctx, visibleScopeErrorFrame("visible_scope_unavailable", "visible-scope authorization is unavailable", requestID, scopeID))
		return nil
	}
	allowed, err := h.visibleScopes.CanReadVisibleScope(ctx, workspaceID, memberID, scopeID)
	if err != nil {
		return err
	}
	if !allowed {
		_ = writer.write(ctx, visibleScopeErrorFrame("unauthorized_visible_scope", "visible scope is not readable", requestID, scopeID))
		return nil
	}
	if !visibleSubscriber.SubscribeVisibleScopeEvents(subscription, workspaceID, memberID, scopeID) {
		_ = writer.write(ctx, visibleScopeErrorFrame("visible_scope_unavailable", "visible-scope subscription failed", requestID, scopeID))
		return nil
	}
	if requestID != "" {
		if err := writer.write(ctx, wsVisibleScopeAckFrame{Kind: "visible_scope.ok", RequestID: requestID, ScopeID: scopeID, Action: "subscribe"}); err != nil {
			return err
		}
	}
	if h.liveEmitter != nil {
		h.liveEmitter.ReplayWakeStatusesForChannel(workspaceID, scopeID, func(event storeim.StreamEvent) {
			if err := writer.write(ctx, wsEventFrame{Kind: "event", Event: toStreamEventResponse(event)}); err != nil {
				h.log().Debug("visible-scope wake-status replay write failed", slog.Any("error", err), slog.String("scope_id", scopeID))
			}
		})
	}
	return nil
}

func (h *EventsWSHandler) unsubscribeVisibleScope(ctx context.Context, writer *lockedWSWriter, subscription *StreamEventSubscription, workspaceID string, memberID string, requestID string, scopeID string) error {
	requestID = strings.TrimSpace(requestID)
	scopeID = strings.TrimSpace(scopeID)
	if scopeID == "" {
		_ = writer.write(ctx, visibleScopeErrorFrame("invalid_visible_scope", "scopeId is required", requestID, scopeID))
		return nil
	}
	visibleSubscriber, ok := h.streamSubscriptions.(VisibleScopeEventSubscriber)
	if subscription == nil || !ok {
		_ = writer.write(ctx, visibleScopeErrorFrame("visible_scope_unavailable", "visible-scope live fanout is unavailable", requestID, scopeID))
		return nil
	}
	visibleSubscriber.UnsubscribeVisibleScopeEvents(subscription, workspaceID, memberID, scopeID)
	if requestID == "" {
		return nil
	}
	return writer.write(ctx, wsVisibleScopeAckFrame{Kind: "visible_scope.ok", RequestID: requestID, ScopeID: scopeID, Action: "unsubscribe"})
}

func visibleScopeErrorFrame(code string, message string, requestID string, scopeID string) wsErrorFrame {
	return wsErrorFrame{
		Kind:      "error",
		Code:      code,
		Message:   message,
		RequestID: strings.TrimSpace(requestID),
		ScopeID:   strings.TrimSpace(scopeID),
	}
}

func (h *EventsWSHandler) validateAckedEvent(ctx context.Context, workspaceID string, memberID string, ackedSeq int64, ackedEventID string) error {
	ackedEventID = strings.TrimSpace(ackedEventID)
	if ackedSeq == 0 {
		if ackedEventID != "" {
			return fmt.Errorf("acked event id is not valid for seq 0")
		}
		return nil
	}
	if ackedEventID == "" || h.streamEvents == nil {
		return nil
	}
	events, err := h.streamEvents.ListEventsAfter(ctx, workspaceID, memberID, ackedSeq-1, 1)
	if err != nil {
		return err
	}
	if len(events) == 0 || events[0].EventSeq != ackedSeq || events[0].EventID != ackedEventID {
		return fmt.Errorf("acked event id does not match acked event seq")
	}
	return nil
}

type lockedWSWriter struct {
	conn *websocket.Conn
	mu   sync.Mutex
}

func (writer *lockedWSWriter) write(ctx context.Context, value any) error {
	if writer == nil || writer.conn == nil {
		return fmt.Errorf("workspace im websocket writer is not initialized")
	}
	writer.mu.Lock()
	defer writer.mu.Unlock()
	return writeWS(ctx, writer.conn, value)
}

func writeWS(ctx context.Context, conn *websocket.Conn, value any) error {
	writeCtx, cancel := context.WithTimeout(ctx, wsWriteTimeout)
	defer cancel()
	return wsjson.Write(writeCtx, conn, value)
}

func toStreamEventResponse(event storeim.StreamEvent) streamEventResponse {
	return streamEventResponse{
		WorkspaceID:      event.WorkspaceID,
		MemberID:         event.MemberID,
		EventSeq:         event.EventSeq,
		EventID:          event.EventID,
		Type:             event.Type,
		ScopeID:          event.ScopeID,
		ConversationID:   event.ConversationID,
		EntityID:         event.EntityID,
		EntitySeq:        event.EntitySeq,
		Payload:          event.Payload,
		Patches:          event.Patches,
		CursorGeneration: event.CursorGeneration,
		CreatedAt:        event.CreatedAt,
		ExpiresAt:        event.ExpiresAt,
	}
}

func parseNonNegativeInt64(raw string) int64 {
	trimmed := strings.TrimSpace(raw)
	if trimmed == "" {
		return 0
	}
	value, err := strconv.ParseInt(trimmed, 10, 64)
	if err != nil || value < 0 {
		return 0
	}
	return value
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}

func isNormalWebSocketClose(err error) bool {
	return websocket.CloseStatus(err) == websocket.StatusNormalClosure || websocket.CloseStatus(err) == websocket.StatusGoingAway
}

func (h *EventsWSHandler) log() *slog.Logger {
	if h != nil && h.logger != nil {
		return h.logger
	}
	return slog.Default()
}