package im

import (
	"context"
	"strings"
	"sync"
	"sync/atomic"

	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	storeim "github.com/tidbcloud/AgentTeam/internal/store/im"
)

const streamHubSubscriptionBuffer = 128

// StreamEventPublisher publishes newly persisted IM stream events to live WS
// subscribers. Persistence stays owned by StreamEventWriter; the hub is only a
// best-effort live delivery side channel.
type StreamEventPublisher interface {
	PublishStreamEvents(events []storeim.StreamEvent)
}

// VisibleScopeEventPublisher publishes scope-write events to connections that
// have explicitly and temporarily joined one readable visible scope.
type VisibleScopeEventPublisher interface {
	PublishVisibleScopeEvents(events []storeim.StreamEvent)
}

// LiveOnlyEventPublisher publishes runtime/control-plane events (agent status,
// machine status, unread updates, wake status, task changes, etc.) as
// non-durable, non-replayable live notifications. eventSeq stays 0; recovery
// after reconnect is via REST bootstrap, never via im_stream_events.
type LiveOnlyEventPublisher interface {
	PublishLiveOnlyEventsByMember(events []storeim.StreamEvent)
	PublishWorkspaceLiveOnlyEvent(event storeim.StreamEvent)
	PublishWorkspaceLiveOnlyEventExceptUser(event storeim.StreamEvent, exceptUserID string)
	PublishUserLiveOnlyEvent(workspaceID string, userID string, event storeim.StreamEvent)
	PublishScopeLiveOnlyEvent(workspaceID string, scopeID string, event storeim.StreamEvent)
}

// UserScopeMembershipSync keeps a user's existing streaming connections in
// sync with channel/DM membership changes. The streaming model normally
// subscribes a connection to its joined scopes only during bootstrap, so
// without these hooks a user who is added to (or removed from) a channel
// after connecting would not receive (or stop receiving) channel-scoped
// live-only events on existing connections until they reconnect. Membership
// mutation handlers call these methods to mirror the old Socket.IO
// SocketsJoin/SocketsLeave-by-user-room behavior.
type UserScopeMembershipSync interface {
	SubscribeScopeForUser(workspaceID string, userID string, scopeID string)
	UnsubscribeScopeForUser(workspaceID string, userID string, scopeID string)
}

// StreamEventSubscriber lets WebSocket handlers subscribe to one member-local
// workspace stream.
type StreamEventSubscriber interface {
	SubscribeStreamEvents(ctx context.Context, workspaceID string, memberID string, userID string) *StreamEventSubscription
}

// VisibleScopeEventSubscriber extends a member-local stream subscription with
// connection-local visible scopes. These subscriptions are intentionally live
// only: they do not create durable scope_memberships or replayable stream rows.
type VisibleScopeEventSubscriber interface {
	SubscribeVisibleScopeEvents(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string) bool
	UnsubscribeVisibleScopeEvents(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string)
}

// ImplicitVisibleScopeEventSubscriber records connection-local scope access
// that comes from bootstrap or durable membership rather than an explicit
// client lease. Explicit unsubscribe frames must not remove this source.
type ImplicitVisibleScopeEventSubscriber interface {
	SubscribeImplicitVisibleScopeEvents(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string) bool
}

type streamHubKey struct {
	workspaceID string
	memberID    string
}

type streamHubScopeKey struct {
	workspaceID string
	scopeID     string
}

type streamHubUserKey struct {
	workspaceID string
	userID      string
}

// StreamEventHub is the in-process live push hub for Workspace IM streams.
// Recoverability still comes from im_stream_events replay; if a subscriber is
// too slow, it is closed so the browser reconnects and resumes from storage.
type StreamEventHub struct {
	mu               sync.Mutex
	nextID           uint64
	subscribers      map[streamHubKey]map[uint64]*streamHubSubscriber
	scopeSubscribers map[streamHubScopeKey]map[uint64]*streamHubScopeSubscriber
	workspaceSubs    map[string]map[uint64]*streamHubSubscriber
	userSubscribers  map[streamHubUserKey]map[uint64]*streamHubSubscriber
}

