Pete/internal/storage/sourcehealth.go

package storage

import (
	"database/sql"
	"fmt"
	"log/slog"
)

// SourceHealth is one source's persisted poll health, as written by the poller.
type SourceHealth struct {
	Source              string
	LastPollAt          int64 // 0 = never polled
	LastSuccessAt       int64 // 0 = never succeeded
	LastError           string
	ConsecutiveFailures int
	LastItemCount       int
	UpdatedAt           int64
}

// SourceContentStat is per-source content derived from the stories table (plus
// post_log for last-posted), independent of poll health.
type SourceContentStat struct {
	Source       string
	Total        int   // stories currently retained for this source
	Classified   int   // of those, how many are classified
	Paywalled    int   // of those, how many are gated
	LastSeenAt   int64 // MAX(seen_at); 0 = none
	LastPostedAt int64 // MAX(post_log.posted_at) joined by guid; 0 = never posted
}

// RecordPollResult upserts a source's health row after a poll attempt. On
// success it clears the error and failure counter and records the item count;
// on failure it bumps consecutive_failures and stores the message while
// preserving the last successful timestamp and item count. A brand-new row
// starts its last_new_at clock at the first poll, so a just-added source gets a
// full quiet window before it can be called quiet. Fire-and-forget:
// a failure here must never disrupt polling, so errors are logged and swallowed.
func RecordPollResult(source string, ok bool, itemCount int, pollErr error) {
	now := nowUnix()
	if ok {
		exec("record poll success",
			`INSERT INTO source_health
			   (source, last_poll_at, last_success_at, last_error, consecutive_failures, last_item_count, updated_at, last_new_at)
			 VALUES (?, ?, ?, '', 0, ?, ?, ?)
			 ON CONFLICT(source) DO UPDATE SET
			   last_poll_at         = excluded.last_poll_at,
			   last_success_at      = excluded.last_success_at,
			   last_error           = '',
			   consecutive_failures = 0,
			   last_item_count      = excluded.last_item_count,
			   updated_at           = excluded.updated_at`,
			source, now, now, itemCount, now, now)
		return
	}
	msg := ""
	if pollErr != nil {
		msg = pollErr.Error()
	}
	exec("record poll failure",
		`INSERT INTO source_health
		   (source, last_poll_at, last_success_at, last_error, consecutive_failures, last_item_count, updated_at, last_new_at)
		 VALUES (?, ?, NULL, ?, 1, 0, ?, ?)
		 ON CONFLICT(source) DO UPDATE SET
		   last_poll_at         = excluded.last_poll_at,
		   last_error           = excluded.last_error,
		   consecutive_failures = source_health.consecutive_failures + 1,
		   updated_at           = excluded.updated_at`,
		source, now, msg, now, now)
}

// ListSourceHealth returns the poll-health row for every source that has been
// polled at least once, keyed by source name.
func ListSourceHealth() (map[string]SourceHealth, error) {
	rows, err := Get().Query(
		`SELECT source, last_poll_at, last_success_at, last_error,
		        consecutive_failures, last_item_count, updated_at
		 FROM source_health`)
	if err != nil {
		return nil, fmt.Errorf("list source health: %w", err)
	}
	defer rows.Close()
	out := make(map[string]SourceHealth)
	for rows.Next() {
		var h SourceHealth
		var lastPoll, lastSuccess sql.NullInt64
		var lastErr sql.NullString
		if err := rows.Scan(&h.Source, &lastPoll, &lastSuccess, &lastErr,
			&h.ConsecutiveFailures, &h.LastItemCount, &h.UpdatedAt); err != nil {
			return nil, err
		}
		h.LastPollAt = lastPoll.Int64
		h.LastSuccessAt = lastSuccess.Int64
		h.LastError = lastErr.String
		out[h.Source] = h
	}
	return out, rows.Err()
}

