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

// mesh-mail: the operator's mail accounts, watched, and each mail said on the bus (novox/hq ADR 0359). One instance,
// on the control-node. Reading is read-only: no flag, move, delete or draft. mail_send is the one way out (novox/hq
// ADR 0361 §2): it sends nothing itself, and reaches an account's outgoing server only on the operator's Approve.
//
//	mesh-mail            the bundle the node's runtime launches: the tools mail_accounts, mail_search, mail_read,
//	                     mail_health, mail_send and mail_send_status, and the events the keeper wrote, said on the
//	                     bus as the module
//	mesh-mail prepare    the preparation step: the index's schema brought to this version's
//	mesh-mail keep       acts: watches every account (IDLE on the inbox, a look at the other folders every five
//	                     minutes), keeps the index and the attachment store, and writes each change's event beside it
//	mesh-mail watch      judges: reads every minute how each account is, and leaves its verdict for mail_health
//	mesh-mail send       the sender (novox/hq ADR 0436 §7): its own bus account, consuming the work queue
//	                     mail-sends; it alone reads a password, and performs the one send an operator's
//	                     Approve authorises, once the tools have recorded it approved and queued it
//
// The two long-running processes are the act and the judge, as mesh-knowledge's are: keep is judged by its unit, watch
// by the tool mail_health (ADR 0359 §6, ADR 0240 §3 and §7); send is judged by its unit too (it holds a bus account
// of its own, never a seat, so no tool answers for it). Only the bundle reaches the bus through the runtime that
// launched it; send reaches it on its own account instead (mesh-sdk go/process), so the keeper hands its events over
// in the index, written in the same step as the change they say, and the bundle says them and removes each once the
// bus took it. stdout is the bundle's MCP channel; what this module says, it says on stderr.
package main

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/asks"
)

// Module is this module's name.
const Module = "mesh-mail"

// Env is where the mesh puts what the module reads, as the manifest's environment names it.
type Env struct {
	Accounts string // the trusted setting accounts
	Grants   string // the trusted setting grants
	Secrets  string // the directory the own secrets mail-credential-<account> are placed in
	Store    string // the attachment store's facts
	Status   string // the watch's verdict
}

func envFromOS() Env {
	return Env{
		Accounts: os.Getenv("MESH_MAIL_ACCOUNTS"),
		Grants:   os.Getenv("MESH_MAIL_GRANTS"),
		Secrets:  os.Getenv("MESH_MAIL_SECRETS"),
		Store:    os.Getenv("MESH_MAIL_STORE"),
		Status:   os.Getenv("MESH_MAIL_STATUS"),
	}
}

func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }

func main() {
	if len(os.Args) > 1 {
		switch os.Args[1] {
		case "prepare":
			os.Exit(prepare())
		case "keep":
			os.Exit(keep(envFromOS()))
		case "watch":
			os.Exit(watch(envFromOS()))
		case "send":
			os.Exit(send(envFromOS()))
		}
	}
	serve(envFromOS())
}

// ---- prepare ------------------------------------------------------------------------------------------------------

func prepare() int {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	s, err := PgStoreFromEnv(ctx)
	if err != nil {
		logf("[mesh-mail] preparing the index: %v", err)
		return 1
	}
	defer s.Close()
	said, err := s.Prepare(ctx)
	for _, line := range said {
		logf("[mesh-mail] %s", line)
	}
	if err != nil {
		logf("[mesh-mail] preparing the index failed: %v", err)
		return 1
	}
	return 0
}

// ---- keep ---------------------------------------------------------------------------------------------------------

// keep watches every account until it is stopped. A setting it cannot read, an index that is behind or a store it
// cannot reach ends it with a fault, which its unit says; the mesh starts it again.
func keep(env Env) int {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	accounts, err := ReadAccounts(env.Accounts)
	if err != nil {
		logf("[mesh-mail] %v", err)
		return 1
	}
	st, err := PgStoreFromEnv(ctx)
	if err != nil {
		logf("[mesh-mail] the index cannot be opened: %v", err)
		return 1
	}
	defer st.Close()
	if err := st.Ready(ctx); err != nil {
		logf("[mesh-mail] %v", err)
		return 1
	}
	facts, err := ReadStoreFacts(env.Store)
	if err != nil {
		logf("[mesh-mail] %v", err)
		return 1
	}
	k := &Keeper{Store: st, Objects: NewS3(facts), Dial: DialServer, SecretsDir: env.Secrets, Logf: logf}
	logf("[mesh-mail] keeping %d account(s)", len(accounts))
	k.Run(ctx, accounts)
	return 0
}