// NewStreamEventHub creates an empty Workspace IM live event hub.
func NewStreamEventHub() *StreamEventHub {
	hub := &StreamEventHub{}
	hub.subscribers = make(map[streamHubKey]map[uint64]*streamHubSubscriber)
	hub.scopeSubscribers = make(map[streamHubScopeKey]map[uint64]*streamHubScopeSubscriber)
	hub.workspaceSubs = make(map[string]map[uint64]*streamHubSubscriber)
	hub.userSubscribers = make(map[streamHubUserKey]map[uint64]*streamHubSubscriber)
	return hub
}

// SubscribeStreamEvents subscribes one member-local stream until ctx is done or
// the subscriber is closed. userID is required for owner-only event routing
// (agent activity/session/trajectory); pass empty when the caller knows the
// stream will never receive user-routed live-only events.
func (hub *StreamEventHub) SubscribeStreamEvents(ctx context.Context, workspaceID string, memberID string, userID string) *StreamEventSubscription {
	if hub == nil {
		return closedStreamEventSubscription()
	}
	key := streamHubKey{workspaceID: strings.TrimSpace(workspaceID), memberID: strings.TrimSpace(memberID)}
	if key.workspaceID == "" || key.memberID == "" {
		return closedStreamEventSubscription()
	}
	id := atomic.AddUint64(&hub.nextID, 1)
	sub := &streamHubSubscriber{
		id:     id,
		key:    key,
		userID: strings.TrimSpace(userID),
		events: make(chan storeim.StreamEvent, streamHubSubscriptionBuffer),
	}
	sub.closeFn = func() {
		hub.removeSubscriber(sub)
	}

	hub.mu.Lock()
	if hub.subscribers[key] == nil {
		hub.subscribers[key] = make(map[uint64]*streamHubSubscriber)
	}
	hub.subscribers[key][id] = sub
	if hub.workspaceSubs[key.workspaceID] == nil {
		hub.workspaceSubs[key.workspaceID] = make(map[uint64]*streamHubSubscriber)
	}
	hub.workspaceSubs[key.workspaceID][id] = sub
	if sub.userID != "" {
		userKey := streamHubUserKey{workspaceID: key.workspaceID, userID: sub.userID}
		if hub.userSubscribers[userKey] == nil {
			hub.userSubscribers[userKey] = make(map[uint64]*streamHubSubscriber)
		}
		hub.userSubscribers[userKey][id] = sub
	}
	hub.mu.Unlock()

	logging.Go(nil, func() {
		<-ctx.Done()
		sub.close()
	})

	return &StreamEventSubscription{sub: sub}
}

// SubscribeVisibleScopeEvents registers an existing member-local subscription
// for live-only events from one currently readable scope.
func (hub *StreamEventHub) SubscribeVisibleScopeEvents(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string) bool {
	source := streamHubScopeSourceExplicit
	return hub.subscribeVisibleScopeEvents(subscription, workspaceID, memberID, scopeID, source)
}

// SubscribeImplicitVisibleScopeEvents records a bootstrap or membership-backed
// scope source. It coexists with an explicit client lease for the same scope.
func (hub *StreamEventHub) SubscribeImplicitVisibleScopeEvents(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string) bool {
	source := streamHubScopeSourceImplicit
	return hub.subscribeVisibleScopeEvents(subscription, workspaceID, memberID, scopeID, source)
}

func (hub *StreamEventHub) subscribeVisibleScopeEvents(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string, source streamHubScopeSource) bool {
	if hub == nil || subscription == nil || subscription.sub == nil {
		return false
	}
	memberKey := streamHubKey{workspaceID: strings.TrimSpace(workspaceID), memberID: strings.TrimSpace(memberID)}
	scopeKey := streamHubScopeKey{workspaceID: memberKey.workspaceID, scopeID: strings.TrimSpace(scopeID)}
	if memberKey.workspaceID == "" || memberKey.memberID == "" || scopeKey.scopeID == "" {
		return false
	}
	sub := subscription.sub
	if sub.key != memberKey || sub.isClosed() {
		return false
	}
	hub.mu.Lock()
	defer hub.mu.Unlock()
	if sub.isClosed() {
		return false
	}
	if hub.scopeSubscribers[scopeKey] == nil {
		hub.scopeSubscribers[scopeKey] = make(map[uint64]*streamHubScopeSubscriber)
	}
	entry := hub.scopeSubscribers[scopeKey][sub.id]
	if entry == nil {
		entry = &streamHubScopeSubscriber{sub: sub}
		hub.scopeSubscribers[scopeKey][sub.id] = entry
	}
	entry.sources |= source
	return true
}

