Start mesh-mail-reader whole against the real bus and store: entries, no body kept, the tell, the cap (issue 1431)
This commit is contained in:
jochen
@@ -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)
|
||||
}
|
||||
@@ -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@<this node>
|
||||
// (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.<account>.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.<account>.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
|
||||
}
|
||||
Reference in New Issue
Block a user