// ---- watch --------------------------------------------------------------------------------------------------------

func watch(env Env) int {
	if env.Status == "" {
		logf("[mesh-mail] watch was started without MESH_MAIL_STATUS")
		return 2
	}
	open := func(ctx context.Context) (Store, func(), error) {
		s, err := PgStoreFromEnv(ctx)
		if err != nil {
			return nil, nil, err
		}
		return s, s.Close, nil
	}
	for {
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		now := time.Now().UTC()
		answer := watchOnce(ctx, env, open, now)
		cancel()
		if err := writeStatus(env.Status, Status{Answer: answer, At: now}); err != nil {
			logf("[mesh-mail] the verdict could not be written: %v", err)
		}
		time.Sleep(WatchEvery)
	}
}

// ---- the bundle ---------------------------------------------------------------------------------------------------

func serve(env Env) {
	var mu sync.Mutex
	var st *PgStore
	b := &Bundle{
		Env: env,
		Index: func(ctx context.Context) (Store, error) {
			mu.Lock()
			defer mu.Unlock()
			if st == nil {
				s, err := PgStoreFromEnv(ctx)
				if err != nil {
					return nil, err
				}
				st = s
			}
			if err := st.Ready(ctx); err != nil {
				return nil, err
			}
			return st, nil
		},
		Objects: func() (Objects, error) {
			facts, err := ReadStoreFacts(env.Store)
			if err != nil {
				return nil, err
			}
			return NewS3(facts), nil
		},
	}
	b.Logf = logf
	mirror := &Mirror{
		Index:    b.Index,
		Accounts: func() ([]Account, error) { return ReadAccounts(env.Accounts) },
		Accts:    keptMirrorState{stdio.State(AccountsState)},
		Mails:    keptMirrorState{stdio.State(MailsState)},
		Logf:     logf,
	}
	trigger := make(chan struct{}, 1)
	p := &Publisher{Index: b.Index, Emit: func(key string, body any) error { return stdio.Emit(key, body) }, Logf: logf,
		Said: func(int) {
			select {
			case trigger <- struct{}{}:
			default:
			}
		}}
	go p.Run(context.Background())
	go mirrorTicker(mirror, trigger)
	go hear(b)
	go sweepSends(b)
	go waitsTellTicker(b)
	if err := stdio.Serve("", b.Tools()); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// sdkTellOperator is the tools' own TellOperator (send.go), wrapping stdio.Ask to the operator channel's
// tell verb, exactly as mesh-files' own pattern (modules/mesh-files/cmd/mesh-files/main.go tellOperator):
// no lines, no subject naming anything the content rule would refuse.
func sdkTellOperator(headline string) error {
	_, err := stdio.Ask("seat:operator-channel.tell", map[string]any{"headline": headline, "lines": []string{}, "subject": "mesh-mail"})
	return err
}

// WaitsTellEvery is how often the tools look for an overdue send still owing their own "waits" tell
// (novox/hq ADR 0436 clause 11, issue 1199): a second ticker, apart from sweepSends, since the two tells —
// the sender's own failure, the tools' own overdue wait — are never the same look.
const WaitsTellEvery = time.Minute

// waitsTellTicker runs the tools' own "waits" tell for every overdue send, at least every WaitsTellEvery,
// until the process ends (novox/hq ADR 0436 clause 11, issue 1199).
func waitsTellTicker(b *Bundle) {
	tick := time.NewTicker(WaitsTellEvery)
	defer tick.Stop()
	for {
		overdue, err := Overdue(b.sends(), time.Now())
		if err != nil {
			logf("[mesh-mail] the overdue check cannot be read: %v", err)
		}
		for _, req := range overdue {
			if err := tellWaits(b.sends(), sdkTellOperator, req.ID, time.Now()); err != nil {
				logf("[mesh-mail] telling the operator that %s waits: %v", req.ID, err)
			}
		}
		<-tick.C
	}
}

// hear subscribes to the router's word on this module's own asks, asked again until the runtime takes it, exactly
// as mesh-knowledge does. Each warrant heard goes to Heard, the same Decide that mail_send_status uses: an
// approved send goes out as soon as the operator answers it, never only when something later calls
// mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790).
func hear(b *Bundle) {
	<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
			var w asks.Warrant
			if json.Unmarshal(e.Body, &w) == nil {
				b.Heard(w)
			}
			return nil
		})
		if err == nil {
			logf("[mesh-mail] hears the operator's answers")
			return
		}
		logf("[mesh-mail] does not hear the operator's answers yet (%v); asking again in %s", err, wait)
		time.Sleep(wait)
	}
}