// UnsubscribeVisibleScopeEvents removes one connection-local visible scope
// subscription. The member-local stream remains active.
func (hub *StreamEventHub) UnsubscribeVisibleScopeEvents(subscription *StreamEventSubscription, workspaceID string, memberID string, scopeID string) {
	if hub == nil || subscription == nil || subscription.sub == nil {
		return
	}
	memberKey := streamHubKey{workspaceID: strings.TrimSpace(workspaceID), memberID: strings.TrimSpace(memberID)}
	scopeKey := streamHubScopeKey{workspaceID: memberKey.workspaceID, scopeID: strings.TrimSpace(scopeID)}
	if memberKey.workspaceID == "" || memberKey.memberID == "" || scopeKey.scopeID == "" {
		return
	}
	sub := subscription.sub
	if sub.key != memberKey {
		return
	}
	hub.mu.Lock()
	defer hub.mu.Unlock()
	byID := hub.scopeSubscribers[scopeKey]
	if len(byID) == 0 {
		return
	}
	entry := byID[sub.id]
	if entry == nil {
		return
	}
	entry.sources &^= streamHubScopeSourceExplicit
	if entry.sources == 0 {
		delete(byID, sub.id)
	}
	if len(byID) == 0 {
		delete(hub.scopeSubscribers, scopeKey)
	}
}

// PublishStreamEvents publishes events to live subscribers keyed by
// (workspace_id, member_id). Events for members without active connections are
// ignored; replay remains available from im_stream_events.
func (hub *StreamEventHub) PublishStreamEvents(events []storeim.StreamEvent) {
	if hub == nil || len(events) == 0 {
		return
	}
	for _, event := range events {
		key := streamHubKey{workspaceID: strings.TrimSpace(event.WorkspaceID), memberID: strings.TrimSpace(event.MemberID)}
		if key.workspaceID == "" || key.memberID == "" {
			continue
		}
		for _, sub := range hub.snapshotSubscribers(key) {
			sub.enqueue(event)
		}
	}
}

// PublishVisibleScopeEvents publishes one live-only copy of each scope-write
// event to every connection that joined that visible scope but did not already
// receive the same event via its member-local stream.
func (hub *StreamEventHub) PublishVisibleScopeEvents(events []storeim.StreamEvent) {
	if hub == nil || len(events) == 0 {
		return
	}
	batches := make(map[visibleScopeEventKey]*visibleScopeEventBatch, len(events))
	order := make([]visibleScopeEventKey, 0, len(events))
	for _, event := range events {
		workspaceID := strings.TrimSpace(event.WorkspaceID)
		scopeID := strings.TrimSpace(event.ScopeID)
		if workspaceID == "" || scopeID == "" {
			continue
		}
		event.WorkspaceID = workspaceID
		event.ScopeID = scopeID
		key := visibleScopeBatchKey(event)
		batch := batches[key]
		if batch == nil {
			batch = &visibleScopeEventBatch{
				event:     event,
				memberIDs: make(map[string]struct{}, 1),
			}
			batches[key] = batch
			order = append(order, key)
		}
		if memberID := strings.TrimSpace(event.MemberID); memberID != "" {
			batch.memberIDs[memberID] = struct{}{}
		}
	}
	for _, key := range order {
		batch := batches[key]
		if batch == nil {
			continue
		}
		scopeKey := streamHubScopeKey{workspaceID: key.workspaceID, scopeID: key.scopeID}
		for _, sub := range hub.snapshotScopeSubscribers(scopeKey) {
			if _, durableRecipient := batch.memberIDs[sub.key.memberID]; durableRecipient {
				continue
			}
			liveEvent := batch.event
			liveEvent.MemberID = sub.key.memberID
			liveEvent.EventSeq = 0
			liveEvent.EventID = ""
			liveEvent.CursorGeneration = ""
			sub.enqueue(liveEvent)
		}
	}
}

