package main // The overview page (novox/hq to-be 55 §11, ADR 0616): one read-only page for the operator with three blocks: every design in // progress with its milestones and their issues; the decisions and research efforts accepted in the last seven days and those // waiting; and what waits on the operator (records waiting for an Approve, the operator channel's open asks, the delivery // groups waiting). Built from mesh paths alone: the module's own ledger, and three other modules' verbs for the first view and // their events after it. It never polls: a source is asked again only when it is marked unanswered, on a bounded backoff. import ( "encoding/json" "fmt" "sort" "strings" "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" ) // The sources of the page, besides the module's own ledger. const ( SourceTracker = "issue-tracker" SourceDelivery = "mesh-delivery" SourceAsks = "operator-channel" ) // The verbs of the first view, and the events that keep each source current. const ( TrackerMilestonesVerb = "seat:issue-tracker.milestones" TrackerListVerb = "seat:issue-tracker.list" DeliveryGroupsVerb = "seat:mesh-delivery.groups" AsksVerb = "seat:operator-channel.asks" ) // OverviewEvents are the events the page consumes, by source, as the manifest declares them. var OverviewEvents = map[string][]string{ SourceTracker: {"mesh-issues.opened", "mesh-issues.moved", "mesh-issues.milestone-set", "mesh-issues.milestone-moved"}, SourceDelivery: {"mesh-delivery-gate.group", "mesh-delivery-gate.transition", "mesh-delivery-gate.merged"}, SourceAsks: {"messenger.offered", "messenger.closed"}, } // The unanswered marks: a verb that did not answer, or a subscription that missed its heartbeat. const ( MarkVerb = "verb" MarkSubscription = "subscription" ) // Backoff for a source marked from its verb: from 10 seconds, doubling, at most every 5 minutes (ADR 0616 §6). const ( BackoffFrom = 10 * time.Second BackoffMost = 5 * time.Minute // HeartbeatWithin is how long a subscription may be silent before its source is marked. HeartbeatWithin = 90 * time.Second // RecentFor is the window of the accepted records block, counted from the ledger's dates when the page is answered. RecentFor = 7 * 24 * time.Hour ) // askRow is an open ask of the operator channel. type askRow struct { Key string `json:"key"` Asker string `json:"asker"` Headline string `json:"headline"` State string `json:"state,omitempty"` Opened string `json:"opened,omitempty"` Offered []string `json:"offered_on,omitempty"` } // sourceMark says a source did not answer, and why. type sourceMark struct { Cause string `json:"cause"` Why string `json:"why"` Since time.Time `json:"since"` // Next is when the verb is asked again (a mark from the verb only). Next time.Time `json:"next,omitempty"` Backoff time.Duration `json:"-"` } // SourceState is one source on the page: answered, or marked, with the time of its last answer. type SourceState struct { Name string `json:"name"` Answered bool `json:"answered"` Mark *sourceMark `json:"mark,omitempty"` Last time.Time `json:"last-answer,omitempty"` } // RecordRow is a record on the page. type RecordRow struct { Record string `json:"record"` Title string `json:"title"` Status string `json:"status"` Date string `json:"date,omitempty"` Waits string `json:"waits,omitempty"` } // MilestoneRow is a milestone with its issues on the page. type MilestoneRow struct { Design int `json:"design"` ID string `json:"id"` Outcome string `json:"outcome"` Status string `json:"status"` Issues []IssueRow `json:"issues"` } // IssueRow is an issue of a milestone. type IssueRow struct { Number int `json:"number"` Title string `json:"title,omitempty"` Status string `json:"status,omitempty"` } // OverviewModel is the page's model, as the page renders it and the tool answers it: the same data. type OverviewModel struct { At time.Time `json:"at"` Designs []MilestoneRow `json:"milestones"` Accepted []RecordRow `json:"accepted-in-seven-days"` Waiting []RecordRow `json:"records-waiting"` Operator OperatorBlock `json:"waiting-on-the-operator"` Sources []SourceState `json:"sources"` LastRebuilt time.Time `json:"last-rebuilt"` } // OperatorBlock is what waits on the operator. type OperatorBlock struct { Records []RecordRow `json:"records"` Asks []askRow `json:"asks"` Groups []groupRow `json:"delivery-groups"` } // Overview is the page's model in memory, kept by events. type Overview struct { // Ask reaches a verb on a seat; Records reads the module's own ledger; Now is the clock. Ask SeatAsk Records func() ([]*Entry, error) Now func() time.Time mu sync.Mutex milestones map[string]trackerMilestone issues map[int]trackerRow groups map[string]groupRow asks map[string]askRow state map[string]*SourceState rebuilt time.Time // verbCalls counts the verbs asked, so a test sees that an event asked none. verbCalls map[string]int } // NewOverview is a model with no source answered: Start asks each source's verb once. func NewOverview(ask SeatAsk, records func() ([]*Entry, error), now func() time.Time) *Overview { o := &Overview{Ask: ask, Records: records, Now: now, milestones: map[string]trackerMilestone{}, issues: map[int]trackerRow{}, groups: map[string]groupRow{}, asks: map[string]askRow{}, state: map[string]*SourceState{}, verbCalls: map[string]int{}} for _, s := range []string{SourceTracker, SourceDelivery, SourceAsks} { o.state[s] = &SourceState{Name: s} } return o } func (o *Overview) now() time.Time { if o.Now != nil { return o.Now().UTC() } return time.Now().UTC() } // rowsOf reads the rows a verb answers: a bare list, or an object holding one list. func rowsOf(raw json.RawMessage, into any, field string) error { if err := json.Unmarshal(raw, into); err == nil { return nil } var wrapped map[string]json.RawMessage if err := json.Unmarshal(raw, &wrapped); err != nil { return err } if v, ok := wrapped[field]; ok { return json.Unmarshal(v, into) } return fmt.Errorf("no %s in the answer", field) } func milestoneKey(design int, id string) string { return fmt.Sprintf("%d/%s", design, id) } // Rebuild asks a source's verb and rebuilds its block from the answer, clearing its mark; a verb that does not answer marks // the source from the verb, with the next try on the backoff. It is the first view at start, the answer to a mark, and the // hand act of the re-read tool. func (o *Overview) Rebuild(source string) error { o.mu.Lock() defer o.mu.Unlock() return o.rebuildLocked(source) } func (o *Overview) rebuildLocked(source string) error { st := o.state[source] if st == nil { return fmt.Errorf("%q is no source of the page (%s)", source, strings.Join(o.sourceNames(), ", ")) } call := func(verb string, body any) (json.RawMessage, error) { o.verbCalls[verb]++ return o.Ask(verb, body) } fail := func(why string) error { prev := st.Mark back := BackoffFrom if prev != nil && prev.Cause == MarkVerb { back = min(prev.Backoff*2, BackoffMost) } since := o.now() if prev != nil { since = prev.Since } st.Answered, st.Mark = false, &sourceMark{Cause: MarkVerb, Why: why, Since: since, Next: o.now().Add(back), Backoff: back} return fmt.Errorf("%s does not answer: %s", source, why) } switch source { case SourceTracker: raw, err := call(TrackerMilestonesVerb, map[string]any{}) if err != nil { return fail(err.Error()) } var ms []trackerMilestone if err := rowsOf(raw, &ms, "milestones"); err != nil { return fail("milestones answered something else: " + err.Error()) } raw, err = call(TrackerListVerb, map[string]any{}) if err != nil { return fail(err.Error()) } var rows []trackerRow if err := rowsOf(raw, &rows, "issues"); err != nil { return fail("list answered something else: " + err.Error()) } o.milestones, o.issues = map[string]trackerMilestone{}, map[int]trackerRow{} for _, m := range ms { o.milestones[milestoneKey(m.Design, m.ID)] = m } for _, r := range rows { o.issues[r.Number] = r } case SourceDelivery: raw, err := call(DeliveryGroupsVerb, map[string]any{}) if err != nil { return fail(err.Error()) } var gs []groupRow if err := rowsOf(raw, &gs, "groups"); err != nil { return fail("groups answered something else: " + err.Error()) } o.groups = map[string]groupRow{} for _, g := range gs { o.groups[g.ID] = g } case SourceAsks: raw, err := call(AsksVerb, map[string]any{}) if err != nil { return fail(err.Error()) } var as []askRow if err := rowsOf(raw, &as, "asks"); err != nil { return fail("asks answered something else: " + err.Error()) } o.asks = map[string]askRow{} for _, a := range as { if a.State == "" || a.State == "open" || a.State == "held" { o.asks[a.Key] = a } } } st.Answered, st.Mark, st.Last = true, nil, o.now() o.rebuilt = o.now() return nil } func (o *Overview) sourceNames() []string { out := make([]string, 0, len(o.state)) for n := range o.state { out = append(out, n) } sort.Strings(out) return out } // Start is the first view: every source's verb asked once. A verb that does not answer marks its source. func (o *Overview) Start() { for _, s := range []string{SourceTracker, SourceDelivery, SourceAsks} { _ = o.Rebuild(s) } } // Due asks again, once, every source marked from its verb whose backoff has run out: the only asks the page makes after start // besides the one a subscription's return makes. It answers the sources asked. func (o *Overview) Due() []string { o.mu.Lock() defer o.mu.Unlock() var asked []string for _, s := range o.sourceNames() { if m := o.state[s].Mark; m != nil && m.Cause == MarkVerb && !o.now().Before(m.Next) { _ = o.rebuildLocked(s) asked = append(asked, s) } } return asked } // Silent marks the sources whose subscription has been silent past its bound: nothing is asked while it is silent. func (o *Overview) Silent() []string { o.mu.Lock() defer o.mu.Unlock() var marked []string for _, s := range o.sourceNames() { st := o.state[s] if st.Answered && st.Mark == nil && !st.Last.IsZero() && o.now().Sub(st.Last) > HeartbeatWithin { st.Answered, st.Mark = false, &sourceMark{Cause: MarkSubscription, Why: fmt.Sprintf("its subscription missed the bus's heartbeat for %s", HeartbeatWithin), Since: o.now()} marked = append(marked, s) } } return marked } // Heard takes an event, or a heartbeat when event is empty, of a source into the model. A source marked from its subscription // is asked once, on its return, to rebuild what it missed; that answer clears the mark. func (o *Overview) Heard(source, event string, body json.RawMessage) { o.mu.Lock() defer o.mu.Unlock() st := o.state[source] if st == nil { return } if st.Mark != nil && st.Mark.Cause == MarkSubscription { _ = o.rebuildLocked(source) if !st.Answered { return } } if st.Mark == nil { st.Last = o.now() } if event == "" { return } var f struct { Number int `json:"number"` Status string `json:"status"` To string `json:"to"` State string `json:"state"` Title string `json:"title"` Design int `json:"design"` ID string `json:"id"` Group string `json:"group"` Outcome string `json:"outcome"` Issues []int `json:"issues"` Members []string `json:"members"` Why string `json:"why"` Key string `json:"key"` Asker string `json:"asker"` Headline string `json:"headline"` Offered []string `json:"offered_on"` Channels []string `json:"channels"` } if json.Unmarshal(body, &f) != nil { return } name := event[strings.LastIndexByte(event, '.')+1:] switch source { case SourceTracker: switch name { case "opened": if f.Number > 0 { o.issues[f.Number] = trackerRow{Number: f.Number, Title: f.Title, Status: firstNonEmpty(f.Status, "open")} } case "moved": if r, ok := o.issues[f.Number]; ok { r.Status = firstNonEmpty(f.To, f.Status, r.Status) o.issues[f.Number] = r } else if f.Number > 0 { o.issues[f.Number] = trackerRow{Number: f.Number, Title: f.Title, Status: firstNonEmpty(f.To, f.Status)} } case "milestone-set", "milestone-moved": k := milestoneKey(f.Design, f.ID) m := o.milestones[k] m.Design, m.ID = f.Design, f.ID m.Outcome, m.Issues = firstNonEmpty(f.Outcome, m.Outcome), append([]int(nil), firstInts(f.Issues, m.Issues)...) m.Status = firstNonEmpty(f.To, f.Status, m.Status) o.milestones[k] = m } case SourceDelivery: id := firstNonEmpty(f.Group, f.ID) switch name { case "group": g := o.groups[id] g.ID, g.State, g.Why = id, firstNonEmpty(f.State, f.To, g.State), firstNonEmpty(f.Why, g.Why) g.Members = firstStrings(f.Members, g.Members) o.groups[id] = g case "transition": if g, ok := o.groups[id]; ok { g.State = firstNonEmpty(f.To, f.State, g.State) g.Why = firstNonEmpty(f.Why, g.Why) o.groups[id] = g } case "merged": delete(o.groups, id) } case SourceAsks: switch name { case "offered": k := firstNonEmpty(f.Key, f.Asker+"."+f.ID) o.asks[k] = askRow{Key: k, Asker: f.Asker, Headline: f.Headline, Offered: firstStrings(f.Offered, f.Channels)} case "closed": delete(o.asks, firstNonEmpty(f.Key, f.Asker+"."+f.ID)) } } o.rebuilt = o.now() } func firstInts(a, b []int) []int { if len(a) > 0 { return a } return b } func firstStrings(a, b []string) []string { if len(a) > 0 { return a } return b } // Model is the page's model at the time asked: the ledger's dates are counted now, the other blocks are what the events kept. func (o *Overview) Model() (*OverviewModel, error) { var entries []*Entry if o.Records != nil { var err error if entries, err = o.Records(); err != nil { return nil, fmt.Errorf("the records cannot be read: %v", err) } } o.mu.Lock() defer o.mu.Unlock() now := o.now() m := &OverviewModel{At: now, LastRebuilt: o.rebuilt} var ms []trackerMilestone for _, t := range o.milestones { ms = append(ms, t) } sort.Slice(ms, func(i, j int) bool { if ms[i].Design != ms[j].Design { return ms[i].Design < ms[j].Design } return ms[i].ID < ms[j].ID }) for _, t := range ms { if t.Status == "done" || t.Status == "dropped" { continue } row := MilestoneRow{Design: t.Design, ID: t.ID, Outcome: t.Outcome, Status: t.Status, Issues: []IssueRow{}} for _, n := range t.Issues { i := o.issues[n] row.Issues = append(row.Issues, IssueRow{Number: n, Title: i.Title, Status: i.Status}) } m.Designs = append(m.Designs, row) } cutoff := now.Add(-RecentFor) for _, e := range entries { r := e.Working() row := RecordRow{Record: r.Key().String(), Title: r.Title, Status: r.Status, Date: r.Date} switch { case e.Current.Accepted() && acceptedAt(e.Current).After(cutoff): m.Accepted = append(m.Accepted, row) case r.Status == StatusProposed: if r.NeedsOperator && r.Ask != nil && r.Ask.Said == "waiting" { row.Waits = "the operator's Approve" m.Operator.Records = append(m.Operator.Records, row) } else if r.Judging != nil && r.Judging.State == JudgingFailed { row.Waits = "its review, which could not be had" } else { row.Waits = "its review" } m.Waiting = append(m.Waiting, row) } } sort.Slice(m.Accepted, func(i, j int) bool { return m.Accepted[i].Record > m.Accepted[j].Record }) for _, a := range o.asks { m.Operator.Asks = append(m.Operator.Asks, a) } sort.Slice(m.Operator.Asks, func(i, j int) bool { return m.Operator.Asks[i].Key < m.Operator.Asks[j].Key }) for _, g := range o.groups { if g.State != "merged" { m.Operator.Groups = append(m.Operator.Groups, g) } } sort.Slice(m.Operator.Groups, func(i, j int) bool { return m.Operator.Groups[i].ID < m.Operator.Groups[j].ID }) for _, n := range o.sourceNames() { m.Sources = append(m.Sources, *o.state[n]) } return m, nil } // acceptedAt is when a record was accepted here, or its day for one imported accepted. func acceptedAt(r *Record) time.Time { if !r.AcceptedAt.IsZero() { return r.AcceptedAt } t, _ := time.Parse("2006-01-02", r.Date) return t } // Unanswered are the sources marked, each with its cause, for the module's health: degraded while any is. func (o *Overview) Unanswered() []SourceState { o.mu.Lock() defer o.mu.Unlock() var out []SourceState for _, n := range o.sourceNames() { if !o.state[n].Answered { out = append(out, *o.state[n]) } } return out } // Calls is how many times a verb was asked. func (o *Overview) Calls(verb string) int { o.mu.Lock() defer o.mu.Unlock() return o.verbCalls[verb] } // Beat takes the runtime's word on the module's consumer between events (stdio.OnHeartbeat): one consumer carries every // subscription, so a beat is heard by every source the page subscribes to, and a missed heartbeat marks each of them // from its subscription at once, with the runtime's why, instead of waiting out HeartbeatWithin. func (o *Overview) Beat(h stdio.Heartbeat) { sources := make([]string, 0, len(OverviewEvents)) for s := range OverviewEvents { sources = append(sources, s) } sort.Strings(sources) if !h.Missed { for _, s := range sources { o.Heard(s, "", nil) } return } why := "its subscription's consumer missed the bus's heartbeat" if h.Why != "" { why += ": " + h.Why } o.mu.Lock() defer o.mu.Unlock() for _, s := range sources { st := o.state[s] switch { case st == nil: case st.Mark == nil: st.Answered, st.Mark = false, &sourceMark{Cause: MarkSubscription, Why: why, Since: o.now()} case st.Mark.Cause == MarkSubscription: st.Mark.Why = why } } }