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

// mesh-watcher: the watcher's watcher (novox/hq to-be 45 §5, S10, ADR 0227 rule 6). A Go bundle the
// node's runtime launches on one machine that is not the control node. It hears the self-check's
// heartbeat and makes a round trip through the bus every minute; when either goes silent
// past its bound, it tells the operator on Telegram directly over HTTPS — the one sender that does not
// pass through the control node — and tells them again when it returns. stdout is the MCP channel;
// what this module says, it says on stderr.
package main

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

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

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

func readChatID(path string) (string, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return "", err
	}
	var s map[string]any
	if err := json.Unmarshal(raw, &s); err != nil {
		return "", fmt.Errorf("the settings file is not JSON: %v", err)
	}
	switch v := s["telegram-chat-id"].(type) {
	case string:
		return strings.TrimSpace(v), nil
	case float64:
		return fmt.Sprintf("%.0f", v), nil
	}
	return "", nil
}

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() {
	settings := os.Getenv("MESH_WATCHER_SETTINGS")
	var said sync.Mutex
	lastSaid := ""
	tg := NewTelegram(TelegramConfig{
		TokenFile: os.Getenv("MESH_WATCHER_TELEGRAM_TOKEN_FILE"),
		ChatID: func() string {
			id, err := readChatID(settings)
			said.Lock()
			defer said.Unlock()
			if err != nil && err.Error() != lastSaid {
				logf("[mesh-watcher] settings cannot be read: %v", err)
			}
			lastSaid = ""
			if err != nil {
				lastSaid = err.Error()
			}
			return id
		},
	})
	node := os.Getenv("MESH_NODE")
	w := NewWatcher(tg, time.Now, logf, node)
	l := &listening{now: "not yet: starting"}
	go run(w, l, node)
	if err := stdio.Serve("", tools(w, l)); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// run keeps time, makes the round trips, reads where the controller is, and listens for heartbeats.
// The clock runs whatever the bus does: it is what notices the bus is gone.
func run(w *Watcher, l *listening, node string) {
	if err := w.Telegram.Ready(); err != nil {
		logf("[mesh-watcher] BLIND until given: %v", err)
	}
	go func() {
		for range time.Tick(time.Minute) {
			go func() { w.BusAnswered(roundTrip(node)) }()
			w.Tick()
		}
	}()
	go func() {
		time.Sleep(2 * time.Second)
		for {
			if nodes, err := controlNodes(); err == nil {
				w.Placed(nodes)
			} else {
				logf("[mesh-watcher] cannot read where the bus runs (%v); asking again in ten minutes", err)
			}
			time.Sleep(10 * time.Minute)
		}
	}()
	handle := func(e stdio.Envelope) error {
		if !isHeartbeat(e.Key) {
			return nil
		}
		b, err := DecodeBeat(e.Body)
		if err != nil {
			w.BadHeartbeat(err)
			return nil
		}
		w.Heartbeat(b)
		return nil
	}
	<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: a heartbeat sent meanwhile is heard
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		err := stdio.Subscribe(health.SelfCheckModule+".*", handle)
		if err == nil {
			l.set("listening")
			logf("[mesh-watcher] listening for the self-check's heartbeat; watching from %s", node)
			return
		}
		l.set("not yet: " + err.Error())
		logf("[mesh-watcher] not hearing heartbeats yet (%v); asking again in %s", err, wait)
		time.Sleep(wait)
	}
}

// roundTrip asks this module's own ping on this machine, through the bus server: an answer is the bus
// carrying a request and its reply.
func roundTrip(node string) error {
	key := "mesh-watcher.watcher_ping"
	if node != "" {
		key += "@" + node
	}
	done := make(chan error, 1)
	go func() {
		_, err := stdio.Ask(key, map[string]any{})
		done <- err
	}()
	select {
	case err := <-done:
		return err
	case <-time.After(30 * time.Second):
		return fmt.Errorf("no answer in 30 s")
	}
}

// controlNodes reads which machines hold the bus, from the placement seat's state of who holds which seat (novox/hq
// ADR 0571 clause 4). The controller is not placed (ADR 0590), so the bus's machines are the control node.
func controlNodes() ([]string, error) {
	st := stdio.State(delivery.SeatHoldersRead)
	names, err := st.Keys()
	if err != nil {
		return nil, err
	}
	keys := map[string][]byte{}
	for _, k := range names {
		e, err := st.Get(k)
		if err != nil {
			return nil, err
		}
		if e != nil {
			keys[k] = e.Value
		}
	}
	return holderNodes(keys, "mesh-broker")
}

// holderNodes is the machines holding any of the seats, in the order the seats are named, each once.
func holderNodes(keys map[string][]byte, seats ...string) ([]string, error) {
	var out []string
	seen := map[string]bool{}
	for _, seat := range seats {
		holders, err := delivery.HoldersOf(keys, seat, "")
		if err != nil {
			return nil, err
		}
		for _, h := range holders {
			if !seen[h.Node] {
				seen[h.Node] = true
				out = append(out, h.Node)
			}
		}
	}
	if len(out) == 0 {
		return nil, fmt.Errorf("the placement seat's state names no holder of %s", strings.Join(seats, " or "))
	}
	return out, nil
}

func tools(w *Watcher, l *listening) []stdio.Tool {
	return []stdio.Tool{
		{Name: "watcher_status",
			Description: "The watcher's own health first — whether it can tell the operator anything (the Telegram token and " +
				"chat id), whether it runs on the machine it watches, whether heartbeats reach it — then each watched " +
				"signal (the self-check's heartbeat, a round trip through the bus): last heard, how long ago, " +
				"its bound, whether it is said silent, and any message not sent yet.",
			Run: func(map[string]any) (any, error) { return w.Status(l.get()), nil }},
		{Name: "watcher_last_heard",
			Description: "When each watched signal was last heard, and what the watcher sent to Telegram lately, newest first.",
			Run:         func(map[string]any) (any, error) { return w.LastHeard(), nil }},
		{Name: "watcher_test",
			Description: "Send a test message to Telegram now, directly: proves the watcher can reach the operator when the " +
				"mesh cannot. Answers sent, or why not.",
			Run: func(map[string]any) (any, error) {
				err := w.send("test", Message{Title: "TEST: mesh-watcher on " + w.Node + " reaches you directly",
					Body: "nothing is wrong; this was asked for"})
				if err != nil {
					return map[string]string{"telegram": "not sent: " + err.Error()}, nil
				}
				return map[string]string{"telegram": "sent"}, nil
			}},
		{Name: "watcher_ping",
			Description: "Answers at once: the round trip the watcher makes through the bus every minute to know the bus carries a request and its reply.",
			Run: func(map[string]any) (any, error) {
				return map[string]string{"pong": time.Now().UTC().Format(time.RFC3339)}, nil
			}},
	}
}