128 lines
5.0 KiB
Go
128 lines
5.0 KiB
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"
|
|
"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")}
|
|
}
|
|
|
|
func main() {
|
|
if len(os.Args) > 1 && os.Args[1] == "prepare" {
|
|
os.Exit(prepare())
|
|
}
|
|
storeFor := func(ctx context.Context) (Store, error) { return PgStoreFromEnv(ctx) }
|
|
b := &Bundle{Env: envFromOS(), Store: storeFor}
|
|
go hearLoop(storeFor, stdio.Subscribe, time.Sleep)
|
|
go batchLoop(storeFor)
|
|
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)
|
|
}
|
|
|
|
// ---- hearing --------------------------------------------------------------------------------------------------
|
|
|
|
// hearLoop subscribes to mesh-mail's added events and hears each one (hear.go), exactly as
|
|
// mesh-mail-notifier's watchWith does (issue 1518): a subscribe the runtime refuses is retried, 2 seconds
|
|
// 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 := 2 * time.Second; ; 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
|
|
}
|