memby/server/internal/logging/logging.go

// Package logging configures the server's human-readable, structured log output.
package logging

import (
	"bufio"
	"context"
	"encoding/json"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// ParseLevel returns a supported slog level, defaulting to INFO for empty or invalid
// values. Keeping this forgiving prevents a typo in Docker configuration from stopping
// the gateway.
func ParseLevel(value string) slog.Level {
	switch strings.ToUpper(strings.TrimSpace(value)) {
	case "DEBUG":
		return slog.LevelDebug
	case "WARN", "WARNING":
		return slog.LevelWarn
	case "ERROR":
		return slog.LevelError
	default:
		return slog.LevelInfo
	}
}

// Format selects how a record is rendered. The wire shape of a log line is operational
// configuration, not a code decision: a person tailing `docker compose logs` and a
// collector shipping the same stream elsewhere want different things from it.
type Format int

const (
	// FormatConsole is the default: one aligned, human-readable line per event.
	FormatConsole Format = iota
	// FormatLogfmt is slog's own `time=… level=… msg=…` text output.
	FormatLogfmt
	// FormatJSON is one object per line, for a log collector.
	FormatJSON
)

// ParseFormat is forgiving for the same reason [ParseLevel] is — a typo in Docker
// configuration must not stop the gateway.
func ParseFormat(value string) Format {
	switch strings.ToUpper(strings.TrimSpace(value)) {
	case "JSON":
		return FormatJSON
	case "LOGFMT", "TEXT":
		return FormatLogfmt
	default:
		return FormatConsole
	}
}

// New returns a logger suited to `docker compose logs`: one compact line per event.
func New(w io.Writer, level slog.Leveler) *slog.Logger {
	logger, _ := NewBuffered(w, level, 0, FormatConsole)
	return logger
}

// Event is the browser-safe representation of one structured server log record.
type Event struct {
	Sequence   int64             `json:"sequence"`
	OccurredAt time.Time         `json:"occurredAt"`
	Level      string            `json:"level"`
	Message    string            `json:"message"`
	Attributes map[string]string `json:"attributes,omitempty"`
}

// EventPage is cursor based so an admin browser can drain bursts without repeatedly
// downloading records it has already rendered.
type EventPage struct {
	Events  []Event `json:"events"`
	Next    int64   `json:"next"`
	Oldest  int64   `json:"oldest"`
	Latest  int64   `json:"latest"`
	Dropped int64   `json:"dropped"`
	HasMore bool    `json:"hasMore"`
}

// Buffer is a bounded, concurrency-safe ring of recent structured log records.
// Bounding is important: a broken TV can generate traffic indefinitely. The configured
// ring retains enough context for an operator without growing or shifting on every write.
type Buffer struct {
	mu       sync.RWMutex
	capacity int
	events   []Event
	start    int
	next     atomic.Int64
	history  *historyFile
}

// historyFile is an append-only JSONL archive of the same structured events the admin
// console reads. It is compacted to the ring's retained tail at startup and after each
// further ringful, so persistence cannot become an unbounded disk cost.
type historyFile struct {
	mu            sync.Mutex
	path          string
	file          *os.File
	capacity      int
	lastCompacted int64
}

// ParseCapacity returns a non-negative log buffer capacity from configuration.
func ParseCapacity(value string, fallback int) int {
	capacity, err := strconv.Atoi(strings.TrimSpace(value))
	if err != nil || capacity < 0 {
		return fallback
	}
	return capacity
}

// NewBuffered writes lines in the requested format and mirrors accepted records into a
// ring buffer, which is what the admin page's live log reads.
func NewBuffered(
	w io.Writer, level slog.Leveler, capacity int, format Format,
) (*slog.Logger, *Buffer) {
	options := &slog.HandlerOptions{
		Level: level,
		ReplaceAttr: func(_ []string, attr slog.Attr) slog.Attr {
			if attr.Key == slog.TimeKey {
				return slog.String(slog.TimeKey, attr.Value.Time().UTC().Format(time.RFC3339))
			}
			return attr
		},
	}
	var written slog.Handler
	switch format {
	case FormatJSON:
		written = slog.NewJSONHandler(w, options)
	case FormatLogfmt:
		written = slog.NewTextHandler(w, options)
	default:
		written = newConsoleHandler(w, level)
	}
	buffer := &Buffer{capacity: capacity}
	if capacity <= 0 {
		return slog.New(written), buffer
	}
	return slog.New(&captureHandler{next: written, buffer: buffer, level: level}), buffer
}

// NewPersistentBuffered restores retained events from path before accepting new ones and
// appends every subsequent accepted record. The caller should close the returned Buffer.
// An empty path keeps the in-memory behaviour used by unit tests and small embeddings.
func NewPersistentBuffered(
	w io.Writer, level slog.Leveler, capacity int, format Format, path string,
) (*slog.Logger, *Buffer, error) {
	logger, buffer := NewBuffered(w, level, capacity, format)
	path = strings.TrimSpace(path)
	if capacity <= 0 || path == "" {
		return logger, buffer, nil
	}
	history, restored, err := openHistory(path, capacity)
	if err != nil {
		return nil, nil, err
	}
	buffer.history = history
	buffer.events = restored
	if len(restored) > 0 {
		buffer.next.Store(restored[len(restored)-1].Sequence)
	}
	return logger, buffer, nil
}

func openHistory(path string, capacity int) (*historyFile, []Event, error) {
	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
		return nil, nil, err
	}
	input, err := os.OpenFile(path, os.O_CREATE|os.O_RDONLY, 0o640)
	if err != nil {
		return nil, nil, err
	}
	restored := make([]Event, 0, capacity)
	scanner := bufio.NewScanner(input)
	// Error attributes are capped upstream, but allow headroom for structured records.
	scanner.Buffer(make([]byte, 64<<10), 1<<20)
	for scanner.Scan() {
		var event Event
		if json.Unmarshal(scanner.Bytes(), &event) != nil || event.Sequence <= 0 {
			continue
		}
		restored = append(restored, event)
		if len(restored) > capacity {
			copy(restored, restored[len(restored)-capacity:])
			restored = restored[:capacity]
		}
	}
	closeErr := input.Close()
	if err := scanner.Err(); err != nil {
		return nil, nil, err
	}
	if closeErr != nil {
		return nil, nil, closeErr
	}
	history := &historyFile{path: path, capacity: capacity}
	if len(restored) > 0 {
		history.lastCompacted = restored[len(restored)-1].Sequence
	}
	if err := history.rewrite(restored); err != nil {
		return nil, nil, err
	}
	return history, restored, nil
}