// MirrorEvery is the Mirror's own round's slowest pace (settlement 6, issue 1255): at least this often, even
// with nothing published in between.
const MirrorEvery = time.Minute

// mirrorTicker runs the Mirror's round at least every MirrorEvery, and at once again whenever trigger fires (a
// Publisher round that said something, settlement 6), until the process ends.
func mirrorTicker(m *Mirror, trigger <-chan struct{}) {
	tick := time.NewTicker(MirrorEvery)
	defer tick.Stop()
	round := func() {
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		if err := m.Round(ctx); err != nil && m.Logf != nil {
			m.Logf("[mesh-mail] the state buckets could not be mirrored: %v", err)
		}
	}
	for {
		select {
		case <-trigger:
			round()
		case <-tick.C:
			round()
		}
	}
}

// sweepSends reads every send still undecided once a minute and asks the router's own record for its answer, so
// an Approve answered while this module could not hear it directly is still acted on (novox/hq issue 790, review
// round 1).
func sweepSends(b *Bundle) {
	tick := time.NewTicker(SweepEvery)
	defer tick.Stop()
	for {
		b.Sweep(time.Now())
		<-tick.C
	}
}

// Publisher says the events the keeper wrote, oldest first, and removes each once the bus took it. One the bus did not
// take stays, and is said again on the next round: delivery is at least once, and a consumer reads the uid and the
// folder to know a repeat.
type Publisher struct {
	Index func(ctx context.Context) (Store, error)
	Emit  func(key string, body any) error
	Logf  func(format string, a ...any)
	Every time.Duration
	// Said, when given, is called after a round that said one event or more (novox/hq issue 1255, settlement
	// 6): the Mirror's own round runs beside it, so the state buckets catch up without waiting for the next
	// tick.
	Said func(n int)
}

// PublishBatch is how many events one round says at most.
const PublishBatch = 100

// Once says what is pending, and answers how many it said.
func (p *Publisher) Once(ctx context.Context) (int, error) {
	st, err := p.Index(ctx)
	if err != nil {
		return 0, err
	}
	pending, err := st.Pending(ctx, PublishBatch)
	if err != nil {
		return 0, err
	}
	var said []int64
	var failed error
	for _, e := range pending {
		if err := p.Emit(e.Key, e.Body); err != nil {
			failed = fmt.Errorf("the bus did not take %s: %v", e.Key, err)
			break
		}
		said = append(said, e.ID)
	}
	if len(said) > 0 {
		if err := st.Said(ctx, said); err != nil {
			return 0, err
		}
		if p.Said != nil {
			p.Said(len(said))
		}
	}
	return len(said), failed
}

// Run says pending events until ctx ends: every few seconds, at once again while a round was full.
func (p *Publisher) Run(ctx context.Context) {
	every := p.Every
	if every <= 0 {
		every = 3 * time.Second
	}
	// The runtime takes the bundle's first calls only once Serve runs.
	time.Sleep(every)
	for ctx.Err() == nil {
		c, cancel := context.WithTimeout(ctx, time.Minute)
		n, err := p.Once(c)
		cancel()
		if err != nil && p.Logf != nil {
			p.Logf("[mesh-mail] events wait to be said: %v", err)
		}
		if n == PublishBatch {
			continue
		}
		select {
		case <-ctx.Done():
		case <-time.After(every):
		}
	}
}