mesh-catalog/modules/messenger/cmd/messenger/status.go

package main

import (
	"fmt"
	"sort"
	"strings"
	"time"
)

// ChannelStatus is one channel as the status says it.
type ChannelStatus struct {
	Name         string `json:"name"`
	CanSend      bool   `json:"can_send"`
	NotReady     string `json:"not_ready,omitempty"`
	LastError    string `json:"last_error,omitempty"`
	LastSent     string `json:"last_sent,omitempty"`
	SentLastHour int    `json:"sent_last_hour"`
	Cap          int    `json:"cap_per_hour"`
	Held         int    `json:"held"` // waiting for their window, the digest or the cap
}

// Status is the holder's whole account of itself.
type Status struct {
	Verdict      string          `json:"verdict"`
	Channels     []ChannelStatus `json:"channels"`
	Open         int             `json:"open"`
	Silenced     int             `json:"silenced"`
	Unsent       []string        `json:"unsent,omitempty"`
	Refused      int             `json:"refused_since_start"`
	Unreadable   int             `json:"unreadable_events_since_start"`
	LastBad      string          `json:"last_unreadable,omitempty"`
	Heard        map[string]int  `json:"events_heard_since_start"`
	LastHeard    string          `json:"last_event_heard,omitempty"`
	Listening    string          `json:"listening"`
	StateRead    string          `json:"state_read,omitempty"`       // when the open conditions were last read from the controller
	StateUnread  string          `json:"state_unreadable,omitempty"` // why they could not be, lately
	History      int             `json:"events_already_in_state"`    // older than the reading: not acted on
	Stale        int             `json:"events_taken_as_state"`      // older than the freshness bound: recorded, never said
	Grace        bool            `json:"in_start_grace,omitempty"`
	StateProblem string          `json:"state_problem,omitempty"`
	TimeZone     string          `json:"time_zone"` // the zone every time in a message is said in
	Rules        []string        `json:"rules"`
}

