multica/server/internal/realtime/metrics.go

package realtime

import (
	"sort"
	"sync"
	"sync/atomic"
)

// Metrics collects lightweight counters describing the realtime subsystem.
//
// Phase 1 (MUL-1138) extends the phase-0 counter set with subscribe / Redis /
// per-scope-room counters. We keep using std-library atomics rather than a
// Prometheus dependency; a future phase can re-export the same numbers.
type Metrics struct {
	ConnectsTotal        atomic.Int64
	DisconnectsTotal     atomic.Int64
	ActiveConnections    atomic.Int64
	SlowEvictionsTotal   atomic.Int64
	MessagesSentTotal    atomic.Int64
	MessagesDroppedTotal atomic.Int64

	// Per-event-type send counters keyed by event type string.
	// Value is *atomic.Int64.
	eventSent sync.Map

	// Per-scope subscribe / unsubscribe / deny counters. Keyed by scope
	// type string ("workspace", "user", "task", "chat"). Value is
	// *atomic.Int64. Scope-room gauges follow the same pattern.
	subscribeTotal       sync.Map
	unsubscribeTotal     sync.Map
	subscribeDeniedTotal sync.Map
	scopeRooms           sync.Map

	// Redis relay counters. Zero unless the Redis broadcaster is enabled.
	RedisXAddTotal             atomic.Int64
	RedisXAddErrors            atomic.Int64
	RedisXReadTotal            atomic.Int64
	RedisXReadErrors           atomic.Int64
	RedisAckTotal              atomic.Int64
	RedisLastXAddLagMicros     atomic.Int64
	RedisMirrorPrimaryErrors   atomic.Int64
	RedisMirrorSecondaryErrors atomic.Int64
	RedisMirrorDivergenceTotal atomic.Int64

	// RedisConnected is set by the relay on startup / reconnect.
	RedisConnected atomic.Bool
	// RedisLastError stores the most recent consumer error message.
	redisLastErrMu sync.RWMutex
	redisLastErr   string

	// NodeID is set once at boot by the relay (or empty in single-node mode).
	NodeID atomic.Value // string
}

// M is the package-level metrics singleton.
var M = &Metrics{}

func loadOrInitCounter(m *sync.Map, key string) *atomic.Int64 {
	if v, ok := m.Load(key); ok {
		return v.(*atomic.Int64)
	}
	c := new(atomic.Int64)
	if existing, loaded := m.LoadOrStore(key, c); loaded {
		return existing.(*atomic.Int64)
	}
	return c
}

// RecordEvent increments the per-event-type send counter.
func (m *Metrics) RecordEvent(eventType string) {
	if eventType == "" {
		return
	}
	loadOrInitCounter(&m.eventSent, eventType).Add(1)
}

// SubscribesTotal returns the per-scope-type counter for successful subscribes.
func (m *Metrics) SubscribesTotal(scopeType string) *atomic.Int64 {
	return loadOrInitCounter(&m.subscribeTotal, scopeType)
}

// UnsubscribesTotal returns the per-scope-type counter for unsubscribes.
func (m *Metrics) UnsubscribesTotal(scopeType string) *atomic.Int64 {
	return loadOrInitCounter(&m.unsubscribeTotal, scopeType)
}

// SubscribeDeniedTotal returns the per-scope-type counter for denied subscribes.
func (m *Metrics) SubscribeDeniedTotal(scopeType string) *atomic.Int64 {
	return loadOrInitCounter(&m.subscribeDeniedTotal, scopeType)
}

// IncRoom / DecRoom adjust the active-rooms gauge for scopeType.
func (m *Metrics) IncRoom(scopeType string) { loadOrInitCounter(&m.scopeRooms, scopeType).Add(1) }
func (m *Metrics) DecRoom(scopeType string) { loadOrInitCounter(&m.scopeRooms, scopeType).Add(-1) }

// SetRedisLastError stores msg as the most recent Redis consumer error. An
// empty msg clears it.
func (m *Metrics) SetRedisLastError(msg string) {
	m.redisLastErrMu.Lock()
	m.redisLastErr = msg
	m.redisLastErrMu.Unlock()
}