type captureHandler struct {
	next   slog.Handler
	buffer *Buffer
	level  slog.Leveler
	attrs  []slog.Attr
	groups []string
}

func (h *captureHandler) Enabled(ctx context.Context, level slog.Level) bool {
	return level >= h.level.Level() && h.next.Enabled(ctx, level)
}

func (h *captureHandler) Handle(ctx context.Context, record slog.Record) error {
	attributes := make(map[string]string, record.NumAttrs()+len(h.attrs))
	for _, attr := range h.attrs {
		addAttribute(attributes, h.groups, attr)
	}
	record.Attrs(func(attr slog.Attr) bool {
		addAttribute(attributes, h.groups, attr)
		return true
	})
	h.buffer.append(Event{
		OccurredAt: record.Time.UTC(),
		Level:      record.Level.String(),
		Message:    record.Message,
		Attributes: attributes,
	})
	return h.next.Handle(ctx, record)
}

func (h *captureHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
	clone := *h
	clone.next = h.next.WithAttrs(attrs)
	clone.attrs = append(append([]slog.Attr{}, h.attrs...), attrs...)
	return &clone
}

func (h *captureHandler) WithGroup(name string) slog.Handler {
	clone := *h
	clone.next = h.next.WithGroup(name)
	clone.groups = append(append([]string{}, h.groups...), name)
	return &clone
}

