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

package main

// mesh-mail send (novox/hq ADR 0436 §7): the sender, a process with a bus account of its own
// (mesh-mail-send), never a seat. It alone of the module's parts reads an account's password (issue 1193);
// the tools never send — they record the operator's Approve and queue it (issue 1194) on the work queue
// mail-sends, which this process consumes through its one durable consumer (issue 1195). For each message
// it verifies the warrant again and claims the send once (issue 1196), performs the one handover to the
// server (issue 1197's classification), and records the outcome, retries a not-taken failure and emits the
// result (issue 1198). The tell to the operator and the overdue health check are issue 1199's.

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/process"

	"mesh-mail/cmd/mesh-mail/internal/secret"
)

// SendProcessID is this process's id, as module.json's resource "send" names it: the controller delivers
// its own bus account as the own secret broker-<SendProcessID> (novox/hq ADR 0436 §2).
const SendProcessID = "send"

// MailSendsStream is the work-queue stream mail-sends carries on the bus (novox/hq ADR 0436 §8): the
// manifest's local name "mail-sends", capitalised with its hyphen an underscore, exactly as the controller
// composes a stream's own bus name (mesh-controller internal/catalogue/streams.go StreamBusName,
// strings.ToUpper(strings.ReplaceAll(local, "-", "_")); confirmed against its own fixture, mesh-controller
// cmd/mesh-controller/process_live_test.go's mailSends, which names it "MAIL_SENDS" — review round 1,
// finding 7).
const MailSendsStream = "MAIL_SENDS"

// MessengerAsksBucket is the operator channel's own records bucket, as the controller's own fixtures name
// it (mesh-controller cmd/mesh-controller/process_live_test.go, internal/broker/process_grants_test.go):
// never hard-coded in mesh-sdk go/process itself, since a process may read more than one seat's records.
const MessengerAsksBucket = "messenger_asks"

// routerRecordOf is the real RouterRecord (send.go), reading the router's record of one of mesh-mail's own
// asks under conn's own account (novox/hq ADR 0436 clause 9) — the same direct-get the tools use
// (stdio.SeatRecord), reached here as the sender's own account instead of the tool runner's. Wired into the
// handler once the claim, the send and its outcome are all in place (issues 1197, 1198).
func routerRecordOf(conn *process.Conn) RouterRecord {
	return func(id string) (*stdio.StateEntry, error) { return conn.Ask(MessengerAsksBucket, id) }
}

// processSends adapts conn's own `sends` state (novox/hq ADR 0436 clause 7's bus.state) to SendStore
// (send.go), the same CAS the tools use, now reached under the sender's own account. Keys is never called
// by the claim or outcome recording (issues 1196, 1198) — only the tools' sweep calls it, over its own
// account — so it is refused by name if ever called by mistake, rather than silently answering empty.
type processSends struct{ state process.State }

func (p processSends) Get(key string) (json.RawMessage, uint64, bool, error) {
	e, err := p.state.Get(key)
	if err != nil || e == nil {
		return nil, 0, false, err
	}
	return e.Value, e.Revision, true, nil
}
func (p processSends) Create(key string, value any) (uint64, error) {
	return p.state.Create(key, value)
}
func (p processSends) Update(key string, value any, revision uint64) (uint64, error) {
	return p.state.Update(key, value, revision)
}
func (p processSends) Keys() ([]string, error) {
	return nil, fmt.Errorf("the sender's own `sends` state is never listed whole; only the tools' sweep does that")
}

