nmon/internal/notify/queue.go

package notify

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"sync"
	"time"

	"github.com/v10networks/nmon/internal/clock"
	"github.com/v10networks/nmon/internal/netpolicy"
	"github.com/v10networks/nmon/internal/selfmon"
	"github.com/v10networks/nmon/internal/store"
)

// Retry schedule: Backoff(n) after the n-th failed attempt, for MaxAge from
// enqueue (the first attempt is due immediately), then the delivery is failed.
const (
	BackoffBase   = 10 * time.Second
	BackoffMax    = 5 * time.Minute
	DefaultMaxAge = time.Hour
)

// DefaultBackoff is 10 s, 20 s, 40 s ... capped at 5 min.
func DefaultBackoff(failed int) time.Duration {
	d := BackoffBase
	for i := 1; i < failed && d < BackoffMax; i++ {
		d *= 2
	}
	return min(d, BackoffMax)
}

// Options configures a Queue.
type Options struct {
	Store     store.Deliveries
	Clock     clock.Clock // default clock.Real()
	Receivers []Receiver
	// Policy gates every outbound connection; default netpolicy.Default().
	Policy *netpolicy.Policy
	// HTTPClient overrides Policy.HTTPClient(Timeout) (tests).
	HTTPClient *http.Client
	// TLSConfig is the base config for SMTP STARTTLS (tests inject roots).
	TLSConfig *tls.Config
	Workers   int           // default 4
	Timeout   time.Duration // per attempt, default 15 s
	Poll      time.Duration // idle poll interval, default 1 s
	MaxAge    time.Duration // default DefaultMaxAge
	Backoff   func(failed int) time.Duration
	Registry  *selfmon.Registry
	Logger    *slog.Logger
}

// Queue is the delivery queue and its worker pool.
type Queue struct {
	o    Options
	clk  clock.Clock
	http *http.Client

	mu   sync.RWMutex
	recv map[string]*compiled

	wake chan struct{}

	// testSlots bounds concurrent SendTest calls.
	testSlots chan struct{}
}

// New validates the receivers (templates are parsed and dry-run) and returns a Queue.
func New(o Options) (*Queue, error) {
	if o.Store == nil {
		return nil, errors.New("notify: no store")
	}
	if o.Clock == nil {
		o.Clock = clock.Real()
	}
	if o.Policy == nil {
		o.Policy = netpolicy.Default()
	}
	if o.Workers <= 0 {
		o.Workers = 4
	}
	if o.Timeout <= 0 {
		o.Timeout = 15 * time.Second
	}
	if o.Poll <= 0 {
		o.Poll = time.Second
	}
	if o.MaxAge <= 0 {
		o.MaxAge = DefaultMaxAge
	}
	if o.Backoff == nil {
		o.Backoff = DefaultBackoff
	}
	if o.Registry == nil {
		o.Registry = selfmon.Default
	}
	if o.Logger == nil {
		o.Logger = slog.Default()
	}
	q := &Queue{o: o, clk: o.Clock, wake: make(chan struct{}, 1), testSlots: make(chan struct{}, 2)}
	q.http = o.HTTPClient
	if q.http == nil {
		q.http = o.Policy.HTTPClient(o.Timeout)
	}
	if err := q.SetReceivers(o.Receivers); err != nil {
		return nil, err
	}
	return q, nil
}

// SetReceivers replaces the receiver set (config reload). Pending deliveries
// to a receiver that no longer exists fail permanently when they come due.
func (q *Queue) SetReceivers(rs []Receiver) error {
	m := make(map[string]*compiled, len(rs))
	for _, r := range rs {
		c, err := compile(r)
		if err != nil {
			return err
		}
		if m[r.Name] != nil {
			return fmt.Errorf("duplicate receiver %q", r.Name)
		}
		m[r.Name] = c
	}
	q.mu.Lock()
	q.recv = m
	q.mu.Unlock()
	return nil
}

func (q *Queue) receiver(name string) *compiled {
	q.mu.RLock()
	defer q.mu.RUnlock()
	return q.recv[name]
}

// Enqueue renders n for the named receiver, redacts it, and stores a pending
// delivery that is due now. It returns the delivery ID. An empty IncidentID is
// filled from AlertID and FiredAt. A template that fails to render is an error
// and nothing is queued.
func (q *Queue) Enqueue(ctx context.Context, receiver string, n Notification) (int64, error) {
	c := q.receiver(receiver)
	if c == nil {
		return 0, fmt.Errorf("notify: unknown receiver %q", receiver)
	}
	if n.AlertID == 0 {
		return 0, errors.New("notify: notification has no alert id")
	}
	if n.IncidentID == "" {
		n.IncidentID = IncidentID(n.AlertID, n.FiredAt)
	}
	payload, err := c.render(n)
	if err != nil {
		return 0, err
	}
	now, exp := q.clk.Now(), q.clk.Now().Add(q.o.MaxAge)
	d := &store.Delivery{AlertID: n.AlertID, Receiver: receiver, IncidentID: n.IncidentID, Payload: payload,
		NextAttemptAt: &now, ExpiresAt: &exp}
	if err := q.o.Store.EnqueueDelivery(ctx, d); err != nil {
		return 0, err
	}
	select {
	case q.wake <- struct{}{}:
	default:
	}
	return d.ID, nil
}

