backrest/internal/api/summarydashboard.go

package api

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"time"

	"connectrpc.com/connect"
	v1 "github.com/garethgeorge/backrest/gen/go/v1"
	"github.com/garethgeorge/backrest/internal/env"
	"github.com/garethgeorge/backrest/internal/oplog"
	"github.com/garethgeorge/backrest/internal/protoutil"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
)

const (
	// summaryHistoryDays is the dashboard window: today plus the prior N-1 days.
	summaryHistoryDays = 30

	// summaryChartBackups caps how many recent backups each summary's chart includes.
	summaryChartBackups = 60

	// overdueGraceFactor pads a schedule's nominal period before a day is flagged
	// overdue, tolerating scheduler jitter and backup run time.
	overdueGraceFactor = 1.25
)

func (s *BackrestHandler) GetSummaryDashboard(ctx context.Context, req *connect.Request[emptypb.Empty]) (*connect.Response[v1.SummaryDashboardResponse], error) {
	cfg, err := s.config.Get()
	if err != nil {
		return nil, fmt.Errorf("failed to get config: %w", err)
	}

	now := time.Now()
	cutoffMidnight := localMidnight(now).AddDate(0, 0, -(summaryHistoryDays - 1))

	// One accumulator per repo and per plan; each operation in the pass below is
	// dispatched to at most one of each.
	repoAccs := make(map[string]*summaryAcc) // keyed by repo GUID
	planAccs := make(map[string]*summaryAcc) // keyed by plan ID
	for _, repo := range cfg.Repos {
		repoAccs[repo.GetGuid()] = newSummaryAcc(cutoffMidnight)
	}
	for _, plan := range cfg.Plans {
		planAccs[plan.Id] = newSummaryAcc(cutoffMidnight)
	}
	// Walk every operation for this instance, newest to oldest, dispatching each
	// backup to its plan's and its repo's accumulator.
	if err := s.oplog.Query(oplog.Query{}.SetInstanceID(cfg.Instance).SetReversed(true), func(op *v1.Operation) error {
		backupOp := op.GetOperationBackup()
		if backupOp == nil {
			return nil
		}
		if acc, ok := planAccs[op.PlanId]; ok {
			acc.observe(op, backupOp)
		}
		if acc, ok := repoAccs[op.RepoGuid]; ok {
			acc.observe(op, backupOp)
		}
		return nil
	}); err != nil {
		return nil, fmt.Errorf("failed to query operations: %w", err)
	}

	response := &v1.SummaryDashboardResponse{
		ConfigPath: env.ConfigFilePath(),
		DataPath:   env.DataDir(),
	}
	for _, repo := range cfg.Repos {
		response.RepoSummaries = append(response.RepoSummaries,
			repoAccs[repo.GetGuid()].finalize(repo.Id, now, repoAllowedStaleness(cfg, repo.Id, now)))
	}
	for _, plan := range cfg.Plans {
		response.PlanSummaries = append(response.PlanSummaries,
			planAccs[plan.Id].finalize(plan.Id, now, allowedStaleness(plan.Schedule, now)))
	}

	return connect.NewResponse(response), nil
}

// allowedStaleness converts a schedule to the maximum acceptable gap between
// OK backups, or 0 when there is no expectation (disabled or unparseable).
func allowedStaleness(sched *v1.Schedule, now time.Time) time.Duration {
	period, err := protoutil.NominalPeriod(sched, now)
	if err != nil {
		if !errors.Is(err, protoutil.ErrScheduleDisabled) {
			zap.S().Warnf("summary dashboard: nominal period: %v", err)
		}
		return 0
	}
	return time.Duration(float64(period) * overdueGraceFactor)
}

// repoAllowedStaleness is the widest allowed staleness among plans targeting
// the repo: a repo day is overdue only when even the slowest plan should have run.
func repoAllowedStaleness(cfg *v1.Config, repoID string, now time.Time) time.Duration {
	var widest time.Duration
	for _, plan := range cfg.Plans {
		if plan.Repo != repoID {
			continue
		}
		if a := allowedStaleness(plan.Schedule, now); a > widest {
			widest = a
		}
	}
	return widest
}

// localMidnight truncates t to midnight in its own location.
func localMidnight(t time.Time) time.Time {
	return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, t.Location())
}

// summaryDayAcc accumulates per-day backup stats for the dashboard history strip.
type summaryDayAcc struct {
	bytesAdded   int64
	bytesScanned int64
	statusCounts map[v1.OperationStatus]int64
}