// Status answers at once from what the holder keeps.
func (h *Holder) Status(listening string) Status {
	now := h.Now()
	var st Status
	for _, ch := range h.kindChannels() {
		cs := ChannelStatus{Name: ch.Name(), Cap: CapPerHour, SentLastHour: h.sentLastHour(ch.Name())}
		if err := ch.Ready(); err != nil {
			cs.NotReady = err.Error()
		} else {
			cs.CanSend = true
		}
		h.mu.Lock()
		cs.LastError = h.chanErr[ch.Name()]
		if t, ok := h.chanOK[ch.Name()]; ok {
			cs.LastSent = t.UTC().Format(time.RFC3339)
		}
		cs.Held = len(h.outbox[ch.Name()])
		h.mu.Unlock()
		if cs.LastError != "" {
			cs.CanSend = false
		}
		st.Channels = append(st.Channels, cs)
	}
	h.mu.Lock()
	for _, r := range h.open {
		if !r.Cleared.IsZero() {
			continue
		}
		st.Open++
		if r.silenced(now) {
			st.Silenced++
		}
		if r.Pending != "" {
			st.Unsent = append(st.Unsent, r.Key+" ("+r.Pending+")")
		}
	}
	st.Refused = len(h.refusals)
	st.Unreadable = h.unreadable
	st.LastBad = h.lastBad
	st.Heard = map[string]int{}
	for k, v := range h.heard {
		st.Heard[k] = v
	}
	if !h.lastHeard.IsZero() {
		st.LastHeard = h.lastHeard.UTC().Format(time.RFC3339)
	}
	st.StateProblem = h.storeErr
	if !h.syncedAt.IsZero() {
		st.StateRead = h.syncedAt.UTC().Format(time.RFC3339)
	}
	st.StateUnread = h.syncErr
	st.History, st.Stale = h.history, h.stale
	st.Grace = h.inGrace(now)
	h.mu.Unlock()
	sort.Strings(st.Unsent)
	st.Listening = listening
	st.TimeZone = h.zone().String()
	st.Rules = []string{
		"a message reads in one glance: the condition's headline as its title, its explanation, and since when in the operator's time zone; no key, id, commit or command — those stay in the controller's conditions and in this holder's history (novox/hq ADR 0253)",
		"a clearance is one line: the condition's resolved line and how long it was open",
		"history is never news: what is open is read from the controller on start, every 10 min and when an old event arrives; an event older than that reading is not acted on, one older than 10 min by its own time is state, never a message",
		"a clearing of something never said says nothing; an urgent condition open and never said is said, all of them in one message",
		"nothing is sent in the first minute after start but that one message",
		"urgent: every channel kind that can send; warning: the desk where it can send, otherwise the channel that reaches the operator away",
		"the channels are kinds of the channel bench, each a module of its own; which exist and what each promises is the controller's word, in this module's membership; whether each can send is its own (novox/hq ADR 0259)",
		"an ask is held here and offered on each kind whose capabilities carry its answers; an answer is judged here and closes the ask with a warrant its asker acts on; nothing is performed here",
		"an account linked as the operator acknowledges at once and approves an hour after it was linked; every link and unlink is said on every channel",
		"the desk proves nothing about who clicked: it answers acknowledgements only",
		"bursts are one message: what arrives within 30 s goes out together, a digest when it is several; urgent goes at once when the channel was quiet for 30 s",
		"the desk is gentle: warnings at most one message every 15 min",
		"deduplicated by the condition's key; reminded once after 1 h (urgent) or 12 h (warning); cleared by editing the first message",
		"at most 20 messages an hour per channel, a last line of defence: reached, it is said once and the rest held for one digest",
		"refused: anything carrying an address, a path or a secret; channel-refused is sent with the words withheld",
		"while a delivery walks, a message about a machine or module it touches is held, and the walk's end is one message: its outcome, what it reached, how long, what is still open; past the walk's bound (twice its largest delivery budget, 15 min with none) what is held and open is said (novox/hq ADR 0308)",
		"an ask, a secret given or replaced, and a walk's own failure are never held",
		"a channel that lost its answers is said again only what is open, asked there and never answered — never a silenced or withheld message — once per restart, and the pass is kept in this module's state",
		"a silenced condition's message says it is silenced, until when and why, and nothing more is said of it while the silence lasts; a silence is the controller's, asked of the operator or set through the mesh",
	}
	anyCan := false
	for _, c := range st.Channels {
		anyCan = anyCan || c.CanSend
	}
	switch {
	case !anyCan:
		st.Verdict = "CANNOT SEND: no channel can reach the operator — see channels"
	case len(st.Unsent) > 0:
		st.Verdict = "BEHIND: some messages could not be sent yet and are tried every minute"
	case listening != "listening":
		st.Verdict = "NOT LISTENING: condition events are not reaching this holder — " + listening
	case st.StateProblem != "":
		st.Verdict = "STATE: open messages are held in memory only — " + st.StateProblem
	default:
		st.Verdict = "ok"
		for _, c := range st.Channels {
			if !c.CanSend {
				st.Verdict = "ok on one channel: " + c.Name + " cannot send"
			}
		}
	}
	return st
}

// OpenMessage is one open message as `open` answers it.
type OpenMessage struct {
	Key      string   `json:"key"`
	Severity string   `json:"severity"`
	Summary  string   `json:"summary"`
	Subject  string   `json:"subject,omitempty"`
	Since    string   `json:"since"`
	Origin   string   `json:"origin"`
	SentOn   []string `json:"sent_on,omitempty"`
	Silenced string   `json:"silenced_until,omitempty"`
	Reminded bool     `json:"reminded"`
	Unsent   string   `json:"unsent,omitempty"`
	Held     bool     `json:"held,omitempty"`
	// HeldBy are the walks holding it until they end or pass their bound (novox/hq ADR 0308).
	HeldBy  []string `json:"held_by_walks,omitempty"`
	History bool     `json:"never_said,omitempty"`
	Refused string   `json:"refused,omitempty"`
	Count   int      `json:"times_opened"`
}

