Bose-SoundTouch/pkg/service/logbuf/logbuf.go

// Package logbuf provides a bounded, in-memory ring buffer for log
// lines. It implements io.Writer so it can be installed as a
// second sink under stdlib log via log.SetOutput(io.MultiWriter(
// os.Stderr, buf)). Each newline-terminated chunk written to the
// buffer becomes one Entry with a monotonic sequence number and a
// capture timestamp; the admin UI polls these entries and renders
// them as a live trace of the running service.
//
// The buffer holds at most Capacity entries; older entries are
// evicted FIFO when new ones arrive. Callers that fall behind can
// detect this via the dropped count returned by Since.
package logbuf

import (
	"bytes"
	"sync"
	"time"
)

// Entry is a single buffered log line. Message excludes the
// trailing newline. Seq is monotonic across the process lifetime
// of the Buffer (it keeps counting up even past evictions, so
// clients can use it as a high-water mark).
type Entry struct {
	Seq     uint64    `json:"seq"`
	Time    time.Time `json:"time"`
	Message string    `json:"message"`
}

// Buffer is a bounded, concurrent-safe in-memory log ring.
type Buffer struct {
	mu       sync.Mutex
	capacity int
	entries  []Entry // ring; oldest first
	nextSeq  uint64  // seq assigned to the next entry written
	partial  bytes.Buffer
	now      func() time.Time
}

// New returns a Buffer that holds up to capacity entries. A
// non-positive capacity is clamped to a small default (16) — a
// zero-capacity buffer would silently drop every line and make
// the feature look broken, which is worse than picking a number.
func New(capacity int) *Buffer {
	if capacity <= 0 {
		capacity = 16
	}

	return &Buffer{
		capacity: capacity,
		entries:  make([]Entry, 0, capacity),
		nextSeq:  1,
		now:      time.Now,
	}
}

// Write implements io.Writer. Bytes are split on '\n'; each
// complete line is appended as an Entry. A trailing partial line
// (no terminating newline) is held until the next Write supplies
// the rest. Returns len(p) on success (never reports a short
// write — the buffer always accepts the bytes, even when older
// entries get evicted).
func (b *Buffer) Write(p []byte) (int, error) {
	if len(p) == 0 {
		return 0, nil
	}

	b.mu.Lock()
	defer b.mu.Unlock()

	remaining := p
	for {
		idx := bytes.IndexByte(remaining, '\n')
		if idx < 0 {
			// No newline in the rest of p — stash it.
			b.partial.Write(remaining)
			break
		}

		// We have a complete line: anything held in `partial` +
		// remaining[:idx]. The newline itself is dropped.
		var line string

		if b.partial.Len() > 0 {
			b.partial.Write(remaining[:idx])
			line = b.partial.String()

			b.partial.Reset()
		} else {
			line = string(remaining[:idx])
		}

		b.append(line)

		remaining = remaining[idx+1:]
		if len(remaining) == 0 {
			break
		}
	}

	return len(p), nil
}

// append adds one Entry under the lock, evicting the oldest when
// the buffer is at capacity.
func (b *Buffer) append(message string) {
	entry := Entry{
		Seq:     b.nextSeq,
		Time:    b.now().UTC(),
		Message: message,
	}
	b.nextSeq++

	if len(b.entries) < b.capacity {
		b.entries = append(b.entries, entry)
		return
	}

	// At capacity: drop the oldest, slide left, append. A
	// circular index would avoid the copy, but at 2k entries
	// the copy is negligible and the slice API is simpler.
	copy(b.entries, b.entries[1:])
	b.entries[len(b.entries)-1] = entry
}

// Snapshot returns a copy of every currently buffered entry, in
// order of increasing Seq.
func (b *Buffer) Snapshot() []Entry {
	b.mu.Lock()
	defer b.mu.Unlock()

	out := make([]Entry, len(b.entries))
	copy(out, b.entries)

	return out
}

// Since returns all entries with Seq strictly greater than since,
// up to limit. nextSince is the highest Seq among the returned
// entries (or the input `since` when none match), suitable for
// the caller's next poll. dropped is the count of entries the
// caller missed — i.e. entries with Seq > since that have already
// been evicted from the buffer. A limit <= 0 means "no limit".
func (b *Buffer) Since(since uint64, limit int) (entries []Entry, nextSince, dropped uint64) {
	b.mu.Lock()
	defer b.mu.Unlock()

	nextSince = since

	if len(b.entries) == 0 {
		return nil, nextSince, 0
	}

	oldest := b.entries[0].Seq
	if since+1 < oldest {
		dropped = oldest - (since + 1)
	}

	// Binary-walking the ring isn't worth it at 2k entries.
	out := make([]Entry, 0, len(b.entries))

	for i := range b.entries {
		if b.entries[i].Seq <= since {
			continue
		}

		out = append(out, b.entries[i])
		nextSince = b.entries[i].Seq

		if limit > 0 && len(out) >= limit {
			break
		}
	}

	return out, nextSince, dropped
}

// Capacity returns the configured maximum number of entries.
func (b *Buffer) Capacity() int {
	return b.capacity
}

// Len returns the current count of buffered entries.
func (b *Buffer) Len() int {
	b.mu.Lock()
	defer b.mu.Unlock()

	return len(b.entries)
}