func addAttribute(target map[string]string, groups []string, attr slog.Attr) {
	attr.Value = attr.Value.Resolve()
	if attr.Equal(slog.Attr{}) {
		return
	}
	key := strings.Join(append(append([]string{}, groups...), attr.Key), ".")
	if attr.Value.Kind() == slog.KindGroup {
		for _, child := range attr.Value.Group() {
			addAttribute(target, append(groups, attr.Key), child)
		}
		return
	}
	target[key] = attributeValue(attr.Value)
}

func (b *Buffer) append(event Event) {
	event.Sequence = b.next.Add(1)
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.events) == b.capacity {
		b.events[b.start] = event
		b.start = (b.start + 1) % b.capacity
	} else {
		b.events = append(b.events, event)
	}
	if b.history != nil {
		ordered := b.orderedEventsLocked()
		_ = b.history.append(event, ordered)
	}
}

func (b *Buffer) orderedEventsLocked() []Event {
	ordered := make([]Event, len(b.events))
	for i := range b.events {
		ordered[i] = b.events[(b.start+i)%len(b.events)]
	}
	return ordered
}

func (h *historyFile) append(event Event, retained []Event) error {
	h.mu.Lock()
	defer h.mu.Unlock()
	if event.Sequence-h.lastCompacted >= int64(h.capacity) {
		if err := h.rewriteLocked(retained); err != nil {
			return err
		}
		h.lastCompacted = event.Sequence
		return nil
	}
	return json.NewEncoder(h.file).Encode(event)
}

func (h *historyFile) rewrite(events []Event) error {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.rewriteLocked(events)
}

func (h *historyFile) rewriteLocked(events []Event) error {
	if h.file != nil {
		if err := h.file.Close(); err != nil {
			return err
		}
	}
	file, err := os.OpenFile(h.path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o640)
	if err != nil {
		return err
	}
	encoder := json.NewEncoder(file)
	for _, event := range events {
		if err := encoder.Encode(event); err != nil {
			_ = file.Close()
			return err
		}
	}
	if err := file.Close(); err != nil {
		return err
	}
	h.file, err = os.OpenFile(h.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o640)
	return err
}

// Close flushes the persistent archive. It is safe for an in-memory Buffer.
func (b *Buffer) Close() error {
	if b == nil || b.history == nil {
		return nil
	}
	b.history.mu.Lock()
	defer b.history.mu.Unlock()
	if b.history.file == nil {
		return nil
	}
	return b.history.file.Close()
}

// Events returns records strictly newer than after, up to limit. If the caller fell
// behind the ring, Dropped reports the gap and delivery resumes at the oldest record.
func (b *Buffer) Events(after int64, limit int) EventPage {
	if limit < 1 {
		limit = 250
	}
	if limit > 1000 {
		limit = 1000
	}
	b.mu.RLock()
	defer b.mu.RUnlock()

	latest := b.next.Load()
	page := EventPage{Next: after, Latest: latest, Events: []Event{}}
	if len(b.events) == 0 {
		return page
	}
	page.Oldest = b.events[b.start].Sequence
	if after < page.Oldest-1 {
		page.Dropped = page.Oldest - after - 1
		after = page.Oldest - 1
	}
	start := len(b.events)
	for i := range b.events {
		event := b.events[(b.start+i)%len(b.events)]
		if event.Sequence > after {
			start = i
			break
		}
	}
	end := min(start+limit, len(b.events))
	for i := start; i < end; i++ {
		page.Events = append(page.Events, b.events[(b.start+i)%len(b.events)])
	}
	if len(page.Events) > 0 {
		page.Next = page.Events[len(page.Events)-1].Sequence
	}
	page.HasMore = page.Next < latest
	return page
}