// SourceContentStats derives per-source content counts from the stories table,
// with last-posted joined from post_log by guid. Keyed by source name. Only
// sources with at least one retained story appear; the caller pads out the rest
// from its configured source list.
func SourceContentStats() (map[string]SourceContentStat, error) {
	out := make(map[string]SourceContentStat)

	rows, err := Get().Query(
		`SELECT source,
		        COUNT(*),
		        COALESCE(SUM(classified), 0),
		        COALESCE(SUM(paywalled), 0),
		        COALESCE(MAX(seen_at), 0)
		 FROM stories
		 GROUP BY source`)
	if err != nil {
		return nil, fmt.Errorf("source content stats: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var st SourceContentStat
		if err := rows.Scan(&st.Source, &st.Total, &st.Classified, &st.Paywalled, &st.LastSeenAt); err != nil {
			return nil, err
		}
		out[st.Source] = st
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	// Last-posted per source, joined by guid. Kept separate so sources with
	// stories but no posts still appear above with a zero last-posted.
	prows, err := Get().Query(
		`SELECT s.source, MAX(p.posted_at)
		 FROM post_log p
		 JOIN stories s ON s.guid = p.guid
		 GROUP BY s.source`)
	if err != nil {
		return nil, fmt.Errorf("source last-posted: %w", err)
	}
	defer prows.Close()
	for prows.Next() {
		var source string
		var lastPosted int64
		if err := prows.Scan(&source, &lastPosted); err != nil {
			return nil, err
		}
		st := out[source]
		st.Source = source
		st.LastPostedAt = lastPosted
		out[source] = st
	}
	return out, prows.Err()
}

// backfillLastNewAt gives rows that predate last_new_at a starting value: the
// newest story still retained for that source, or, when none is, the time of
// the migration. The second case is a grace period rather than a guess: a
// source with no retained stories might be long dead or might have been added
// yesterday, and nothing on disk says which, so it gets a full quiet window
// before it can be called quiet.
func backfillLastNewAt(d *sql.DB) {
	if _, err := d.Exec(
		`UPDATE source_health
		    SET last_new_at = COALESCE(
		          (SELECT MAX(seen_at) FROM stories WHERE stories.source = source_health.source),
		          ?)
		  WHERE last_new_at IS NULL`, nowUnix()); err != nil {
		slog.Error("backfill source_health.last_new_at failed", "err", err)
	}
}

// RecordNewStories notes that a poll of source kept n genuinely new stories.
// Only a non-zero n moves last_new_at: a poll that fetched fine but found
// nothing new is exactly what "gone quiet" is made of.
func RecordNewStories(source string, n int) {
	if n <= 0 {
		return
	}
	exec("record new stories",
		`UPDATE source_health SET last_new_at = ? WHERE source = ?`, nowUnix(), source)
}

// FeedAlertRow is the slice of source_health the feed watcher reads.
type FeedAlertRow struct {
	Source              string
	ConsecutiveFailures int
	LastError           string
	LastSuccessAt       int64 // 0 = never fetched successfully
	LastNewAt           int64 // 0 = unknown
	DownAlertedAt       int64 // 0 = owner not told it's down
	QuietAlertedAt      int64 // 0 = owner not told it's quiet
}

// ListFeedAlertRows returns the alert-relevant health of every polled source.
func ListFeedAlertRows() ([]FeedAlertRow, error) {
	rows, err := Get().Query(
		`SELECT source, consecutive_failures, COALESCE(last_error, ''),
		        COALESCE(last_success_at, 0), COALESCE(last_new_at, 0),
		        COALESCE(down_alerted_at, 0), COALESCE(quiet_alerted_at, 0)
		   FROM source_health ORDER BY source`)
	if err != nil {
		return nil, fmt.Errorf("list feed alert rows: %w", err)
	}
	defer rows.Close()
	var out []FeedAlertRow
	for rows.Next() {
		var r FeedAlertRow
		if err := rows.Scan(&r.Source, &r.ConsecutiveFailures, &r.LastError,
			&r.LastSuccessAt, &r.LastNewAt, &r.DownAlertedAt, &r.QuietAlertedAt); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// SetFeedDownAlerted records (at > 0) or clears (at == 0) that the owner has
// been told source is down.
func SetFeedDownAlerted(source string, at int64) error {
	_, err := Get().Exec(`UPDATE source_health SET down_alerted_at = NULLIF(?, 0) WHERE source = ?`, at, source)
	return err
}

// SetFeedQuietAlerted records (at > 0) or clears (at == 0) that the owner has
// been told source has gone quiet.
func SetFeedQuietAlerted(source string, at int64) error {
	_, err := Get().Exec(`UPDATE source_health SET quiet_alerted_at = NULLIF(?, 0) WHERE source = ?`, at, source)
	return err
}