panel/agent/internal/dispatch/logcoalesce.go

package dispatch

import (
	"regexp"
	"sync"
	"time"

	"github.com/dbledeez/panel/agent/internal/runtime"
)

// logCoalescer throttles repetitive progress-style log lines while letting
// genuine content through unchanged.
//
// Why this exists: containers under install/update pressure (SteamCMD
// download ticks, preallocation, LGSM step banners) can emit 40+ lines per
// second. Forwarding every one to the controller saturates the gRPC send
// window, fills the event bus's per-subscriber buffer (which silently drops
// on overflow), and burns CPU re-rendering DOM. The browser only needs to
// see the *latest* progress value, not every interstitial frame.
//
// Behaviour:
//   - Lines with progressKey() == "": emit immediately.
//   - Lines with a key: if it's been less than `interval` since the last
//     emit of that key, replace the currently-buffered line for that key.
//     A single background timer flushes any pending lines every `interval`.
type logCoalescer struct {
	interval time.Duration
	emit     func(stream runtime.LogStream, line string, at time.Time)

	mu      sync.Mutex
	pending map[string]*pendingLine
	lastAt  map[string]time.Time
	timer   *time.Timer
	stopped bool
}

type pendingLine struct {
	stream runtime.LogStream
	line   string
	at     time.Time
}

func newLogCoalescer(interval time.Duration, emit func(stream runtime.LogStream, line string, at time.Time)) *logCoalescer {
	c := &logCoalescer{
		interval: interval,
		emit:     emit,
		pending:  map[string]*pendingLine{},
		lastAt:   map[string]time.Time{},
	}
	return c
}

// offer delivers a line to the coalescer. Safe to call concurrently, though
// in practice the caller is the single log-scan goroutine.
func (c *logCoalescer) offer(stream runtime.LogStream, line string, at time.Time) {
	key := progressKey(line)
	if key == "" {
		// Non-progress — emit immediately, but first flush any buffered
		// progress lines so they don't appear AFTER a later non-progress line.
		c.mu.Lock()
		c.flushLocked()
		c.mu.Unlock()
		c.emit(stream, line, at)
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.stopped {
		return
	}
	last := c.lastAt[key]
	if last.IsZero() || at.Sub(last) >= c.interval {
		// Enough quiet time — emit right away.
		c.emit(stream, line, at)
		c.lastAt[key] = at
		delete(c.pending, key)
		return
	}
	// Replace the buffered line for this key (we only care about the newest).
	c.pending[key] = &pendingLine{stream: stream, line: line, at: at}
	// Make sure the timer is armed. Single timer drains all pending keys.
	if c.timer == nil {
		c.timer = time.AfterFunc(c.interval, c.tick)
	}
}

func (c *logCoalescer) tick() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.flushLocked()
	// If more pending arrived during flush (or we expect more), re-arm the timer.
	if len(c.pending) > 0 && !c.stopped {
		c.timer = time.AfterFunc(c.interval, c.tick)
	} else {
		c.timer = nil
	}
}

func (c *logCoalescer) flushLocked() {
	now := time.Now()
	for key, p := range c.pending {
		c.emit(p.stream, p.line, p.at)
		c.lastAt[key] = now
		delete(c.pending, key)
	}
}

// stop flushes any pending lines and prevents further work. Call from the
// streamLogs goroutine's defer so a stopping instance doesn't leave buffered
// progress frames undelivered.
func (c *logCoalescer) stop() {
	c.mu.Lock()
	c.stopped = true
	if c.timer != nil {
		c.timer.Stop()
		c.timer = nil
	}
	c.flushLocked()
	c.mu.Unlock()
}

// Regexes for lines we treat as part of a repeating progress stream.
// Each group returns a stable key — same key = same progress counter.
var (
	// SteamCMD: "Update state (0x61) downloading, progress: 47.95 (...)"
	// Also covers "verifying", "preallocating", "committing" stages.
	reSteamcmdProgress = regexp.MustCompile(`Update state \([^)]+\) (\w+), progress:`)
	// Generic "Progress: 64%" or "[ 64%] Installing ...".
	reGenericProgress = regexp.MustCompile(`^\s*(?:Progress:|\[\s*\d+%\s*\])`)
	// LGSM step banner: "Starting sdtdserver: <step>". vinanrra emits
	// several per step as the banner animates ([ .... ] → [ INFO ] → [  OK  ]).
	reLGSMStarting = regexp.MustCompile(`^\s*Starting sdtdserver:`)
	reLGSMStopping = regexp.MustCompile(`^\s*Stopping sdtdserver:`)
	// Docker pull: "Downloading [=>] 12.3MB / 100MB"
	reDockerPull = regexp.MustCompile(`^\s*(?:Downloading|Extracting|Pulling)\s+[\[(]`)
)

// progressKey returns a stable identifier for a progress-style line, or ""
// if the line should be forwarded unchanged. The returned key collapses
// every frame of the same counter onto a single slot so we keep only the
// latest within each coalesce window.
func progressKey(text string) string {
	if m := reSteamcmdProgress.FindStringSubmatch(text); m != nil {
		return "steamcmd:" + m[1]
	}
	if reGenericProgress.MatchString(text) {
		return "progress"
	}
	if reLGSMStarting.MatchString(text) {
		return "lgsm-starting"
	}
	if reLGSMStopping.MatchString(text) {
		return "lgsm-stopping"
	}
	if reDockerPull.MatchString(text) {
		return "docker-pull"
	}
	return ""
}