tapir/internal/progress/tracker.go

// Package progress tracks the phases of one upload job (ADR-035). A Tracker is the
// only writer of a job's progress: every transition is saved to the job's row and
// logged by the same call, so what the page shows and what the log says cannot
// disagree. The engines and the reconciler report through a Reporter carried in the
// context, so the Transcriber interface stays what it was.
package progress

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"git.d-ma.be/mathias/tapir/internal/domain"
	"git.d-ma.be/mathias/tapir/internal/errkind"
)

// Planned is one phase of a job, declared up front so the page can show what is coming.
type Planned struct {
	Key, Label string
}

// Reporter is what a long-running component reports to. Begin starts a phase (total
// is the number of countable units, 0 when there is no honest count), Advance says how
// many are done, Done closes the phase. Fail closes a phase that did not work while
// the job goes on (an engine that was down, the other one's text used instead) and
// says why in the log. Note is a log line for the operator that costs no write.
type Reporter interface {
	Begin(key string, total int)
	Advance(key string, done int)
	Done(key string)
	Fail(key string, err error)
	Note(msg string, kv ...any)
}

// Sink is where a job's progress is persisted. Both calls are best effort: progress
// is for people, so a failed write is logged and never stops the job.
type Sink interface {
	SaveProgress(ctx context.Context, steps []domain.ProgressStep, at time.Time) error
	Beat(ctx context.Context, at time.Time) error
}

const (
	defaultMinSave     = 2 * time.Second
	defaultSaveTimeout = 5 * time.Second
	logAdvanceEach     = time.Minute
)

// Option tunes a Tracker.
type Option func(*Tracker)

// WithClock fixes the clock, for tests.
func WithClock(now func() time.Time) Option { return func(t *Tracker) { t.now = now } }

// WithMinSave sets the least time between two saves caused by Advance alone. Begin,
// Done, Fail and Finish always save.
func WithMinSave(d time.Duration) Option { return func(t *Tracker) { t.minSave = d } }

// WithSaveTimeout bounds one save or heartbeat write. Progress is cosmetic, so a database
// that stops answering costs a dropped write rather than an unbounded stall. It is still a
// delay: saves run under the tracker's lock, which Begin and Advance (called by the engines)
// need, so each transition can wait up to this long, and a hung database slows a job by
// that much per call, out of the job's own time budget (#51). A dropped final save is not
// retried; the row keeps its last saved step.
func WithSaveTimeout(d time.Duration) Option { return func(t *Tracker) { t.saveTimeout = d } }

// Tracker implements Reporter for one job.
type Tracker struct {
	ctx     context.Context
	sink    Sink
	log     *slog.Logger
	now     func() time.Time
	minSave time.Duration

	saveTimeout time.Duration

	mu              sync.Mutex
	steps           []domain.ProgressStep
	lastAdvanceSave time.Time
	lastAdvanceLog  time.Time
}

// NewTracker plans a job's phases. ctx is used for the saves, so it should outlive the
// job's own timeout: a job that timed out still gets its failure recorded.
func NewTracker(ctx context.Context, sink Sink, log *slog.Logger, plan []Planned, opts ...Option) *Tracker {
	t := &Tracker{ctx: ctx, sink: sink, log: log, now: time.Now, minSave: defaultMinSave, saveTimeout: defaultSaveTimeout}
	for _, o := range opts {
		o(t)
	}
	for _, p := range plan {
		t.steps = append(t.steps, domain.ProgressStep{Key: p.Key, Label: p.Label, State: domain.StepPending})
	}
	return t
}

// Start saves the plan, every phase pending.
func (t *Tracker) Start() {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.log.Info("job started", "steps", len(t.steps))
	t.save()
}

// Begin implements Reporter.
func (t *Tracker) Begin(key string, total int) {
	t.mu.Lock()
	defer t.mu.Unlock()
	s := t.find(key)
	if s == nil {
		return
	}
	s.State, s.StartedAt, s.FinishedAt, s.Done, s.Total = domain.StepRunning, t.now(), time.Time{}, 0, total
	t.log.Info("step started", "step", key, "total", total)
	t.save()
}

// Advance implements Reporter. The count is always kept; it is saved at most once per
// minimum interval and logged at most once a minute, so a counted phase does not turn
// into a write and a log line per unit.
func (t *Tracker) Advance(key string, done int) {
	t.mu.Lock()
	defer t.mu.Unlock()
	s := t.find(key)
	if s == nil {
		return
	}
	s.Done = done
	now := t.now()
	if t.lastAdvanceLog.IsZero() || now.Sub(t.lastAdvanceLog) >= logAdvanceEach {
		t.lastAdvanceLog = now
		t.log.Info("step progress", "step", key, "done", done, "total", s.Total)
	}
	if t.lastAdvanceSave.IsZero() || now.Sub(t.lastAdvanceSave) >= t.minSave {
		t.lastAdvanceSave = now
		t.save()
	}
}