// summaryAcc accumulates the backup operations for one plan or repo, observed
// newest to oldest, into a dashboard summary.
type summaryAcc struct {
	cutoffMidnight time.Time

	backupsExamined  int64
	bytesScanned30   int64
	bytesAdded30     int64
	backupsFailed30  int64
	backupsSuccess30 int64
	backupsWarning30 int64
	nextBackupTime   int64
	protectedBytes   int64
	backupChart      *v1.SummaryDashboardResponse_BackupChart

	// Per-day accumulators keyed by the day's local-midnight unix millis. oldestDay
	// tracks the earliest in-window day with a backup; reachedCutoff means backups
	// exist beyond the window, so the history strip spans the full window.
	perDay        map[int64]*summaryDayAcc
	oldestDay     time.Time
	reachedCutoff bool

	// Times of in-window OK backups (success or warning, not a dry run), which
	// reset the staleness clock, plus the most recent OK backup before the window.
	okBackupDates            []time.Time
	lastOkBackupBeforeWindow time.Time
}

func newSummaryAcc(cutoffMidnight time.Time) *summaryAcc {
	return &summaryAcc{
		cutoffMidnight: cutoffMidnight,
		backupChart:    &v1.SummaryDashboardResponse_BackupChart{},
		perDay:         make(map[int64]*summaryDayAcc),
	}
}

func (a *summaryAcc) observe(op *v1.Operation, backupOp *v1.OperationBackup) {
	startTime := time.UnixMilli(op.UnixTimeStartMs)
	opMidnight := localMidnight(startTime)
	// Dry runs don't reset the staleness clock, matching the scheduler's view.
	isOkBackup := (op.Status == v1.OperationStatus_STATUS_SUCCESS ||
		op.Status == v1.OperationStatus_STATUS_WARNING) && !backupOp.DryRun

	// Backups older than the window only contribute the staleness anchor; walking
	// newest-first, the first OK backup seen here is the most recent.
	if opMidnight.Before(a.cutoffMidnight) {
		a.reachedCutoff = true
		if isOkBackup && a.lastOkBackupBeforeWindow.IsZero() {
			a.lastOkBackupBeforeWindow = startTime
		}
		return
	}
	if op.GetStatus() == v1.OperationStatus_STATUS_PENDING {
		a.nextBackupTime = op.UnixTimeStartMs
		return
	}
	a.backupsExamined++

	switch op.Status {
	case v1.OperationStatus_STATUS_SUCCESS:
		a.backupsSuccess30++
	case v1.OperationStatus_STATUS_ERROR:
		a.backupsFailed30++
	case v1.OperationStatus_STATUS_WARNING:
		a.backupsWarning30++
	}

	if isOkBackup {
		a.okBackupDates = append(a.okBackupDates, startTime)
	}

	summary := backupOp.GetLastStatus().GetSummary()
	if summary != nil {
		a.bytesScanned30 += summary.TotalBytesProcessed
		a.bytesAdded30 += summary.DataAdded
	}

	// protected_bytes: the most recent (first seen) good backup's total size.
	if a.protectedBytes == 0 && summary != nil && isOkBackup {
		a.protectedBytes = summary.TotalBytesProcessed
	}

	// Update the per-day aggregate for this backup's day.
	dayMs := opMidnight.UnixMilli()
	acc := a.perDay[dayMs]
	if acc == nil {
		acc = &summaryDayAcc{statusCounts: make(map[v1.OperationStatus]int64)}
		a.perDay[dayMs] = acc
	}
	acc.statusCounts[op.Status]++
	if summary != nil {
		acc.bytesAdded += summary.DataAdded
		acc.bytesScanned += summary.TotalBytesProcessed
	}
	if a.oldestDay.IsZero() || opMidnight.Before(a.oldestDay) {
		a.oldestDay = opMidnight
	}

	if len(a.backupChart.TimestampMs) < summaryChartBackups {
		duration := op.UnixTimeEndMs - op.UnixTimeStartMs
		if duration <= 1000 {
			duration = 1000
		}

		a.backupChart.FlowId = append(a.backupChart.FlowId, op.FlowId)
		a.backupChart.TimestampMs = append(a.backupChart.TimestampMs, op.UnixTimeStartMs)
		a.backupChart.DurationMs = append(a.backupChart.DurationMs, duration)
		a.backupChart.Status = append(a.backupChart.Status, op.Status)
		a.backupChart.BytesAdded = append(a.backupChart.BytesAdded, summary.GetDataAdded())
	}
}

