mesh-catalog/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go

//go:build integration

package main

// Release 3's acceptance test (novox/hq ADR 0436, issue 1198 AC4): the whole module, over a JetStream bus
// of the test's own (nats-server v2.11.17, copied from modules/mesh-delivery/cmd/mesh-delivery/bus_test.go
// as newTestBus already does in sender_integration_test.go), a fake SMTP server, the router's records
// bucket with a closed Approve record, the account file and a password file written by the test. Compiled
// by the merge gate's `go vet -tags integration`; never run on this machine (the context file's hard rule).
//
// "The tools queue an approved send" is played here directly against the real bus (Decide, send.go) rather
// than through the tool runner's own RPC — the tool runner's part (stdio.Serve, grants, the ask's wording)
// is proved elsewhere (send_test.go, ask_test.go); what this test proves is the handover from an approved,
// queued send through to the sender's own claim, send and outcome, end to end over a real bus.

import (
	"context"
	"encoding/json"
	"errors"
	"path/filepath"
	"strings"
	"testing"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// kvSends adapts a raw jetstream.KeyValue to SendStore (send.go): the test's own stand-in for both the
// tools' and the sender's `sends` state, which in production are the same bucket reached under two
// different accounts.
type kvSends struct{ kv jetstream.KeyValue }

func (k kvSends) Get(key string) (json.RawMessage, uint64, bool, error) {
	e, err := k.kv.Get(context.Background(), key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, 0, false, nil
	}
	if err != nil {
		return nil, 0, false, err
	}
	return json.RawMessage(e.Value()), e.Revision(), true, nil
}
func (k kvSends) Create(key string, value any) (uint64, error) {
	data, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	rev, err := k.kv.Create(context.Background(), key, data)
	if errors.Is(err, jetstream.ErrKeyExists) {
		return 0, stdio.ErrChanged
	}
	return rev, err
}
func (k kvSends) Update(key string, value any, revision uint64) (uint64, error) {
	data, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	rev, err := k.kv.Update(context.Background(), key, data, revision)
	if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
		return 0, stdio.ErrChanged
	}
	return rev, err
}
func (k kvSends) Keys() ([]string, error) { return nil, errors.New("not used by this test") }

// acceptanceBus is everything the test built on a fresh bus: ready-made buckets and streams, and the raw
// clients to act as the tools with.
type acceptanceBus struct {
	nc    *nats.Conn
	js    jetstream.JetStream
	sends jetstream.KeyValue
	asks  jetstream.KeyValue
}

func newAcceptanceBus(t *testing.T) (*acceptanceBus, string) {
	t.Helper()
	url := newTestBus(t)
	ctx := context.Background()
	nc, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	sends, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: Module + "_" + SendsState})
	if err != nil {
		t.Fatal(err)
	}
	asksKV, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: MessengerAsksBucket})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, mailSendsStreamConfig()); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod." + Module + ".event.>"}}); err != nil {
		t.Fatal(err)
	}
	return &acceptanceBus{nc: nc, js: js, sends: sends, asks: asksKV}, url
}

