worker/internal/distworker/results.go

package distworker

import (
	"sync"
	"time"

	"rocketgit.ru/rsmon/worker/internal/wire"
)

// recentResultsSize is the in-memory ring buffer capacity for the
// last N check results produced by the worker. The webapp reads
// from this buffer for the /checks page and the recent-results
// counters on /overview.
const recentResultsSize = 200

// recentNotificationsSize mirrors recentResultsSize for emitted
// notifications.
const recentNotificationsSize = 100

// ResultRow is one row from the worker's in-memory result ring
// buffer. Kept in the distworker package so the webapp can read it
// without going through wire (which is a payload envelope, not a
// stable render type).
type ResultRow struct {
	MonitorID  int64
	CheckID    int64
	Kind       string
	Host       string
	State      string
	DurationMs int64
	Error      string
	At         time.Time
}

// NotificationRow is one row from the worker's notification ring buffer.
// Delegated rows use only JobID, Method, Status, DurationMs, and At. They
// deliberately omit delivery inputs and provider output.
type NotificationRow struct {
	Kind    string // "email", "telegram_private", "telegram_group"
	Channel string
	Subject string
	Body    string
	OK      bool
	Error   string
	At      time.Time

	JobID      string
	Method     string
	Status     string
	DurationMs int
}

// resultBuffer is a thread-safe FIFO ring buffer of ResultRow. The
// Runner owns one; both the dispatcher goroutine and the webapp
// readers can touch it concurrently.
type resultBuffer struct {
	mu   sync.RWMutex
	buf  []ResultRow
	head int
	size int
}

func newResultBuffer() *resultBuffer {
	return &resultBuffer{buf: make([]ResultRow, recentResultsSize)}
}

// add appends one entry, evicting the oldest when full.
func (b *resultBuffer) add(r *ResultRow) {
	if b == nil {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.buf[b.head] = *r
	b.head = (b.head + 1) % len(b.buf)
	if b.size < len(b.buf) {
		b.size++
	}
}

// snapshot returns the most recent n rows in chronological order.
// n <= 0 returns an empty slice.
func (b *resultBuffer) snapshot(n int) []ResultRow {
	return ringSnapshot(&b.mu, &b.head, &b.size, &b.buf, n)
}

// notificationBuffer mirrors resultBuffer for NotificationRow.
type notificationBuffer struct {
	mu   sync.RWMutex
	buf  []NotificationRow
	head int
	size int
}

func newNotificationBuffer() *notificationBuffer {
	return &notificationBuffer{buf: make([]NotificationRow, recentNotificationsSize)}
}

func (b *notificationBuffer) add(r *NotificationRow) {
	if b == nil {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.buf[b.head] = *r
	b.head = (b.head + 1) % len(b.buf)
	if b.size < len(b.buf) {
		b.size++
	}
}

func (b *notificationBuffer) snapshot(n int) []NotificationRow {
	return ringSnapshot(&b.mu, &b.head, &b.size, &b.buf, n)
}

// ringSnapshot is a generic FIFO ring-buffer snapshot. The caller
// passes the mutex/head/size/buf by pointer; the lock is taken
// while reading. n <= 0 returns nil. The returned slice is a copy
// so callers can hand it to non-locking code paths (e.g. the
// webapp's templates) without holding the buffer lock.
func ringSnapshot[T any](mu *sync.RWMutex, head, size *int, buf *[]T, n int) []T {
	if mu == nil || n <= 0 {
		return nil
	}
	mu.RLock()
	defer mu.RUnlock()
	if *size == 0 {
		return nil
	}
	if n > *size {
		n = *size
	}
	out := make([]T, 0, n)
	start := (*head - n + len(*buf)) % len(*buf)
	for i := 0; i < n; i++ {
		idx := (start + i) % len(*buf)
		out = append(out, (*buf)[idx])
	}
	return out
}

// resultRowFromReport converts a wire.CheckResultReport into the
// internal ResultRow type the webapp renders.
func resultRowFromReport(env *resultEnvelope, report *wire.CheckResultReport, at time.Time) *ResultRow {
	row := &ResultRow{
		MonitorID:  report.MonitorID,
		CheckID:    report.CheckID,
		Kind:       env.job.Kind,
		Host:       env.job.Host,
		State:      report.State,
		DurationMs: report.DurationMs,
		At:         at,
	}
	if report.Error != nil {
		row.Error = *report.Error
	}
	return row
}