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

package main

// The outbox: nothing is sent the moment it is decided. Each channel holds what is to be said for a
// short window and sends it as one message — the message itself when it is alone, a digest when it
// is not — so a burst of conditions is one notification, never one each (novox/hq issue 271).

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

// item is one thing to say on one channel: a record's message (Key, What), or a message of the
// holder's own (Msg).
type item struct {
	Key      string
	What     string
	Severity string
	At       time.Time
	Msg      *Message
	// Carries are the records a walk's end or bound message tells (ADR 0308 §4): each is marked told on a
	// channel only once that message was sent there (issue 538 A), never when it is held to be sent.
	Carries []string
}

func (i item) urgent() bool { return i.Severity == Urgent && i.What != "cleared" }

// deliver decides where a record's message goes and holds it there. Urgent goes to every channel
// that can send; a warning to the desktop when it is configured, otherwise to Telegram — and from
// the desktop to Telegram when no session there takes it. No channel at all: it is unsent, said in
// the status, and tried again every minute.
func (h *Holder) deliver(r *Record, what string) {
	if needsNothing(r) && what != "escalated" && what != "quiet-reminder" {
		// Nothing for the operator to do, and only a warning: kept, not sent (ADR 0253).
		h.record(Sent{At: h.Now(), Channel: "-", Key: r.Key, What: what, Outcome: "quiet: nothing for the operator to do"})
		return
	}
	if h.hold(r, what) {
		// An open walk touches it: told when the walk ends or passes its bound (ADR 0308 §2).
		return
	}
	var to []Channel
	if kinds := h.askKinds(r); len(kinds) > 0 {
		// An ask goes to every channel that carries one of its answers (novox/hq ADR 0259): the desk for an
		// acknowledgement, the channel that proves its sender for an approval — whatever its severity.
		for _, ch := range h.channels() {
			if kinds[ch.Name()] && ch.Ready() == nil {
				to = append(to, ch)
			}
		}
	} else if r.Severity == Urgent {
		for _, ch := range h.channels() {
			if ch.Ready() == nil {
				to = append(to, ch)
			}
		}
	} else {
		switch {
		case h.Desk != nil && h.Desk.Ready() == nil:
			to = []Channel{h.Desk}
		case h.Away != nil && h.Away.Ready() == nil:
			to = []Channel{h.Away}
		}
	}
	if len(to) == 0 {
		h.mu.Lock()
		r.Pending = what
		h.mu.Unlock()
		h.Logf("[messenger] could not send %s %s on any channel; trying again every minute: %s", what, r.Key, h.whyNot())
		return
	}
	now := h.Now()
	for _, ch := range to {
		h.enqueue(ch.Name(), item{Key: r.Key, What: what, Severity: r.Severity, At: now})
	}
	h.flush(now, false)
}

// enqueue holds an item on a channel; an item for the same key replaces the one held before.
func (h *Holder) enqueue(channel string, it item) {
	h.mu.Lock()
	defer h.mu.Unlock()
	list := h.outbox[channel]
	for i := range list {
		if list[i].Msg == nil && it.Msg == nil && list[i].Key == it.Key {
			at := list[i].At
			list[i] = it
			list[i].At = at
			return
		}
	}
	h.outbox[channel] = append(list, it)
}

// unqueue drops what is held for a key, on every channel: it is over before it was said.
func (h *Holder) unqueue(key string) int {
	h.mu.Lock()
	defer h.mu.Unlock()
	n := 0
	for ch, list := range h.outbox {
		kept := list[:0]
		for _, it := range list {
			if it.Msg == nil && it.Key == key {
				n++
				continue
			}
			kept = append(kept, it)
		}
		h.outbox[ch] = kept
	}
	return n
}

// queuedWhat is what is held to be said of a record, not its clearing: "" when nothing is.
func (h *Holder) queuedWhat(key string) string {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, list := range h.outbox {
		for _, it := range list {
			if it.Msg == nil && it.Key == key && it.What != "cleared" {
				return it.What
			}
		}
	}
	return ""
}

