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

// messenger: the holder of the operator-channel seat (novox/hq to-be 45 §5, ADR 0227, research 028).
// A Go bundle the node's runtime launches. It consumes the controller's condition events and decides
// what is said to the operator, on Telegram and on the desktop notifier of the machine the operator
// is at. It keeps its open messages in its own state, so a restart forgets nothing. stdout is the MCP
// channel; what this module says, it says on stderr.
package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync"
	"time"

	_ "time/tzdata" // the operator's zone is known whatever this machine has installed

	stdio "git.novox.be/novox/mesh-sdk/go"
)

func errorf(format string, a ...any) error { return fmt.Errorf(format, a...) }

func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }

// Settings are the operator's values for this module, merged by the mesh into one JSON file.
type Settings struct {
	TelegramChatID  string   `json:"telegram-chat-id"`
	DesktopMachines []string `json:"desktop-machines"`
	// TimeZone is the operator's, an IANA name such as Europe/Brussels: every time a message says is said
	// in it. Empty is this machine's own zone.
	TimeZone string `json:"time-zone"`
}

// zoneOf is the location a time-zone setting names, or this machine's own when it names none or one
// that does not exist (said, never guessed silently).
func zoneOf(name string) (*time.Location, error) {
	if name == "" {
		return time.Local, nil
	}
	loc, err := time.LoadLocation(name)
	if err != nil {
		return time.Local, fmt.Errorf("the time-zone %q is not one this machine knows; times are said in its own zone", name)
	}
	return loc, nil
}

func readSettings(path string) (Settings, error) {
	var s Settings
	if path == "" {
		return s, errors.New("started without a settings file")
	}
	raw, err := os.ReadFile(path)
	if err != nil {
		return s, err
	}
	// The chat id may be given as a number.
	var loose map[string]any
	if err := json.Unmarshal(raw, &loose); err != nil {
		return s, fmt.Errorf("the settings file is not JSON: %v", err)
	}
	switch v := loose["telegram-chat-id"].(type) {
	case string:
		s.TelegramChatID = strings.TrimSpace(v)
	case float64:
		s.TelegramChatID = fmt.Sprintf("%.0f", v)
	}
	if tz, ok := loose["time-zone"].(string); ok {
		s.TimeZone = strings.TrimSpace(tz)
	}
	if list, ok := loose["desktop-machines"].([]any); ok {
		for _, m := range list {
			if name, ok := m.(string); ok && strings.TrimSpace(name) != "" {
				s.DesktopMachines = append(s.DesktopMachines, strings.TrimSpace(name))
			}
		}
	}
	return s, nil
}

// stateStore keeps the open messages in the module's declared state `open`, and the recent sends
// under one key of `sent` (ADR 0201).
type stateStore struct{}

// kvKey is a condition key as a bucket key: only the characters a key may carry.
func kvKey(key string) string {
	var b strings.Builder
	for _, r := range key {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_', r == '.', r == '=':
			b.WriteRune(r)
		default:
			b.WriteRune('_')
		}
	}
	return strings.Trim(b.String(), ".")
}

func (stateStore) Put(r Record) error {
	_, err := stdio.State("open").Put(kvKey(r.Key), r)
	return err
}
func (stateStore) Delete(key string) error { return stdio.State("open").Delete(kvKey(key)) }
func (stateStore) All() ([]Record, error) {
	keys, err := stdio.State("open").Keys()
	if err != nil {
		return nil, err
	}
	var out []Record
	for _, k := range keys {
		e, err := stdio.State("open").Get(k)
		if err != nil {
			return nil, err
		}
		if e == nil {
			continue
		}
		var r Record
		if err := json.Unmarshal(e.Value, &r); err != nil {
			logf("[messenger] the open message kept as %s cannot be read (%v); left as it is", k, err)
			continue
		}
		out = append(out, r)
	}
	return out, nil
}
func (stateStore) PutRecent(s []Sent) error {
	_, err := stdio.State("sent").Put("recent", s)
	return err
}
func (stateStore) Recent() ([]Sent, error) {
	e, err := stdio.State("sent").Get("recent")
	if err != nil || e == nil {
		return nil, err
	}
	var s []Sent
	return s, json.Unmarshal(e.Value, &s)
}

