mesh-catalog/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go

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