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- (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) }