// PublishLiveOnlyEventsByMember routes live-only runtime events to subscribers
// keyed by (workspace_id, member_id). eventSeq is forced to 0 and eventID is
// cleared; nothing is persisted to im_stream_events.
func (hub *StreamEventHub) PublishLiveOnlyEventsByMember(events []storeim.StreamEvent) {
	if hub == nil || len(events) == 0 {
		return
	}
	for _, event := range events {
		key := streamHubKey{workspaceID: strings.TrimSpace(event.WorkspaceID), memberID: strings.TrimSpace(event.MemberID)}
		if key.workspaceID == "" || key.memberID == "" {
			continue
		}
		event.EventSeq = 0
		event.EventID = ""
		event.CursorGeneration = ""
		for _, sub := range hub.snapshotSubscribers(key) {
			sub.enqueue(event)
		}
	}
}

// PublishWorkspaceLiveOnlyEvent broadcasts a single live-only event to every
// subscriber in the workspace. eventSeq is forced to 0 and eventID is cleared.
func (hub *StreamEventHub) PublishWorkspaceLiveOnlyEvent(event storeim.StreamEvent) {
	hub.publishWorkspaceLiveOnly(event, "")
}

// PublishWorkspaceLiveOnlyEventExceptUser broadcasts a live-only event to
// workspace subscribers whose user ID differs from exceptUserID. Use this for
// coarse owner-only events (e.g. agent:status) that are paired with a detailed
// PublishUserLiveOnlyEvent to the owner.
func (hub *StreamEventHub) PublishWorkspaceLiveOnlyEventExceptUser(event storeim.StreamEvent, exceptUserID string) {
	hub.publishWorkspaceLiveOnly(event, strings.TrimSpace(exceptUserID))
}

func (hub *StreamEventHub) publishWorkspaceLiveOnly(event storeim.StreamEvent, exceptUserID string) {
	if hub == nil {
		return
	}
	workspaceID := strings.TrimSpace(event.WorkspaceID)
	if workspaceID == "" {
		return
	}
	event.WorkspaceID = workspaceID
	event.EventSeq = 0
	event.EventID = ""
	event.CursorGeneration = ""
	for _, sub := range hub.snapshotWorkspaceSubscribers(workspaceID) {
		if exceptUserID != "" && sub.userID == exceptUserID {
			continue
		}
		sub.enqueue(event)
	}
}

// PublishUserLiveOnlyEvent routes a single live-only event to every subscriber
// matching (workspace_id, user_id). This supports owner-only runtime events
// (agent activity/session/trajectory) without expanding the durable stream.
func (hub *StreamEventHub) PublishUserLiveOnlyEvent(workspaceID string, userID string, event storeim.StreamEvent) {
	if hub == nil {
		return
	}
	userKey := streamHubUserKey{workspaceID: strings.TrimSpace(workspaceID), userID: strings.TrimSpace(userID)}
	if userKey.workspaceID == "" || userKey.userID == "" {
		return
	}
	event.WorkspaceID = userKey.workspaceID
	event.EventSeq = 0
	event.EventID = ""
	event.CursorGeneration = ""
	for _, sub := range hub.snapshotUserSubscribers(userKey) {
		// Replicate the event per recipient so the live copy carries the
		// subscriber's own memberID, matching the visible-scope live pattern.
		event.MemberID = sub.key.memberID
		sub.enqueue(event)
	}
}

