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

// mesh-mail-reader: the default claimant of mail-reading (novox/hq ADR 0459 §1), built from mesh-mail's own
// source, like mesh-mail-web and mesh-mail-notifier. One instance, held once for the mesh, on the machine
// mesh-mail runs on. It hears mesh-mail's added events under its own runs-as account (ADR 0459 §3, issue
// 1518), judges a batch of them at a time (1427, 1520), and serves the six mail-reading verbs (entries,
// groups, correct, lessons, undo, status) and its own health tool, mail_reader_health (1519). It reads mail;
// it moves, flags, deletes and sends nothing (novox/hq ADR 0459 §11).
//
//	mesh-mail-reader            the bundle the node's runtime launches: the hearing loop (hear.go), the
//	                            batch loop (batch.go) and the six mail-reading verbs plus mail_reader_health,
//	                            served over stdio (tools.go)
//	mesh-mail-reader prepare    the preparation step: the index's schema (store.go, pgstore.go) brought to
//	                            this version's (issue 1517)
//
// stdout is the bundle's MCP channel; what this module says, it says on stderr.
package main

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

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

// Module is this module's name.
const Module = "mesh-mail-reader"

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

func envFromOS() Env {
	return Env{Callers: os.Getenv("MESH_MAIL_READER_CALLERS")}
}

// cachedStore opens the index once, lazily, and hands every later caller that SAME store (review round 1
// finding 2, issue 1517/1518): the Bundle's own tool calls (one per mesh/ask, tools.go), hearLoop's handler
// (one per event heard) and batchLoop. Before this, each of those called storeFor itself, which opened a
// fresh *pgxpool.Pool (PgStoreFromEnv) every time and never closed it — one connection pool leaked per tool
// call and per heard event, for the life of the process. Mirrors mesh-mail's own main.go Index (serve):
// a failed open is never retried here on its own; it is retried the next time any caller calls get, exactly
// as hearLoop's own backoff or a tool call's own caller already does.
type cachedStore struct {
	open func(ctx context.Context) (Store, error)
	mu   sync.Mutex
	s    Store
}

func newCachedStore(open func(ctx context.Context) (Store, error)) *cachedStore {
	return &cachedStore{open: open}
}

// get is this cachedStore's own storeFor: the open store once it has one, or a fresh attempt to open one.
func (c *cachedStore) get(ctx context.Context) (Store, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.s != nil {
		return c.s, nil
	}
	s, err := c.open(ctx)
	if err != nil {
		return nil, err
	}
	c.s = s
	return c.s, nil
}

func main() {
	if len(os.Args) > 1 && os.Args[1] == "prepare" {
		os.Exit(prepare())
	}
	store := newCachedStore(func(ctx context.Context) (Store, error) { return PgStoreFromEnv(ctx) })
	b := &Bundle{Env: envFromOS(), Store: store.get}
	go hearLoop(store.get, stdio.Subscribe, time.Sleep)
	go batchLoop(store.get)
	if err := stdio.ServeOn("", b.Tools(), os.Stdin, os.Stdout); err != nil {
		logf("[mesh-mail-reader] %v", err)
		os.Exit(1)
	}
}

// ---- batches (1427, 1520) --------------------------------------------------------------------------------------

// bundleAsk is the Ask (batch.go) over the runtime's own stdio.Ask.
func bundleAsk(key string, body map[string]any) (json.RawMessage, error) { return stdio.Ask(key, body) }

// batchLoop runs the reader's batches (RunBatches, batch.go) against the module's own store, judging on this
// machine (MESH_NODE, the lead's decision at this milestone's prepare step). Started before stdio.ServeOn,
// exactly as hearLoop is: a batch failure must never take the six mail-reading verbs or mail_reader_health
// down with it (RunBatches logs and keeps going; see its own doc comment).
func batchLoop(storeFor func(context.Context) (Store, error)) {
	store, err := storeFor(context.Background())
	if err != nil {
		logf("[mesh-mail-reader] batches: the index cannot be reached: %v", err)
		return
	}
	node := os.Getenv("MESH_NODE")
	readText := NewReadText(bundleAsk)
	judge := NewJudge(bundleAsk, node, time.Now)
	check := NewMessengerCheck(bundleAsk)
	tell := NewTell(bundleAsk)
	RunBatches(context.Background(), store, readText, judge, check, tell, time.Now, time.Sleep, logf, BatchGap)
}

// ---- hearing --------------------------------------------------------------------------------------------------

// hearFirstRetryWait is the hearing loop's first retry wait (issue 1518, fixed per the lead's try feedback
// on bb594652): 250ms, doubling to at most a minute, so a bundle hears within about a second of stdio.ServeOn
// starting to serve rather than mesh-mail-notifier's own 2s start — a window during which a plain NATS
// subject (no durable consumer) loses an event published before the subscribe lands; the mesh's own
// controller-made durable consumer keeps it regardless, so this only shortens an unnecessary wait, it does
// not paper over a production loss.
const hearFirstRetryWait = 250 * time.Millisecond

// hearLoop subscribes to mesh-mail's added events and hears each one (hear.go), mirroring mesh-mail-notifier's
// watchWith (issue 1518): a subscribe the runtime refuses is retried, starting at hearFirstRetryWait and
// doubling to at most a minute, logging each refusal, never fatal — a refusal here must never take the six
// mail-reading verbs or mail_reader_health down with it. Started before stdio.ServeOn, so it keeps running
// whatever stdio.ServeOn does with the stream it is given.
func hearLoop(storeFor func(context.Context) (Store, error), subscribe func(string, func(stdio.Envelope) error) error, sleep func(time.Duration)) {
	for wait := hearFirstRetryWait; ; wait = minDuration(wait*2, time.Minute) {
		err := subscribe(AddedPattern, func(e stdio.Envelope) error {
			ctx := context.Background()
			store, err := storeFor(ctx)
			if err != nil {
				return err
			}
			return Hear(ctx, store, e.Body, time.Now)
		})
		if err == nil {
			return
		}
		logf("[mesh-mail-reader] does not hear %s yet (%v); asking again in %s", AddedPattern, err, wait)
		sleep(wait)
	}
}

func minDuration(a, b time.Duration) time.Duration {
	if a < b {
		return a
	}
	return b
}

// ---- prepare ------------------------------------------------------------------------------------------------------

// prepare is the preparation step (1517/AC2): the schema brought to what this version needs, run once
// before the bundle itself starts (the manifest's run-once process `prepare`).
func prepare() int {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	s, err := PgStoreFromEnv(ctx)
	if err != nil {
		logf("[mesh-mail-reader] preparing the index: %v", err)
		return 1
	}
	defer s.Close()
	said, err := s.Prepare(ctx)
	for _, line := range said {
		logf("[mesh-mail-reader] %s", line)
	}
	if err != nil {
		logf("[mesh-mail-reader] preparing the index: %v", err)
		return 1
	}
	return 0
}