// send is `mesh-mail send`: it reads its own bus account and the accounts setting, connects, and consumes
// mail-sends until stopped. It refuses to start naming what is missing when the account or the accounts
// setting cannot be read — never a password, which no error here or anywhere in this process logs
// (novox/hq ADR 0436 §7, §12).
func send(env Env) int {
	if env.Secrets == "" {
		logf("[mesh-mail] send was started without its secrets directory (MESH_MAIL_SECRETS)")
		return 2
	}
	if _, err := ReadAccounts(env.Accounts); err != nil {
		logf("[mesh-mail] send: %v", err)
		return 1
	}
	account, err := process.ReadAccount(secret.Path(env.Secrets, "broker-"+SendProcessID))
	if err != nil {
		logf("[mesh-mail] send: %v", err)
		return 1
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	conn, err := process.Connect(account)
	if err != nil {
		logf("[mesh-mail] send: the bus cannot be reached: %v", err)
		return 1
	}
	defer conn.Close()
	h := &senderHandler{
		Store:      processSends{state: conn.State(SendsState)},
		Record:     routerRecordOf(conn),
		Accounts:   func() ([]Account, error) { return ReadAccounts(env.Accounts) },
		SecretsDir: env.Secrets,
		Emit:       emitOf(conn),
		Tell:       tellOperatorOf(conn),
		Logf:       logf,
	}
	go h.tellTicker(ctx)
	logf("[mesh-mail] send is consuming %s as %s", MailSendsStream, account.User)
	if err := conn.Consume(ctx, MailSendsStream, h.handle); err != nil && ctx.Err() == nil {
		logf("[mesh-mail] send stopped: %v", err)
		return 1
	}
	return 0
}

// tellOperatorOf is the real TellOperator (send.go), wrapping conn's own Call to the operator channel's
// tell verb (novox/hq ADR 0380 §8, ADR 0436 clause 11): no lines, no subject naming anything the content
// rule would refuse — the headline alone carries the one line.
func tellOperatorOf(conn *process.Conn) TellOperator {
	return func(headline string) error {
		_, err := conn.Call("seat:operator-channel.tell", map[string]any{"headline": headline, "lines": []string{}, "subject": "mesh-mail"})
		return err
	}
}

// TellEvery is how often the sender looks for a send still owing its own tell (novox/hq ADR 0436 clause
// 11: "at least every 10 minutes").
const TellEvery = 10 * time.Minute

// tellTicker runs every send still owing the sender's own tell at least every TellEvery, until ctx ends
// (novox/hq ADR 0436 clause 11, issue 1199): once right away, so a tell left over a restart is not made to
// wait a full interval.
func (h *senderHandler) tellTicker(ctx context.Context) {
	t := time.NewTicker(TellEvery)
	defer t.Stop()
	for {
		h.tellPending()
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

func (h *senderHandler) tellPending() {
	ids, err := tellPendingIDs(h.Store)
	if err != nil {
		h.logf("[mesh-mail] send: the tells-pending index cannot be read: %v", err)
		return
	}
	for _, id := range ids {
		if err := tellSend(h.Store, h.Tell, id, h.now()); err != nil {
			h.logf("[mesh-mail] send: telling the operator about %s: %v", id, err)
		}
	}
}

// runSend connects as the sender's own account and consumes mail-sends until ctx ends, handing each message
// to handle. Apart from send so a test can drive it over a bus of its own, with a handler of its own
// (novox/hq issue 1195, AC2): on ctx ending (a signal), Consume stops pulling and returns without
// acknowledging a message still in handle's hands — only handle decides a message's fate (process/queue.go).
func runSend(ctx context.Context, account process.Account, handle func(*process.Message)) error {
	conn, err := process.Connect(account)
	if err != nil {
		return fmt.Errorf("the bus cannot be reached: %w", err)
	}
	defer conn.Close()
	return conn.Consume(ctx, MailSendsStream, handle)
}

// emitOf is the real Emit (send.go), wrapping conn's own Emit with the given id as both x-event-id and the
// bus's own dedup id (novox/hq ADR 0436 clauses 4, 11).
func emitOf(conn *process.Conn) Emit {
	return func(eventType, id string, body any) error {
		return conn.Emit(eventType, body, stdio.EmitOptions{MessageID: id})
	}
}

// senderHandler is everything handleQueuedSend needs: the `sends` state and the router's record, both
// reached under the sender's own account, the accounts setting (to find the server to connect to) and the
// secrets directory (to read the one password this process holds), and how it sends and emits — real
// unless a test gives its own (novox/hq ADR 0436 §7, issues 1196 to 1198).
type senderHandler struct {
	Store      SendStore
	Record     RouterRecord
	Accounts   func() ([]Account, error)
	SecretsDir string
	Dial       SMTPDialer
	Emit       Emit
	Tell       TellOperator
	Now        func() time.Time
	Logf       func(format string, a ...any)
}

func (h *senderHandler) now() time.Time {
	if h.Now != nil {
		return h.Now()
	}
	return time.Now()
}

func (h *senderHandler) logf(format string, a ...any) {
	if h.Logf != nil {
		h.Logf(format, a...)
	}
}

// handle is the handler mesh-mail send's Consume hands every message to mail-sends (novox/hq ADR 0436
// clauses 9, 10, 11): the warrant verified again and the send claimed (claimQueuedSend, send.go), the one
// handover to the server attempted, and the outcome recorded, retried or emitted (recordOutcome, send.go) —
// then, and only then, the message is acknowledged, asked redelivered, or terminated, exactly as that
// outcome says.
func (h *senderHandler) handle(msg *process.Message) {
	var q QueuedSend
	if err := json.Unmarshal(msg.Body, &q); err != nil {
		h.logf("[mesh-mail] send: a queued message on %s could not be read: %v", msg.Subject, err)
		if err := msg.Term(); err != nil {
			h.logf("[mesh-mail] send: terminating %s: %v", msg.Subject, err)
		}
		return
	}
	req, claimed, rev, outcome, err := claimQueuedSend(h.Store, h.Record, h.Emit, q.Send, h.now())
	if err != nil {
		// Logged, never returned on alone: an index-bookkeeping error (novox/hq review round 1 finding 3)
		// can stand beside a real decision (claimRefused, claimNotProven) that the switch below still
		// must act on; a fatal read or claim failure, with no decision made, always reads claimLostRace,
		// which the switch already leaves to the bus.
		h.logf("[mesh-mail] send: claiming %s: %v", q.Send, err)
	}
	switch outcome {
	case claimLostRace:
		return // another delivery of the same message already owns it.
	case claimRefused, claimNotProven, claimAlreadyFinished:
		// claimAlreadyFinished: a repeat of a message whose send already finished (the operator was
		// already told, or not, exactly once) — terminated, and nothing else touched (novox/hq review,
		// operator-approved change to clause 10).
		if err := msg.Term(); err != nil {
			h.logf("[mesh-mail] send: terminating %s: %v", q.Send, err)
		}
		return
	}
	sendErr := h.attempt(req)
	_, action, delay, err := recordOutcome(h.Store, h.Emit, req, claimed, rev, sendErr, h.now())
	if err != nil {
		h.logf("[mesh-mail] send: recording the outcome of %s: %v", q.Send, err)
	}
	switch action {
	case actionAck:
		if err := msg.Ack(); err != nil {
			h.logf("[mesh-mail] send: acknowledging %s: %v", q.Send, err)
		}
	case actionRetry:
		if err := msg.NakWithDelay(delay); err != nil {
			h.logf("[mesh-mail] send: asking %s redelivered: %v", q.Send, err)
		}
	case actionTerm:
		if err := msg.Term(); err != nil {
			h.logf("[mesh-mail] send: terminating %s: %v", q.Send, err)
		}
	case actionLeave:
		// The outcome's own write could not be kept, even retried on a changed revision (novox/hq review
		// round 1 finding 5): nothing was emitted above, and nothing is done to the message either — left
		// entirely to the bus, so a later redelivery finds the record still at its previous outcome.
	}
}

// attempt is the one handover to req.Account's outgoing server (novox/hq ADR 0436 clause 10): it reads the
// accounts setting and the account's own password fresh, never from the queue's message, and every way it
// cannot even be attempted (the account vanished, its password not given, the setting unreadable) is a
// *SendError classified NotTaken, exactly as SendMail's own would be — safe to retry.
func (h *senderHandler) attempt(req SendRequest) error {
	accounts, err := h.Accounts()
	if err != nil {
		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the accounts setting cannot be read: %w", err)}
	}
	acc, found := accountByName(accounts, req.Account)
	if !found {
		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the account %s no longer exists", req.Account)}
	}
	pw, given, err := secret.Read(h.SecretsDir, acc.Secret())
	if err != nil {
		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the account %s's password cannot be read: %w", acc.Name, err)}
	}
	if !given {
		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the account %s has no password yet; the operator gives it at the desk", acc.Name)}
	}
	dial := h.Dial
	if dial == nil {
		dial = DialSMTP
	}
	return SendMail(dial, acc, pw, req.To, req.Subject, req.Body, h.now(), req.ID)
}