broad-announce/internal/tailhub/hub.go

// Package tailhub is the in-process fan-out for ingestd's
// "live tail" WebSocket endpoint (M5). It is the seam between
// the producers (HTTP, MQTT, WS ingest paths, all of which call
// Publish after a successful ProcessAlert) and the consumers
// (one or more /v1/tail/ws clients).
//
// Why in-process and not NATS: the tail is an operator's
// debugging tool, not a feature for end users. The event rate
// is at most the ingest rate (~k/s in dev), and a single
// ingestd replica can handle thousands of subscribers. M9
// promotes this to NATS KV if the fan-out ever needs to span
// replicas.
package tailhub

import (
	"sync"
	"sync/atomic"

	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
)

// Event is the compact representation of an accepted alert that
// the tail endpoint streams to subscribers. The full alert
// payload (which can be 256 KB) is intentionally NOT included —
// the operator sees enough metadata to filter, then drills into
// the per-alert GET endpoint (M9) for the body.
type Event struct {
	AlertID    string  `json:"alert_id"`
	CompanyID  string  `json:"company_id"`
	SourceID   string  `json:"source_id"`
	Severity   string  `json:"severity"`
	Title      string  `json:"title"`
	ReceivedAt string  `json:"received_at"`
	Transport  string  `json:"transport"`
	DedupeCount uint32 `json:"dedupe_count"`
}

// FromAlert is the constructor used by ingestd after a
// successful ProcessAlert. We pull the title out of the data
// field; the alert package doesn't currently expose a top-level
// Title, so we use the first 120 chars of body if available.
func FromAlert(a *alert.Alert, transport string) *Event {
	title := a.Title
	if title == "" {
		title = a.Category + " alert"
	}
	if len(title) > 120 {
		title = title[:120]
	}
	recv := ""
	if !a.ReceivedAt.IsZero() {
		recv = a.ReceivedAt.UTC().Format("2006-01-02T15:04:05.000000000Z")
	}
	return &Event{
		AlertID:    a.ID,
		CompanyID:  a.CompanyID,
		SourceID:   a.SourceID,
		Severity:   string(a.Severity),
		Title:      title,
		ReceivedAt: recv,
		Transport:  transport,
		DedupeCount: a.DedupeCount,
	}
}

// Filter is the subscription filter. Empty CompanyID matches all.
type Filter struct {
	CompanyID string
}

// Hub is the in-process pub/sub for tail events. Producers call
// Publish; consumers call Subscribe and read from the returned
// channel until they Close it.
type Hub struct {
	mu         sync.RWMutex
	subs       map[*Subscription]struct{}
	nextID     atomic.Uint64
	droppedTotal atomic.Uint64
	publishedTotal atomic.Uint64
}

// Subscription is a per-consumer handle. The consumer reads from
// C until closed; missed events while C is full are counted in
// Hub.Stats() and logged by the consumer (which can choose to
// disconnect on its own).
type Subscription struct {
	id     uint64
	hub    *Hub
	filter Filter
	C      chan *Event
	// Drops is the per-subscription counter; bumped when the
	// producer side can't write to C (consumer too slow).
	Drops atomic.Uint64
}

const subscriberBuffer = 64

// NewHub creates an empty hub.
func NewHub() *Hub {
	return &Hub{subs: make(map[*Subscription]struct{})}
}

// Subscribe registers a new consumer with the given filter.
// The returned channel is closed when the hub is closed or the
// caller invokes Unsubscribe.
func (h *Hub) Subscribe(filter Filter) *Subscription {
	s := &Subscription{
		id:     h.nextID.Add(1),
		hub:    h,
		filter: filter,
		C:      make(chan *Event, subscriberBuffer),
	}
	h.mu.Lock()
	h.subs[s] = struct{}{}
	h.mu.Unlock()
	return s
}

// Unsubscribe removes the subscription and closes its channel.
// Safe to call multiple times.
func (s *Subscription) Unsubscribe() {
	s.hub.mu.Lock()
	if _, ok := s.hub.subs[s]; ok {
		delete(s.hub.subs, s)
		close(s.C)
	}
	s.hub.mu.Unlock()
}

// Publish fans out the event to every matching subscriber.
// A subscriber whose channel is full has the event dropped (and
// the per-subscription Drops counter incremented). The producer
// never blocks.
func (h *Hub) Publish(ev *Event) {
	if ev == nil {
		return
	}
	h.publishedTotal.Add(1)
	h.mu.RLock()
	defer h.mu.RUnlock()
	for s := range h.subs {
		if s.filter.CompanyID != "" && s.filter.CompanyID != ev.CompanyID {
			continue
		}
		select {
		case s.C <- ev:
		default:
			s.Drops.Add(1)
			h.droppedTotal.Add(1)
		}
	}
}

// Stats is a snapshot of hub-level counters (for /v1/debug/tail
// and the Prometheus gauge).
type Stats struct {
	Subscribers     int
	PublishedTotal  uint64
	DroppedTotal    uint64
}

// Stats returns a snapshot of the hub's counters.
func (h *Hub) Stats() Stats {
	h.mu.RLock()
	n := len(h.subs)
	h.mu.RUnlock()
	return Stats{
		Subscribers:    n,
		PublishedTotal: h.publishedTotal.Load(),
		DroppedTotal:   h.droppedTotal.Load(),
	}
}