// listening is whether condition events reach this holder, in words.
type listening struct {
	mu  sync.Mutex
	now string
}

func (l *listening) set(s string) { l.mu.Lock(); l.now = s; l.mu.Unlock() }
func (l *listening) get() string  { l.mu.Lock(); defer l.mu.Unlock(); return l.now }

func main() {
	settingsFile := os.Getenv("MESH_MESSENGER_SETTINGS")
	var said sync.Mutex
	lastSaid, zoneSaid := "", ""
	settings := func() Settings {
		s, err := readSettings(settingsFile)
		said.Lock()
		defer said.Unlock()
		if err != nil && err.Error() != lastSaid {
			logf("[messenger] settings cannot be read: %v", err)
		}
		lastSaid = ""
		if err != nil {
			lastSaid = err.Error()
		}
		return s
	}
	h := &Holder{
		Telegram: NewTelegram(TelegramConfig{
			TokenFile: os.Getenv("MESH_MESSENGER_TELEGRAM_TOKEN_FILE"),
			ChatID:    func() string { return settings().TelegramChatID },
		}),
		Desktop: &Desktop{
			Machines: func() []string { return settings().DesktopMachines },
			Ask:      stdio.Ask,
		},
		Store: stateStore{},
		Ask:   stdio.Ask,
		Logf:  logf,
		Zone: func() *time.Location {
			loc, err := zoneOf(settings().TimeZone)
			said.Lock()
			defer said.Unlock()
			if err != nil && err.Error() != zoneSaid {
				logf("[messenger] %v", err)
			}
			zoneSaid = ""
			if err != nil {
				zoneSaid = err.Error()
			}
			return loc
		},
		Emit: func(event string, body any) error { return stdio.Emit(event, body) },
	}
	h.init()
	l := &listening{now: "not yet: starting"}
	go run(h, l)
	if err := stdio.Serve("", tools(h, l)); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// run reads back the state, listens for condition events, and keeps time — each retried, and each
// failure said, never given up on quietly.
func run(h *Holder, l *listening) {
	time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		err := h.Load()
		if err == nil {
			break
		}
		logf("[messenger] cannot read back its open messages yet (%v); asking again in %s", err, wait)
		time.Sleep(wait)
	}
	for _, ch := range h.channels() {
		if err := ch.Ready(); err != nil {
			logf("[messenger] %s cannot send: %v", ch.Name(), err)
		}
	}
	// The state first: what is open now is read from the controller before any event is taken, so
	// what the stream replays is history (novox/hq issue 271). Not reachable yet, it is asked for
	// again in the loop below, and old events meanwhile are kept as state, never said.
	h.Start()
	syncFromController(h)
	go func() {
		tick := time.NewTicker(5 * time.Second)
		defer tick.Stop()
		lastTick, lastAsk := time.Now(), time.Now()
		for now := range tick.C {
			h.Flush()
			if now.Sub(lastTick) >= time.Minute {
				lastTick = now
				h.Tick()
			}
			if h.WantsSync() && now.Sub(lastAsk) >= time.Minute {
				lastAsk = now
				syncFromController(h)
			}
		}
	}()
	handle := func(e stdio.Envelope) error {
		event, ok := eventOf(e.Key)
		if !ok {
			return nil
		}
		c, err := DecodeCondition(event, e.Body)
		if err != nil {
			h.Unreadable(e.Key, err)
			return nil
		}
		h.Condition(event, c)
		return nil
	}
	go listenForAnswers(h)
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		err := stdio.Subscribe(ControllerSeat+".*", handle)
		if err == nil {
			l.set("listening")
			logf("[messenger] listening for the controller's condition events")
			return
		}
		l.set("not yet: " + err.Error())
		logf("[messenger] not hearing condition events yet (%v); asking again in %s", err, wait)
		time.Sleep(wait)
	}
}