// Done implements Reporter.
func (t *Tracker) Done(key string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	s := t.find(key)
	if s == nil {
		return
	}
	t.close(s, domain.StepDone)
	t.log.Info("step finished", "step", key, "elapsed", s.FinishedAt.Sub(s.StartedAt))
	t.save()
}

// Fail implements Reporter: one phase failed and the job carries on.
func (t *Tracker) Fail(key string, err error) {
	t.mu.Lock()
	defer t.mu.Unlock()
	s := t.find(key)
	if s == nil {
		return
	}
	t.close(s, domain.StepFailed)
	t.log.Warn("step failed", "step", key, "elapsed", s.FinishedAt.Sub(s.StartedAt), "kind", errkind.Of(err))
	t.save()
}

// Note implements Reporter.
func (t *Tracker) Note(msg string, kv ...any) { t.log.Info(msg, kv...) }

// FailRunning marks the phase that was running as failed, because the whole job
// failed. What never started stays pending, which is what shows the user where the
// job stopped.
func (t *Tracker) FailRunning() {
	t.mu.Lock()
	defer t.mu.Unlock()
	for i := range t.steps {
		if t.steps[i].State == domain.StepRunning {
			t.close(&t.steps[i], domain.StepFailed)
			t.log.Warn("step failed", "step", t.steps[i].Key, "elapsed", t.steps[i].FinishedAt.Sub(t.steps[i].StartedAt))
		}
	}
	t.save()
}

// Finish closes the job: a phase still running is done, a phase that never ran is
// skipped.
func (t *Tracker) Finish() {
	t.mu.Lock()
	defer t.mu.Unlock()
	for i := range t.steps {
		switch t.steps[i].State {
		case domain.StepRunning:
			t.close(&t.steps[i], domain.StepDone)
		case domain.StepPending:
			t.steps[i].State = domain.StepSkipped
		}
	}
	t.log.Info("job finished")
	t.save()
}

// Annotate attaches a short plain note to a phase and saves it at once. It is for a
// remark the reader should see although nothing failed. Durations and counts only: a note
// is shown on a page about a private recording.
func (t *Tracker) Annotate(key, note string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	s := t.find(key)
	if s == nil {
		return
	}
	s.Note = note
	t.save()
}

// Heartbeat stamps the job's row every beatEvery and logs which phase is still running
// every stillEvery, until ctx ends. It runs on its own, independent of any engine call:
// a heartbeat that stopped means the process is gone, not that a request is slow.
func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Duration) {
	beat, still := time.NewTicker(beatEvery), time.NewTicker(stillEvery)
	defer beat.Stop()
	defer still.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-beat.C:
			beatCtx, cancel := context.WithTimeout(ctx, t.saveTimeout)
			err := t.sink.Beat(beatCtx, t.now())
			cancel()
			if err != nil && ctx.Err() == nil {
				t.log.Warn("progress: heartbeat not saved", "err", err)
			}
		case <-still.C:
			t.logStill()
		}
	}
}

func (t *Tracker) logStill() {
	t.mu.Lock()
	defer t.mu.Unlock()
	for _, s := range t.steps {
		if s.State == domain.StepRunning {
			t.log.Info("still running", "step", s.Key, "elapsed", t.now().Sub(s.StartedAt), "done", s.Done, "total", s.Total)
			return
		}
	}
}

// close ends a phase. Callers hold t.mu.
func (t *Tracker) close(s *domain.ProgressStep, state domain.StepState) {
	s.State, s.FinishedAt = state, t.now()
	if s.StartedAt.IsZero() {
		s.StartedAt = s.FinishedAt
	}
}

// find returns the phase with this key, or nil. Callers hold t.mu.
func (t *Tracker) find(key string) *domain.ProgressStep {
	for i := range t.steps {
		if t.steps[i].Key == key {
			return &t.steps[i]
		}
	}
	return nil
}

// save persists a copy of the phases. Callers hold t.mu. A failed write is logged and
// dropped: progress is cosmetic and must never fail the job it describes.
func (t *Tracker) save() {
	snapshot := append([]domain.ProgressStep(nil), t.steps...)
	ctx, cancel := context.WithTimeout(t.ctx, t.saveTimeout)
	defer cancel()
	if err := t.sink.SaveProgress(ctx, snapshot, t.now()); err != nil && t.ctx.Err() == nil {
		t.log.Warn("progress: not saved", "err", err)
	}
}

type ctxKey struct{}

// With returns a context that carries r.
func With(ctx context.Context, r Reporter) context.Context {
	return context.WithValue(ctx, ctxKey{}, r)
}

// From returns the Reporter carried by ctx, or one that does nothing, so a component
// can always report without checking whether anyone is listening.
func From(ctx context.Context) Reporter {
	if r, ok := ctx.Value(ctxKey{}).(Reporter); ok {
		return r
	}
	return noop{}
}

type noop struct{}

func (noop) Begin(string, int)   {}
func (noop) Advance(string, int) {}
func (noop) Done(string)         {}
func (noop) Fail(string, error)  {}
func (noop) Note(string, ...any) {}