// queued says whether something is held to be said of a record: its own message, or a walk's message that
// carries it.
func (h *Holder) queued(key string) bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, list := range h.outbox {
		for _, it := range list {
			if it.Msg == nil && it.Key == key {
				return true
			}
			for _, k := range it.Carries {
				if k == key {
					return true
				}
			}
		}
	}
	return false
}

// edit changes the first message on each channel that showed it. A channel that cannot, or whose
// edit fails, or that said it inside a digest, is given a new message instead — held like any other.
// silentOnly: only where an edit notifies nobody; elsewhere nothing.
func (h *Holder) edit(r *Record, what string, silentOnly bool) {
	if h.hold(r, what) {
		return // an open walk touches it (ADR 0308 §2); a clearing or a silence is never held
	}
	h.mu.Lock()
	sent := map[string]string{}
	for k, v := range r.Sent {
		sent[k] = v
	}
	h.mu.Unlock()
	now := h.Now()
	for _, ch := range h.channels() {
		id, shown := sent[ch.Name()]
		if !shown {
			continue
		}
		silent := editsSilently(ch)
		if silentOnly && (!silent || id == "") {
			continue
		}
		if id != "" && ch.CanEdit() {
			err := ch.Edit(id, h.sayOn(r, what, ch.Name()))
			h.noteChannel(ch.Name(), err)
			if err == nil {
				h.record(Sent{At: now, Channel: ch.Name(), Key: r.Key, What: what, Outcome: "edited"})
				continue
			}
			h.Logf("[messenger] could not edit the message for %s on %s (%v)", r.Key, ch.Name(), err)
			if silentOnly {
				continue
			}
		}
		h.enqueue(ch.Name(), item{Key: r.Key, What: what, Severity: r.Severity, At: now})
	}
	h.flush(now, false)
}

// sayOnce holds a message of the holder's own at most once per interval, on the first channel ready.
func (h *Holder) sayOnce(key string, every time.Duration, m Message) {
	now := h.Now()
	h.mu.Lock()
	if last, said := h.saidOnce[key]; said && now.Sub(last) < every {
		h.mu.Unlock()
		return
	}
	h.saidOnce[key] = now
	h.mu.Unlock()
	for _, ch := range h.channels() {
		if ch.Ready() != nil {
			continue
		}
		msg := m
		h.enqueue(ch.Name(), item{Key: key, What: "notice", Severity: Warning, At: now, Msg: &msg})
		h.flush(now, false)
		return
	}
}

// flush sends what is due on every channel. force sends what is held whatever its window — for the
// one message of urgent conditions found open at a reading of the state.
func (h *Holder) flush(now time.Time, force bool) {
	for _, ch := range h.channels() {
		h.flushChannel(ch, now, force)
	}
}

// due says whether a channel's held items go out now.
func (h *Holder) due(ch Channel, list []item, now time.Time) bool {
	h.mu.Lock()
	grace := h.inGrace(now)
	lastTry, lastSend, lastWarn := h.lastTry[ch.Name()], h.lastSend[ch.Name()], h.lastWarn[ch.Name()]
	h.mu.Unlock()
	if len(list) == 0 || grace {
		return false
	}
	if !lastTry.IsZero() && now.Sub(lastTry) < RetryEvery {
		return false
	}
	oldest := list[0].At
	urgent := false
	for _, it := range list {
		if it.At.Before(oldest) {
			oldest = it.At
		}
		urgent = urgent || it.urgent()
	}
	if urgent {
		// At once when it is alone and the channel was quiet; otherwise with the burst it is in.
		alone := len(list) == 1 && (lastSend.IsZero() || now.Sub(lastSend) >= BurstWindow)
		return alone || now.Sub(oldest) >= BurstWindow
	}
	if now.Sub(oldest) < BurstWindow {
		return false
	}
	if ch == h.Desk && !lastWarn.IsZero() && now.Sub(lastWarn) < DesktopWarningEvery && !carriesWalk(list) {
		// A walk's message is what the rhythm would have spared the operator many of: not held to it.
		return false
	}
	return true
}