// AnswerEvent is the desk's notifier saying which answer was chosen, and StartedEvent that it started (so
// the answers it waited for are lost): the dunst holder of node-notifier emits them as its module's events
// until seats publish events of their own (to-be 46 phase 3).
const (
	AnswerEvent  = "dunst.action-chosen"
	StartedEvent = "dunst.started"
)

// listenForAnswers takes the answers the operator chooses on the desk, retried until it can.
func listenForAnswers(h *Holder) {
	handle := func(e stdio.Envelope) error {
		if e.Key == StartedEvent {
			var body struct {
				Machine string `json:"machine"`
			}
			_ = json.Unmarshal(e.Body, &body)
			if body.Machine == "" {
				body.Machine = e.Node
			}
			h.Reoffer(body.Machine)
			return nil
		}
		if e.Key != AnswerEvent {
			return nil
		}
		var body struct {
			Token   string `json:"token"`
			Machine string `json:"machine"`
		}
		if err := json.Unmarshal(e.Body, &body); err != nil || body.Token == "" {
			logf("[messenger] an answer from the desk could not be read: %s", firstLineOf(string(e.Body)))
			return nil
		}
		machine := body.Machine
		if machine == "" {
			machine = e.Node
		}
		// The machine is the event's own node, said by the runtime: the body's word is only a hint.
		if e.Node != "" {
			machine = e.Node
		}
		_, _ = h.Chosen("desktop", machine, body.Token, e.Headers["x-event-id"])
		return nil
	}
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		if err := stdio.Subscribe("dunst.*", handle); err == nil {
			logf("[messenger] listening for the operator's answers on the desk")
			return
		} else {
			logf("[messenger] not hearing the desk's answers yet (%v); asking again in %s", err, wait)
		}
		time.Sleep(wait)
	}
}

// syncFromController reads the controller's open conditions and gives them to the holder.
func syncFromController(h *Holder) {
	asked := h.Now()
	type answer struct {
		raw json.RawMessage
		err error
	}
	done := make(chan answer, 1)
	go func() {
		raw, err := stdio.Ask("seat:"+ControllerSeat+".conditions", map[string]any{})
		done <- answer{raw, err}
	}()
	var a answer
	select {
	case a = <-done:
	case <-time.After(30 * time.Second):
		a.err = errors.New("no answer in 30 s")
	}
	if a.err != nil {
		h.SyncFailed(a.err)
		return
	}
	open, err := DecodeOpen(a.raw)
	if err != nil {
		h.SyncFailed(err)
		return
	}
	h.Sync(open, asked)
}

func str(description string) map[string]any {
	return map[string]any{"type": "string", "description": description}
}

func strArg(a map[string]any, k string) string { s, _ := a[k].(string); return strings.TrimSpace(s) }

func limitArg(a map[string]any, def int) int {
	if v, ok := a["limit"].(float64); ok && v >= 1 {
		return min(int(v), KeptSends)
	}
	return def
}

