mesh-catalog/modules/mesh-mail-notifier/cmd/mesh-mail-notifier/main.go

// mesh-mail-notifier: the default claimant of mail-notifier, built from mesh-mail's own event shape (novox/hq
// ADR 0378 §9, issue 800 part B). A Go bundle the node's runtime launches, its own account (runs-as, ADR 0376
// §2): the one process serves mail-notifier.status and the module's own health tool (mail_notifier_health,
// review round 1 of issue 800), and, beside stdio.Serve, hears mesh-mail's added events and shows them
// (exactly as modules/mesh-knowledge's serve starts go hear(k) and go background(k) before stdio.Serve) — not
// a separate `notify` process, which systemd gave an empty stdin and so exited at once, every five seconds
// (novox/hq issue 859). What this module says, it says on stderr; stdout is the MCP channel.
package main

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

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

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

// AddedPattern is the mesh-mail stream's event this module consumes (its own `consumes` entry, novox/hq ADR
// 0363, ADR 0378 §9): a new, changed or removed message's `added` only.
const AddedPattern = "mesh-mail.mail.*.message.added"

// StatusTool is the mail-notifier seat's one verb this module serves.
const StatusTool = "mail-notifier.status"

// ShownEvent is this module's own event (its `emits`), one per notice shown.
const ShownEvent = "shown"

func main() {
	n := &Notifier{
		Send: func(args map[string]any) error {
			_, err := stdio.Ask(sendKey(), args)
			return err
		},
		Accounts: readAccounts,
		Store:    busStore{},
		EmitShown: func(account, messageID string) error {
			return stdio.Emit(ShownEvent, map[string]any{"account": account, "message-id": messageID})
		},
		Logf: logf,
	}
	n.init()

	tools := []stdio.Tool{
		{Name: StatusTool,
			Description: "The accounts mesh-mail-notifier announces on this machine, and when it last showed a " +
				"notice of new mail (novox/hq ADR 0378 §7). Read-only.",
			Run: func(map[string]any) (any, error) { return n.Status() }},
		{Name: HealthTool,
			Description: "How mesh-mail-notifier is: unhealthy once a notice has failed to show for more than " +
				"ten minutes, healthy otherwise, even mid-retry (review round 1 of issue 800). Read-only.",
			Run: func(map[string]any) (any, error) {
				failures, err := busStore{}.Failures()
				if err != nil {
					return nil, err
				}
				return Judge(failures, time.Now()), nil
			}},
	}
	if err := run(n, watch, os.Stdin, os.Stdout, tools); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// sendKey is the ask key Send uses: this machine's own node-notifier, named exactly as desk-channel asks
// another machine's ("seat:node-notifier.send@"+machine) and as the SDK's own Emit reads this machine's name
// (MESH_NODE, which the runtime sets) — not the bare seat, which any machine's node-notifier may answer
// first, leaving this one's notices undelivered (novox/hq issue 865).
func sendKey() string { return "seat:node-notifier.send@" + os.Getenv("MESH_NODE") }

// run is main's startup path, factored so a test can run it: the watcher started before stdio.Serve, exactly
// as mesh-knowledge's serve starts go hear(k) and go background(k) before stdio.Serve — so the watcher keeps
// running whatever stdio.Serve does with the stream it is given, including returning at once when stdin is
// already at EOF, as systemd gave the old separate `notify` process (novox/hq issue 859). The node's runtime
// launches this one process and keeps it alive.
func run(n *Notifier, watcher func(*Notifier), in io.Reader, out io.Writer, tools []stdio.Tool) error {
	go watcher(n)
	return stdio.ServeOn("", tools, in, out)
}

// watch hears mesh-mail's added events and, once a second, flushes what Notifier.Flush finds ready: an
// account's first mails of its window at once, and its folded overflow once the window is over (ADR 0378 §9,
// the lead's reading, review round 1 of issue 800).
func watch(n *Notifier) { watchWith(n, stdio.Subscribe, time.Sleep) }

// watchWith is watch with its subscribe and sleep factored out, so a test can make a subscribe fail without
// the real bus and without the real wait. A subscribe the runtime refuses is retried, 2 seconds doubling to
// at most a minute, logging each refusal — exactly as modules/mesh-knowledge's hear does — never fatal: a
// refusal here must never take mail-notifier.status and mail_notifier_health down with it (novox/hq issue
// 859, review round 1; the old os.Exit(1) did exactly that).
func watchWith(n *Notifier, subscribe func(string, func(stdio.Envelope) error) error, sleep func(time.Duration)) {
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		sleep(wait)
		err := subscribe(AddedPattern, func(e stdio.Envelope) error {
			var body EventBody
			if err := json.Unmarshal(e.Body, &body); err != nil {
				logf("[mesh-mail-notifier] an added event could not be read; taken and dropped: %v", err)
				return nil
			}
			return n.Add(body)
		})
		if err == nil {
			break
		}
		logf("[mesh-mail-notifier] does not hear %s yet (%v); asking again in %s", AddedPattern, err, wait)
	}
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	for range ticker.C {
		if err := n.Flush(); err != nil {
			logf("[mesh-mail-notifier] a buffered notice could not be shown: %v", err)
		}
	}
}