// carriesWalk says whether held items carry a walk's end or bound message.
func carriesWalk(list []item) bool {
	for _, it := range list {
		if it.What == WalkEndWhat {
			return true
		}
	}
	return false
}

// line is what an item says, resolved against the record now: nil when it is no longer to be said.
func (h *Holder) resolve(it item, now time.Time, kind string) (*Record, *Message) {
	if it.Msg != nil {
		return nil, it.Msg
	}
	h.mu.Lock()
	r := h.open[it.Key]
	h.mu.Unlock()
	if r == nil || r.silenced(now) {
		return nil, nil
	}
	if (it.What == "cleared") != !r.Cleared.IsZero() {
		return nil, nil
	}
	m := h.sayOn(r, it.What, kind)
	return r, &m
}

func (h *Holder) flushChannel(ch Channel, now time.Time, force bool) {
	name := ch.Name()
	h.mu.Lock()
	list := append([]item(nil), h.outbox[name]...)
	h.mu.Unlock()
	if len(list) == 0 || (!force && !h.due(ch, list, now)) {
		return
	}
	taken := len(list)
	// Asked before the messages are composed: composing one makes its Details ticket, and a channel that
	// stays unready would make one at every try, pushing out the tickets of messages already shown (issue 371).
	if err := ch.Ready(); err != nil {
		h.noteChannel(name, err)
		h.failed(ch, list, taken, now, err)
		return
	}
	var entries []entry
	for _, it := range list {
		r, m := h.resolve(it, now, name)
		if m == nil {
			continue
		}
		entries = append(entries, entry{it, r, *m})
	}
	if len(entries) == 0 {
		h.drop(name, taken)
		return
	}
	if !h.allow(name) {
		// The last line of defence: said once, in a message of its own, and the rest held.
		h.mu.Lock()
		for _, e := range entries {
			if e.r != nil {
				e.r.Folded = true
			}
		}
		said := h.capSaid[name]
		h.mu.Unlock()
		if said.IsZero() || now.Sub(said) >= time.Hour {
			m := Message{Title: "Mesh messages paused for this hour",
				Body: fmt.Sprintf("The mesh sent %d messages here this hour, its limit. %d more are held and come as "+
					"one message when the hour allows.", CapPerHour, len(entries)), Urgent: true}
			_, err := ch.Send(m)
			h.noteChannel(name, err)
			if err == nil {
				h.mu.Lock()
				h.capSaid[name] = now
				h.mu.Unlock()
				h.record(Sent{At: now, Channel: name, Key: "(cap)", What: "cap reached", Outcome: "said once"})
			}
		}
		return
	}
	if len(entries) > 1 {
		// A message that asks the operator something goes alone, so its answers come with it: one with several
		// answers, and one carrying an ask's answer even when only one can be given here (the lab's proof found
		// an ask whose approval was not offered folded into a digest, and its one answer lost with it).
		var alone, rest []entry
		for _, e := range entries {
			// A message said again about a record shown here goes alone too: it replaces that record's message,
			// so a reminder is never a digest line beside the message it follows (issue 373).
			// A walk's end or bound message goes alone too: its lines are what it says (ADR 0308 §4).
			if len(e.msg.Actions) > 1 || (len(e.msg.Actions) > 0 && e.r != nil && e.r.Ask != "") || h.replacing(ch, e) != "" ||
				e.it.What == WalkEndWhat {
				alone = append(alone, e)
			} else {
				rest = append(rest, e)
			}
		}
		if len(alone) > 0 {
			for _, e := range alone {
				if !h.sendEntries(ch, []entry{e}, now) {
					h.failed(ch, list, taken, now, errorf("could not send"))
					return
				}
			}
			if len(rest) > 0 && !h.sendEntries(ch, rest, now) {
				h.failed(ch, list, taken, now, errorf("could not send"))
				return
			}
			h.drop(name, taken)
			return
		}
	}
	if !h.sendEntries(ch, entries, now) {
		h.failed(ch, list, taken, now, errorf("could not send"))
		return
	}
	h.drop(name, taken)
}

// entry is one held item resolved to its record and message.
type entry struct {
	it  item
	r   *Record
	msg Message
}