// finalize builds the summary proto. allowedStaleness > 0 enables overdue
// detection on the day history.
func (a *summaryAcc) finalize(id string, now time.Time, allowedStaleness time.Duration) *v1.SummaryDashboardResponse_Summary {
	todayMidnight := localMidnight(now)

	backupsExamined := a.backupsExamined
	if backupsExamined == 0 {
		backupsExamined = 1 // prevent division by zero for avg calculations
	}

	// OK backups ascending: these reset the staleness clock. The pre-window
	// anchor counts only when the window has no OK backup of its own, so a fully
	// stalled plan still flags every day, while an overdue stretch that ended
	// before the window's first OK backup is resolved history and stays muted.
	okBackupDates := a.okBackupDates
	if len(okBackupDates) == 0 && !a.lastOkBackupBeforeWindow.IsZero() {
		okBackupDates = append(okBackupDates, a.lastOkBackupBeforeWindow)
	}
	slices.SortFunc(okBackupDates, func(x, y time.Time) int { return x.Compare(y) })

	// Flatten the per-day map into buckets ordered oldest-first, one per consecutive day
	// from the oldest active day (or the window start, if older backups exist) through
	// today; absent days get empty buckets. The client matches buckets to days by their
	// distance from the newest bucket, and renders days before the span as "before start".
	start := todayMidnight
	if a.reachedCutoff {
		start = a.cutoffMidnight
	} else if !a.oldestDay.IsZero() {
		start = a.oldestDay
	}
	var history []*v1.SummaryDashboardResponse_DayStatusBucket
	var lastOkBackup time.Time
	nextOkBackup := 0
	for day := start; !day.After(todayMidnight); day = day.AddDate(0, 0, 1) {
		bucket := &v1.SummaryDashboardResponse_DayStatusBucket{
			TimestampMs: day.UnixMilli(),
		}
		if acc := a.perDay[day.UnixMilli()]; acc != nil {
			bucket.BytesAdded = acc.bytesAdded
			bucket.BytesScanned = acc.bytesScanned
			for status, count := range acc.statusCounts {
				bucket.StatusCounts = append(bucket.StatusCounts, &v1.SummaryDashboardResponse_StatusAndCount{
					Status: status,
					Count:  count,
				})
			}
		}

		// A day is overdue when, at its close (clamped to now for today), the newest
		// OK backup is older than the allowed staleness AND the day had no OK backup
		// of its own. Days before the first OK backup ever are exempt: the scheduler
		// anchors to the first real run. Requiring the day itself to be empty keeps
		// sub-daily schedules honest: an hourly plan whose last run was at noon still
		// backed the day up even though midnight is well past the allowed gap, so the
		// day must render as backed-up, not overdue.
		if allowedStaleness > 0 {
			checkpoint := day.AddDate(0, 0, 1)
			if checkpoint.After(now) {
				checkpoint = now
			}
			for nextOkBackup < len(okBackupDates) && !okBackupDates[nextOkBackup].After(checkpoint) {
				lastOkBackup = okBackupDates[nextOkBackup]
				nextOkBackup++
			}
			// lastOkBackup <= checkpoint <= day+1, so lastOkBackup.Before(day) is
			// exactly "the newest OK backup landed on an earlier day" — this day had none.
			bucket.Overdue = !lastOkBackup.IsZero() &&
				lastOkBackup.Before(day) &&
				checkpoint.Sub(lastOkBackup) > allowedStaleness
		}

		history = append(history, bucket)
	}

	return &v1.SummaryDashboardResponse_Summary{
		Id:                        id,
		BytesScannedLast_30Days:   a.bytesScanned30,
		BytesAddedLast_30Days:     a.bytesAdded30,
		BackupsFailed_30Days:      a.backupsFailed30,
		BackupsWarningLast_30Days: a.backupsWarning30,
		BackupsSuccessLast_30Days: a.backupsSuccess30,
		BytesScannedAvg:           a.bytesScanned30 / backupsExamined,
		BytesAddedAvg:             a.bytesAdded30 / backupsExamined,
		NextBackupTimeMs:          a.nextBackupTime,
		RecentBackups:             a.backupChart,
		ProtectedBytes:            a.protectedBytes,
		HistoryLast_30Days:        history,
	}
}