// readAccounts asks mail-store.accounts for the accounts this module is granted, by name to own address.
func readAccounts() (map[string]string, error) {
	raw, err := stdio.Ask("seat:mail-store.accounts", map[string]any{})
	if err != nil {
		return nil, err
	}
	var answer struct {
		Accounts []struct {
			Account string `json:"account"`
			Address string `json:"address"`
		} `json:"accounts"`
	}
	if err := json.Unmarshal(raw, &answer); err != nil {
		return nil, err
	}
	out := map[string]string{}
	for _, a := range answer.Accounts {
		out[a.Account] = a.Address
	}
	return out, nil
}

// busStore keeps the notifier's own state (the `shown` bucket the manifest declares): which messages were
// shown already, each account's last notice, and each account's notice still failing to send.
type busStore struct{}

func stateKey(parts ...string) string {
	words := make([]string, len(parts))
	for i, p := range parts {
		var b strings.Builder
		for _, r := range p {
			switch {
			case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-':
				b.WriteRune(r)
			default:
				b.WriteRune('_')
			}
		}
		words[i] = b.String()
	}
	return strings.Join(words, "_")
}

func (busStore) HandleSeen(handle string) (bool, error) {
	e, err := stdio.State("shown").Get("handle_" + stateKey(strings.Split(handle, "\x1f")...))
	return e != nil, err
}

func (busStore) MarkHandle(handle string) error {
	_, err := stdio.State("shown").Put("handle_"+stateKey(strings.Split(handle, "\x1f")...), true)
	return err
}

func (busStore) PutStatus(account string, s AccountStatus) error {
	_, err := stdio.State("shown").Put("status_"+stateKey(account), s)
	return err
}

func (busStore) Statuses() (map[string]AccountStatus, error) {
	keys, err := stdio.State("shown").Keys()
	if err != nil {
		return nil, err
	}
	out := map[string]AccountStatus{}
	for _, k := range keys {
		account, ok := strings.CutPrefix(k, "status_")
		if !ok {
			continue
		}
		e, err := stdio.State("shown").Get(k)
		if err != nil || e == nil {
			continue
		}
		var s AccountStatus
		if json.Unmarshal(e.Value, &s) == nil {
			out[account] = s
		}
	}
	return out, nil
}

func (busStore) PutFailure(account string, f FailureStatus) error {
	_, err := stdio.State("shown").Put("failure_"+stateKey(account), f)
	return err
}

func (busStore) ClearFailure(account string) error {
	return stdio.State("shown").Delete("failure_" + stateKey(account))
}

func (busStore) Failures() (map[string]FailureStatus, error) {
	keys, err := stdio.State("shown").Keys()
	if err != nil {
		return nil, err
	}
	out := map[string]FailureStatus{}
	for _, k := range keys {
		account, ok := strings.CutPrefix(k, "failure_")
		if !ok {
			continue
		}
		e, err := stdio.State("shown").Get(k)
		if err != nil || e == nil {
			continue
		}
		var f FailureStatus
		if json.Unmarshal(e.Value, &f) == nil {
			out[account] = f
		}
	}
	return out, nil
}