// replacing is the id of the message an entry replaces on a channel: the record's own message there, when the
// entry says it again (not its clearing) and the channel can replace; "" for a new message.
func (h *Holder) replacing(ch Channel, e entry) string {
	if e.r == nil || e.it.What == "cleared" || !replaces(ch) {
		return ""
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	return e.r.Sent[ch.Name()]
}

// sendEntries sends entries as one message — itself when alone, a digest when several — and records it. A
// record said again where its message is shown replaces that message (issue 373): one message per problem.
func (h *Holder) sendEntries(ch Channel, entries []entry, now time.Time) bool {
	name := ch.Name()
	var m Message
	prior := ""
	if len(entries) == 1 {
		m = entries[0].msg
		prior = h.replacing(ch, entries[0])
	} else {
		items := make([]item, len(entries))
		msgs := make([]Message, len(entries))
		for i, e := range entries {
			items[i], msgs[i] = e.it, e.msg
		}
		m = digest(items, msgs)
	}
	var id string
	var err error
	if prior != "" {
		id, err = prior, ch.(replacer).Replace(prior, m)
	} else {
		id, err = ch.Send(m)
	}
	h.noteChannel(name, err)
	if err != nil {
		return false
	}
	h.mu.Lock()
	h.lastSend[name] = now
	h.lastTry[name] = time.Time{}
	for _, e := range entries {
		if ch == h.Desk && !e.it.urgent() {
			// The desktop's warning rhythm counts from a message that carried a warning.
			h.lastWarn[name] = now
		}
	}
	for _, e := range entries {
		r := e.r
		if r == nil {
			continue
		}
		r.Pending, r.Folded, r.History = "", false, false
		if e.it.What != "cleared" {
			r.Owed = ""
			if r.Sent == nil {
				r.Sent = map[string]string{}
			}
			if len(entries) == 1 {
				// Its own message: the one later words replace or edit — also after it was said in a digest.
				r.Sent[name] = id
			} else if _, has := r.Sent[name]; !has {
				r.Sent[name] = ""
			}
			if r.FirstSent.IsZero() {
				r.FirstSent = now
			}
			r.LastSaid = now
		}
	}
	type carried struct {
		r    *Record
		walk string
	}
	var told []carried
	for _, e := range entries {
		for _, key := range e.it.Carries {
			r := h.open[key]
			if r == nil {
				continue
			}
			if r.Sent == nil {
				r.Sent = map[string]string{}
			}
			if _, has := r.Sent[name]; !has {
				r.Sent[name] = ""
			}
			if r.FirstSent.IsZero() {
				r.FirstSent = now
			}
			r.LastSaid, r.Pending, r.History, r.Owed = now, "", false, ""
			if !r.Cleared.IsZero() && !r.ClearUnsaid {
				// It cleared while the message saying it open waited: its clearing is said too.
				r.ClearUnsaid = true
			}
			told = append(told, carried{r, strings.TrimPrefix(e.it.Key, "walk.")})
		}
	}
	h.mu.Unlock()
	for _, c := range told {
		h.record(Sent{At: now, Channel: name, Key: c.r.Key, What: "released", Outcome: "told in the message of walk " + c.walk})
		h.persist(c.r)
	}
	key, what := "(digest)", fmt.Sprintf("digest of %d", len(entries))
	if len(entries) == 1 {
		key, what = entries[0].it.Key, entries[0].it.What
		if prior != "" {
			what += ", in place of its earlier message"
		}
	}
	h.record(Sent{At: now, Channel: name, Key: key, What: what, Outcome: "sent"})
	for _, e := range entries {
		if len(entries) > 1 {
			h.record(Sent{At: now, Channel: name, Key: e.it.Key, What: e.it.What, Outcome: "in digest"})
		}
		if e.r != nil {
			h.persist(e.r)
		}
	}
	return true
}

// drop removes the first n held items of a channel: those this flush took.
func (h *Holder) drop(channel string, n int) {
	h.mu.Lock()
	defer h.mu.Unlock()
	list := h.outbox[channel]
	if n > len(list) {
		n = len(list)
	}
	h.outbox[channel] = append([]item(nil), list[n:]...)
}

// failed keeps what a channel could not send, to be tried again in a minute. A warning the desktop
// could not show — no session there — goes to Telegram instead.
func (h *Holder) failed(ch Channel, list []item, taken int, now time.Time, err error) {
	name := ch.Name()
	h.mu.Lock()
	h.lastTry[name] = now
	h.mu.Unlock()
	h.record(Sent{At: now, Channel: name, Key: "(held)", What: fmt.Sprintf("%d held", taken), Outcome: "failed: " + err.Error()})
	if ch == h.Desk && h.Away != nil && h.Away.Ready() == nil {
		var keep []item
		for _, it := range list[:taken] {
			if it.Severity == Urgent {
				keep = append(keep, it)
				continue
			}
			h.enqueue(h.Away.Name(), it)
		}
		h.mu.Lock()
		h.outbox[name] = append(keep, h.outbox[name][taken:]...)
		h.mu.Unlock()
	}
	h.mu.Lock()
	for _, it := range list[:taken] {
		if r := h.open[it.Key]; r != nil && it.Msg == nil && r.Cleared.IsZero() {
			r.Pending = it.What
		}
	}
	h.mu.Unlock()
}

// digest is several messages as one: what they are, counted, then each in a line.
func digest(items []item, msgs []Message) Message {
	var newU, newW, still, cleared, notices int
	urgent, quiet := false, true
	for _, it := range items {
		switch {
		case it.Msg != nil:
			notices++
		case it.What == "cleared":
			cleared++
		case it.What == "reminder":
			still++
		case it.Severity == Urgent:
			newU++
		default:
			newW++
		}
		urgent = urgent || it.urgent()
		quiet = quiet && it.What == "cleared"
	}
	var parts []string
	add := func(n int, one, many string) {
		switch {
		case n == 1:
			parts = append(parts, "1 "+one)
		case n > 1:
			parts = append(parts, fmt.Sprintf("%d %s", n, many))
		}
	}
	add(newU, "urgent problem", "urgent problems")
	add(newW, "new problem", "new problems")
	add(still, "still open", "still open")
	add(cleared, "resolved", "resolved")
	add(notices, "notice", "notices")
	prefix := "Mesh"
	if urgent {
		prefix = "Urgent"
	}
	// Each line is one message's title: a headline, or a resolved line — plain words, no key (ADR 0253).
	var lines []string
	for i, m := range msgs {
		if i == DigestLines {
			lines = append(lines, fmt.Sprintf("and %d more", len(msgs)-DigestLines))
			break
		}
		lines = append(lines, "- "+m.Title)
	}
	return Message{Title: prefix + ": " + strings.Join(parts, ", "), Body: strings.Join(lines, "\n"), Urgent: urgent, Quiet: quiet}
}

// sendNow sends one message on one channel at once, under the cap: for what was asked for (a test).
func (h *Holder) sendNow(ch Channel, key, what string, m Message) bool {
	if err := ch.Ready(); err != nil {
		h.noteChannel(ch.Name(), err)
		return false
	}
	_, err := ch.Send(m)
	h.noteChannel(ch.Name(), err)
	now := h.Now()
	if err != nil {
		h.record(Sent{At: now, Channel: ch.Name(), Key: key, What: what, Outcome: "failed: " + err.Error()})
		return false
	}
	h.mu.Lock()
	h.lastSend[ch.Name()] = now
	h.mu.Unlock()
	h.record(Sent{At: now, Channel: ch.Name(), Key: key, What: what, Outcome: "sent"})
	return true
}

// allow says whether a channel is under its cap: messages sent in the last hour, as recorded.
func (h *Holder) allow(channel string) bool {
	return h.sentLastHour(channel) < CapPerHour
}

func (h *Holder) sentLastHour(channel string) int {
	h.mu.Lock()
	defer h.mu.Unlock()
	since := h.Now().Add(-time.Hour)
	n := 0
	for _, s := range h.recent {
		if s.Channel == channel && s.Outcome == "sent" && s.At.After(since) {
			n++
		}
	}
	return n
}