func tools(h *Holder, l *listening) []stdio.Tool {
	return []stdio.Tool{
		{Name: "operator-channel.open",
			Description: "What is open now: every message the operator was sent about something still wrong, urgent first, " +
				"oldest first — its key, severity, summary, since when, the channels it went to, whether it is silenced, " +
				"reminded, held (for its burst window, the desktop's rhythm or the cap), refused, not sent yet, or " +
				"never said (known from the state or an old event: history, not news).",
			Run: func(map[string]any) (any, error) { return h.Open(), nil }},
		{Name: "operator-channel.history",
			Description: "What was said to the operator lately, newest first — each send, digest, edit and failure with its " +
				"channel, key and outcome — and every message refused for carrying an address, a path or a secret.",
			Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many of each (default 50)"}},
			Run:   func(a map[string]any) (any, error) { return h.History(limitArg(a, 50)), nil }},
		{Name: "operator-channel.notify",
			Description: "Tell the operator something, as a module that uses the seat: a key (the same key is the same " +
				"message), urgent or warning, one line in the mesh's words, and what it is about; and best, in plain " +
				"words, a headline, an explanation and the line said when it is over. clear says it is over. " +
				"Roles and words only: an address, a path or a secret is refused. Deduplicated, capped and routed " +
				"like the controller's conditions.",
			Input: map[string]any{
				"key":      str("what makes it the same message the next time, e.g. backup.ace.failed"),
				"severity": map[string]any{"type": "string", "enum": []string{Urgent, Warning}},
				"summary":  str("one line in the mesh's words"),
				"subject":  str("what it is about: a machine's role, a module, a plan"),
				"headline": str("a few plain words the operator reads as the title, e.g. backup failed on ace: no ids, keys or commands"),
				"explanation": str("one or two plain sentences: what happened, what it means for the operator, and whether " +
					"they need to act"),
				"resolved": str("the one short line said when it is over, e.g. backups run again on ace"),
				"clear":    map[string]any{"type": "boolean", "description": "it is over: the message is edited to say so"},
			},
			Run: func(a map[string]any) (any, error) {
				clear, _ := a["clear"].(bool)
				return h.Notify(strArg(a, "key"), strArg(a, "severity"), strArg(a, "summary"), strArg(a, "subject"), clear,
					strArg(a, "headline"), strArg(a, "explanation"), strArg(a, "resolved"))
			}},
		{Name: "messenger_status",
			Description: "Whether the operator can be reached, and why not: each channel — can it send, what it lacks " +
				"(the Telegram bot token and chat id, the desktop machines), its last error, how many it sent in the " +
				"last hour against the cap, how many it holds — the open and silenced count, messages not sent yet, " +
				"refusals, unreadable events, whether condition events arrive, and the rules it applies. check asks " +
				"Telegram who the bot is, sending nothing.",
			Input: map[string]any{"check": map[string]any{"type": "boolean", "description": "also ask Telegram whether the token works"}},
			Run: func(a map[string]any) (any, error) {
				st := h.Status(l.get())
				if check, _ := a["check"].(bool); check {
					if t, ok := h.Telegram.(*Telegram); ok {
						who, err := t.Who()
						if err != nil {
							return map[string]any{"status": st, "telegram_check": "failed: " + err.Error()}, nil
						}
						return map[string]any{"status": st, "telegram_check": "the token works: the bot is @" + who}, nil
					}
				}
				return st, nil
			}},
		{Name: "messenger_recent",
			Description: "The recent sends — each message, digest, edit and failure on each channel, newest first — and the refusals.",
			Input:       map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many (default 20)"}},
			Run:         func(a map[string]any) (any, error) { return h.History(limitArg(a, 20)), nil }},
		{Name: "messenger_test",
			Description: "Send a test message now, to telegram, desktop or both (default both), through the content rule " +
				"and the cap: proves a channel reaches the operator. Answers per channel: sent, or why not.",
			Input: map[string]any{
				"channel": map[string]any{"type": "string", "enum": []string{"telegram", "desktop", "both"}},
				"text":    str("the words (default: a line saying it is a test)"),
			},
			Run: func(a map[string]any) (any, error) { return h.Test(strArg(a, "channel"), strArg(a, "text")), nil }},
		{Name: "messenger_check",
			Description: "Would these words be allowed to leave the mesh? Runs the content rule — no address, path or " +
				"secret — and answers allowed, or the class of what it carried. Sends nothing.",
			Input: map[string]any{"text": str("the words to judge")},
			Run: func(a map[string]any) (any, error) {
				if r, ok := Check(strArg(a, "text")); !ok {
					return map[string]any{"allowed": false, "class": r.Class, "carried": r.What}, nil
				}
				return map[string]any{"allowed": true}, nil
			}},
	}
}