diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/copies_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/copies_test.go new file mode 100644 index 00000000..996b10f4 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/copies_test.go @@ -0,0 +1,33 @@ +package main + +// mesh-mail-reader carries a byte-identical copy of mesh-mail's event shape (novox/hq ADR 0459 §3, issue 1518), +// the way mesh-mail-notifier carries it too (ADR 0378 §9, copies_test.go) and modules/networkmanager's +// uplink-tools carries ifwatchd's own file (ADR 0241): each builds alone, so each holds its own copy, and this +// test holds them to one text. + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +// source is where mesh-mail's own copy of the file lives. +var source = filepath.Join("..", "..", "..", "mesh-mail", "cmd", "mesh-mail") + +// 1518/AC1. +func TestCarriesTheSameMailEventAsMeshMail(t *testing.T) { + for _, f := range []string{"mailevent.go"} { + mine, err := os.ReadFile(f) + if err != nil { + t.Fatal(err) + } + theirs, err := os.ReadFile(filepath.Join(source, f)) + if err != nil { + t.Fatalf("mesh-mail's copy of %s could not be read: %v", f, err) + } + if !bytes.Equal(mine, theirs) { + t.Errorf("1518/AC1: this copy of %s differs from mesh-mail's: change both together", f) + } + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear.go new file mode 100644 index 00000000..082e0c91 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear.go @@ -0,0 +1,64 @@ +package main + +// Hearing mesh-mail's added events (novox/hq ADR 0459 §3, §12, issue 1518): mesh-mail-reader consumes +// mesh-mail.mail.*.message.added under its own runs-as account, exactly as mesh-mail-notifier does (main.go +// AddedPattern, watchWith), but records each one as a waiting mail (store.go, pgstore.go) BEFORE the handler +// returns success — so a mail heard is never lost, even across a restart: the controller composes a durable +// consumer (MaxDeliver 5, AckWait 30s) that redelivers an event five times before it is dead-lettered, and a +// store failure here must make that redelivery happen, never swallow the event. + +import ( + "context" + "encoding/json" + "time" +) + +// EventBody is what mesh-mail's added event says of a message (metadata only; mailevent.go names where it is +// said). Unknown fields are ignored: the reader reads only what it needs, never a body. +type EventBody struct { + Account string `json:"account"` + Folder string `json:"folder"` + UID uint32 `json:"uid"` + UIDValidity uint32 `json:"uidvalidity"` + MessageID string `json:"message-id"` + Date string `json:"date"` + From []Person `json:"from"` + Subject string `json:"subject"` + Size int64 `json:"size"` + Flags []string `json:"flags"` +} + +// Hear records one added event as a waiting mail (1518/AC2): the store is written to before this returns +// nil, so the caller (the bus subscription's handler) only then acknowledges the event. A store failure is +// returned unchanged, so the event is redelivered rather than lost (1518/AC2). The same event heard twice +// (1518/AC3) is recorded once: AddWaiting itself is idempotent on the mail's own identity. +func Hear(ctx context.Context, store Store, raw json.RawMessage, now func() time.Time) error { + var body EventBody + if err := json.Unmarshal(raw, &body); err != nil { + // An event this reader cannot even read is taken and dropped, exactly as mesh-mail-notifier does + // (main.go watchWith): a malformed event would otherwise be redelivered forever, never succeeding. + return nil + } + date, err := time.Parse(time.RFC3339, body.Date) + if err != nil { + date = now() + } + w := WaitingMail{ + Account: body.Account, + Folder: body.Folder, + MessageID: body.MessageID, + UIDValidity: body.UIDValidity, + UID: body.UID, + Date: date, + From: body.From, + Subject: body.Subject, + Flags: body.Flags, + Size: body.Size, + Heard: now(), + } + return store.AddWaiting(ctx, w) +} + +// AddedPattern is the mesh-mail stream's event this module consumes (its own `consumes` entry, novox/hq ADR +// 0459 §3): a new message's `added` only. +const AddedPattern = "mesh-mail.mail.*.message.added" diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear_test.go new file mode 100644 index 00000000..806b8e72 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear_test.go @@ -0,0 +1,113 @@ +package main + +import ( + "context" + "encoding/json" + "testing" + "time" +) + +func addedEventBody(t *testing.T, account, folder, messageID string, uid uint32) json.RawMessage { + t.Helper() + raw, err := json.Marshal(map[string]any{ + "account": account, "folder": folder, "uid": uid, "uidvalidity": uint32(1), + "message-id": messageID, "date": "2026-10-20T10:00:00Z", + "from": []Person{{Name: "Alice", Address: "alice@example.org"}}, + "subject": "hello", "size": int64(1234), "flags": []string{}, + }) + if err != nil { + t.Fatal(err) + } + return raw +} + +// 1518/AC2: an added event is recorded as a waiting mail before the handler returns success. +func TestHearRecordsTheEventAsAWaitingMailBeforeSucceeding(t *testing.T) { + store := newMemStore() + now := func() time.Time { return time.Date(2026, 10, 20, 10, 0, 1, 0, time.UTC) } + raw := addedEventBody(t, "bob", "INBOX", "", 7) + if err := Hear(context.Background(), store, raw, now); err != nil { + t.Fatal(err) + } + waiting, err := store.Waiting(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(waiting) != 1 { + t.Fatalf("waiting = %v, want one waiting mail", waiting) + } + w := waiting[0] + if w.Account != "bob" || w.Folder != "INBOX" || w.MessageID != "" || w.UID != 7 || + w.Subject != "hello" || w.Size != 1234 || len(w.From) != 1 || w.From[0].Address != "alice@example.org" { + t.Fatalf("waiting mail: %+v", w) + } +} + +// 1518/AC2: a store failure is returned unchanged, so the event is redelivered rather than lost. +func TestHearReturnsAStoreFailureSoTheEventIsRedelivered(t *testing.T) { + store := newMemStore() + store.failNext = errStoreFake + raw := addedEventBody(t, "bob", "INBOX", "", 8) + err := Hear(context.Background(), store, raw, time.Now) + if err != errStoreFake { + t.Fatalf("Hear with a failing store: %v, want %v", err, errStoreFake) + } + waiting, _ := store.Waiting(context.Background()) + if len(waiting) != 0 { + t.Fatalf("waiting = %v, want none (the failed add kept nothing)", waiting) + } +} + +// 1518/AC3: the same event heard twice gives one waiting mail. +func TestHearTwiceGivesOneWaitingMail(t *testing.T) { + store := newMemStore() + raw := addedEventBody(t, "bob", "INBOX", "", 9) + if err := Hear(context.Background(), store, raw, time.Now); err != nil { + t.Fatal(err) + } + if err := Hear(context.Background(), store, raw, time.Now); err != nil { + t.Fatal(err) + } + waiting, _ := store.Waiting(context.Background()) + if len(waiting) != 1 { + t.Fatalf("waiting = %v, want exactly one", waiting) + } +} + +// 1518/AC4: waiting mails are handed out in arrival order (the order they were heard), oldest first — not by +// their own date, and regardless of whether reading was healthy when they arrived (Hear never consults +// health; a mail heard while unhealthy is recorded exactly the same way). +func TestHearKeepsArrivalOrder(t *testing.T) { + store := newMemStore() + // Heard out of date order: the third mail (by date) arrives first. + first := addedEventBody(t, "bob", "INBOX", "", 1) + second := addedEventBody(t, "bob", "INBOX", "", 2) + third := addedEventBody(t, "bob", "INBOX", "", 3) + for _, raw := range []json.RawMessage{first, second, third} { + if err := Hear(context.Background(), store, raw, time.Now); err != nil { + t.Fatal(err) + } + } + waiting, _ := store.Waiting(context.Background()) + if len(waiting) != 3 { + t.Fatalf("waiting = %v, want three", waiting) + } + wantOrder := []string{"", "", ""} + for i, w := range waiting { + if w.MessageID != wantOrder[i] { + t.Fatalf("waiting[%d].MessageID = %q, want %q (arrival order)", i, w.MessageID, wantOrder[i]) + } + } +} + +// A malformed event is taken and dropped, never redelivered forever (mirrors mesh-mail-notifier's watchWith). +func TestHearDropsAMalformedEvent(t *testing.T) { + store := newMemStore() + if err := Hear(context.Background(), store, json.RawMessage(`not json`), time.Now); err != nil { + t.Fatalf("Hear on a malformed event: %v, want it taken and dropped", err) + } + waiting, _ := store.Waiting(context.Background()) + if len(waiting) != 0 { + t.Fatalf("waiting = %v, want none", waiting) + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go new file mode 100644 index 00000000..abb021ee --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go @@ -0,0 +1,78 @@ +package main + +// What the tests stand on: an in-memory double of Store (store.go), for the hearing (1518) and health (1519) +// tests alike. The postgres-backed PgStore (pgstore.go) is proven only by pgstore_test.go's unit tests and by +// pgstore_integration_test.go, which this context never runs. + +import ( + "context" + "errors" + "sync" +) + +type memStore struct { + mu sync.Mutex + waiting []WaitingMail + keys map[string]bool + nextSeq int64 + failNext error // AddWaiting fails once with this, then clears, so a test can prove a failure is returned + health ReaderHealth +} + +func newMemStore() *memStore { + return &memStore{keys: map[string]bool{}, health: ReaderHealth{Healthy: true}} +} + +func (s *memStore) AddWaiting(_ context.Context, w WaitingMail) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.failNext != nil { + err := s.failNext + s.failNext = nil + return err + } + key := w.Key() + if s.keys[key] { + return nil // 1518/AC3: the same mail heard twice is recorded once. + } + s.keys[key] = true + s.nextSeq++ + w.Seq = s.nextSeq + s.waiting = append(s.waiting, w) + return nil +} + +func (s *memStore) Waiting(context.Context) ([]WaitingMail, error) { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]WaitingMail, len(s.waiting)) + copy(out, s.waiting) // already in arrival order: appended in the order AddWaiting succeeded (1518/AC4). + return out, nil +} + +func (s *memStore) TakeWaiting(_ context.Context, seq int64) error { + s.mu.Lock() + defer s.mu.Unlock() + for i, w := range s.waiting { + if w.Seq == seq { + s.waiting = append(s.waiting[:i], s.waiting[i+1:]...) + return nil + } + } + return nil +} + +func (s *memStore) ReaderHealth(context.Context) (ReaderHealth, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.health, nil +} + +func (s *memStore) SetReaderHealth(_ context.Context, h ReaderHealth) error { + s.mu.Lock() + defer s.mu.Unlock() + s.health = h + return nil +} + +var errStoreFake = errors.New("the store refused (fake, for a test)") diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/mailevent.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/mailevent.go new file mode 100644 index 00000000..fb9596fb --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/mailevent.go @@ -0,0 +1,13 @@ +package main + +// The shape of a mail-store message event (novox/hq ADR 0378 §2, issue 800 part A), moved here on its own so +// mesh-mail-notifier (issue 800 part B) carries a byte-identical copy, the way modules/networkmanager's +// uplink-tools carries ifwatchd's own file (novox/hq ADR 0241, copies_test.go). Nothing here reaches a mailbox or +// a server; it only says where an account's change is told on the bus and what the three changes are called. + +// EventKey is where an account's change is said: mail..message. (research 049 +// document 02 §5), the account by its name in the setting, never its address. +func EventKey(account, change string) string { return "mail." + account + ".message." + change } + +// Changes are the three things said of a message. +var Changes = []string{"added", "changed", "removed"} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go index ace6f32a..883645ad 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go @@ -1,13 +1,12 @@ // 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 — not yet built), judges a batch of them at a time (1427, 1520 — not yet built), and serves the six -// mail-reading verbs (entries, groups, correct, lessons, undo, status) and its own health tool, -// mail_reader_health (1519 — not yet built). It reads mail; it moves, flags, deletes and sends nothing -// (novox/hq ADR 0459 §11). +// 1518), judges a batch of them at a time (1427, 1520 — not yet built), and serves the six mail-reading +// verbs (entries, groups, correct, lessons, undo, status) and its own health tool, mail_reader_health +// (1519 — not yet built). 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 six mail-reading verbs plus -// mail_reader_health, served over stdio (tools.go) +// mesh-mail-reader the bundle the node's runtime launches: the hearing loop (hear.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) // @@ -36,13 +35,47 @@ 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()} + go hearLoop(storeFor, stdio.Subscribe, time.Sleep) if err := stdio.ServeOn("", b.Tools(), os.Stdin, os.Stdout); err != nil { logf("[mesh-mail-reader] %v", err) os.Exit(1) } } +// ---- 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 diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go index b94e9d9c..5d7ed598 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go @@ -23,6 +23,7 @@ type manifest struct { RunsAs string `json:"runs-as"` Claims []seat `json:"claims"` Invokes []string `json:"invokes"` + Consumes []string `json:"consumes"` Tools []string `json:"tools"` Requires []string `json:"requires"` Contributes map[string]map[string]any `json:"contributes"` @@ -115,3 +116,14 @@ func TestTheManifestRequiresItsOwnDatabase(t *testing.T) { t.Fatalf("prepare's DATABASE_URL_FILE is %v, database-url's own path is %v", env["DATABASE_URL_FILE"], databaseURL["path"]) } } + +// 1518/AC5: the manifest consumes mesh-mail.mail.*.message.added and runs-as mesh-mail-reader. +func TestTheManifestConsumesMeshMailsAddedEvent(t *testing.T) { + m, _ := readManifest(t) + if !reflect.DeepEqual(m.Consumes, []string{AddedPattern}) { + t.Fatalf("consumes %v, want [%s]", m.Consumes, AddedPattern) + } + if m.RunsAs != Module { + t.Fatalf("runs-as %q, want %q", m.RunsAs, Module) + } +} diff --git a/modules/mesh-mail-reader/module.json b/modules/mesh-mail-reader/module.json index d00618ad..ff16ca7f 100644 --- a/modules/mesh-mail-reader/module.json +++ b/modules/mesh-mail-reader/module.json @@ -9,6 +9,9 @@ "serves": ["entries", "groups", "correct", "lessons", "undo", "status"] } ], + "consumes": [ + "mesh-mail.mail.*.message.added" + ], "invokes": [ "seat:mail-store.read", "seat:mail-store.accounts"