hub.go

package centrifuge

import (
	"context"
	"io"
	"sync"
	"sync/atomic"
	"time"

	"github.com/centrifugal/centrifuge/internal/convert"
	"github.com/centrifugal/centrifuge/internal/filter"

	"github.com/centrifugal/protocol"
	"github.com/segmentio/encoding/json"
	fdelta "github.com/shadowspore/fossil-delta"
)

const numHubShards = 64

var pushPool = sync.Pool{
	New: func() any {
		return &protocol.Push{}
	},
}

var replyPool = sync.Pool{
	New: func() any {
		return &protocol.Reply{}
	},
}

func getPush() *protocol.Push {
	return pushPool.Get().(*protocol.Push)
}

func putPush(p *protocol.Push) {
	p.Channel = ""
	p.Id = 0
	p.Pub = nil
	p.Join = nil
	p.Leave = nil
	p.Message = nil
	p.Subscribe = nil
	p.Unsubscribe = nil
	p.Connect = nil
	p.Disconnect = nil
	p.Refresh = nil
	pushPool.Put(p)
}

func getReply() *protocol.Reply {
	return replyPool.Get().(*protocol.Reply)
}

func putReply(r *protocol.Reply) {
	r.Id = 0
	r.Error = nil
	r.Push = nil
	r.Connect = nil
	r.Subscribe = nil
	r.Unsubscribe = nil
	r.Publish = nil
	r.Presence = nil
	r.PresenceStats = nil
	r.History = nil
	r.Ping = nil
	r.Rpc = nil
	r.Refresh = nil
	r.SubRefresh = nil
	replyPool.Put(r)
}

// Hub tracks Client connections on the current Node.
type Hub struct {
	connShards [numHubShards]*connShard
	subShards  [numHubShards]*subShard
	sessionsMu sync.RWMutex
	sessions   map[string]*Client
}

// newHub initializes Hub.
func newHub(logger *logger, metrics *metrics, maxTimeLagMilli int64) *Hub {
	h := &Hub{
		sessions: map[string]*Client{},
	}
	for i := 0; i < numHubShards; i++ {
		h.connShards[i] = newConnShard()
		h.subShards[i] = newSubShard(logger, metrics, maxTimeLagMilli, i)
	}
	return h
}

func (h *Hub) clientBySession(session string) (*Client, bool) {
	h.sessionsMu.RLock()
	defer h.sessionsMu.RUnlock()
	c, ok := h.sessions[session]
	return c, ok
}

// shutdown unsubscribes users from all channels and disconnects them.
func (h *Hub) shutdown(ctx context.Context) error {
	// Limit concurrency here to prevent resource usage burst on shutdown.
	sem := make(chan struct{}, hubShutdownSemaphoreSize)

	var errMu sync.Mutex
	var shutdownErr error

	var wg sync.WaitGroup
	wg.Add(numHubShards)
	for i := 0; i < numHubShards; i++ {
		go func(i int) {
			defer wg.Done()
			err := h.connShards[i].shutdown(ctx, sem)
			if err != nil {
				errMu.Lock()
				if shutdownErr == nil {
					shutdownErr = err
				}
				errMu.Unlock()
			}
		}(i)
	}
	wg.Wait()
	return shutdownErr
}

// Add connection into clientHub connections registry.
// add registers connection in clientHub. Returns true if this is a new
// registration (uid not previously present).
func (h *Hub) add(c *Client) bool {
	h.sessionsMu.Lock()
	if c.sessionID() != "" {
		h.sessions[c.sessionID()] = c
	}
	h.sessionsMu.Unlock()
	return h.connShards[index(c.UserID(), numHubShards)].add(c)
}

// Remove connection from clientHub connections registry.
// Returns true if found and really removed from registry.
func (h *Hub) remove(c *Client) bool {
	h.sessionsMu.Lock()
	if c.sessionID() != "" {
		delete(h.sessions, c.sessionID())
	}
	h.sessionsMu.Unlock()
	return h.connShards[index(c.UserID(), numHubShards)].remove(c)
}

// Connections returns all user connections to the current Node.
func (h *Hub) Connections() map[string]*Client {
	connections := make(map[string]*Client)
	for _, shard := range h.connShards {
		shard.mu.RLock()
		for clientID, c := range shard.clients {
			connections[clientID] = c
		}
		shard.mu.RUnlock()
	}
	return connections
}

// UserConnections returns all user connections to the current Node.
func (h *Hub) UserConnections(userID string) map[string]*Client {
	return h.connShards[index(userID, numHubShards)].userConnections(userID)
}

func (h *Hub) refresh(userID string, clientID, sessionID string, labelFilter *FilterNode, opts ...RefreshOption) error {
	return h.connShards[index(userID, numHubShards)].refresh(userID, clientID, sessionID, labelFilter, opts...)
}

func (h *Hub) subscribe(userID string, ch string, clientID string, sessionID string, labelFilter *FilterNode, opts ...SubscribeOption) error {
	return h.connShards[index(userID, numHubShards)].subscribe(userID, ch, clientID, sessionID, labelFilter, opts...)
}

func (h *Hub) unsubscribe(userID string, ch string, unsubscribe Unsubscribe, clientID string, sessionID string, labelFilter *FilterNode) error {
	return h.connShards[index(userID, numHubShards)].unsubscribe(userID, ch, unsubscribe, clientID, sessionID, labelFilter)
}