// PublishScopeLiveOnlyEvent routes a single live-only event to every connection
// that has joined the supplied visible scope via SubscribeVisibleScopeEvents.
// This is the streaming-native equivalent of a Socket.IO channel-room emit:
// channel-scoped runtime events (task_*, agent_wake_status) reach only members
// currently viewing the channel, so workspace members outside the scope do not
// receive task or wake-status payloads they are not supposed to see. eventSeq
// stays 0 and eventID is cleared; recovery for non-viewers remains via REST
// bootstrap when they open the channel.
func (hub *StreamEventHub) PublishScopeLiveOnlyEvent(workspaceID string, scopeID string, event storeim.StreamEvent) {
	if hub == nil {
		return
	}
	scopeKey := streamHubScopeKey{
		workspaceID: strings.TrimSpace(workspaceID),
		scopeID:     strings.TrimSpace(scopeID),
	}
	if scopeKey.workspaceID == "" || scopeKey.scopeID == "" {
		return
	}
	event.WorkspaceID = scopeKey.workspaceID
	event.ScopeID = scopeKey.scopeID
	event.EventSeq = 0
	event.EventID = ""
	event.CursorGeneration = ""
	for _, sub := range hub.snapshotScopeSubscribers(scopeKey) {
		event.MemberID = sub.key.memberID
		sub.enqueue(event)
	}
}

// SubscribeScopeForUser adds the supplied scope to every active streaming
// subscription owned by (workspaceID, userID). It is the streaming equivalent
// of the old Socket.IO SocketsJoin-by-user-room call: when a user is added to
// a channel or a new DM is opened for them after they have already connected,
// every existing connection starts receiving that scope's channel-scoped
// live-only events (runtime.task_*, agent_wake_status, etc.) without needing
// to reconnect.
func (hub *StreamEventHub) SubscribeScopeForUser(workspaceID string, userID string, scopeID string) {
	if hub == nil {
		return
	}
	userKey := streamHubUserKey{
		workspaceID: strings.TrimSpace(workspaceID),
		userID:      strings.TrimSpace(userID),
	}
	scopeKey := streamHubScopeKey{
		workspaceID: userKey.workspaceID,
		scopeID:     strings.TrimSpace(scopeID),
	}
	if userKey.workspaceID == "" || userKey.userID == "" || scopeKey.scopeID == "" {
		return
	}
	hub.mu.Lock()
	if hub.scopeSubscribers[scopeKey] == nil {
		hub.scopeSubscribers[scopeKey] = make(map[uint64]*streamHubScopeSubscriber)
	}
	added := false
	for _, sub := range hub.userSubscribers[userKey] {
		if sub == nil || sub.isClosed() {
			continue
		}
		entry := hub.scopeSubscribers[scopeKey][sub.id]
		if entry == nil {
			entry = &streamHubScopeSubscriber{sub: sub}
			hub.scopeSubscribers[scopeKey][sub.id] = entry
		}
		entry.sources |= streamHubScopeSourceImplicit
		added = true
	}
	hub.mu.Unlock()
	_ = added
}

// UnsubscribeScopeForUser removes the supplied scope from every active
// streaming subscription owned by (workspaceID, userID). It mirrors the old
// Socket.IO SocketsLeave-by-user-room call: when a user leaves a channel or
// is removed by an admin, every existing connection stops receiving that
// scope's channel-scoped live-only events immediately, without waiting for
// reconnect.
func (hub *StreamEventHub) UnsubscribeScopeForUser(workspaceID string, userID string, scopeID string) {
	if hub == nil {
		return
	}
	userKey := streamHubUserKey{
		workspaceID: strings.TrimSpace(workspaceID),
		userID:      strings.TrimSpace(userID),
	}
	scopeKey := streamHubScopeKey{
		workspaceID: userKey.workspaceID,
		scopeID:     strings.TrimSpace(scopeID),
	}
	if userKey.workspaceID == "" || userKey.userID == "" || scopeKey.scopeID == "" {
		return
	}
	hub.mu.Lock()
	byID := hub.scopeSubscribers[scopeKey]
	if len(byID) > 0 {
		for _, sub := range hub.userSubscribers[userKey] {
			if sub == nil {
				continue
			}
			delete(byID, sub.id)
		}
		if len(byID) == 0 {
			delete(hub.scopeSubscribers, scopeKey)
		}
	}
	hub.mu.Unlock()
}