// queueAnApprovedSend plays the tools' own part (send.go's Decide, the queue publish) directly against the
// real bus: a send asked, approved by a warrant whose digest really is the one req.Ask() binds, recorded
// approved and queued — exactly what issue 1194 made the tools do, without needing the tool runner's RPC.
func queueAnApprovedSend(t *testing.T, bus *acceptanceBus, account string, approvedAt time.Time) (SendRequest, string) {
	t.Helper()
	ctx := context.Background()
	req := SendRequest{ID: newSendID(), Account: account, To: []string{"ops@example.org"}, Subject: "The nightly gate is red",
		Body: "See the job log.", Caller: "person.op", At: approvedAt, AskTo: "o… at example org"}
	store := kvSends{kv: bus.sends}
	if _, err := store.Create(req.ID, req); err != nil {
		t.Fatalf("keeping the send: %v", err)
	}
	w := warrantFor(req, approvedAt)
	rec := routerRecord{State: string(w.Outcome), Warrant: &w}
	raw, _ := json.Marshal(rec)
	if _, err := bus.asks.Put(ctx, Module+"."+req.ID, raw); err != nil {
		t.Fatalf("writing the router's record: %v", err)
	}
	// The tools' own publish path (send.go's QueuePublish, wired for real to stdio.Emit/sdkQueuePublish):
	// the real wire subject the controller composes for mail-sends' one event (mailSendApprovedSubject),
	// the send id as the message id, so the stream's own duplicate window is the one actually exercised
	// (novox/hq review round 1, finding 7).
	publish := func(id string, body any) error {
		data, err := json.Marshal(body)
		if err != nil {
			return err
		}
		_, err = bus.js.Publish(ctx, mailSendApprovedSubject, data, jetstream.WithMsgID(id))
		return err
	}
	if _, err := Decide(store, publish, w, approvedAt); err != nil {
		t.Fatalf("the tools' own Decide: %v", err)
	}
	kept, _, ok, err := getSend(store, req.ID)
	if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "queued" {
		t.Fatalf("not left queued: %v %v %+v", err, ok, kept.Decided)
	}
	// Published a second time under the same message id, exactly as the sweep would for a record it finds
	// still queued past its grace (Sweep, send.go): the stream's 24-hour duplicate window keeps it one
	// message (novox/hq review round 1, finding 7) — the sender's own claim (clause 9) is what then keeps
	// it one send, proved by this test's own "sent once" assertion below.
	if err := publish(req.ID, QueuedSend{Send: req.ID, Account: req.Account, Ask: req.ID}); err != nil {
		t.Fatalf("the duplicate publish: %v", err)
	}
	return req, req.ID
}

// 1198/AC4: when the tools queue an approved send and the send process runs, the fake server receives
// exactly one mail carrying the send id's Message-ID, sends says sent, and mail.send.sent is on the stream.
func TestMailSenderAcceptanceAnApprovedSendIsSentOnceAndSaidOnce(t *testing.T) {
	bus, url := newAcceptanceBus(t)
	fake := newFakeSMTPServer(t)

	dir := t.TempDir()
	accountsPath := filepath.Join(dir, "accounts.json")
	writeFile(t, accountsPath, `{"accounts": {"mesh": {"kind": "imap", "address": "jschoubben@novox.be",
	  "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"},
	  "outgoing": {"host": "127.0.0.1", "port": 1, "security": "tls"}}}}`)
	writeFile(t, filepath.Join(dir, "mail-credential-mesh.secret"), "the-password\n")

	approvedAt := judgedAt
	req, id := queueAnApprovedSend(t, bus, "mesh", approvedAt)

	account := process.Account{URL: url, Node: "g14", Module: Module, User: "g14." + Module + "." + SendProcessID, Password: "pw"}
	if _, err := bus.js.CreateConsumer(context.Background(), MailSendsStream, jetstream.ConsumerConfig{
		Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: 5 * time.Second, MaxDeliver: MostAttempts}); err != nil {
		t.Fatal(err)
	}
	conn, err := process.Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	h := &senderHandler{
		Store:      kvSends{kv: bus.sends},
		Record:     routerRecordOf(conn),
		Accounts:   func() ([]Account, error) { return ReadAccounts(accountsPath) },
		SecretsDir: dir,
		Dial:       fake.dialer(),
		Emit:       emitOf(conn),
		Now:        func() time.Time { return approvedAt.Add(time.Minute) },
	}

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- conn.Consume(ctx, MailSendsStream, h.handle) }()

	eventually(t, 10*time.Second, "the mail sent", func() (bool, string) {
		return fake.count() == 1, "received so far: " + describeReceived(fake)
	})
	cancel()
	<-done

	store := kvSends{kv: bus.sends}
	kept, _, ok, err := getSend(store, id)
	if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sent" {
		t.Fatalf("the kept send: %v %v %+v", err, ok, kept.Decided)
	}
	got := fake.last()
	if got.to[0] != "ops@example.org" {
		t.Fatalf("received: %+v", got)
	}
	if !strings.Contains(got.data, "Message-Id: <"+req.ID+"@novox.be>") {
		t.Fatalf("the mail's Message-ID does not name the send: %s", got.data)
	}

	// mail.send.sent is on the EVENTS stream, naming the send id, never the recipient, subject or body.
	eventually(t, 10*time.Second, "mail.send.sent on the stream", func() (bool, string) {
		msg, err := fetchOneRaw(bus, "mesh.mod."+Module+".event."+MailSendSent)
		if err != nil || msg == nil {
			return false, "not yet"
		}
		var e OutcomeEvent
		_ = json.Unmarshal(msg, &e)
		return e.Send == req.ID && e.Account == "mesh", "got " + string(msg)
	})
}