// Open is what is unresolved now, urgent first, oldest first.
func (h *Holder) Open() []OpenMessage {
	now := h.Now()
	h.mu.Lock()
	defer h.mu.Unlock()
	queued := map[string]bool{}
	for _, list := range h.outbox {
		for _, it := range list {
			queued[it.Key] = true
		}
	}
	var out []OpenMessage
	for _, r := range h.open {
		if !r.Cleared.IsZero() {
			continue
		}
		o := OpenMessage{Key: r.Key, Severity: r.Severity, Summary: r.Summary, Subject: r.Subject,
			Since: r.Raised.UTC().Format(time.RFC3339), Origin: r.Origin, Reminded: r.Reminded,
			Unsent: r.Pending, Held: r.Folded || queued[r.Key], History: r.History, Refused: r.Refused, Count: r.Count}
		for ch := range r.Sent {
			o.SentOn = append(o.SentOn, ch)
		}
		sort.Strings(o.SentOn)
		o.HeldBy = h.heldBy(r.Key, now)
		if r.silenced(now) {
			o.Silenced = r.SilencedTill.UTC().Format(time.RFC3339)
		}
		out = append(out, o)
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Severity != out[j].Severity {
			return out[i].Severity == Urgent
		}
		return out[i].Since < out[j].Since
	})
	return out
}

// History is what went out, newest first, and the refusals.
func (h *Holder) History(limit int) map[string]any {
	h.mu.Lock()
	defer h.mu.Unlock()
	sends := []Sent{}
	for i := len(h.recent) - 1; i >= 0 && len(sends) < limit; i-- {
		sends = append(sends, h.recent[i])
	}
	refusals := []RefusalNote{}
	for i := len(h.refusals) - 1; i >= 0 && len(refusals) < limit; i-- {
		refusals = append(refusals, h.refusals[i])
	}
	return map[string]any{"sends": sends, "refusals": refusals}
}

// Test sends a test message on one channel or both, through the content rule and the cap.
func (h *Holder) Test(channel, text string) map[string]string {
	h.work.Lock()
	defer h.work.Unlock()
	if text == "" {
		text = "a test from the operator-channel's holder: this channel reaches you"
	}
	out := map[string]string{}
	if refusal, ok := Check(text); !ok {
		out["refused"] = "it carried " + refusal.String() + "; nothing was sent"
		return out
	}
	m := Message{Title: "Test message: " + text, Body: "Nothing is wrong; this was asked for."}
	for _, ch := range h.kindChannels() {
		if channel != "" && channel != "both" && channel != ch.Name() {
			continue
		}
		if err := ch.Ready(); err != nil {
			out[ch.Name()] = "not sent: " + err.Error()
			continue
		}
		if !h.allow(ch.Name()) {
			out[ch.Name()] = "not sent: the channel is at its cap of 20 an hour"
			continue
		}
		if h.sendNow(ch, "messenger.test", "test", m) {
			out[ch.Name()] = "sent"
		} else {
			h.mu.Lock()
			out[ch.Name()] = "failed: " + h.chanErr[ch.Name()]
			h.mu.Unlock()
		}
	}
	return out
}

// Notify takes a message from a module that uses the seat (research 028 Q5, modules as sources):
// the same shape, rule, cap and deduplication as a condition. Its key is put under `notify.` so it
// can never be taken for one of the controller's conditions.
func (h *Holder) Notify(key, severity, summary, subject string, clear bool, plain ...string) (map[string]any, error) {
	if key == "" {
		return nil, errorf("key is required: the same key is the same message")
	}
	if len(key) < 7 || key[:7] != "notify." {
		key = "notify." + key
	}
	h.work.Lock()
	defer h.work.Unlock()
	h.mu.Lock()
	h.init()
	h.mu.Unlock()
	if clear {
		h.cleared(key, false)
		return map[string]any{"key": key, "cleared": true}, nil
	}
	switch severity {
	case Urgent, Warning:
	case "":
		return nil, errorf("severity is required: urgent or warning")
	default:
		return nil, errorf("severity %q is neither urgent nor warning", severity)
	}
	if summary == "" {
		return nil, errorf("summary is required: one line in the mesh's words")
	}
	rec := Record{Key: key, Kind: "notice", Subject: subject, Severity: severity, Summary: summary,
		Origin: "notify", More: "operator-channel.open"}
	// The plain words a module may give (novox/hq ADR 0253): a headline, an explanation, a resolved line.
	for i, w := range plain {
		switch i {
		case 0:
			rec.Headline = w
		case 1:
			rec.Explanation = w
		case 2:
			rec.Resolved = w
		}
	}
	h.raised(rec, false)
	h.mu.Lock()
	r := h.open[key]
	var sent []string
	refused, pending := "", ""
	if r != nil {
		for ch := range r.Sent {
			sent = append(sent, ch)
		}
		refused, pending = r.Refused, r.Pending
	}
	h.mu.Unlock()
	sort.Strings(sent)
	held := h.queued(key)
	return map[string]any{"key": key, "sent_on": sent, "held": held, "refused": refused, "unsent": pending}, nil
}

