tapir/internal/progress/tracker_test.go

package progress_test

import (
	"context"
	"errors"
	"log/slog"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

// fakeSink records every saved snapshot and every heartbeat.
type fakeSink struct {
	mu    sync.Mutex
	saves [][]domain.ProgressStep
	at    []time.Time
	beats int
	err   error
}

func (f *fakeSink) SaveProgress(_ context.Context, steps []domain.ProgressStep, at time.Time) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.saves = append(f.saves, append([]domain.ProgressStep(nil), steps...))
	f.at = append(f.at, at)
	return f.err
}

func (f *fakeSink) Beat(context.Context) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.beats++
	return f.err
}

func (f *fakeSink) last(t *testing.T) []domain.ProgressStep {
	t.Helper()
	f.mu.Lock()
	defer f.mu.Unlock()
	require.NotEmpty(t, f.saves, "nothing was saved")
	return f.saves[len(f.saves)-1]
}

func (f *fakeSink) saveCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.saves)
}

func (f *fakeSink) beatCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.beats
}

// syncBuf is a log sink safe to read while the heartbeat goroutine is still writing.
type syncBuf struct {
	mu sync.Mutex
	b  strings.Builder
}

func (s *syncBuf) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.b.Write(p)
}

func (s *syncBuf) String() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.b.String()
}

// clock is a hand-wound clock so durations and throttling are exact.
type clock struct{ t time.Time }

func (c *clock) now() time.Time       { return c.t }
func (c *clock) tick(d time.Duration) { c.t = c.t.Add(d) }
func newClock() *clock                { return &clock{t: time.Date(2026, 10, 2, 8, 0, 0, 0, time.UTC)} }

func plan() []progress.Planned {
	return []progress.Planned{{Key: "audio", Label: "Preparing audio"}, {Key: "decide", Label: "Deciding"}, {Key: "save", Label: "Saving"}}
}

func newTracker(t *testing.T) (*progress.Tracker, *fakeSink, *clock, *syncBuf) {
	t.Helper()
	sink, clk, buf := &fakeSink{}, newClock(), &syncBuf{}
	log := slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
	return progress.NewTracker(context.Background(), sink, log, plan(), progress.WithClock(clk.now), progress.WithMinSave(2*time.Second)), sink, clk, buf
}

func stateOf(steps []domain.ProgressStep, key string) domain.StepState {
	for _, s := range steps {
		if s.Key == key {
			return s.State
		}
	}
	return ""
}

func TestTracker_PlanIsShownUpFrontInOrderAllPending(t *testing.T) {
	tr, sink, _, _ := newTracker(t)

	tr.Start()

	steps := sink.last(t)
	require.Len(t, steps, 3)
	require.Equal(t, []string{"audio", "decide", "save"}, []string{steps[0].Key, steps[1].Key, steps[2].Key})
	for _, s := range steps {
		require.Equal(t, domain.StepPending, s.State, s.Key)
	}
	require.Equal(t, "Preparing audio", steps[0].Label)
}

func TestTracker_BeginRunsAStepAndDoneRecordsItsDuration(t *testing.T) {
	tr, sink, clk, _ := newTracker(t)
	tr.Start()

	tr.Begin("audio", 0)
	require.Equal(t, domain.StepRunning, stateOf(sink.last(t), "audio"))
	clk.tick(90 * time.Second)
	tr.Done("audio")

	s := sink.last(t)[0]
	require.Equal(t, domain.StepDone, s.State)
	require.Equal(t, 90*time.Second, s.FinishedAt.Sub(s.StartedAt), "the finished step must carry how long it took")
}

func TestTracker_AdvanceCountsAreSavedAndThrottled(t *testing.T) {
	tr, sink, clk, _ := newTracker(t)
	tr.Start()
	tr.Begin("decide", 282)
	base := sink.saveCount()

	tr.Advance("decide", 1)
	clk.tick(500 * time.Millisecond)
	tr.Advance("decide", 2)
	clk.tick(500 * time.Millisecond)
	tr.Advance("decide", 3)

	require.Equal(t, base+1, sink.saveCount(), "advances inside the minimum interval must not each hit the database")
	clk.tick(2 * time.Second)
	tr.Advance("decide", 4)
	require.Equal(t, base+2, sink.saveCount())
	got := sink.last(t)[1]
	require.Equal(t, 4, got.Done)
	require.Equal(t, 282, got.Total)
}