// Run starts the worker pool and blocks until ctx is done. Pending deliveries
// left in the database by an earlier process are picked up like new ones.
func (q *Queue) Run(ctx context.Context) error {
	var wg sync.WaitGroup
	for range q.o.Workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			q.worker(ctx)
		}()
	}
	wg.Wait()
	return ctx.Err()
}

func (q *Queue) worker(ctx context.Context) {
	for ctx.Err() == nil {
		n, err := q.runBatch(ctx, 1)
		if err != nil && ctx.Err() == nil {
			q.o.Logger.Error("notify: claim failed", "err", err)
		}
		if n > 0 {
			continue
		}
		select {
		case <-ctx.Done():
			return
		case <-q.wake:
		case <-q.clk.After(q.o.Poll):
		}
	}
}

// RunOnce delivers every delivery due now, serially, and returns how many
// attempts it made. Run is the production loop; RunOnce serves tests and
// shutdown flushes.
func (q *Queue) RunOnce(ctx context.Context) (int, error) {
	total := 0
	for {
		n, err := q.runBatch(ctx, 50)
		total += n
		if err != nil || n == 0 {
			return total, err
		}
	}
}

func (q *Queue) runBatch(ctx context.Context, limit int) (int, error) {
	lease := q.o.Timeout*2 + 30*time.Second
	ds, err := q.o.Store.ClaimDueDeliveries(ctx, q.clk.Now(), lease, limit)
	if err != nil {
		return 0, err
	}
	for _, d := range ds {
		q.deliver(ctx, d)
	}
	return len(ds), nil
}

// permanent marks an error that retrying cannot fix.
type permanent struct{ err error }

func (p permanent) Error() string { return p.err.Error() }
func (p permanent) Unwrap() error { return p.err }

// deliver makes one attempt for d and records the outcome.
func (q *Queue) deliver(ctx context.Context, d store.Delivery) {
	err := q.send(ctx, d)
	if err != nil && ctx.Err() != nil {
		return // shutting down: the lease expires and the next process retries
	}
	a := q.outcome(d, err)
	if rerr := q.o.Store.RecordDeliveryAttempt(ctx, d.ID, a); rerr != nil {
		q.o.Logger.Error("notify: recording attempt failed", "receiver", d.Receiver, "incident", d.IncidentID, "err", rerr)
		return
	}
	q.report(d, a)
}

// send runs the transport for d's receiver under the per-attempt timeout.
func (q *Queue) send(ctx context.Context, d store.Delivery) error {
	c := q.receiver(d.Receiver)
	if c == nil {
		return permanent{fmt.Errorf("receiver %q is not configured", d.Receiver)}
	}
	actx, cancel := context.WithTimeout(ctx, q.o.Timeout)
	defer cancel()
	switch c.Kind {
	case KindWebhook:
		return q.sendWebhook(actx, c, d)
	case KindSMTP:
		return q.sendSMTP(actx, c, d)
	}
	return nil
}

// outcome classifies an attempt's error: success, permanent failure, expired
// (gave up) or retry with backoff.
func (q *Queue) outcome(d store.Delivery, err error) store.DeliveryAttempt {
	now := q.clk.Now()
	a := store.DeliveryAttempt{Status: store.AttemptOK, At: now}
	var perm permanent
	switch {
	case err == nil:
	case errors.As(err, &perm) || netpolicy.IsDenied(err):
		a.Status, a.Error = store.AttemptFailed, sanitizeErr(err)
	case d.ExpiresAt != nil && !now.Before(*d.ExpiresAt):
		a.Status, a.Error = store.AttemptFailed, fmt.Sprintf("gave up after %s: %s", q.o.MaxAge, sanitizeErr(err))
	default:
		next := now.Add(q.o.Backoff(int(d.Attempt) + 1))
		if d.ExpiresAt != nil && next.After(*d.ExpiresAt) {
			next = *d.ExpiresAt // one last try at the deadline
		}
		a.Status, a.Error, a.NextAttemptAt = store.AttemptRetry, sanitizeErr(err), next
	}
	return a
}

// report updates the counters and logs a recorded attempt.
func (q *Queue) report(d store.Delivery, a store.DeliveryAttempt) {
	reg := q.o.Registry
	reg.Counter("nmon_notification_attempts_total", "receiver", d.Receiver, "result", a.Status).Inc()
	switch a.Status {
	case store.AttemptOK:
		reg.Counter("nmon_notifications_sent_total", "receiver", d.Receiver).Inc()
	case store.AttemptFailed:
		reg.Counter("nmon_notifications_failed_total", "receiver", d.Receiver).Inc()
	}
	attrs := []any{"receiver", d.Receiver, "incident", d.IncidentID, "attempt", d.Attempt + 1, "result", a.Status, "error", a.Error}
	if a.Status == store.AttemptRetry {
		attrs = append(attrs, "next", a.NextAttemptAt)
	}
	q.o.Logger.Info("notify attempt", attrs...)
	q.o.Logger.Debug("notify payload", "receiver", d.Receiver, "incident", d.IncidentID, "payload", d.Payload) // redacted at enqueue
}

// sanitizeErr is err's text without request URLs, which may embed tokens.
func sanitizeErr(err error) string {
	var ue *url.Error
	if errors.As(err, &ue) {
		err = ue.Err
	}
	s := RedactText(err.Error())
	if len(s) > 500 {
		s = s[:500]
	}
	return s
}