Pete/internal/feedwatch/feedwatch.go

// Package feedwatch tells the owner when a feed breaks, when it comes back, and
// when it goes quiet.
//
// The poller has always recorded every poll's outcome in source_health, and
// /status shows it, but only to someone who goes and looks. Feeds died that way
// for months: Siliconera and Digital Foundry had been refused on every poll
// thousands of times over before anyone noticed.
//
// The watcher reads source_health on a timer rather than hooking the pollers, so
// that one pass can batch every change into a single message, and so that what
// has already been said lives in the database (down_alerted_at,
// quiet_alerted_at) instead of in a goroutine that forgets it on restart. Each
// episode is reported once: a feed that goes down is announced, stays silent
// while it stays down, and is announced again when it comes back.
package feedwatch

import (
	"context"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"pete/internal/config"
	"pete/internal/storage"
)

const (
	// DownAfter is how many failed polls in a row make a feed "down". One or
	// two failures is weather; five is a feed that has stopped working.
	DownAfter = 5
	// QuietAfter is how long a feed can fetch fine and yield nothing new before
	// it is "quiet". Every healthy source Pete carries produces something within
	// about four days.
	QuietAfter = 7 * 24 * time.Hour
	// Every is how often the watcher looks.
	Every = 15 * time.Minute
)

// Report is what one pass found that the owner hasn't been told yet.
type Report struct {
	Down  []storage.FeedAlertRow
	Back  []storage.FeedAlertRow
	Quiet []storage.FeedAlertRow
	// quietOver are feeds whose quiet spell has ended. They are cleared without
	// a message: a feed that simply started posting again needs no fanfare.
	quietOver []string
}

// Empty reports whether there is nothing to say.
func (r Report) Empty() bool { return len(r.Down)+len(r.Back)+len(r.Quiet) == 0 }

// Check reads source_health and works out what has changed, without recording
// anything. Only sources that are enabled in the config are considered, so a
// feed that has been switched off or removed never alerts.
func Check(now time.Time, sources []config.SourceConfig) (Report, error) {
	enabled := make(map[string]bool, len(sources))
	for _, s := range sources {
		if s.Enabled {
			enabled[s.Name] = true
		}
	}
	rows, err := storage.ListFeedAlertRows()
	if err != nil {
		return Report{}, err
	}
	var r Report
	quietCutoff := now.Add(-QuietAfter).Unix()
	for _, row := range rows {
		if !enabled[row.Source] {
			continue
		}
		down := row.ConsecutiveFailures >= DownAfter
		switch {
		case down && row.DownAlertedAt == 0:
			r.Down = append(r.Down, row)
		// Back means "has fetched since we said it was down", not "is fetching
		// right now": a feed that recovers and breaks again between two passes
		// still gets its all-clear, and then a fresh down alert next pass.
		case row.DownAlertedAt != 0 && row.LastSuccessAt > row.DownAlertedAt:
			r.Back = append(r.Back, row)
		}
		// A feed that isn't fetching can't be judged on whether it has news;
		// "down" already covers it.
		if row.ConsecutiveFailures > 0 {
			continue
		}
		quiet := row.LastNewAt > 0 && row.LastNewAt < quietCutoff
		switch {
		case quiet && row.QuietAlertedAt == 0:
			r.Quiet = append(r.Quiet, row)
		case !quiet && row.QuietAlertedAt != 0:
			r.quietOver = append(r.quietOver, row.Source)
		}
	}
	return r, nil
}

// Commit records that the owner has now been told everything in r.
func Commit(now time.Time, r Report) error {
	at := now.Unix()
	for _, row := range r.Down {
		if err := storage.SetFeedDownAlerted(row.Source, at); err != nil {
			return err
		}
	}
	for _, row := range r.Back {
		if err := storage.SetFeedDownAlerted(row.Source, 0); err != nil {
			return err
		}
	}
	for _, row := range r.Quiet {
		if err := storage.SetFeedQuietAlerted(row.Source, at); err != nil {
			return err
		}
	}
	for _, src := range r.quietOver {
		if err := storage.SetFeedQuietAlerted(src, 0); err != nil {
			return err
		}
	}
	return nil
}

