gotail/backend/internal/broker/broker.go

package broker

import (
	"context"
	"sync"
	"time"

	"github.com/rs/zerolog/log"
	"github.com/seu-usuario/go-react-web-tail/internal/models"
)

// Broker gerencia clientes SSE e broadcasting de logs
type Broker struct {
	clients    map[chan models.LogEntry]struct{}
	mu         sync.RWMutex
	register   chan chan models.LogEntry
	unregister chan chan models.LogEntry
	broadcast  chan models.LogEntry
	ctx        context.Context
	cancel     context.CancelFunc

	// Métricas
	totalBroadcasts uint64
	activeClients   int
}

// New cria uma nova instância do Broker
func New(ctx context.Context) *Broker {
	ctx, cancel := context.WithCancel(ctx)
	b := &Broker{
		clients:    make(map[chan models.LogEntry]struct{}),
		register:   make(chan chan models.LogEntry),
		unregister: make(chan chan models.LogEntry),
		broadcast:  make(chan models.LogEntry, 1000),
		ctx:        ctx,
		cancel:     cancel,
	}
	go b.run()
	return b
}

// run é o loop principal do broker
func (b *Broker) run() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	log.Info().Msg("Broker started")

	for {
		select {
		case <-b.ctx.Done():
			log.Info().Msg("Broker shutting down")
			b.closeAllClients()
			return

		case client := <-b.register:
			b.addClient(client)

		case client := <-b.unregister:
			b.removeClient(client)

		case entry := <-b.broadcast:
			b.broadcastToClients(entry)

		case <-ticker.C:
			b.logMetrics()
		}
	}
}

// Subscribe registra um novo cliente para receber logs
func (b *Broker) Subscribe() chan models.LogEntry {
	ch := make(chan models.LogEntry, 100)
	b.register <- ch
	return ch
}

// Unsubscribe remove um cliente
func (b *Broker) Unsubscribe(ch chan models.LogEntry) {
	b.unregister <- ch
}

// Broadcast envia uma entrada de log para todos os clientes
func (b *Broker) Broadcast(entry models.LogEntry) {
	select {
	case b.broadcast <- entry:
	case <-b.ctx.Done():
	default:
		log.Warn().Msg("Broadcast channel full, dropping message")
	}
}

// Shutdown encerra o broker graciosamente
func (b *Broker) Shutdown() {
	log.Info().Msg("Broker shutdown requested")
	b.cancel()
}

// addClient adiciona um novo cliente
func (b *Broker) addClient(client chan models.LogEntry) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.clients[client] = struct{}{}
	b.activeClients = len(b.clients)
	log.Info().Int("active_clients", b.activeClients).Msg("Client subscribed")
}

// removeClient remove um cliente
func (b *Broker) removeClient(client chan models.LogEntry) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if _, ok := b.clients[client]; ok {
		delete(b.clients, client)
		close(client)
		b.activeClients = len(b.clients)
		log.Info().Int("active_clients", b.activeClients).Msg("Client unsubscribed")
	}
}

// broadcastToClients envia uma entrada para todos os clientes conectados
func (b *Broker) broadcastToClients(entry models.LogEntry) {
	b.mu.RLock()
	defer b.mu.RUnlock()

	b.totalBroadcasts++

	for clientCh := range b.clients {
		select {
		case clientCh <- entry:
			// Enviado com sucesso
		default:
			// Cliente está lento, remover
			go b.Unsubscribe(clientCh)
			log.Warn().Msg("Removed slow client")
		}
	}
}

// closeAllClients fecha todos os canais de clientes
func (b *Broker) closeAllClients() {
	b.mu.Lock()
	defer b.mu.Unlock()
	for client := range b.clients {
		close(client)
	}
	b.clients = make(map[chan models.LogEntry]struct{})
	log.Info().Msg("All clients disconnected")
}

// logMetrics registra métricas do broker
func (b *Broker) logMetrics() {
	b.mu.RLock()
	defer b.mu.RUnlock()
	log.Debug().
		Int("active_clients", b.activeClients).
		Uint64("total_broadcasts", b.totalBroadcasts).
		Int("broadcast_buffer", len(b.broadcast)).
		Msg("Broker metrics")
}

// GetMetrics retorna métricas atuais do broker
func (b *Broker) GetMetrics() (activeClients int, totalBroadcasts uint64) {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return b.activeClients, b.totalBroadcasts
}