func TestTracker_DoneFlushesTheFinalCountEvenInsideTheThrottle(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	tr.Start()
	tr.Begin("decide", 10)
	tr.Advance("decide", 9)

	tr.Advance("decide", 10)
	tr.Done("decide")

	got := sink.last(t)[1]
	require.Equal(t, domain.StepDone, got.State)
	require.Equal(t, 10, got.Done, "a finished step must show its full count, not the last throttled one")
}

func TestTracker_FinishSkipsWhatNeverRanAndClosesWhatDid(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	tr.Start()
	tr.Begin("audio", 0)
	tr.Done("audio")
	tr.Begin("decide", 5)

	tr.Finish()

	steps := sink.last(t)
	require.Equal(t, domain.StepDone, stateOf(steps, "audio"))
	require.Equal(t, domain.StepDone, stateOf(steps, "decide"), "a step still running when the job completes is closed")
	require.Equal(t, domain.StepSkipped, stateOf(steps, "save"), "a step that never ran is skipped, not left pending")
}

func TestTracker_FailRunningMarksTheRunningStepFailedAndLeavesTheRestPending(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	tr.Start()
	tr.Begin("audio", 0)
	tr.Done("audio")
	tr.Begin("decide", 5)

	tr.FailRunning()

	steps := sink.last(t)
	require.Equal(t, domain.StepDone, stateOf(steps, "audio"))
	require.Equal(t, domain.StepFailed, stateOf(steps, "decide"))
	require.Equal(t, domain.StepPending, stateOf(steps, "save"), "what never started stays pending: it shows where the job stopped")
}

func TestTracker_AKeyedFailMarksOneStepFailedLogsItsKindAndTheJobGoesOn(t *testing.T) {
	tr, sink, _, buf := newTracker(t)
	tr.Start()
	tr.Begin("audio", 0)

	tr.Fail("audio", errors.New("gateway said no"))
	tr.Begin("decide", 5)

	steps := sink.last(t)
	require.Equal(t, domain.StepFailed, stateOf(steps, "audio"))
	require.Equal(t, domain.StepRunning, stateOf(steps, "decide"), "a soft failure must not stop the next step from running")
	require.Contains(t, buf.String(), "level=WARN")
	require.NotContains(t, buf.String(), "gateway said no", "an error's text can carry provider output or a path: the log gets its kind")
	require.Contains(t, buf.String(), "kind=error")
	require.Contains(t, buf.String(), "step=audio")
}

func TestTracker_NoteIsALogLineAndNeverASave(t *testing.T) {
	tr, sink, _, buf := newTracker(t)
	tr.Start()
	before := sink.saveCount()

	tr.Note("reconciled", "disputes", 362, "auto_resolved", 80)

	require.Equal(t, before, sink.saveCount(), "a note is for the operator; it must not cost a database write")
	require.Contains(t, buf.String(), "reconciled")
	require.Contains(t, buf.String(), "disputes=362")
	require.Contains(t, buf.String(), "auto_resolved=80")
}

func TestTracker_AnUnknownKeyIsIgnored(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	tr.Start()
	before := sink.saveCount()

	require.NotPanics(t, func() {
		tr.Begin("no-such-step", 0)
		tr.Advance("no-such-step", 1)
		tr.Done("no-such-step")
		tr.Fail("no-such-step", errors.New("x"))
	})
	require.Equal(t, before, sink.saveCount())
}

func TestTracker_ASinkFailureIsLoggedAndNeverStopsTheJob(t *testing.T) {
	tr, sink, _, buf := newTracker(t)
	sink.err = errors.New("db down")

	require.NotPanics(t, func() {
		tr.Start()
		tr.Begin("audio", 0)
		tr.Done("audio")
	})
	require.Contains(t, buf.String(), "level=WARN", "a lost progress write is worth a WARN")
	require.Contains(t, buf.String(), "db down")
}

func TestTracker_EveryTransitionIsLoggedByTheSameCall(t *testing.T) {
	tr, _, clk, buf := newTracker(t)
	tr.Start()
	tr.Begin("audio", 0)
	clk.tick(3 * time.Second)
	tr.Done("audio")
	tr.Begin("decide", 5)
	clk.tick(2 * time.Second)
	tr.FailRunning()
	tr.Finish()

	out := buf.String()
	for _, want := range []string{
		`msg="job started"`,
		`msg="step started" step=audio`,
		`msg="step finished" step=audio elapsed=3s`,
		`msg="step started" step=decide total=5`,
		`msg="step failed" step=decide elapsed=2s`,
		`msg="job finished"`,
	} {
		require.Contains(t, out, want, "a transition with no log line is invisible to the operator")
	}
}

