visionA/local-agent/server/pkg/logger/broadcaster.go

package logger

import (
	"sync"
	"time"
)

// LogEntry represents a single structured log entry for WebSocket streaming.
type LogEntry struct {
	Timestamp string `json:"timestamp"`
	Level     string `json:"level"`
	Message   string `json:"message"`
}

// BroadcastFunc is called whenever a new log entry is produced.
type BroadcastFunc func(entry LogEntry)

// Broadcaster captures log output, maintains a ring buffer of recent entries,
// and notifies subscribers (via BroadcastFunc) in real time.
type Broadcaster struct {
	mu        sync.RWMutex
	buffer    []LogEntry
	bufSize   int
	pos       int
	full      bool
	broadcast BroadcastFunc
}

// NewBroadcaster creates a broadcaster with the given ring buffer capacity.
func NewBroadcaster(bufferSize int, fn BroadcastFunc) *Broadcaster {
	return &Broadcaster{
		buffer:    make([]LogEntry, bufferSize),
		bufSize:   bufferSize,
		broadcast: fn,
	}
}

// Push adds a log entry to the ring buffer and broadcasts it.
func (b *Broadcaster) Push(level, message string) {
	entry := LogEntry{
		Timestamp: time.Now().Format("2006-01-02 15:04:05"),
		Level:     level,
		Message:   message,
	}

	b.mu.Lock()
	b.buffer[b.pos] = entry
	b.pos = (b.pos + 1) % b.bufSize
	if b.pos == 0 {
		b.full = true
	}
	b.mu.Unlock()

	if b.broadcast != nil {
		b.broadcast(entry)
	}
}

// Recent returns a copy of all buffered log entries in chronological order.
func (b *Broadcaster) Recent() []LogEntry {
	b.mu.RLock()
	defer b.mu.RUnlock()

	if !b.full {
		result := make([]LogEntry, b.pos)
		copy(result, b.buffer[:b.pos])
		return result
	}
	result := make([]LogEntry, b.bufSize)
	copy(result, b.buffer[b.pos:])
	copy(result[b.bufSize-b.pos:], b.buffer[:b.pos])
	return result
}