From 89025bee59bf67ec88533e131b824e5da37634e3 Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Thu, 22 Oct 2026 21:00:45 -0700 Subject: [PATCH] Derive pending review stages directly from dependency state --- internal/store/review.go | 55 ++++++++++++++++++---------------------- 1 file changed, 25 insertions(+), 30 deletions(-) diff --git a/internal/store/review.go b/internal/store/review.go index 28b04b5..9a9906e 100644 --- a/internal/store/review.go +++ b/internal/store/review.go @@ -189,57 +189,53 @@ func pendingStages(ctx context.Context, tx *sql.Tx) ([]StageStart, error) { if err != nil { return nil, err } - type defined struct { - review api.Review - defID string + type reviewRun struct { + defID string + byStage map[string]api.Review } - found := []defined{} + runs := map[SnapshotRef]reviewRun{} for rows.Next() { - d := defined{} - d.review, err = scanReview(trailingScanner{rows, []any{&d.defID}}) + defID := "" + r, err := scanReview(trailingScanner{rows, []any{&defID}}) if err != nil { rows.Close() return nil, err } - found = append(found, d) + if defID == "" { + continue + } + ref := SnapshotRef{TaskID: r.TaskId, N: r.SnapshotN} + run, found := runs[ref] + if !found { + run = reviewRun{defID: defID, byStage: map[string]api.Review{}} + runs[ref] = run + } + run.byStage[r.Stage] = r } err = errors.Join(rows.Err(), rows.Close()) if err != nil { return nil, err } - byTask := map[string]map[string]api.Review{} - for _, d := range found { - if byTask[d.review.TaskId] == nil { - byTask[d.review.TaskId] = map[string]api.Review{} - } - byTask[d.review.TaskId][d.review.Stage] = d.review - } defs := map[string]api.WorkflowDefinition{} starts := []StageStart{} - seen := map[string]bool{} - for _, d := range found { - if d.review.State != api.ReviewStateDone || d.defID == "" { - continue - } - def, loaded := defs[d.defID] + for ref, run := range runs { + def, loaded := defs[run.defID] if !loaded { - def, err = loadDefinition(ctx, tx, d.defID) + def, err = loadDefinition(ctx, tx, run.defID) if err != nil { return nil, err } - defs[d.defID] = def + defs[run.defID] = def } - byStage := byTask[d.review.TaskId] - for _, next := range workflow.Next(def, d.review.Stage) { - key := d.review.TaskId + "\x00" + next - if _, started := byStage[next]; started || seen[key] { + for _, stage := range workflow.Stages(def) { + if _, started := run.byStage[stage.Stage]; started { continue } inputs := []api.Review{} - after := workflow.After(def, next) + after := workflow.After(def, stage.Stage) for _, a := range after { - in := byStage[a] + in := run.byStage[a] if in.State != api.ReviewStateDone { break } @@ -248,8 +244,7 @@ func pendingStages(ctx context.Context, tx *sql.Tx) ([]StageStart, error) { if len(inputs) != len(after) { continue } - seen[key] = true - starts = append(starts, StageStart{TaskID: d.review.TaskId, SnapshotN: d.review.SnapshotN, DefinitionID: d.defID, Definition: def, Stage: next, Inputs: inputs}) + starts = append(starts, StageStart{TaskID: ref.TaskID, SnapshotN: ref.N, DefinitionID: run.defID, Definition: def, Stage: stage.Stage, Inputs: inputs}) } } return starts, nil