func (m *Metrics) lastRedisErr() string {
	m.redisLastErrMu.RLock()
	defer m.redisLastErrMu.RUnlock()
	return m.redisLastErr
}

func snapshotCounters(s *sync.Map) map[string]int64 {
	out := map[string]int64{}
	s.Range(func(k, v any) bool {
		out[k.(string)] = v.(*atomic.Int64).Load()
		return true
	})
	keys := make([]string, 0, len(out))
	for k := range out {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	ordered := make(map[string]int64, len(out))
	for _, k := range keys {
		ordered[k] = out[k]
	}
	return ordered
}

// Snapshot returns a JSON-friendly copy of the current counter values.
func (m *Metrics) Snapshot() map[string]any {
	nodeID := ""
	if v := m.NodeID.Load(); v != nil {
		nodeID, _ = v.(string)
	}
	return map[string]any{
		"connects_total":         m.ConnectsTotal.Load(),
		"disconnects_total":      m.DisconnectsTotal.Load(),
		"active_connections":     m.ActiveConnections.Load(),
		"slow_evictions_total":   m.SlowEvictionsTotal.Load(),
		"messages_sent_total":    m.MessagesSentTotal.Load(),
		"messages_dropped_total": m.MessagesDroppedTotal.Load(),
		"events_sent_by_type":    snapshotCounters(&m.eventSent),
		"subscribes_total":       snapshotCounters(&m.subscribeTotal),
		"unsubscribes_total":     snapshotCounters(&m.unsubscribeTotal),
		"subscribe_denied_total": snapshotCounters(&m.subscribeDeniedTotal),
		"active_scope_rooms":     snapshotCounters(&m.scopeRooms),
		"redis": map[string]any{
			"connected":               m.RedisConnected.Load(),
			"node_id":                 nodeID,
			"xadd_total":              m.RedisXAddTotal.Load(),
			"xadd_errors":             m.RedisXAddErrors.Load(),
			"xread_total":             m.RedisXReadTotal.Load(),
			"xread_errors":            m.RedisXReadErrors.Load(),
			"ack_total":               m.RedisAckTotal.Load(),
			"last_xadd_lag_micros":    m.RedisLastXAddLagMicros.Load(),
			"mirror_primary_errors":   m.RedisMirrorPrimaryErrors.Load(),
			"mirror_secondary_errors": m.RedisMirrorSecondaryErrors.Load(),
			"mirror_divergence_total": m.RedisMirrorDivergenceTotal.Load(),
			"last_error":              m.lastRedisErr(),
		},
	}
}

// Reset zeroes all counters. Tests only.
func (m *Metrics) Reset() {
	m.ConnectsTotal.Store(0)
	m.DisconnectsTotal.Store(0)
	m.ActiveConnections.Store(0)
	m.SlowEvictionsTotal.Store(0)
	m.MessagesSentTotal.Store(0)
	m.MessagesDroppedTotal.Store(0)
	m.eventSent.Range(func(k, _ any) bool { m.eventSent.Delete(k); return true })
	m.subscribeTotal.Range(func(k, _ any) bool { m.subscribeTotal.Delete(k); return true })
	m.unsubscribeTotal.Range(func(k, _ any) bool { m.unsubscribeTotal.Delete(k); return true })
	m.subscribeDeniedTotal.Range(func(k, _ any) bool { m.subscribeDeniedTotal.Delete(k); return true })
	m.scopeRooms.Range(func(k, _ any) bool { m.scopeRooms.Delete(k); return true })
	m.RedisXAddTotal.Store(0)
	m.RedisXAddErrors.Store(0)
	m.RedisXReadTotal.Store(0)
	m.RedisXReadErrors.Store(0)
	m.RedisAckTotal.Store(0)
	m.RedisLastXAddLagMicros.Store(0)
	m.RedisMirrorPrimaryErrors.Store(0)
	m.RedisMirrorSecondaryErrors.Store(0)
	m.RedisMirrorDivergenceTotal.Store(0)
	m.RedisConnected.Store(false)
	m.SetRedisLastError("")
}