Jarvis/internal/server/worker.go

package server

import (
	"context"
	"crypto/sha1"
	"encoding/hex"
	"encoding/json"
	"log/slog"
	"time"

	"gitea.alexandru.macocian.me/amacocian/jarvis/internal/alert"
	"gitea.alexandru.macocian.me/amacocian/jarvis/internal/diagnose"
	"gitea.alexandru.macocian.me/amacocian/jarvis/internal/store"
)

// job is a single alert queued for diagnosis.
type job struct {
	alert       alert.Alert
	groupLabels map[string]string
	externalURL string
	raw         []byte
}

// Worker consumes queued alerts, runs the diagnostic, and persists entries.
//
// A single worker processes alerts serially. That keeps concurrent Copilot
// CLI invocations (each heavyweight) to one at a time, which is the right
// default for a homelab; the queue absorbs bursts from grouped alerts.
type Worker struct {
	q     chan job
	diag  diagnose.Diagnoser
	store *store.Store
	log   *slog.Logger
}

// NewWorker builds a Worker with a buffered queue.
func NewWorker(diag diagnose.Diagnoser, st *store.Store, log *slog.Logger, queueSize int) *Worker {
	if queueSize <= 0 {
		queueSize = 128
	}
	return &Worker{
		q:     make(chan job, queueSize),
		diag:  diag,
		store: st,
		log:   log,
	}
}

// enqueue adds an alert to the queue. It reports whether the alert was
// accepted (false means the queue is full and the alert was dropped).
func (w *Worker) enqueue(j job) bool {
	select {
	case w.q <- j:
		return true
	default:
		return false
	}
}

// Run processes the queue until ctx is cancelled.
func (w *Worker) Run(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case j := <-w.q:
			w.process(ctx, j)
		}
	}
}

func (w *Worker) process(ctx context.Context, j job) {
	a := j.alert
	log := w.log.With("alert", a.Name(), "fingerprint", a.Fingerprint, "status", a.Status)

	diagMD, err := w.diag.Diagnose(ctx, diagnose.Request{
		Alert:       a,
		GroupLabels: j.groupLabels,
		ExternalURL: j.externalURL,
	})
	if err != nil {
		// Never drop the feed entry on diagnostic failure: record the alert
		// with the error so it still surfaces in the reader.
		log.Error("diagnose failed", "err", err)
		diagMD = "_Diagnostic failed: " + err.Error() + "_"
	}

	e := entryFor(a, j, diagMD)
	if err := w.store.Insert(ctx, e); err != nil {
		log.Error("store insert failed", "err", err)
		return
	}
	log.Info("recorded feed entry", "id", e.ID)
}

// entryFor builds a store.Entry from an alert and its diagnostic.
func entryFor(a alert.Alert, j job, diagMD string) store.Entry {
	now := time.Now().UTC()
	return store.Entry{
		ID:          entryID(a, now),
		Fingerprint: a.Fingerprint,
		Status:      a.Status,
		Title:       feedTitle(a),
		Alertname:   a.Labels["alertname"],
		Severity:    a.Labels["severity"],
		Host:        hostOf(a),
		Summary:     a.Summary(),
		Verdict:     diagnose.ParseVerdict(diagMD),
		Confidence:  diagnose.ParseConfidence(diagMD),
		Diagnostic:  diagMD,
		Payload:     string(j.raw),
		CreatedAt:   now,
	}
}

// entryID is a stable id per (alert instance, firing time). Two distinct
// firings of the same alert get distinct entries; a retried delivery of the
// same firing collapses to one (fingerprint+startsAt are identical).
func entryID(a alert.Alert, fallback time.Time) string {
	seed := a.Fingerprint + "|" + a.Status + "|"
	if !a.StartsAt.IsZero() {
		seed += a.StartsAt.UTC().Format(time.RFC3339)
	} else {
		seed += fallback.Format(time.RFC3339)
	}
	sum := sha1.Sum([]byte(seed))
	return hex.EncodeToString(sum[:])
}

func feedTitle(a alert.Alert) string {
	status := a.Status
	if status == "" {
		status = "firing"
	}
	sev := a.Labels["severity"]
	prefix := "[" + status
	if sev != "" {
		prefix += "/" + sev
	}
	prefix += "] "
	return prefix + a.Summary()
}

// hostOf best-effort extracts the host a host-resources alert refers to.
func hostOf(a alert.Alert) string {
	for _, k := range []string{"host_name", "instance", "hostname", "host"} {
		if v := a.Labels[k]; v != "" {
			return v
		}
	}
	return ""
}

// marshalAlert re-serializes a single alert for the entry's audit payload.
func marshalAlert(a alert.Alert) []byte {
	b, err := json.Marshal(a)
	if err != nil {
		return nil
	}
	return b
}