// Tell sends a plain notice that needs no answer and opens no condition (ADR 0380 §8): one message with a
// headline and one or two lines, said at once on every channel the router sends on — bypassing the severity
// routing, the burst window and the walk hold that a condition's message goes through, and never reminded,
// escalated or kept open. Composed and held to the content rule exactly as notify's words are (allowedWords'
// reduction, withheld parts as today). It offers Details, like any message, and is kept in the history so
// `history` shows it — but its record is already marked cleared, so `open` never shows it and it is forgotten
// after ReopenWindow, as any cleared record is. Its cap, TellCapPerHour a channel an hour, is its own and
// never shared with CapPerHour: past it, a tell is dropped, said once an hour in the log and named in this
// call's own "dropped" answer, but never kept in the bounded send history the conditions' cap counts from
// (KeptSends, outbox.go's sentLastHour) — a flood of drops must never evict a real, still-recent send from it
// and let that cap stop holding (review round 2 of issue 800, issue 797's area).
func (h *Holder) Tell(headline string, lines []string, subject string) (map[string]any, error) {
	headline = strings.TrimSpace(headline)
	if headline == "" {
		return nil, errorf("headline is required")
	}
	if len(lines) == 0 || len(lines) > 2 {
		return nil, errorf("lines takes one or two lines")
	}
	clean := make([]string, 0, len(lines))
	for _, l := range lines {
		l = strings.TrimSpace(l)
		if l == "" {
			return nil, errorf("a line is empty")
		}
		clean = append(clean, l)
	}
	h.work.Lock()
	defer h.work.Unlock()
	h.mu.Lock()
	h.init()
	h.mu.Unlock()
	now := h.Now()
	h.mu.Lock()
	h.tellSeq++
	seq := h.tellSeq
	h.mu.Unlock()
	key := fmt.Sprintf("tell.%d.%d", now.UnixNano(), seq)
	rec := &Record{Key: key, Kind: "tell", Subject: subject, Origin: "tell", Headline: headline,
		Explanation: strings.Join(clean, "\n"), Summary: headline, Raised: now, Seen: now, Count: 1,
		Sent: map[string]string{}, Cleared: now}
	h.mu.Lock()
	h.open[key] = rec
	h.mu.Unlock()
	var sentOn, dropped []string
	for _, ch := range h.channels() {
		name := ch.Name()
		if ch.Ready() != nil {
			continue
		}
		if !h.allowTell(name) {
			dropped = append(dropped, name)
			h.mu.Lock()
			said := h.tellCapSaid[name]
			sayIt := said.IsZero() || now.Sub(said) >= time.Hour
			if sayIt {
				h.tellCapSaid[name] = now
			}
			h.mu.Unlock()
			if sayIt {
				h.Logf("[messenger] a tell was dropped on %s: it is at its own cap of %d tells an hour", name, TellCapPerHour)
			}
			// Not h.record'd: that history is bounded (KeptSends) and is what the conditions' cap counts from
			// (sentLastHour, outbox.go); a flood of drops must never evict a real send from it and let the
			// conditions' cap stop holding (review round 2 of issue 800, issue 797's area). This call's own
			// "dropped" answer, and the once-an-hour log line above, already say it happened.
			continue
		}
		m := h.sayOn(rec, "tell", name)
		id, err := ch.Send(m)
		h.noteChannel(name, err)
		if err != nil {
			h.record(Sent{At: now, Channel: name, Key: key, What: "tell", Outcome: "failed: " + err.Error()})
			continue
		}
		h.mu.Lock()
		rec.Sent[name] = id
		h.mu.Unlock()
		sentOn = append(sentOn, name)
		h.record(Sent{At: now, Channel: name, Key: key, What: "tell", Outcome: "sent"})
	}
	h.persist(rec)
	sort.Strings(sentOn)
	sort.Strings(dropped)
	return map[string]any{"key": key, "sent_on": sentOn, "dropped": dropped, "refused": rec.Refused}, nil
}