Pete/internal/storage/sourcehealth.go

package storage

import (
	"database/sql"
	"fmt"
)

// 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. 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)
			 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)
		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)
		 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)
}

// 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()
}