// UserConnectionController forcibly tears down every active streaming
// subscription owned by (workspaceID, userID). It is the streaming equivalent
// of the old Socket.IO SocketsDisconnect-by-user-room call used when a user's
// workspace membership is revoked: closing every subscription makes the
// /events/stream handler return so the underlying WebSocket closes, and the
// browser's reconnect attempt is then rejected by the workspace auth
// middleware.
type UserConnectionController interface {
	DisconnectUser(workspaceID string, userID string)
}

// DisconnectUser closes every active streaming subscription owned by
// (workspaceID, userID). Each closed subscription makes the owning
// /events/stream handler return; the cascading cleanup in removeSubscriber
// drops the subscription from every hub index.
func (hub *StreamEventHub) DisconnectUser(workspaceID string, userID string) {
	if hub == nil {
		return
	}
	userKey := streamHubUserKey{
		workspaceID: strings.TrimSpace(workspaceID),
		userID:      strings.TrimSpace(userID),
	}
	if userKey.workspaceID == "" || userKey.userID == "" {
		return
	}
	hub.mu.Lock()
	subs := make([]*streamHubSubscriber, 0, len(hub.userSubscribers[userKey]))
	for _, sub := range hub.userSubscribers[userKey] {
		if sub == nil {
			continue
		}
		subs = append(subs, sub)
	}
	hub.mu.Unlock()
	for _, sub := range subs {
		sub.close()
	}
}

type visibleScopeEventKey struct {
	workspaceID string
	scopeID     string
	eventID     string
	memberID    string
	eventSeq    int64
	eventType   string
	entityID    string
}

type visibleScopeEventBatch struct {
	event     storeim.StreamEvent
	memberIDs map[string]struct{}
}

func visibleScopeBatchKey(event storeim.StreamEvent) visibleScopeEventKey {
	key := visibleScopeEventKey{
		workspaceID: strings.TrimSpace(event.WorkspaceID),
		scopeID:     strings.TrimSpace(event.ScopeID),
		eventID:     strings.TrimSpace(event.EventID),
	}
	if key.eventID != "" {
		return key
	}
	key.memberID = strings.TrimSpace(event.MemberID)
	key.eventSeq = event.EventSeq
	key.eventType = strings.TrimSpace(event.Type)
	key.entityID = strings.TrimSpace(event.EntityID)
	return key
}

func (hub *StreamEventHub) snapshotSubscribers(key streamHubKey) []*streamHubSubscriber {
	hub.mu.Lock()
	defer hub.mu.Unlock()
	byID := hub.subscribers[key]
	if len(byID) == 0 {
		return nil
	}
	subscribers := make([]*streamHubSubscriber, 0, len(byID))
	for _, sub := range byID {
		subscribers = append(subscribers, sub)
	}
	return subscribers
}

func (hub *StreamEventHub) snapshotScopeSubscribers(key streamHubScopeKey) []*streamHubSubscriber {
	hub.mu.Lock()
	defer hub.mu.Unlock()
	byID := hub.scopeSubscribers[key]
	if len(byID) == 0 {
		return nil
	}
	subscribers := make([]*streamHubSubscriber, 0, len(byID))
	for _, entry := range byID {
		if entry == nil || entry.sub == nil || entry.sources == 0 {
			continue
		}
		subscribers = append(subscribers, entry.sub)
	}
	return subscribers
}

func (hub *StreamEventHub) snapshotWorkspaceSubscribers(workspaceID string) []*streamHubSubscriber {
	hub.mu.Lock()
	defer hub.mu.Unlock()
	byID := hub.workspaceSubs[workspaceID]
	if len(byID) == 0 {
		return nil
	}
	subscribers := make([]*streamHubSubscriber, 0, len(byID))
	for _, sub := range byID {
		subscribers = append(subscribers, sub)
	}
	return subscribers
}

func (hub *StreamEventHub) snapshotUserSubscribers(key streamHubUserKey) []*streamHubSubscriber {
	hub.mu.Lock()
	defer hub.mu.Unlock()
	byID := hub.userSubscribers[key]
	if len(byID) == 0 {
		return nil
	}
	subscribers := make([]*streamHubSubscriber, 0, len(byID))
	for _, sub := range byID {
		subscribers = append(subscribers, sub)
	}
	return subscribers
}

