From 95704c741089e7f613cc3ba22efc31d0af1d9a99 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 00:53:37 +0200 Subject: [PATCH] Start mesh-mail-reader whole against the real bus and store: entries, no body kept, the tell, the cap (issue 1431) --- .../acceptance_integration_test.go | 270 +++++++++ .../runtime_player_integration_test.go | 537 ++++++++++++++++++ modules/mesh-mail-reader/go.mod | 16 +- modules/mesh-mail-reader/go.sum | 33 +- 4 files changed, 850 insertions(+), 6 deletions(-) create mode 100644 modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go create mode 100644 modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go new file mode 100644 index 00000000..0773ccba --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go @@ -0,0 +1,270 @@ +//go:build integration + +package main + +// mesh-mail-reader started whole against the real bus and store (novox/hq ADR 0459, 0461, 0472; issue +// 1431, design 59 tests 10, 11, 13 — this issue's own acceptance criteria note replaces its stale report, +// which named the grant split ADR 0472 made won't-fix). Compiled by `go vet -tags integration ./...`; never +// run on this machine — the lead's try runs it for real. + +import ( + "context" + "encoding/json" + "fmt" + "regexp" + "strconv" + "strings" + "testing" + "time" +) + +// mailIDPattern finds a mail's own id in a batch's question (question.go's mailHeaderBlock: "mail %d: +// account ..."), so a stand-in judge can answer about exactly the mails it was actually asked, without +// knowing their ids ahead of time (postgres assigns them). +var mailIDPattern = regexp.MustCompile(`mail (\d+):`) + +func extractMailIDs(question string) []int64 { + matches := mailIDPattern.FindAllStringSubmatch(question, -1) + out := make([]int64, 0, len(matches)) + for _, m := range matches { + id, _ := strconv.ParseInt(m[1], 10, 64) + out = append(out, id) + } + return out +} + +// judgeEveryMailAs is a stand-in Judge that marks every mail of the batch's own question the same way — +// enough for a test that cares about one outcome, not a per-mail distinction. +func judgeEveryMailAs(mark, group, why, dueDate string) func(args map[string]any) (any, error) { + return func(args map[string]any) (any, error) { + question, _ := args["question"].(string) + ids := extractMailIDs(question) + mails := make([]map[string]any, len(ids)) + for i, id := range ids { + m := map[string]any{"id": id, "mark": mark, "summary": "s"} + if group != "" { + m["group"] = group + } + if why != "" { + m["why"] = why + } + if dueDate != "" { + m["due-date"] = dueDate + } + mails[i] = m + } + raw, _ := json.Marshal(map[string]any{"mails": mails}) + return map[string]any{"answer": json.RawMessage(raw)}, nil + } +} + +func oneAccountAnswer(account string) func(args map[string]any) (any, error) { + return func(map[string]any) (any, error) { + return map[string]any{"accounts": []map[string]any{{"account": account, "kind": "imap", "address": account + "@example.org"}}}, nil + } +} + +// assertNeverSendsMovesFlagsOrDeletes is 1431/AC2: across the run, the reader never asks a verb that +// changes the mailbox. Every ask this harness's stand-ins do not know already fails the test loudly +// (readerPlayer.handleAsk); this is the belt the AC's own words ask for, on top of that suspenders. +func assertNeverSendsMovesFlagsOrDeletes(t *testing.T, r *startedReader) { + t.Helper() + for _, k := range r.stand.Keys() { + lower := strings.ToLower(k) + for _, forbidden := range []string{"send", "move", "flag", "delete"} { + if strings.Contains(lower, forbidden) { + t.Fatalf("asked %q, which names %q — the reader must never change a mailbox", k, forbidden) + } + } + } +} + +// 1431/AC1 (design 59 test 10): a published added event gives a mail entry in a mail group, and a search +// of every text column of the reader's own database for a known line of that mail's body finds nothing. +func TestAcceptancePublishedMailBecomesAnEntryNeverKeepingItsBody(t *testing.T) { + r := startReader(t) + r.stand.MailAccounts = oneAccountAnswer("bob") + knownLine := "the-known-body-line-xyz123-never-kept" + r.stand.MailRead = func(map[string]any) (any, error) { + return map[string]any{"text": "hello, " + knownLine + ", goodbye"}, nil + } + r.stand.Judge = judgeEveryMailAs("neither", "newsletters", "", "") + r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil } + r.stand.Tell = func(map[string]any) (any, error) { return map[string]any{"sent_on": ""}, nil } + + publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now()) + + conn := pgConnect(t, r.dbURL) + ctx := context.Background() + waitFor(t, 30*time.Second, "the mail entry to appear", func() bool { + var n int + if err := conn.QueryRow(ctx, `select count(*) from mail_entry`).Scan(&n); err != nil { + return false + } + return n == 1 + }) + + var groupID *int64 + if err := conn.QueryRow(ctx, `select mail_group from mail_entry limit 1`).Scan(&groupID); err != nil { + t.Fatal(err) + } + if groupID == nil { + t.Fatal("the entry carries no mail group") + } + + // A search of every text-ish column of the whole database for the known body line finds nothing. + rows, err := conn.Query(ctx, `select table_name, column_name from information_schema.columns + where table_schema = 'public' and data_type in ('text', 'character varying')`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + checked := 0 + for rows.Next() { + var table, column string + if err := rows.Scan(&table, &column); err != nil { + t.Fatal(err) + } + var found int + q := fmt.Sprintf(`select count(*) from %q where %q like $1`, table, column) + if err := conn.QueryRow(ctx, q, "%"+knownLine+"%").Scan(&found); err != nil { + t.Fatal(err) + } + checked++ + if found > 0 { + t.Fatalf("the known body line was found in %s.%s", table, column) + } + } + if checked == 0 { + t.Fatal("no text column was checked at all") + } + assertNeverSendsMovesFlagsOrDeletes(t, r) +} + +// 1431/AC3 (design 59 test 13, first half): a batch with one mail the stand-in judge marks as mattering +// gives exactly one tell naming it, its words passing the content rule (messenger_check). +func TestAcceptanceOneMatteringMailSendsExactlyOneTellNamingIt(t *testing.T) { + r := startReader(t) + r.stand.MailAccounts = oneAccountAnswer("bob") + r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "pay by friday"}, nil } + r.stand.Judge = judgeEveryMailAs("matters", "", "due friday", "2026-11-01") + + var checked []string + r.stand.MessengerCheck = func(args map[string]any) (any, error) { + text, _ := args["text"].(string) + checked = append(checked, text) + // The content rule's own shape (ADR 0461 §4): refuse words carrying a URL, an address or a host name. + allowed := !strings.Contains(text, "://") && !strings.Contains(text, "example.com") + return map[string]any{"allowed": allowed}, nil + } + var tells []map[string]any + r.stand.Tell = func(args map[string]any) (any, error) { + tells = append(tells, args) + return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil + } + + publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "invoice", time.Now()) + + waitFor(t, 30*time.Second, "exactly one tell", func() bool { return len(tells) >= 1 }) + time.Sleep(2 * time.Second) // give a stray second tell a chance to prove this wrong, if it would happen + if len(tells) != 1 { + t.Fatalf("tells = %d, want exactly 1", len(tells)) + } + if tells[0]["headline"] != "1 mail matters" { + t.Fatalf("headline = %v, want '1 mail matters'", tells[0]["headline"]) + } + if tells[0]["subject"] != "mail" { + t.Fatalf("subject = %v, want mail", tells[0]["subject"]) + } + lines, _ := tells[0]["lines"].([]any) + named := false + for _, l := range lines { + if s, ok := l.(string); ok && strings.Contains(s, "Alice") { + named = true + } + } + if !named { + t.Fatalf("lines %v do not name Alice", lines) + } + for _, text := range checked { + if strings.Contains(text, "://") || strings.Contains(text, "example.com") { + t.Fatalf("messenger_check was asked a word carrying a URL or host name: %q", text) + } + } + assertNeverSendsMovesFlagsOrDeletes(t, r) +} + +// 1431/AC3 (design 59 test 13, second half): a sender whose name carries a host name gives the safe form. +func TestAcceptanceSenderNameWithAHostNameGivesTheSafeForm(t *testing.T) { + r := startReader(t) + r.stand.MailAccounts = oneAccountAnswer("bob") + r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "hi"}, nil } + r.stand.Judge = judgeEveryMailAs("matters", "", "", "") + r.stand.MessengerCheck = func(args map[string]any) (any, error) { + text, _ := args["text"].(string) + return map[string]any{"allowed": !strings.Contains(text, "evil.example.com")}, nil + } + var tells []map[string]any + r.stand.Tell = func(args map[string]any) (any, error) { + tells = append(tells, args) + return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil + } + + publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "mail.evil.example.com", "address": "x@example.org"}}, "hi", time.Now()) + + waitFor(t, 30*time.Second, "one tell", func() bool { return len(tells) >= 1 }) + lines, _ := tells[0]["lines"].([]any) + if len(lines) != 1 || lines[0] != "open the mail page" { + t.Fatalf("lines = %v, want the safe form (just 'open the mail page')", lines) + } + assertNeverSendsMovesFlagsOrDeletes(t, r) +} + +// 1431/AC4: with the stand-in tell answering dropped (sent_on empty, the cap of 10 an hour), the mail stays +// untold, and the next mail signal — after a second batch — counts it. +func TestAcceptanceDroppedTellLeavesMailUntoldAndCountedNextTime(t *testing.T) { + r := startReader(t) + r.stand.MailAccounts = oneAccountAnswer("bob") + r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "hi"}, nil } + r.stand.Judge = judgeEveryMailAs("matters", "", "", "") + r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil } + var tellCalls int + r.stand.Tell = func(map[string]any) (any, error) { + tellCalls++ + if tellCalls == 1 { + return map[string]any{"sent_on": ""}, nil // dropped: the operator channel's own cap + } + return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil + } + + publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now()) + waitFor(t, 30*time.Second, "the first (dropped) tell attempt", func() bool { return tellCalls >= 1 }) + + conn := pgConnect(t, r.dbURL) + ctx := context.Background() + waitFor(t, 10*time.Second, "the entry to be written", func() bool { + var n int + _ = conn.QueryRow(ctx, `select count(*) from mail_entry`).Scan(&n) + return n == 1 + }) + var told bool + if err := conn.QueryRow(ctx, `select told from mail_entry limit 1`).Scan(&told); err != nil { + t.Fatal(err) + } + if told { + t.Fatal("the mail was marked told even though the tell was dropped") + } + + // A second mail, and its own batch (at least BatchGap after the first started), triggers the next mail + // signal — which counts the still-untold first mail too (1429/AC4, 1522/AC2). + publishAdded(t, r.nc, "bob", "INBOX", 2, []map[string]any{{"name": "Bob", "address": "bob2@example.org"}}, "second", time.Now()) + waitFor(t, BatchGap+60*time.Second, "a second tell attempt", func() bool { return tellCalls >= 2 }) + waitFor(t, 10*time.Second, "the first entry to become told", func() bool { + var firstTold bool + if err := conn.QueryRow(ctx, `select told from mail_entry order by id limit 1`).Scan(&firstTold); err != nil { + return false + } + return firstTold + }) + assertNeverSendsMovesFlagsOrDeletes(t, r) +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go new file mode 100644 index 00000000..4c58abd8 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go @@ -0,0 +1,537 @@ +//go:build integration + +package main + +// The runtime player for mesh-mail-reader (novox/hq ADR 0192, ADR 0198; issue 1431), mirroring +// mesh-mail-web's own harness (modules/mesh-mail-web/cmd/mesh-mail-web/runtime_integration_test.go, issue +// 1285) closely: a built binary, played over its real stdin/stdout (initialize, tools/list, tools/call; +// mesh/ask answered by stand-ins for mail-store, claude-code's judge and judged, messenger and +// operator-channel). mesh-mail-reader also HEARS mesh-mail's added events (stdio.Subscribe), which +// mesh-mail-web never does: this player also plays the runtime's other half of that — mesh/subscribe +// (acked at once; this harness needs no pattern filtering of its own, since mesh-mail-reader subscribes to +// exactly one pattern and the bundle itself filters by it) and mesh/event, bridged from a REAL nats-server +// (newTestBus, copied from mesh-mail's own sender_integration_test.go) on the exact wire subject the +// controller composes for mesh-mail's own stream ("mesh.stream."+module+".event."+event, +// modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go's own comment) — so a test's own publish is +// genuinely carried over a real bus, never handed to the bundle directly. Against the real store +// MESH_TEST_POSTGRES names, never a database started by hand (this context's hard rule); it fails, never +// skips, without it (pgstore_integration_test.go's own testDatabaseURL). Compiled by +// `go vet -tags integration ./...`; never run on this machine — the lead's try runs it for real. + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" +) + +// ---- the wire, copied from the SDK's own stdio.go (its unexported `message`) -------------------------------- + +type wireMessage struct { + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *wireError `json:"error,omitempty"` +} + +type wireError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// ---- the stand-ins (issue 1431's own AC1-AC4) ---------------------------------------------------------------- + +// askRecord is one ask the bundle made, as the harness recorded it: its key and its body, whole. +type askRecord struct { + Key string + Body json.RawMessage +} + +// testNode is MESH_NODE for these tests: the judge's own key is claude-code.claude_code_judge@ +// (the lead's decision at this milestone's prepare step), and the judged key's own node must answer the same +// for a pending answer to be collected from the right place. +const testNode = "test-node" + +// standIns is every bus call mesh-mail-reader makes, faked (issue 1431's AC2: a verb this struct was given +// nothing for is unknown to the player, which fails the test naming the ask — nothing here moves, flags, +// deletes, sends or asks send-status, because nothing here answers those keys at all). +type standIns struct { + mu sync.Mutex + + MailAccounts func(args map[string]any) (any, error) + MailRead func(args map[string]any) (any, error) + Judge func(args map[string]any) (any, error) + Judged func(args map[string]any) (any, error) + MessengerCheck func(args map[string]any) (any, error) + Tell func(args map[string]any) (any, error) + + calls []askRecord +} + +func (s *standIns) answer(key string, args map[string]any) (any, error, bool) { + raw, _ := json.Marshal(args) + s.mu.Lock() + s.calls = append(s.calls, askRecord{Key: key, Body: raw}) + s.mu.Unlock() + var f func(args map[string]any) (any, error) + switch { + case key == "seat:mail-store.accounts": + f = s.MailAccounts + case key == "seat:mail-store.read": + f = s.MailRead + case key == "claude-code.claude_code_judge@"+testNode: + f = s.Judge + case strings.HasPrefix(key, "claude-code.claude_code_judged@"): + f = s.Judged + case key == "messenger.messenger_check": + f = s.MessengerCheck + case key == "seat:operator-channel.tell": + f = s.Tell + default: + return nil, nil, false + } + if f == nil { + return nil, nil, false + } + out, err := f(args) + return out, err, true +} + +// Calls is every ask this stand-in has answered so far, a copy. +func (s *standIns) Calls() []askRecord { + s.mu.Lock() + defer s.mu.Unlock() + return append([]askRecord(nil), s.calls...) +} + +// Keys is every distinct key asked so far, in the order first seen (1431/AC2: "across the run the reader +// asks no send, send-status, move, flag or delete"). +func (s *standIns) Keys() []string { + s.mu.Lock() + defer s.mu.Unlock() + seen := map[string]bool{} + var out []string + for _, c := range s.calls { + if !seen[c.Key] { + seen[c.Key] = true + out = append(out, c.Key) + } + } + return out +} + +// ---- the real bus (issue 1431: "against the real bus and store") -------------------------------------------- + +// newTestBus starts a JetStream-enabled nats-server for the test alone — copied from mesh-mail's own +// cmd/mesh-mail/sender_integration_test.go (itself from modules/mesh-delivery/cmd/mesh-delivery/bus_test.go). +func newTestBus(t *testing.T) string { + t.Helper() + opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), + NoLog: true, NoSigs: true} + s, err := server.NewServer(opts) + if err != nil { + t.Fatalf("a bus for this test could not be made: %v", err) + } + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + s.Shutdown() + t.Fatal("a bus for this test did not come up within 30s") + } + t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() }) + return s.ClientURL() +} + +// addedWireSubject is the real wire subject the controller composes for mesh-mail's own stream mail's added +// event, for one account ("mesh.stream."+module+".event."+event, mesh-mail's own sender_integration_test.go +// comment; mesh-mail's own EventKey, mailevent.go): "mesh.stream.mesh-mail.event.mail..message.added". +func addedWireSubject(account string) string { + return "mesh.stream.mesh-mail.event.mail." + account + ".message.added" +} + +// ---- building the bundle's own binary (issue 1431, mirrors issue 1285's buildMailWebBinary) ------------------- + +func buildReaderBinary(t *testing.T) string { + t.Helper() + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "mw-reader-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + bin := filepath.Join(dir, "mesh-mail-reader") + cmd := exec.Command("go", "build", "-o", bin, ".") + cmd.Dir, err = os.Getwd() + if err != nil { + t.Fatal(err) + } + cmd.Env = os.Environ() + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("building mesh-mail-reader: %v\n%s", err, out) + } + return bin +} + +// runReaderPrepare runs the bundle's own `prepare` subcommand against databaseURL, exactly as the manifest's +// run-once prepare process would (1517/AC2). +func runReaderPrepare(t *testing.T, binary, databaseURL string) { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "database.url") + if err := os.WriteFile(path, []byte(databaseURL), 0o600); err != nil { + t.Fatal(err) + } + cmd := exec.Command(binary, "prepare") + cmd.Env = append(os.Environ(), "DATABASE_URL_FILE="+path) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("mesh-mail-reader prepare: %v\n%s", err, out) + } +} + +// ---- the runtime player itself (issue 1431, mirrors issue 1285's runtimePlayer) -------------------------------- + +type askFailer interface { + Helper() + Errorf(format string, args ...any) +} + +type readerPlayer struct { + fail askFailer + cmd *exec.Cmd + in io.WriteCloser + + mu sync.Mutex + nextID int + waiting map[string]chan wireMessage + + stand *standIns + nc *nats.Conn +} + +// startReaderPlayer builds and starts mesh-mail-reader's own binary with env, plays the handshake, and begins +// bridging the real bus (nc) to the bundle's own hearing (mesh/subscribe, mesh/event), exactly as the +// node-engine would. t is used for setup and cleanup always; failer is where an unknown ask is reported — t +// itself when nil. +func startReaderPlayer(t *testing.T, binary string, env []string, stand *standIns, nc *nats.Conn, failer askFailer) *readerPlayer { + t.Helper() + if failer == nil { + failer = t + } + cmd := exec.Command(binary) + cmd.Env = env + stderr := &prefixedLog{t: t, prefix: "mesh-mail-reader stderr: "} + cmd.Stderr = stderr + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + p := &readerPlayer{fail: failer, cmd: cmd, in: stdin, waiting: map[string]chan wireMessage{}, stand: stand, nc: nc} + go p.readLoop(stdout) + t.Cleanup(func() { + _ = cmd.Process.Kill() + _ = cmd.Wait() + }) + + if _, err := p.ask("initialize", map[string]any{ + "protocolVersion": "2025-03-26", "capabilities": map[string]any{}, "clientInfo": map[string]any{"name": "test", "version": "1"}, + }); err != nil { + t.Fatalf("initialize: %v", err) + } + if err := p.send(map[string]any{"jsonrpc": "2.0", "method": "notifications/initialized", "params": map[string]any{}}); err != nil { + t.Fatalf("notifications/initialized: %v", err) + } + + // Bridge the real bus to the bundle's own hearing (issue 1518's AddedPattern), exactly once: every + // message on the wildcard subject becomes one mesh/event, regardless of which pattern the bundle itself + // asked mesh/subscribe for — the bundle filters by its own pattern internally (stdio.go's own + // TopicMatches), so an unmatched forward here is simply never dispatched there. + if nc != nil { + _, err := nc.Subscribe("mesh.stream.mesh-mail.event.>", func(msg *nats.Msg) { + rest := strings.TrimPrefix(msg.Subject, "mesh.stream.mesh-mail.event.") + key := "mesh-mail." + rest + if _, err := p.ask("mesh/event", map[string]any{ + "envelope": map[string]any{"key": key, "node": testNode, "body": json.RawMessage(msg.Data)}, + }); err != nil { + p.fail.Errorf("delivering %s to mesh-mail-reader: %v", key, err) + } + }) + if err != nil { + t.Fatal(err) + } + } + return p +} + +type prefixedLog struct { + t *testing.T + prefix string + buf []byte + mu sync.Mutex +} + +func (p *prefixedLog) Write(b []byte) (int, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.buf = append(p.buf, b...) + for { + i := strings.IndexByte(string(p.buf), '\n') + if i < 0 { + break + } + p.t.Log(p.prefix + string(p.buf[:i])) + p.buf = p.buf[i+1:] + } + return len(b), nil +} + +func (p *readerPlayer) readLoop(stdout io.Reader) { + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 1<<20), 1<<24) + for scanner.Scan() { + var m wireMessage + if json.Unmarshal(scanner.Bytes(), &m) != nil { + continue + } + if m.Method == "" && len(m.ID) > 0 { + var key string + _ = json.Unmarshal(m.ID, &key) + p.mu.Lock() + ch := p.waiting[key] + delete(p.waiting, key) + p.mu.Unlock() + if ch != nil { + ch <- m + } + continue + } + if m.Method != "" { + go p.handleBundleRequest(m) + } + } +} + +func (p *readerPlayer) send(m map[string]any) error { + raw, err := json.Marshal(m) + if err != nil { + return err + } + p.mu.Lock() + defer p.mu.Unlock() + _, err = p.in.Write(append(raw, '\n')) + return err +} + +func (p *readerPlayer) ask(method string, params any) (json.RawMessage, error) { + p.mu.Lock() + p.nextID++ + id := "player-" + strconv.Itoa(p.nextID) + ch := make(chan wireMessage, 1) + p.waiting[id] = ch + p.mu.Unlock() + + if err := p.send(map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params}); err != nil { + return nil, err + } + select { + case m := <-ch: + if m.Error != nil { + return nil, fmt.Errorf("%s: %s", method, m.Error.Message) + } + return m.Result, nil + case <-time.After(60 * time.Second): + p.mu.Lock() + delete(p.waiting, id) + p.mu.Unlock() + return nil, fmt.Errorf("%s: no answer within 60s", method) + } +} + +// ToolsCall is tools/call, answering the tool's own text content and whether the runtime said isError. +func (p *readerPlayer) ToolsCall(name string, args map[string]any) (text string, isError bool, err error) { + raw, askErr := p.ask("tools/call", map[string]any{"name": name, "arguments": args}) + if askErr != nil { + return "", false, askErr + } + var out struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + IsError bool `json:"isError"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return "", false, err + } + if len(out.Content) == 0 { + return "", out.IsError, nil + } + return out.Content[0].Text, out.IsError, nil +} + +func (p *readerPlayer) handleBundleRequest(m wireMessage) { + p.fail.Helper() + reply := func(result any, errMsg string) { + out := map[string]any{"jsonrpc": "2.0", "id": json.RawMessage(m.ID)} + if errMsg != "" { + out["error"] = map[string]any{"code": -32000, "message": errMsg} + } else { + raw, _ := json.Marshal(result) + out["result"] = json.RawMessage(raw) + } + _ = p.send(out) + } + switch m.Method { + case "mesh/ask": + p.handleAsk(m, reply) + case "mesh/subscribe": + // Acked at once: this harness needs no durable consumer semantics of its own (1518's own catch-up is + // the reader's own store, proven in kit_test.go's unit tests, not this bus bridge). + reply(map[string]any{}, "") + default: + p.fail.Errorf("the bundle asked %s, which the helper does not know (issue 1431)", m.Method) + reply(nil, "the helper does not know "+m.Method) + } +} + +func (p *readerPlayer) handleAsk(m wireMessage, reply func(result any, errMsg string)) { + p.fail.Helper() + var params struct { + Key string `json:"key"` + Body map[string]any `json:"body"` + } + _ = json.Unmarshal(m.Params, ¶ms) + if p.stand == nil { + p.fail.Errorf("the bundle asked %s, which the helper does not know (issue 1431)", params.Key) + reply(nil, "the helper does not know "+params.Key) + return + } + result, err, known := p.stand.answer(params.Key, params.Body) + if !known { + p.fail.Errorf("the bundle asked %s, which the helper does not know (issue 1431)", params.Key) + reply(nil, "the helper does not know "+params.Key) + return + } + if err != nil { + reply(nil, err.Error()) + return + } + reply(result, "") +} + +// ---- a started mesh-mail-reader, ready for a test -------------------------------------------------------------- + +type startedReader struct { + player *readerPlayer + stand *standIns + nc *nats.Conn + dbURL string +} + +// startReader builds mesh-mail-reader's own binary, prepares a fresh database (MESH_TEST_POSTGRES; fails, +// never skips, without it), starts a real bus, and plays the runtime on the bundle until initialize and +// notifications/initialized are done. The stand-ins answer nothing until a test sets their own fields. +func startReader(t *testing.T) *startedReader { + t.Helper() + binary := buildReaderBinary(t) + dbURL := freshIntegrationDatabase(t) + runReaderPrepare(t, binary, dbURL) + + busURL := newTestBus(t) + nc, err := nats.Connect(busURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + + dir := t.TempDir() + dbPath := filepath.Join(dir, "database.url") + if err := os.WriteFile(dbPath, []byte(dbURL), 0o600); err != nil { + t.Fatal(err) + } + callersPath := filepath.Join(dir, "callers.json") + callersJSON, _ := json.Marshal(map[string]any{"callers": []string{"mesh-mail-web"}}) + if err := os.WriteFile(callersPath, callersJSON, 0o600); err != nil { + t.Fatal(err) + } + + stand := &standIns{} + env := append(os.Environ(), + "MESH_NODE="+testNode, + "MESH_SERVED_MODULE=mesh-mail-reader", + "DATABASE_URL_FILE="+dbPath, + "MESH_MAIL_READER_CALLERS="+callersPath, + ) + player := startReaderPlayer(t, binary, env, stand, nc, nil) + return &startedReader{player: player, stand: stand, nc: nc, dbURL: dbURL} +} + +// publishAdded publishes one mail..message.added event on the real bus, exactly on the wire subject +// the controller composes (addedWireSubject): account, folder, uid, date, from, subject — headers only, +// never a body (mesh-mail's own event shape, metadata() in imap.go). +func publishAdded(t *testing.T, nc *nats.Conn, account, folder string, uid uint32, from []map[string]any, subject string, date time.Time) { + t.Helper() + body := map[string]any{ + "account": account, "folder": folder, "uid": uid, "uidvalidity": 1, + "message-id": fmt.Sprintf("<%d@example.org>", uid), "date": date.UTC().Format(time.RFC3339), + "from": from, "subject": subject, "size": int64(1234), "flags": []string{}, + } + raw, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + if err := nc.Publish(addedWireSubject(account), raw); err != nil { + t.Fatal(err) + } + if err := nc.Flush(); err != nil { + t.Fatal(err) + } +} + +// waitFor polls cond every 200ms until it answers true or timeout passes, failing the test otherwise (every +// test that waits on the reader's own asynchronous hearing and batching does so through this, never a bare +// sleep). +func waitFor(t *testing.T, timeout time.Duration, why string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for { + if cond() { + return + } + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", why) + } + time.Sleep(200 * time.Millisecond) + } +} + +// pgConnect is a plain connection to the reader's own database, for a test to read its tables directly +// (1431/AC1: "a search of every text column... finds nothing"). +func pgConnect(t *testing.T, url string) *pgx.Conn { + t.Helper() + conn, err := pgx.Connect(context.Background(), url) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close(context.Background()) }) + return conn +} diff --git a/modules/mesh-mail-reader/go.mod b/modules/mesh-mail-reader/go.mod index cd92bafa..8db18754 100644 --- a/modules/mesh-mail-reader/go.mod +++ b/modules/mesh-mail-reader/go.mod @@ -5,12 +5,24 @@ go 1.26.0 require ( git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019062050-34392e74916c github.com/jackc/pgx/v5 v5.11.0 + github.com/nats-io/nats-server/v2 v2.11.17 + github.com/nats-io/nats.go v1.54.0 ) require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect - golang.org/x/sync v0.17.0 // indirect - golang.org/x/text v0.29.0 // indirect + github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.42.0 // indirect + golang.org/x/time v0.15.0 // indirect ) diff --git a/modules/mesh-mail-reader/go.sum b/modules/mesh-mail-reader/go.sum index 860456a6..56e9e3c2 100644 --- a/modules/mesh-mail-reader/go.sum +++ b/modules/mesh-mail-reader/go.sum @@ -1,8 +1,12 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019062050-34392e74916c h1:/f6f2ugss172hDJSL3EFZqb9PdfO5DKzYVfPiX2LdAM= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019062050-34392e74916c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= @@ -11,6 +15,20 @@ github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -18,10 +36,17 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= -golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= -golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= -golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=