func TestTracker_AKeyedFailIsSavedAtOnceNotOnTheNextTransition(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	tr.Start()
	tr.Begin("audio", 0)
	before := sink.saveCount()

	tr.Fail("audio", errors.New("down"))

	require.Equal(t, before+1, sink.saveCount(), "a soft-failed phase must show as failed on the page now")
	require.Equal(t, domain.StepFailed, stateOf(sink.last(t), "audio"))
}

// retainingSink keeps the very slice it is handed, like a sink that queues it for later.
type retainingSink struct{ got [][]domain.ProgressStep }

func (r *retainingSink) SaveProgress(_ context.Context, steps []domain.ProgressStep, _ time.Time) error {
	r.got = append(r.got, steps)
	return nil
}
func (r *retainingSink) Beat(context.Context) error { return nil }

func TestTracker_ASnapshotHandedToTheSinkIsNotChangedByLaterTransitions(t *testing.T) {
	sink := &retainingSink{}
	tr := progress.NewTracker(context.Background(), sink, slog.New(slog.NewTextHandler(&syncBuf{}, nil)), plan(),
		progress.WithClock(newClock().now))
	tr.Start()
	first := sink.got[0]

	tr.Begin("audio", 0)
	tr.Done("audio")

	require.Equal(t, domain.StepPending, first[0].State, "a sink that keeps a snapshot must not see it change under it")
}

func TestTracker_AdvanceIsLoggedAtMostOncePerMinute(t *testing.T) {
	tr, _, clk, buf := newTracker(t)
	tr.Start()
	tr.Begin("decide", 100)

	for i := 1; i <= 30; i++ {
		clk.tick(1 * time.Second)
		tr.Advance("decide", i)
	}
	require.LessOrEqual(t, strings.Count(buf.String(), "step progress"), 1,
		"30 seconds of progress is at most one log line, not thirty")
	clk.tick(61 * time.Second)
	tr.Advance("decide", 31)
	require.Equal(t, 2, strings.Count(buf.String(), "step progress"))
	require.Contains(t, buf.String(), "done=31")
	require.Contains(t, buf.String(), "total=100")
}

func TestTracker_HeartbeatBeatsOnItsOwnUntilStopped(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})

	go func() {
		defer close(done)
		tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
	}()

	require.Eventually(t, func() bool { return sink.beatCount() >= 3 }, 2*time.Second, 5*time.Millisecond,
		"a heartbeat that does not depend on any engine call is the point")
	cancel()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("Heartbeat did not stop when its context ended")
	}
}

func TestTracker_HeartbeatSaysWhichStepIsStillRunningAndForHowLong(t *testing.T) {
	tr, _, clk, buf := newTracker(t)
	tr.Start()
	tr.Begin("decide", 100)
	clk.tick(7 * time.Minute)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go tr.Heartbeat(ctx, time.Hour, 5*time.Millisecond)

	require.Eventually(t, func() bool { return strings.Contains(buf.String(), "still running") }, 2*time.Second, 5*time.Millisecond)
	require.Contains(t, buf.String(), "step=decide")
	require.Contains(t, buf.String(), "elapsed=7m0s")
}

func TestFrom_WithoutAReporterReturnsOneThatDoesNothing(t *testing.T) {
	r := progress.From(context.Background())

	require.NotNil(t, r)
	require.NotPanics(t, func() {
		r.Begin("x", 3)
		r.Advance("x", 1)
		r.Done("x")
		r.Fail("x", errors.New("e"))
		r.Note("n", "k", 1)
	})
}

func TestWithAndFrom_ACarriedReporterIsTheOneReturned(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	tr.Start()
	ctx := progress.With(context.Background(), tr)

	progress.From(ctx).Begin("audio", 0)

	require.Equal(t, domain.StepRunning, stateOf(sink.last(t), "audio"))
}

func TestTracker_AnnotateAttachesANoteToAStepAndSavesItAtOnce(t *testing.T) {
	tr, sink, _, _ := newTracker(t)
	tr.Start()
	tr.Begin("audio", 0)
	before := sink.saveCount()

	tr.Annotate("audio", "20 min of the 79 the file claimed")

	require.Equal(t, before+1, sink.saveCount())
	require.Equal(t, "20 min of the 79 the file claimed", sink.last(t)[0].Note)
	require.NotPanics(t, func() { tr.Annotate("no-such-step", "x") })
}