func (hub *StreamEventHub) removeSubscriber(sub *streamHubSubscriber) {
	hub.mu.Lock()
	defer hub.mu.Unlock()
	if byID := hub.subscribers[sub.key]; len(byID) > 0 {
		delete(byID, sub.id)
		if len(byID) == 0 {
			delete(hub.subscribers, sub.key)
		}
	}
	if byID := hub.workspaceSubs[sub.key.workspaceID]; len(byID) > 0 {
		delete(byID, sub.id)
		if len(byID) == 0 {
			delete(hub.workspaceSubs, sub.key.workspaceID)
		}
	}
	if sub.userID != "" {
		userKey := streamHubUserKey{workspaceID: sub.key.workspaceID, userID: sub.userID}
		if byID := hub.userSubscribers[userKey]; len(byID) > 0 {
			delete(byID, sub.id)
			if len(byID) == 0 {
				delete(hub.userSubscribers, userKey)
			}
		}
	}
	for key, byID := range hub.scopeSubscribers {
		delete(byID, sub.id)
		if len(byID) == 0 {
			delete(hub.scopeSubscribers, key)
		}
	}
}

type streamHubSubscriber struct {
	id                 uint64
	key                streamHubKey
	userID             string
	events             chan storeim.StreamEvent
	mu                 sync.Mutex
	closed             bool
	closedBackpressure bool
	droppedEvent       storeim.StreamEvent
	closeFn            func()
}

type streamHubScopeSource uint8

const (
	streamHubScopeSourceExplicit streamHubScopeSource = 1 << iota
	streamHubScopeSourceImplicit
)

type streamHubScopeSubscriber struct {
	sub     *streamHubSubscriber
	sources streamHubScopeSource
}

func (sub *streamHubSubscriber) enqueue(event storeim.StreamEvent) {
	var closeFn func()
	sub.mu.Lock()
	if sub.closed {
		sub.mu.Unlock()
		return
	}
	select {
	case sub.events <- event:
		sub.mu.Unlock()
	default:
		sub.closed = true
		sub.closedBackpressure = true
		sub.droppedEvent = event
		close(sub.events)
		closeFn = sub.closeFn
		sub.mu.Unlock()
	}
	if closeFn != nil {
		closeFn()
	}
}

func (sub *streamHubSubscriber) close() {
	if sub == nil {
		return
	}
	var closeFn func()
	sub.mu.Lock()
	if sub.closed {
		sub.mu.Unlock()
		return
	}
	sub.closed = true
	close(sub.events)
	closeFn = sub.closeFn
	sub.mu.Unlock()
	if closeFn != nil {
		closeFn()
	}
}

func (sub *streamHubSubscriber) isClosed() bool {
	if sub == nil {
		return true
	}
	sub.mu.Lock()
	defer sub.mu.Unlock()
	return sub.closed
}

// StreamEventSubscription is a live event subscription returned by the hub.
type StreamEventSubscription struct {
	sub *streamHubSubscriber
}

func closedStreamEventSubscription() *StreamEventSubscription {
	events := make(chan storeim.StreamEvent)
	close(events)
	return &StreamEventSubscription{sub: &streamHubSubscriber{events: events, closed: true}}
}

// Events returns live events for the subscribed member stream.
func (subscription *StreamEventSubscription) Events() <-chan storeim.StreamEvent {
	if subscription == nil || subscription.sub == nil {
		events := make(chan storeim.StreamEvent)
		close(events)
		return events
	}
	return subscription.sub.events
}

func (subscription *StreamEventSubscription) backpressureEvent() (storeim.StreamEvent, bool) {
	if subscription == nil || subscription.sub == nil {
		return storeim.StreamEvent{}, false
	}
	subscription.sub.mu.Lock()
	defer subscription.sub.mu.Unlock()
	return subscription.sub.droppedEvent, subscription.sub.closedBackpressure
}

// Close unregisters the live subscription.
func (subscription *StreamEventSubscription) Close() {
	if subscription != nil && subscription.sub != nil {
		subscription.sub.close()
	}
}