// 1198/AC4: when the fake server refuses sign-in, then after the attempts (shortened here) the record says
// failed and nothing was received.
func TestMailSenderAcceptanceARefusedSignInFailsAfterItsAttemptsAndNothingIsReceived(t *testing.T) {
	orig := RetryDelays
	RetryDelays = []time.Duration{10 * time.Millisecond, 10 * time.Millisecond, 10 * time.Millisecond, 10 * time.Millisecond}
	t.Cleanup(func() { RetryDelays = orig })

	bus, url := newAcceptanceBus(t)
	s := newScriptedSMTPServer(t)
	s.auth = reply(535, "authentication failed")

	dir := t.TempDir()
	accountsPath := filepath.Join(dir, "accounts.json")
	writeFile(t, accountsPath, `{"accounts": {"mesh": {"kind": "imap", "address": "jschoubben@novox.be",
	  "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"},
	  "outgoing": {"host": "127.0.0.1", "port": 1, "security": "tls"}}}}`)
	writeFile(t, filepath.Join(dir, "mail-credential-mesh.secret"), "the-wrong-password\n")

	approvedAt := judgedAt
	_, id := queueAnApprovedSend(t, bus, "mesh", approvedAt)

	account := process.Account{URL: url, Node: "g14", Module: Module, User: "g14." + Module + "." + SendProcessID, Password: "pw"}
	if _, err := bus.js.CreateConsumer(context.Background(), MailSendsStream, jetstream.ConsumerConfig{
		Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: 2 * time.Second, MaxDeliver: MostAttempts}); err != nil {
		t.Fatal(err)
	}
	conn, err := process.Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	h := &senderHandler{
		Store:      kvSends{kv: bus.sends},
		Record:     routerRecordOf(conn),
		Accounts:   func() ([]Account, error) { return ReadAccounts(accountsPath) },
		SecretsDir: dir,
		Dial:       s.dialer(),
		Emit:       emitOf(conn),
		Now:        func() time.Time { return approvedAt.Add(time.Minute) },
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- conn.Consume(ctx, MailSendsStream, h.handle) }()

	store := kvSends{kv: bus.sends}
	eventually(t, 25*time.Second, "the send to fail after its attempts", func() (bool, string) {
		kept, _, ok, err := getSend(store, id)
		if err != nil || !ok || kept.Decided == nil {
			return false, "not yet decided"
		}
		// "queued" is the send's own starting state (queueAnApprovedSend's own write, before the sender
		// ever claims it): without excluding it here too, this check reads done on the very first look,
		// long before the sender connects or claims anything at all (novox/hq review: build-1792400882114092431,
		// 1198/AC4).
		o := kept.Decided.Outcome
		return o != "queued" && o != "retrying" && o != "sending", o
	})
	cancel()
	<-done

	kept, _, ok, err := getSend(store, id)
	if err != nil || !ok || kept.Decided == nil || !strings.HasPrefix(kept.Decided.Outcome, "failed: ") {
		t.Fatalf("the kept send did not end failed: %v %v %+v", err, ok, kept.Decided)
	}
	if kept.Decided.Attempt != MostAttempts {
		t.Fatalf("failed after %d attempt(s), want all %d", kept.Decided.Attempt, MostAttempts)
	}
	if n := s.receivedCount(); n != 0 {
		t.Fatalf("the scripted server received %d message(s), want none: a refused sign-in never reaches DATA", n)
	}
}

// fetchOneRaw pulls at most one message off a stream's subject, or nil within a short wait, for a test to
// check an emitted event without binding a durable consumer of its own.
func fetchOneRaw(bus *acceptanceBus, subject string) ([]byte, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	cons, err := bus.js.OrderedConsumer(ctx, "EVENTS", jetstream.OrderedConsumerConfig{FilterSubjects: []string{subject}})
	if err != nil {
		return nil, err
	}
	batch, err := cons.Fetch(1, jetstream.FetchMaxWait(time.Second))
	if err != nil {
		return nil, err
	}
	msg, ok := <-batch.Messages()
	if !ok {
		return nil, nil
	}
	return msg.Data(), nil
}

func describeReceived(f *fakeSMTPServer) string {
	if f.count() == 0 {
		return "none"
	}
	got := f.last()
	return got.from
}