// Message renders r as the plain-text note Pete sends the owner.
func Message(now time.Time, r Report) string {
	var b strings.Builder
	b.WriteString("Feed check")
	if len(r.Down) > 0 {
		b.WriteString("\n\nNot fetching:")
		for _, row := range r.Down {
			since := "has never fetched successfully"
			if row.LastSuccessAt > 0 {
				since = "last worked " + ago(now, row.LastSuccessAt)
			}
			fmt.Fprintf(&b, "\n• %s: %s failed polls in a row, %s. Last error: %s",
				row.Source, count(row.ConsecutiveFailures), since, short(row.LastError))
		}
	}
	if len(r.Back) > 0 {
		b.WriteString("\n\nWorking again:")
		for _, row := range r.Back {
			fmt.Fprintf(&b, "\n• %s is fetching again.", row.Source)
		}
	}
	if len(r.Quiet) > 0 {
		b.WriteString("\n\nGone quiet:")
		for _, row := range r.Quiet {
			fmt.Fprintf(&b, "\n• %s fetches fine but hasn't had anything new in %s.",
				row.Source, span(now, row.LastNewAt))
		}
	}
	return b.String()
}

// Send delivers a message to the owner. An error means it didn't arrive, and
// the watcher will try again next pass rather than mark it said.
type Send func(ctx context.Context, msg string) error

// Run checks once shortly after startup (so pollers have had a first go) and
// then every Every until ctx ends.
func Run(ctx context.Context, sources []config.SourceConfig, send Send) {
	first := time.NewTimer(2 * time.Minute)
	defer first.Stop()
	t := time.NewTicker(Every)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-first.C:
		case <-t.C:
		}
		pass(ctx, time.Now(), sources, send)
	}
}

func pass(ctx context.Context, now time.Time, sources []config.SourceConfig, send Send) {
	r, err := Check(now, sources)
	if err != nil {
		slog.Error("feedwatch: check failed", "err", err)
		return
	}
	if r.Empty() {
		// Still commit: a quiet spell may have ended silently.
		if err := Commit(now, r); err != nil {
			slog.Error("feedwatch: commit failed", "err", err)
		}
		return
	}
	msg := Message(now, r)
	if err := send(ctx, msg); err != nil {
		slog.Error("feedwatch: alert not delivered, will retry", "err", err)
		return
	}
	slog.Info("feedwatch: alert sent", "down", len(r.Down), "back", len(r.Back), "quiet", len(r.Quiet))
	if err := Commit(now, r); err != nil {
		slog.Error("feedwatch: commit failed", "err", err)
	}
}

// span renders the time since unix as a rough human length ("9 days").
func span(now time.Time, unix int64) string {
	d := now.Sub(time.Unix(unix, 0))
	switch {
	case d < 2*time.Hour:
		return fmt.Sprintf("%d minutes", int(d.Minutes()))
	case d < 48*time.Hour:
		return fmt.Sprintf("%d hours", int(d.Hours()))
	default:
		return fmt.Sprintf("%d days", int(d.Hours()/24))
	}
}

// ago is span as a point in the past ("9 days ago").
func ago(now time.Time, unix int64) string { return span(now, unix) + " ago" }

// count renders n with thousands separators.
func count(n int) string {
	s := fmt.Sprint(n)
	for i := len(s) - 3; i > 0; i -= 3 {
		s = s[:i] + "," + s[i:]
	}
	return s
}

// short trims an error to one readable line.
func short(s string) string {
	s = strings.TrimSpace(strings.ReplaceAll(s, "\n", " "))
	if s == "" {
		return "none recorded"
	}
	if len(s) > 140 {
		s = s[:140] + "…"
	}
	return s
}