292 lines
8.9 KiB
Go
292 lines
8.9 KiB
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
|
|
//
|
|
// 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). Only the bundle reaches the bus, through the runtime that
|
|
// launched it, 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()))
|
|
}
|
|
}
|
|
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
|
|
p := &Publisher{Index: b.Index, Emit: func(key string, body any) error { return stdio.Emit(key, body) }, Logf: logf}
|
|
go p.Run(context.Background())
|
|
go hear(b)
|
|
go sweepSends(b)
|
|
if err := stdio.Serve("", b.Tools()); err != nil {
|
|
logf("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// 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) {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
time.Sleep(wait)
|
|
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)
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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
|
|
}
|
|
}
|
|
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):
|
|
}
|
|
}
|
|
}
|