Derive pending review stages directly from dependency state
This commit is contained in:
+24
-29
@@ -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
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user