// 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 controller's // self-check 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" ) 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 controller 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 } time.Sleep(500 * time.Millisecond) for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.Subscribe(ControllerSeat+".*", 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 controller and the bus. func controlNodes() ([]string, error) { raw, err := stdio.Ask("seat:mesh-controller.seats", map[string]any{}) if err != nil { return nil, err } return holdersOf(raw, "mesh-controller", "mesh-broker") } // holdersOf reads the seats verb's answer, bare or inside a tool reply, for the named seats' machines. func holdersOf(raw json.RawMessage, seats ...string) ([]string, error) { var answer struct { Seats []struct { Seat string `json:"seat"` Holders []struct { Node string `json:"node"` } `json:"holders"` } `json:"seats"` Output string `json:"output"` Content []struct { Text string `json:"text"` } `json:"content"` } if err := json.Unmarshal(raw, &answer); err != nil { var s string if json.Unmarshal(raw, &s) == nil { return holdersOf(json.RawMessage(s), seats...) } return nil, fmt.Errorf("the seats answer is not JSON") } if len(answer.Seats) == 0 { switch { case answer.Output != "": return holdersOf(json.RawMessage(answer.Output), seats...) case len(answer.Content) > 0: return holdersOf(json.RawMessage(answer.Content[0].Text), seats...) } return nil, fmt.Errorf("the seats answer lists no seat") } var out []string seen := map[string]bool{} for _, s := range answer.Seats { for _, want := range seats { if s.Seat != want { continue } for _, h := range s.Holders { if !seen[h.Node] { seen[h.Node] = true out = append(out, h.Node) } } } } 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 controller's self-check 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 }}, } }