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")) }