// refreshAcrossUsers iterates every connection on every shard, applying
// clientID/sessionID/labelFilter narrowers. Used by Node.Refresh when userID
// is empty and the allUsers flag is set. Not used through the normal per-user
// dispatch path; see Node.Refresh for the routing decision.
func (h *Hub) refreshAcrossUsers(clientID, sessionID string, labelFilter *FilterNode, opts ...RefreshOption) error {
	var firstErr error
	for _, shard := range h.connShards {
		if err := shard.refreshAcrossUsers(clientID, sessionID, labelFilter, opts...); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// subscribeAcrossUsers — see refreshAcrossUsers.
func (h *Hub) subscribeAcrossUsers(ch string, clientID, sessionID string, labelFilter *FilterNode, opts ...SubscribeOption) error {
	var firstErr error
	for _, shard := range h.connShards {
		if err := shard.subscribeAcrossUsers(ch, clientID, sessionID, labelFilter, opts...); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// unsubscribeAcrossUsers — see refreshAcrossUsers.
func (h *Hub) unsubscribeAcrossUsers(ch string, unsubscribe Unsubscribe, clientID, sessionID string, labelFilter *FilterNode) error {
	for _, shard := range h.connShards {
		_ = shard.unsubscribeAcrossUsers(ch, unsubscribe, clientID, sessionID, labelFilter)
	}
	return nil
}

// disconnectAcrossUsers — see refreshAcrossUsers.
func (h *Hub) disconnectAcrossUsers(disconnect Disconnect, clientID, sessionID string, whitelist []string, labelFilter *FilterNode) error {
	for _, shard := range h.connShards {
		_ = shard.disconnectAcrossUsers(disconnect, clientID, sessionID, whitelist, labelFilter)
	}
	return nil
}

func (h *Hub) disconnect(userID string, disconnect Disconnect, clientID, sessionID string, whitelist []string, labelFilter *FilterNode) error {
	return h.connShards[index(userID, numHubShards)].disconnect(userID, disconnect, clientID, sessionID, whitelist, labelFilter)
}

func (h *Hub) addSub(ch string, sub subInfo) (int64, bool, error) {
	return h.subShards[index(ch, numHubShards)].addSub(ch, sub)
}

// removeSub removes connection from clientHub subscriptions registry.
// Returns (isEmpty, wasRemoved, wasKeyed).
func (h *Hub) removeSub(ch string, c *Client, subGen uint64) (bool, bool, bool) {
	return h.subShards[index(ch, numHubShards)].removeSub(ch, c, subGen)
}

func (h *Hub) updateServerTagsFilter(ch string, clientID string, tf *tagsFilter) (bool, bool) {
	return h.subShards[index(ch, numHubShards)].updateServerTagsFilter(ch, clientID, tf)
}

func (h *Hub) removeSubID(ch string) {
	h.subShards[index(ch, numHubShards)].removeSubID(ch)
}

// BroadcastPublication sends message to all clients subscribed on a channel on the current Node.
// Usually this is NOT what you need since in most cases you should use Node.Publish method which
// uses a Broker to deliver publications to all Nodes in a cluster and maintains publication history
// in a channel with incremental offset. By calling BroadcastPublication messages will only be sent
// to the current node subscribers without any defined offset semantics, without delta support.
func (h *Hub) BroadcastPublication(ch string, pub *Publication, sp StreamPosition) error {
	return h.broadcastPublication(ch, sp, pub, nil, nil, ChannelBatchConfig{})
}

// BroadcastPublicationDelta is like BroadcastPublication but supports delta compression.
// When prevPub is non-nil, subscribers with delta enabled receive a computed delta
// instead of the full publication data. Only sent to the current node subscribers.
func (h *Hub) BroadcastPublicationDelta(ch string, pub *Publication, prevPub *Publication, sp StreamPosition) error {
	return h.broadcastPublication(ch, sp, pub, prevPub, prevPub, ChannelBatchConfig{})
}

func (h *Hub) broadcastPublication(
	ch string, sp StreamPosition, pub, prevPub, localPrevPub *Publication,
	batchConfig ChannelBatchConfig,
) error {
	return h.subShards[index(ch, numHubShards)].broadcastPublication(
		ch, sp, pub, prevPub, localPrevPub, batchConfig)
}

// broadcastJoin sends message to all clients subscribed on channel.
func (h *Hub) broadcastJoin(ch string, info *ClientInfo, batchConfig ChannelBatchConfig) error {
	return h.subShards[index(ch, numHubShards)].broadcastJoin(ch, &protocol.Join{Info: infoToProto(info)}, batchConfig)
}

func (h *Hub) broadcastLeave(ch string, info *ClientInfo, batchConfig ChannelBatchConfig) error {
	return h.subShards[index(ch, numHubShards)].broadcastLeave(ch, &protocol.Leave{Info: infoToProto(info)}, batchConfig)
}

// NumSubscribers returns number of current subscribers for a given channel.
func (h *Hub) NumSubscribers(ch string) int {
	return h.subShards[index(ch, numHubShards)].NumSubscribers(ch)
}

// Channels returns a slice of all active channels.
func (h *Hub) Channels() []string {
	channels := make([]string, 0, h.NumChannels())
	for i := 0; i < numHubShards; i++ {
		channels = append(channels, h.subShards[i].Channels()...)
	}
	return channels
}

// NumClients returns total number of client connections.
func (h *Hub) NumClients() int {
	var total int
	for i := 0; i < numHubShards; i++ {
		total += h.connShards[i].NumClients()
	}
	return total
}

// NumUsers returns a number of unique users connected.
func (h *Hub) NumUsers() int {
	var total int
	for i := 0; i < numHubShards; i++ {
		// users do not overlap among shards.
		total += h.connShards[i].NumUsers()
	}
	return total
}

// NumSubscriptions returns a total number of subscriptions.
func (h *Hub) NumSubscriptions() int {
	var total int
	for i := 0; i < numHubShards; i++ {
		// users do not overlap among shards.
		total += h.subShards[i].NumSubscriptions()
	}
	return total
}

// NumChannels returns a total number of different channels.
func (h *Hub) NumChannels() int {
	var total int
	for i := 0; i < numHubShards; i++ {
		// channels do not overlap among shards.
		total += h.subShards[i].NumChannels()
	}
	return total
}

type connShard struct {
	mu sync.RWMutex
	// match client ID with actual client connection.
	clients map[string]*Client
	// registry to hold active client connections grouped by user.
	users map[string]map[string]struct{}
}

func newConnShard() *connShard {
	return &connShard{
		clients: make(map[string]*Client),
		users:   make(map[string]map[string]struct{}),
	}
}

const (
	// hubShutdownSemaphoreSize limits graceful disconnects concurrency
	// on node shutdown.
	hubShutdownSemaphoreSize = 128
)

// shutdown unsubscribes users from all channels and disconnects them.
func (h *connShard) shutdown(ctx context.Context, sem chan struct{}) error {
	advice := DisconnectShutdown
	h.mu.RLock()
	// At this moment node won't accept new client connections, so we can
	// safely copy existing clients and release lock.
	clients := make([]*Client, 0, len(h.clients))
	for _, client := range h.clients {
		clients = append(clients, client)
	}
	h.mu.RUnlock()

	closeFinishedCh := make(chan struct{}, len(clients))
	finished := 0

	if len(clients) == 0 {
		return nil
	}

	for _, client := range clients {
		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
			return ctx.Err()
		}
		go func(cc *Client) {
			defer func() { <-sem }()
			defer func() { closeFinishedCh <- struct{}{} }()
			_ = cc.close(advice)
		}(client)
	}

	for {
		select {
		case <-closeFinishedCh:
			finished++
			if finished == len(clients) {
				return nil
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

func stringInSlice(str string, slice []string) bool {
	for _, s := range slice {
		if s == str {
			return true
		}
	}
	return false
}

// matchLabelFilter returns true when c should be included in a label-filtered
// operation. A nil filter matches every client. c.labels is set once before the
// client is published to the hub (see Client connect flow) and never mutated,
// so the read is safe without taking c.mu.
func matchLabelFilter(c *Client, f *FilterNode) bool {
	if f == nil {
		return true
	}
	ok, _ := filter.Match(f, c.labels)
	return ok
}

func (h *connShard) subscribe(user string, ch string, clientID string, sessionID string, labelFilter *FilterNode, opts ...SubscribeOption) error {
	userConnections := h.userConnections(user)

	var firstErr error
	var errMu sync.Mutex

	var wg sync.WaitGroup
	for _, c := range userConnections {
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		wg.Add(1)
		go func(c *Client) {
			defer wg.Done()
			err := c.Subscribe(ch, opts...)
			errMu.Lock()
			defer errMu.Unlock()
			if err != nil && err != io.EOF && firstErr == nil {
				firstErr = err
			}
		}(c)
	}
	wg.Wait()
	return firstErr
}

func (h *connShard) refresh(user string, clientID string, sessionID string, labelFilter *FilterNode, opts ...RefreshOption) error {
	userConnections := h.userConnections(user)

	var firstErr error
	var errMu sync.Mutex

	var wg sync.WaitGroup
	for _, c := range userConnections {
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		wg.Add(1)
		go func(c *Client) {
			defer wg.Done()
			err := c.Refresh(opts...)
			errMu.Lock()
			defer errMu.Unlock()
			if err != nil && err != io.EOF && firstErr == nil {
				firstErr = err
			}
		}(c)
	}
	wg.Wait()
	return firstErr
}

func (h *connShard) unsubscribe(user string, ch string, unsubscribe Unsubscribe, clientID string, sessionID string, labelFilter *FilterNode) error {
	userConnections := h.userConnections(user)

	var wg sync.WaitGroup
	for _, c := range userConnections {
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		wg.Add(1)
		go func(c *Client) {
			defer wg.Done()
			c.Unsubscribe(ch, unsubscribe)
		}(c)
	}
	wg.Wait()
	return nil
}

// connShard.*AcrossUsers methods iterate the shard's full client map applying
// clientID/sessionID/labelFilter narrowers. Used by the fleet-wide dispatch
// path in Node ops when userID is empty AND the allUsers flag is set. The
// shard read-lock is released before invoking client operations to avoid
// holding it during downstream IO; in flight changes to the connection set
// during the loop are acceptable (the op is best-effort over a roughly-current
// snapshot of the hub, same guarantee the regular paths give).
func (h *connShard) refreshAcrossUsers(clientID, sessionID string, labelFilter *FilterNode, opts ...RefreshOption) error {
	clients := h.allClientsSnapshot()
	var firstErr error
	var errMu sync.Mutex
	var wg sync.WaitGroup
	for _, c := range clients {
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		wg.Add(1)
		go func(c *Client) {
			defer wg.Done()
			err := c.Refresh(opts...)
			errMu.Lock()
			defer errMu.Unlock()
			if err != nil && err != io.EOF && firstErr == nil {
				firstErr = err
			}
		}(c)
	}
	wg.Wait()
	return firstErr
}

func (h *connShard) subscribeAcrossUsers(ch string, clientID, sessionID string, labelFilter *FilterNode, opts ...SubscribeOption) error {
	clients := h.allClientsSnapshot()
	var firstErr error
	var errMu sync.Mutex
	var wg sync.WaitGroup
	for _, c := range clients {
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		wg.Add(1)
		go func(c *Client) {
			defer wg.Done()
			err := c.Subscribe(ch, opts...)
			errMu.Lock()
			defer errMu.Unlock()
			if err != nil && err != io.EOF && firstErr == nil {
				firstErr = err
			}
		}(c)
	}
	wg.Wait()
	return firstErr
}

func (h *connShard) unsubscribeAcrossUsers(ch string, unsubscribe Unsubscribe, clientID, sessionID string, labelFilter *FilterNode) error {
	clients := h.allClientsSnapshot()
	var wg sync.WaitGroup
	for _, c := range clients {
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		wg.Add(1)
		go func(c *Client) {
			defer wg.Done()
			c.Unsubscribe(ch, unsubscribe)
		}(c)
	}
	wg.Wait()
	return nil
}

func (h *connShard) disconnectAcrossUsers(disconnect Disconnect, clientID, sessionID string, whitelist []string, labelFilter *FilterNode) error {
	clients := h.allClientsSnapshot()
	for _, c := range clients {
		if stringInSlice(c.ID(), whitelist) {
			continue
		}
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		c.Disconnect(disconnect)
	}
	return nil
}

// allClientsSnapshot copies the current shard's client map under read-lock,
// then releases the lock so the per-client op work isn't done under it.
func (h *connShard) allClientsSnapshot() []*Client {
	h.mu.RLock()
	defer h.mu.RUnlock()
	out := make([]*Client, 0, len(h.clients))
	for _, c := range h.clients {
		out = append(out, c)
	}
	return out
}

func (h *connShard) disconnect(user string, disconnect Disconnect, clientID string, sessionID string, whitelist []string, labelFilter *FilterNode) error {
	userConnections := h.userConnections(user)
	for _, c := range userConnections {
		if stringInSlice(c.ID(), whitelist) {
			continue
		}
		if clientID != "" && c.ID() != clientID {
			continue
		}
		if sessionID != "" && c.sessionID() != sessionID {
			continue
		}
		if !matchLabelFilter(c, labelFilter) {
			continue
		}
		c.Disconnect(disconnect)
	}
	return nil
}

// userConnections returns all connections of user with specified User.
func (h *connShard) userConnections(userID string) map[string]*Client {
	h.mu.RLock()
	defer h.mu.RUnlock()

	userConnections, ok := h.users[userID]
	if !ok {
		return map[string]*Client{}
	}

	connections := make(map[string]*Client, len(userConnections))
	for uid := range userConnections {
		c, ok := h.clients[uid]
		if !ok {
			continue
		}
		connections[uid] = c
	}

	return connections
}

// Add connection into clientHub connections registry.
// add registers a client connection. Returns true if this was a new
// registration (the uid was not already present), so callers can keep the
// connectionsInflight gauge in lockstep with the clients map.
func (h *connShard) add(c *Client) bool {
	h.mu.Lock()
	defer h.mu.Unlock()

	uid := c.ID()
	user := c.UserID()

	_, existed := h.clients[uid]
	h.clients[uid] = c

	if _, ok := h.users[user]; !ok {
		h.users[user] = make(map[string]struct{})
	}
	h.users[user][uid] = struct{}{}
	return !existed
}

// Remove connection from clientHub connections registry.
// Returns true if found and really removed from registry.
// remove deregisters a client connection. Returns true if the uid was present
// in the clients map — the same condition add() reports as new — so the
// connectionsInflight Dec pairs exactly with its Inc regardless of the users map.
func (h *connShard) remove(c *Client) bool {
	h.mu.Lock()
	defer h.mu.Unlock()

	uid := c.ID()
	user := c.UserID()

	_, existed := h.clients[uid]
	delete(h.clients, uid)

	// Clean up the user grouping if present.
	if userConns, ok := h.users[user]; ok {
		delete(userConns, uid)
		if len(userConns) == 0 {
			delete(h.users, user)
		}
	}

	return existed
}

// NumClients returns total number of client connections.
func (h *connShard) NumClients() int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	// clients holds exactly one entry per connection, which is the same total as
	// summing the per-user sets. Iterating users instead would make this O(users)
	// while holding the read lock — at a million connections that is a periodic
	// scan (updateGauges runs every 10s) which also blocks connects/disconnects.
	return len(h.clients)
}

// NumUsers returns a number of unique users connected.
func (h *connShard) NumUsers() int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return len(h.users)
}

type DeltaType string

const (
	deltaTypeNone DeltaType = ""
	// DeltaTypeFossil is Fossil delta encoding. See https://fossil-scm.org/home/doc/tip/www/delta_encoder_algorithm.wiki.
	DeltaTypeFossil DeltaType = "fossil"
)

var stringToDeltaType = map[string]DeltaType{
	"fossil": DeltaTypeFossil,
}

type tagsFilter struct {
	filter *protocol.FilterNode
	hash   [32]byte
}

type subInfo struct {
	client           *Client
	deltaType        DeltaType
	useID            bool
	tagsFilter       *tagsFilter
	serverTagsFilter *tagsFilter
	isMap            bool // true for map subscriptions.
	// protoType and unidirectional mirror the client's transport. Both are fixed
	// for the connection's lifetime, but reading them goes through two interface
	// dispatches — a cost paid once per subscriber per broadcast. Snapshotting
	// them at addSub time turns the broadcast loop's per-subscriber protocol
	// dispatch into plain field loads. Set by addSub, not by callers.
	protoType      protocol.Type
	unidirectional bool
	// subGen identifies this specific subscription of the client to the channel.
	// A resubscribe gets a fresh subGen, so a stale unsubscribe (carrying an older
	// subGen it read from c.channels) will not remove a newer subscription's hub
	// entry — which is how hub routing stays consistent with c.channels without
	// holding a lock across both.
	subGen uint64
}

type subShard struct {
	mu sync.RWMutex
	// registry to hold active subscriptions of clients to channels with some additional info.
	subs map[string]map[string]subInfo
	// numSubs is the total number of subscriptions across subs, maintained
	// incrementally under mu. Summing subs on demand would be O(channels) while
	// holding the read lock, and NumSubscriptions is read periodically by the
	// node gauge update.
	numSubs         int
	maxTimeLagMilli int64
	logger          *logger
	metrics         *metrics
	shardIndex      int

	chanIDs     map[string]int64
	lastChanID  atomic.Int64
	mapChannels map[string]bool // tracks which channels are keyed subscriptions
}

func newSubShard(logger *logger, metrics *metrics, maxTimeLagMilli int64, shardIndex int) *subShard {
	return &subShard{
		subs:            make(map[string]map[string]subInfo),
		logger:          logger,
		metrics:         metrics,
		maxTimeLagMilli: maxTimeLagMilli,
		shardIndex:      shardIndex,
		chanIDs:         make(map[string]int64),
		mapChannels:     make(map[string]bool),
	}
}

// addSub adds connection into clientHub subscriptions registry.
// Returns (chanID, isFirst, error) where isFirst is true if this is the first subscriber.
func (s *subShard) addSub(ch string, sub subInfo) (int64, bool, error) {
	// Snapshot the transport's immutable protocol properties so the broadcast
	// loop reads fields instead of making interface calls per subscriber.
	sub.protoType = sub.client.transport.Protocol().toProto()
	sub.unidirectional = sub.client.transport.Unidirectional()

	s.mu.Lock()
	defer s.mu.Unlock()

	uid := sub.client.ID()

	_, ok := s.subs[ch]
	if !ok {
		s.subs[ch] = make(map[string]subInfo)
		// Track if this channel is keyed (first subscriber determines this).
		if sub.isMap {
			s.mapChannels[ch] = true
		}
	}
	if _, exists := s.subs[ch][uid]; !exists {
		s.numSubs++
		// Inflight is co-located with numSubs (same condition) so the two stay in
		// lockstep. Doing it here, rather than unconditionally in addSubscription,
		// avoids over-counting when a resubscribe overwrites an existing
		// client+channel entry (no new subscription) — its stale generation's
		// removeSub is a no-op that never Dec's, which would otherwise drift.
		baseLabels := []string{sub.client.metricName, s.metrics.getChannelNamespaceLabel(ch)}
		s.metrics.subscriptionsInflight.WithLabelValues(s.metrics.appendClientLabels(baseLabels, sub.client)...).Inc()
	}
	s.subs[ch][uid] = sub

	var chanID int64
	if sub.useID {
		existingChanID, hasChanID := s.chanIDs[ch]
		if !hasChanID {
			// Generate unique ID using shard index + (counter * numHubShards)
			// This ensures each shard generates non-overlapping ID ranges
			counter := s.lastChanID.Add(1)
			chanID = int64(s.shardIndex) + ((counter - 1) * numHubShards)
			s.chanIDs[ch] = chanID
		} else {
			chanID = existingChanID
		}
	}

	if !ok {
		return chanID, true, nil
	}
	return chanID, false, nil
}

// updateServerTagsFilter updates the server-side tags filter for a specific
// client subscription. Returns (found, changed) where changed is true only
// if the filter hash differs from the current one.
func (s *subShard) updateServerTagsFilter(ch string, clientID string, tf *tagsFilter) (bool, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	chSubs, ok := s.subs[ch]
	if !ok {
		return false, false
	}
	sub, ok := chSubs[clientID]
	if !ok {
		return false, false
	}
	if sub.serverTagsFilter != nil && sub.serverTagsFilter.hash == tf.hash {
		return true, false
	}
	if sub.serverTagsFilter == nil && tf == nil {
		return true, false
	}
	sub.serverTagsFilter = tf
	chSubs[clientID] = sub
	return true, true
}

func (s *subShard) removeSubID(ch string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.chanIDs, ch)
}

// removeSub removes connection from clientHub subscriptions registry.
// Returns (isEmpty, wasRemoved, wasKeyed) where:
// - isEmpty: true if channel has no subscribers left
// - wasRemoved: true if subscription was found and removed
// - wasMap: true if the now-empty channel was a keyed subscription channel
// anySubGen removes whatever generation is currently registered. No production
// path passes it anymore — every rollback identity-matches its own generation
// so a stalled attempt can never remove a fresh resubscribe's entry. Kept as
// the named zero value so an accidental gen-0 removal is at least explicit in
// removeSub. Real generations start at 1.
const anySubGen uint64 = 0

func (s *subShard) removeSub(ch string, c *Client, subGen uint64) (bool, bool, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()

	uid := c.ID()

	// try to find subscription to delete, return early if not found.
	if _, ok := s.subs[ch]; !ok {
		return true, false, false
	}
	sub, ok := s.subs[ch][uid]
	if !ok {
		return true, false, false
	}
	if subGen != anySubGen && sub.subGen != subGen {
		// A newer subscription generation is registered for this client+channel —
		// the caller is unsubscribing an older generation that a concurrent
		// resubscribe already superseded. Leave the newer entry in place so hub
		// routing stays consistent with c.channels. Report not-removed, not-empty.
		return false, false, false
	}

	// actually remove subscription from hub.
	delete(s.subs[ch], uid)
	s.numSubs--
	// Mirror of the Inc in addSub — only when a subscription is actually removed
	// (matched generation), so a stale-generation removeSub does not Dec.
	baseLabels := []string{c.metricName, s.metrics.getChannelNamespaceLabel(ch)}
	s.metrics.subscriptionsInflight.WithLabelValues(s.metrics.appendClientLabels(baseLabels, c)...).Dec()

	// clean up subs map if it's needed.
	if len(s.subs[ch]) == 0 {
		delete(s.subs, ch)
		wasMap := s.mapChannels[ch]
		delete(s.mapChannels, ch)
		return true, true, wasMap
	}

	return false, true, false
}

type encodeError struct {
	client string
	user   string
	error  error
}

type preparedKey struct {
	ProtocolType   protocol.Type
	Unidirectional bool
	DeltaType      DeltaType
	UseID          bool
	WasFiltered    bool
}

type preparedData struct {
	fullData        []byte
	brokerDeltaData []byte
	localDeltaData  []byte
	deltaSub        bool
	wasFiltered     bool
	filteredPub     *protocol.Publication
	// For keyed channel lazy delta encoding (set by buildPreparedPollData).
	keyedDeltaPatch       []byte // raw fossil delta data (patch or full data if patch >= full)
	keyedDeltaIsReal      bool   // true when the patch is a real delta (smaller than full)
	keyedDeltaPrevVersion uint64 // version corresponding to the delta's base data (entry.version BEFORE the publish)
}

func getDeltaPub(prevPub *Publication, fullPub *protocol.Publication, key preparedKey) *protocol.Publication {
	deltaPub := fullPub
	if prevPub != nil && key.DeltaType == DeltaTypeFossil {
		patch := fdelta.Create(prevPub.Data, fullPub.Data)
		delta := true
		deltaData := patch
		if len(patch) >= len(fullPub.Data) {
			delta = false
			deltaData = fullPub.Data
		}
		if key.ProtocolType == protocol.TypeJSON {
			deltaData = json.Escape(convert.BytesToString(deltaData))
		}
		deltaPub = &protocol.Publication{
			Offset:  fullPub.Offset,
			Data:    deltaData,
			Info:    fullPub.Info,
			Tags:    fullPub.Tags,
			Delta:   delta,
			Key:     fullPub.Key,
			Removed: fullPub.Removed,
			Score:   fullPub.Score,
			Channel: fullPub.Channel,
		}
	} else if prevPub == nil && key.ProtocolType == protocol.TypeJSON && key.DeltaType == DeltaTypeFossil {
		// In JSON and Fossil case we need to send full state in JSON string format.
		deltaPub = &protocol.Publication{
			Offset:  fullPub.Offset,
			Data:    json.Escape(convert.BytesToString(fullPub.Data)),
			Info:    fullPub.Info,
			Tags:    fullPub.Tags,
			Key:     fullPub.Key,
			Removed: fullPub.Removed,
			Score:   fullPub.Score,
			Channel: fullPub.Channel,
		}
	}
	return deltaPub
}

// jsonEncodeErr is a **encodeError, not *encodeError, deliberately: the caller
// holds a nil *encodeError and needs it REPLACED on failure. Taking it by value
// meant this function dereferenced nil (panicking the whole broadcast) and,
// had it not panicked, would have written to the pointee rather than the
// caller's variable — so the caller's `jsonEncodeErr != nil` check could never
// fire. Matches how the non-delta paths do `jsonEncodeErr = &encodeError{...}`.
func getDeltaData(sub subInfo, key preparedKey, channel string, deltaPub *protocol.Publication, channelSubID int64, jsonEncodeErr **encodeError) ([]byte, error) {
	var deltaData []byte
	if key.ProtocolType == protocol.TypeJSON {
		if sub.client.transport.Unidirectional() {
			push := getPush()
			push.Pub = deltaPub
			if key.UseID {
				push.Id = channelSubID
			} else {
				push.Channel = channel
			}
			var err error
			deltaData, err = protocol.DefaultJsonPushEncoder.Encode(push)
			putPush(push)
			if err != nil {
				*jsonEncodeErr = &encodeError{client: sub.client.ID(), user: sub.client.UserID(), error: err}
			}
		} else {
			push := getPush()
			push.Pub = deltaPub
			if key.UseID {
				push.Id = channelSubID
			} else {
				push.Channel = channel
			}
			reply := getReply()
			reply.Push = push
			var err error
			deltaData, err = protocol.DefaultJsonReplyEncoder.Encode(reply)
			putReply(reply)
			putPush(push)
			if err != nil {
				*jsonEncodeErr = &encodeError{client: sub.client.ID(), user: sub.client.UserID(), error: err}
			}
		}
	} else if key.ProtocolType == protocol.TypeProtobuf {
		if sub.client.transport.Unidirectional() {
			push := getPush()
			push.Pub = deltaPub
			if key.UseID {
				push.Id = channelSubID
			} else {
				push.Channel = channel
			}
			var err error
			deltaData, err = protocol.DefaultProtobufPushEncoder.Encode(push)
			putPush(push)
			if err != nil {
				return nil, err
			}
		} else {
			push := getPush()
			push.Pub = deltaPub
			if key.UseID {
				push.Id = channelSubID
			} else {
				push.Channel = channel
			}
			reply := getReply()
			reply.Push = push
			var err error
			deltaData, err = protocol.DefaultProtobufReplyEncoder.Encode(reply)
			putReply(reply)
			putPush(push)
			if err != nil {
				return nil, err
			}
		}
	}
	return deltaData, nil
}

// broadcastPublication sends message to all clients subscribed on a channel.
func (s *subShard) broadcastPublication(
	channel string, sp StreamPosition, pub, prevPub, localPrevPub *Publication,
	batchConfig ChannelBatchConfig,
) error {
	pubTime := pub.Time
	// Check lag in PUB/SUB processing. We use it to notify subscribers with positioning enabled
	// about insufficient state in the stream.
	var maxLagExceeded bool
	now := time.Now()
	if pubTime > 0 {
		timeLagMilli := now.UnixMilli() - pubTime
		if s.maxTimeLagMilli > 0 && timeLagMilli > s.maxTimeLagMilli {
			maxLagExceeded = true
		}
		s.metrics.observePubSubDeliveryLag(timeLagMilli, channel)
	}

	fullPub := pubToProto(pub)
	// Deliver the channel epoch in the first publication (offset==1) so clients
	// learn it. Stamped on this per-broadcast copy rather than on the source pub,
	// which the MemoryBroker shares with concurrent recovery readers (see
	// Node.HandlePublication).
	if fullPub.Offset == 1 && sp.Epoch != "" {
		fullPub.Epoch = sp.Epoch
	}
	// Encoded payloads are shared by all subscribers with the same
	// protocol/delta/useID/filtered combination. Nearly every channel has just
	// one such combination, so a single-entry front cache serves every
	// subscriber after the first and the map — whose key hashing showed up as a
	// per-subscriber cost, preparedKey holding a string — is only built when a
	// channel really does mix combinations.
	var (
		lastKey           preparedKey
		lastData          preparedData
		haveLast          bool
		preparedDataByKey map[preparedKey]preparedData
	)

	var filteredPub *protocol.Publication

	s.mu.RLock()
	defer s.mu.RUnlock()

	channelSubscribers, ok := s.subs[channel]
	if !ok {
		return nil
	}

	if pub.Channel != channel {
		fullPub.Channel = pub.Channel
	}

	var (
		jsonEncodeErr     *encodeError
		tagsFilterDropped int
	)

	// Get subID for this channel if it exists
	channelSubID, hasSubID := s.chanIDs[channel]

	for _, sub := range channelSubscribers {
		useChannelID := sub.useID && hasSubID

		wasFiltered := false
		if sub.serverTagsFilter != nil {
			match, _ := filter.Match(sub.serverTagsFilter.filter, pub.Tags)
			if !match {
				wasFiltered = true
				tagsFilterDropped++
			}
		}
		if !wasFiltered && sub.tagsFilter != nil {
			match, _ := filter.Match(sub.tagsFilter.filter, pub.Tags)
			if !match {
				wasFiltered = true
				tagsFilterDropped++
			}
		}

		key := preparedKey{
			ProtocolType:   sub.protoType,
			Unidirectional: sub.unidirectional,
			DeltaType:      sub.deltaType,
			UseID:          useChannelID,
			WasFiltered:    wasFiltered,
		}
		var (
			prepValue     preparedData
			prepDataFound bool
		)
		if haveLast && lastKey == key {
			prepValue, prepDataFound = lastData, true
		} else if preparedDataByKey != nil {
			prepValue, prepDataFound = preparedDataByKey[key]
		}
		if !prepDataFound {
			var brokerDeltaPub *protocol.Publication
			if fullPub.Offset > 0 {
				brokerDeltaPub = getDeltaPub(prevPub, fullPub, key)
			}
			localDeltaPub := getDeltaPub(localPrevPub, fullPub, key)

			var brokerDeltaData []byte
			var localDeltaData []byte
			if key.DeltaType != deltaTypeNone {
				var err error
				brokerDeltaData, err = getDeltaData(sub, key, channel, brokerDeltaPub, channelSubID, &jsonEncodeErr)
				if err != nil {
					return err
				}
				localDeltaData, err = getDeltaData(sub, key, channel, localDeltaPub, channelSubID, &jsonEncodeErr)
				if err != nil {
					return err
				}
			}

			var fullData []byte

			if key.ProtocolType == protocol.TypeJSON {
				if sub.client.transport.Unidirectional() {
					pubToUse := fullPub
					if key.ProtocolType == protocol.TypeJSON && key.DeltaType == DeltaTypeFossil {
						pubToUse = &protocol.Publication{
							Offset:  fullPub.Offset,
							Data:    json.Escape(convert.BytesToString(fullPub.Data)),
							Info:    fullPub.Info,
							Tags:    fullPub.Tags,
							Channel: fullPub.Channel,
							Key:     fullPub.Key,
							Removed: fullPub.Removed,
							Score:   fullPub.Score,
						}
					}
					push := getPush()
					push.Pub = pubToUse
					if key.UseID {
						push.Id = channelSubID
					} else {
						push.Channel = channel
					}
					var err error
					fullData, err = protocol.DefaultJsonPushEncoder.Encode(push)
					putPush(push)
					if err != nil {
						jsonEncodeErr = &encodeError{client: sub.client.ID(), user: sub.client.UserID(), error: err}
					}
				} else {
					pubToUse := fullPub
					if key.ProtocolType == protocol.TypeJSON && key.DeltaType == DeltaTypeFossil {
						pubToUse = &protocol.Publication{
							Offset:  fullPub.Offset,
							Data:    json.Escape(convert.BytesToString(fullPub.Data)),
							Info:    fullPub.Info,
							Tags:    fullPub.Tags,
							Channel: fullPub.Channel,
							Key:     fullPub.Key,
							Removed: fullPub.Removed,
							Score:   fullPub.Score,
						}
					}
					push := getPush()
					push.Pub = pubToUse
					if key.UseID {
						push.Id = channelSubID
					} else {
						push.Channel = channel
					}
					reply := getReply()
					reply.Push = push
					var err error
					fullData, err = protocol.DefaultJsonReplyEncoder.Encode(reply)
					putReply(reply)
					putPush(push)
					if err != nil {
						jsonEncodeErr = &encodeError{client: sub.client.ID(), user: sub.client.UserID(), error: err}
					}
				}
			} else if key.ProtocolType == protocol.TypeProtobuf {
				if sub.client.transport.Unidirectional() {
					push := getPush()
					push.Pub = fullPub
					if key.UseID {
						push.Id = channelSubID
					} else {
						push.Channel = channel
					}
					var err error
					fullData, err = protocol.DefaultProtobufPushEncoder.Encode(push)
					putPush(push)
					if err != nil {
						return err
					}
				} else {
					push := getPush()
					push.Pub = fullPub
					if key.UseID {
						push.Id = channelSubID
					} else {
						push.Channel = channel
					}
					reply := getReply()
					reply.Push = push
					var err error
					fullData, err = protocol.DefaultProtobufReplyEncoder.Encode(reply)
					putReply(reply)
					putPush(push)
					if err != nil {
						return err
					}
				}
			}

			prepValue = preparedData{
				fullData:        fullData,
				brokerDeltaData: brokerDeltaData,
				localDeltaData:  localDeltaData,
				deltaSub:        key.DeltaType != deltaTypeNone,
				wasFiltered:     wasFiltered,
			}
			if wasFiltered {
				// One marker per publication is enough (same offset for all filtered
				// keys), but it must be attached to EVERY filtered key's prepValue.
				// Attaching it only to the first filtered key left later filtered keys
				// (e.g. a different protocol/delta/useID combo) with a nil filteredPub,
				// which the buffering path turns into a nil syncPub and then a nil entry
				// in the recovery merge — a nil-pointer panic.
				if filteredPub == nil {
					filteredPub = &protocol.Publication{
						Offset: fullPub.Offset,
						Time:   -1, // Use -1 for indicating filtered publication.
					}
				}
				prepValue.filteredPub = filteredPub
			}
			// Promote to the front cache. Only once a second distinct key shows up
			// is the map created, seeded with the entry the front cache is about to
			// evict, so an alternating pattern still hits the map rather than
			// re-encoding.
			if haveLast {
				if preparedDataByKey == nil {
					preparedDataByKey = make(map[preparedKey]preparedData, 4)
				}
				preparedDataByKey[lastKey] = lastData
				preparedDataByKey[key] = prepValue
			}
			lastKey, lastData, haveLast = key, prepValue, true
		}
		if sub.protoType == protocol.TypeJSON {
			if jsonEncodeErr != nil {
				go func(c *Client) { c.Disconnect(DisconnectInappropriateProtocol) }(sub.client)
				continue
			}
		}

		// Even filtered publications need to reach writePublication for offset tracking,
		// but they will be marked as filtered so the client can skip them without adding to the queue.
		_ = sub.client.writePublication(channel, fullPub, prepValue, sp, maxLagExceeded, batchConfig)
	}
	if jsonEncodeErr != nil && s.logger.enabled(LogLevelWarn) {
		// Log that we had clients with inappropriate protocol, and point to the first such client.
		s.logger.log(newLogEntry(LogLevelWarn, "inappropriate protocol publication", map[string]any{
			"channel": channel,
			"user":    jsonEncodeErr.user,
			"client":  jsonEncodeErr.client,
			"error":   jsonEncodeErr.error,
		}))
	}

	if tagsFilterDropped > 0 {
		s.metrics.incTagsFilterDropped(channel, tagsFilterDropped)
	}
	s.metrics.observeBroadcastDuration(now, channel)
	return nil
}

// broadcastJoin sends message to all clients subscribed on channel.
func (s *subShard) broadcastJoin(channel string, join *protocol.Join, batchConfig ChannelBatchConfig) error {
	s.mu.RLock()
	defer s.mu.RUnlock()

	channelSubscribers, ok := s.subs[channel]
	if !ok {
		return nil
	}

	var (
		jsonEncodeErr *encodeError
	)

	// Get subID for this channel if it exists
	channelSubID, hasSubID := s.chanIDs[channel]

	// See broadcastPublication: single-entry front cache, map only on mixed keys.
	var (
		lastKey           preparedKey
		lastData          []byte
		haveLast          bool
		preparedDataByKey map[preparedKey][]byte
	)

	for _, sub := range channelSubscribers {
		useChannelID := sub.useID && hasSubID
		key := preparedKey{
			ProtocolType:   sub.protoType,
			Unidirectional: sub.unidirectional,
			DeltaType:      deltaTypeNone, // Join messages don't use delta
			UseID:          useChannelID,
		}

		var (
			prepValue     []byte
			prepDataFound bool
		)
		if haveLast && lastKey == key {
			prepValue, prepDataFound = lastData, true
		} else if preparedDataByKey != nil {
			prepValue, prepDataFound = preparedDataByKey[key]
		}
		if !prepDataFound {
			var data []byte

			if key.ProtocolType == protocol.TypeJSON {
				push := getPush()
				push.Join = join
				if key.UseID {
					push.Id = channelSubID
				} else {
					push.Channel = channel
				}

				var err error
				if key.Unidirectional {
					data, err = protocol.DefaultJsonPushEncoder.Encode(push)
					putPush(push)
				} else {
					reply := getReply()
					reply.Push = push
					data, err = protocol.DefaultJsonReplyEncoder.Encode(reply)
					putReply(reply)
					putPush(push)
				}
				if err != nil {
					jsonEncodeErr = &encodeError{client: sub.client.ID(), user: sub.client.UserID(), error: err}
				}
			} else if key.ProtocolType == protocol.TypeProtobuf {
				push := getPush()
				push.Join = join
				if key.UseID {
					push.Id = channelSubID
				} else {
					push.Channel = channel
				}

				var err error
				if key.Unidirectional {
					data, err = protocol.DefaultProtobufPushEncoder.Encode(push)
					putPush(push)
				} else {
					reply := getReply()
					reply.Push = push
					data, err = protocol.DefaultProtobufReplyEncoder.Encode(reply)
					putReply(reply)
					putPush(push)
				}
				if err != nil {
					return err
				}
			}

			prepValue = data
			if haveLast {
				if preparedDataByKey == nil {
					preparedDataByKey = make(map[preparedKey][]byte, 4)
				}
				preparedDataByKey[lastKey] = lastData
				preparedDataByKey[key] = data
			}
			lastKey, lastData, haveLast = key, data, true
		}

		if sub.protoType == protocol.TypeJSON && jsonEncodeErr != nil {
			go func(c *Client) { c.Disconnect(DisconnectInappropriateProtocol) }(sub.client)
			continue
		}

		_ = sub.client.writeJoin(channel, join, prepValue, batchConfig)
	}

	if jsonEncodeErr != nil && s.logger.enabled(LogLevelWarn) {
		// Log that we had clients with inappropriate protocol, and point to the first such client.
		s.logger.log(newLogEntry(LogLevelWarn, "inappropriate protocol join", map[string]any{
			"channel": channel,
			"user":    jsonEncodeErr.user,
			"client":  jsonEncodeErr.client,
			"error":   jsonEncodeErr.error,
		}))
	}
	return nil
}

// broadcastLeave sends message to all clients subscribed on channel.
func (s *subShard) broadcastLeave(channel string, leave *protocol.Leave, batchConfig ChannelBatchConfig) error {
	s.mu.RLock()
	defer s.mu.RUnlock()

	channelSubscribers, ok := s.subs[channel]
	if !ok {
		return nil
	}

	var (
		jsonEncodeErr *encodeError
	)

	// Get subID for this channel if it exists
	channelSubID, hasSubID := s.chanIDs[channel]

	// See broadcastPublication: single-entry front cache, map only on mixed keys.
	var (
		lastKey           preparedKey
		lastData          []byte
		haveLast          bool
		preparedDataByKey map[preparedKey][]byte
	)

	for _, sub := range channelSubscribers {
		useChannelID := sub.useID && hasSubID
		key := preparedKey{
			ProtocolType:   sub.protoType,
			Unidirectional: sub.unidirectional,
			DeltaType:      deltaTypeNone, // Leave messages don't use delta
			UseID:          useChannelID,
		}

		var (
			prepValue     []byte
			prepDataFound bool
		)
		if haveLast && lastKey == key {
			prepValue, prepDataFound = lastData, true
		} else if preparedDataByKey != nil {
			prepValue, prepDataFound = preparedDataByKey[key]
		}
		if !prepDataFound {
			var data []byte

			if key.ProtocolType == protocol.TypeJSON {
				push := getPush()
				push.Leave = leave
				if key.UseID {
					push.Id = channelSubID
				} else {
					push.Channel = channel
				}

				var err error
				if key.Unidirectional {
					data, err = protocol.DefaultJsonPushEncoder.Encode(push)
					putPush(push)
				} else {
					reply := getReply()
					reply.Push = push
					data, err = protocol.DefaultJsonReplyEncoder.Encode(reply)
					putReply(reply)
					putPush(push)
				}
				if err != nil {
					jsonEncodeErr = &encodeError{client: sub.client.ID(), user: sub.client.UserID(), error: err}
				}
			} else if key.ProtocolType == protocol.TypeProtobuf {
				push := getPush()
				push.Leave = leave
				if key.UseID {
					push.Id = channelSubID
				} else {
					push.Channel = channel
				}

				var err error
				if key.Unidirectional {
					data, err = protocol.DefaultProtobufPushEncoder.Encode(push)
					putPush(push)
				} else {
					reply := getReply()
					reply.Push = push
					data, err = protocol.DefaultProtobufReplyEncoder.Encode(reply)
					putReply(reply)
					putPush(push)
				}
				if err != nil {
					return err
				}
			}

			prepValue = data
			if haveLast {
				if preparedDataByKey == nil {
					preparedDataByKey = make(map[preparedKey][]byte, 4)
				}
				preparedDataByKey[lastKey] = lastData
				preparedDataByKey[key] = data
			}
			lastKey, lastData, haveLast = key, data, true
		}

		if sub.protoType == protocol.TypeJSON && jsonEncodeErr != nil {
			go func(c *Client) { c.Disconnect(DisconnectInappropriateProtocol) }(sub.client)
			continue
		}

		_ = sub.client.writeLeave(channel, leave, prepValue, batchConfig)
	}

	if jsonEncodeErr != nil && s.logger.enabled(LogLevelWarn) {
		// Log that we had clients with inappropriate protocol, and point to the first such client.
		s.logger.log(newLogEntry(LogLevelWarn, "inappropriate protocol leave", map[string]any{
			"channel": channel,
			"user":    jsonEncodeErr.user,
			"client":  jsonEncodeErr.client,
			"error":   jsonEncodeErr.error,
		}))
	}
	return nil
}

// NumChannels returns a total number of different channels.
func (s *subShard) NumChannels() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return len(s.subs)
}

// NumSubscriptions returns total number of subscriptions.
func (s *subShard) NumSubscriptions() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.numSubs
}

// Channels returns a slice of all active channels.
func (s *subShard) Channels() []string {
	s.mu.RLock()
	defer s.mu.RUnlock()
	channels := make([]string, len(s.subs))
	i := 0
	for ch := range s.subs {
		channels[i] = ch
		i++
	}
	return channels
}

// NumSubscribers returns number of current subscribers for a given channel.
func (s *subShard) NumSubscribers(ch string) int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	clients, ok := s.subs[ch]
	if !ok {
		return 0
	}
	return len(clients)
}