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

package main

// mesh-mail-notifier: the default holder of mail-notifier (novox/hq ADR 0378 §7, as amended by ADR 0379 §4,
// issue 800 part B). It hears mesh-mail's added events on its own account (runs-as, ADR 0376 §2) and shows each
// new inbox mail at most once, through node-notifier.send: quiet during a first sync (mesh-mail already says
// nothing then, imap.go ~387), for mail from the account's own address, for mail already marked read (\Seen),
// and for anything outside the inbox. For each new qualifying mail of an account, arriving at t, N counts that
// account's qualifying mail — shown or folded alike, each counted once — that arrived in the 60 seconds up to
// and including t: N <= FoldOver is shown at once; N > FoldOver is folded. The first folded mail opens the
// account's fold; every folded mail joins it; the fold is shown, as one notice naming how many it holds and
// the latest sender, once Window has passed since its own first mail — after which a new fold may open. The
// count is a true sliding window of arrivals, not a per-session quota that only resets after a quiet Window:
// review round 2's reading kept a session's count alive, and so folding, for as long as any gap stayed under
// Window, which folds a steady trickle (one mail every 50s, say) forever even though no burst ever happened;
// review round 3 of issue 800 replaces that reading with this one. Nothing pending — a folded mail awaiting
// its fold's deadline, or a failed send awaiting its retry — is ever dropped (novox/hq ADR 0378 §9, review
// round 3 of issue 800, which replaces the round 1 and round 2 readings).
//
// Nothing here names Betterbird (ADR 0379 §3).

import (
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

// Inbox is the one folder a new mail is shown for.
const Inbox = "INBOX"

// Seen is the flag that means a mail was already read: shown for nothing it marks.
const SeenFlag = "\\Seen"

// FoldOver is how many new mails from one account, within one window, are still shown one by one; the rest of
// that window is folded into a single notice (novox/hq ADR 0378 §9: "more than three").
const FoldOver = 3

// Window is the trailing 60 seconds ADR 0378 §9 counts a mail's N over (every qualifying mail of the account,
// shown or folded alike, that arrived in the 60 seconds up to and including this one), and the time a fold
// waits, from its own first mail, before it is shown (review round 3 of issue 800, which replaces the round 1
// and round 2 readings: the count is a true sliding window of arrivals, not a per-session quota that stays
// alive, and so keeps folding, for as long as any gap between mails stays under Window).
const Window = time.Minute

// RetryBackoff is how long the notifier waits, after an account's notice fails to send, before it tries that
// account again: at most once a minute (review round 1 of issue 800), so a send that keeps failing is not
// retried on every Flush.
const RetryBackoff = time.Minute

// AccountsCacheFor is how long mail-store.accounts's answer is reused before it is asked again (review round
// 1 of issue 800, bullet 5): neither the quiet rule nor status asks it more than once a minute.
const AccountsCacheFor = time.Minute

// Person is a sender or a recipient, as mesh-mail's event says it.
type Person struct {
	Name    string `json:"name,omitempty"`
	Address string `json:"address"`
}

// EventBody is what mesh-mail's added event says of a message (metadata only; mailevent.go names where it is
// said). Unknown fields are ignored: the notifier reads only what it needs.
type EventBody struct {
	Account     string   `json:"account"`
	Folder      string   `json:"folder"`
	UID         uint32   `json:"uid"`
	UIDValidity uint32   `json:"uidvalidity"`
	MessageID   string   `json:"message-id"`
	Subject     string   `json:"subject"`
	From        []Person `json:"from"`
	Flags       []string `json:"flags"`
}

// Handle names one message, stably, for the dedupe state (novox/hq ADR 0378 §9: "a redelivered event is not
// shown twice"). The message id is the key wherever a message carries one, so a UIDVALIDITY reset — which
// changes every UID in a folder — can never make an already-shown mail look new, or a genuinely new one look
// already shown; only a message with no message id falls back to its folder, UIDVALIDITY and UID (review
// round 1 of issue 800).
func (e EventBody) Handle() string {
	if e.MessageID != "" {
		return e.Account + "\x1f" + e.MessageID
	}
	return e.Account + "\x1f" + e.Folder + "\x1f" +
		strconv.FormatUint(uint64(e.UIDValidity), 10) + "\x1f" + strconv.FormatUint(uint64(e.UID), 10)
}

func (e EventBody) hasFlag(flag string) bool {
	for _, f := range e.Flags {
		if strings.EqualFold(f, flag) {
			return true
		}
	}
	return false
}

// sender is how a message's first sender reads in a notice: its name and address, or its address alone.
func (e EventBody) sender() string {
	if len(e.From) == 0 {
		return "(unknown sender)"
	}
	p := e.From[0]
	if p.Name == "" {
		return p.Address
	}
	return p.Name + " <" + p.Address + ">"
}

func (e EventBody) fromAddress() string {
	if len(e.From) == 0 {
		return ""
	}
	return e.From[0].Address
}

// Notice is one notification the notifier shows: its args for node-notifier.send, and the message its `shown`
// event names (the latest one, for a folded notice).
type Notice struct {
	Account string
	Args    map[string]any
	Handles []string // every message this notice accounts for, marked seen once shown
	Latest  EventBody
}

// FailureStatus is what the notifier knows of an account's notice still failing to send (review round 1 of
// issue 800): how many attempts, the last error, and since when. Kept in the Store, not only in memory, so
// the health tool — its own process, asked separately by the node-engine — sees the same account the live
// notifier does.
type FailureStatus struct {
	Count     int       `json:"count"`
	LastError string    `json:"last-error"`
	Since     time.Time `json:"since"`
}

// Store is what the notifier keeps of its own (the `shown` state the manifest declares): whether a message
// was shown already, each account's last notice, and each account's notice still failing to send.
type Store interface {
	HandleSeen(handle string) (bool, error)
	MarkHandle(handle string) error
	PutStatus(account string, s AccountStatus) error
	Statuses() (map[string]AccountStatus, error)
	PutFailure(account string, f FailureStatus) error
	ClearFailure(account string) error
	Failures() (map[string]FailureStatus, error)
}

// AccountStatus is what the notifier last showed of one account.
type AccountStatus struct {
	At      time.Time `json:"at"`
	Summary string    `json:"summary"`
}

// accountBuffer is one account's state: history is every qualifying mail's arrival time, kept only for the
// trailing Window, that N is counted against (review round 3 of issue 800); immediate holds mail waiting to be
// shown one by one at the next Flush; folded holds the account's current fold (its first FoldOver mails within
// Window are never folded, so a fold only ever holds the overflow from the fourth qualifying mail of its own
// window onward), due as one notice once Window has passed since foldStart, its first mail's arrival.
type accountBuffer struct {
	history   []time.Time
	foldStart time.Time // when the current fold's first mail arrived; zero while nothing is folded
	immediate []EventBody
	folded    []EventBody
}

// Notifier buffers qualifying mail per account and shows it: the first FoldOver mails of a window one by one,
// the rest folded once the window ends. A notice that fails to send is tried again, at most once a minute,
// without holding up any other account's notice.
type Notifier struct {
	// Send calls node-notifier.send on this machine.
	Send func(args map[string]any) error
	// Accounts reads the accounts the notifier is granted, by mail-store.accounts: each one's own address, for
	// the quiet rule (novox/hq ADR 0378 §9's LEAD note). An account not known there is shown anyway.
	Accounts func() (map[string]string, error)
	Store    Store
	// EmitShown says mesh-mail-notifier's own event `shown`: account and message id, never sender or subject.
	EmitShown func(account, messageID string) error
	Now       func() time.Time
	Logf      func(string, ...any)

	mu       sync.Mutex
	pending  map[string]*accountBuffer
	buffered map[string]bool      // a message's Handle while it is waiting to be shown: never counted twice
	cooldown map[string]time.Time // account -> not retried before this time, after a failed send

	accountsAt  time.Time
	accountsVal map[string]string
	accountsErr error
}

func (n *Notifier) init() {
	n.mu.Lock()
	defer n.mu.Unlock()
	if n.pending == nil {
		n.pending = map[string]*accountBuffer{}
		n.buffered = map[string]bool{}
		n.cooldown = map[string]time.Time{}
	}
	if n.Now == nil {
		n.Now = time.Now
	}
	if n.Logf == nil {
		n.Logf = func(string, ...any) {}
	}
}

// Add takes one added event. It returns an error only to have the event offered again (the handler
// Subscribe wraps this in follows that rule).
func (n *Notifier) Add(e EventBody) error {
	n.init()
	handle := e.Handle()
	seen, err := n.Store.HandleSeen(handle)
	if err != nil {
		return err
	}
	if seen {
		return nil // a redelivery of a message already shown (or already judged not to be)
	}
	own, known, err := n.ownAddress(e.Account)
	if err != nil {
		return err
	}
	if !n.qualifies(e, own, known) {
		return nil
	}

	n.mu.Lock()
	defer n.mu.Unlock()
	if n.buffered[handle] {
		return nil // a redelivery of a message still waiting to be shown: never counted twice
	}
	now := n.Now()
	b := n.pending[e.Account]
	if b == nil {
		b = &accountBuffer{}
		n.pending[e.Account] = b
	}
	// N is every qualifying mail of this account — shown or folded alike, each counted once — that arrived in
	// the 60 seconds up to and including now: a true sliding window over arrivals, not a per-session quota that
	// stays alive, and so keeps folding, for as long as any gap stays under Window (review round 3 of issue
	// 800, which replaces the round 1 and round 2 readings; nothing in this count depends on whether an earlier
	// mail has been shown yet).
	cutoff := now.Add(-Window)
	history := make([]time.Time, 0, len(b.history)+1)
	for _, at := range b.history {
		if !at.Before(cutoff) {
			history = append(history, at)
		}
	}
	b.history = append(history, now)
	n.buffered[handle] = true
	if len(b.history) <= FoldOver {
		b.immediate = append(b.immediate, e)
	} else {
		if len(b.folded) == 0 {
			b.foldStart = now
		}
		b.folded = append(b.folded, e)
	}
	return nil
}

// qualifies says whether a mail is shown at all: a new inbox mail, not the account's own, not already read.
func (n *Notifier) qualifies(e EventBody, ownAddress string, ownKnown bool) bool {
	if !strings.EqualFold(e.Folder, Inbox) {
		return false
	}
	if e.hasFlag(SeenFlag) {
		return false
	}
	if ownKnown && ownAddress != "" && strings.EqualFold(e.fromAddress(), ownAddress) {
		return false
	}
	return true
}

func (n *Notifier) ownAddress(account string) (string, bool, error) {
	accounts, err := n.readAccounts()
	if err != nil {
		// Not known: ADR 0378 §9's LEAD note says the notice is shown when the address cannot be read. Logged,
		// not retried: mail-store being unreachable is not this event's fault.
		n.Logf("[mesh-mail-notifier] the account's own address could not be read (%v); the notice is shown", err)
		return "", false, nil
	}
	addr, ok := accounts[account]
	return addr, ok, nil
}

// readAccounts asks mail-store.accounts, reusing its last answer for AccountsCacheFor so neither the quiet
// rule nor status reads it more than once a minute (review round 1 of issue 800, bullet 5).
func (n *Notifier) readAccounts() (map[string]string, error) {
	n.mu.Lock()
	if n.accountsVal != nil && n.Now().Sub(n.accountsAt) < AccountsCacheFor {
		val, err := n.accountsVal, n.accountsErr
		n.mu.Unlock()
		return val, err
	}
	n.mu.Unlock()

	val, err := n.Accounts()
	n.mu.Lock()
	n.accountsVal, n.accountsErr, n.accountsAt = val, err, n.Now()
	n.mu.Unlock()
	return val, err
}

// Flush shows what is ready: each account's first FoldOver mails of its window, one by one, as soon as Flush
// runs, and the window's folded overflow once the window has run its minute. Accounts are judged in a sorted
// order, not a map's iteration order, so the tests do not depend on it (review round 1 of issue 800, BLOCKING
// 1). A notice that fails to send leaves its mail pending, tried again once RetryBackoff has passed, and never
// holds up another account's notice (BLOCKING 2): nothing is deleted before it is shown.
func (n *Notifier) Flush() error {
	n.init()
	now := n.Now()

	n.mu.Lock()
	accounts := make([]string, 0, len(n.pending))
	for account := range n.pending {
		accounts = append(accounts, account)
	}
	n.mu.Unlock()
	sort.Strings(accounts)

	for _, account := range accounts {
		n.flushAccount(account, now)
	}
	return nil
}

// flushAccount ships one account's ready notices (its immediate mail, and its fold once Window has passed
// since the fold's own first mail), and keeps whatever fails to send pending, under cooldown, for the next
// Flush. Nothing is deleted before it is shown (review round 1 BLOCKING 2), and nothing pending is ever lost
// (review round 2 BLOCKING, kept under review round 3's rule; see Add).
func (n *Notifier) flushAccount(account string, now time.Time) {
	n.mu.Lock()
	if cd, ok := n.cooldown[account]; ok && now.Before(cd) {
		n.mu.Unlock()
		return // backing off after a failed send; tried again once RetryBackoff has passed
	}
	b := n.pending[account]
	if b == nil {
		n.mu.Unlock()
		return
	}
	immediate := append([]EventBody(nil), b.immediate...)
	// The fold is due Window after its own first mail (review round 3 of issue 800).
	foldDue := len(b.folded) > 0 && now.Sub(b.foldStart) >= Window
	var fold []EventBody
	if foldDue {
		fold = append([]EventBody(nil), b.folded...)
	}
	n.mu.Unlock()

	if len(immediate) == 0 && len(fold) == 0 {
		return
	}

	sentImmediate := 0
	var failErr error
	for _, e := range immediate {
		if err := n.show(n.singleNotice(account, e)); err != nil {
			failErr = err
			break
		}
		sentImmediate++
	}
	foldSent := false
	if failErr == nil && len(fold) > 0 {
		if err := n.show(n.foldNotice(account, fold, len(fold))); err != nil {
			failErr = err
		} else {
			foldSent = true
		}
	}

	n.mu.Lock()
	if cur := n.pending[account]; cur == b {
		cur.immediate = cur.immediate[sentImmediate:]
		if foldSent {
			cur.folded = cur.folded[len(fold):]
			if len(cur.folded) == 0 {
				cur.foldStart = time.Time{} // nothing left folded: a new fold may open (review round 3 of issue 800)
			} else {
				// More mail folded in while this batch was being sent: its own Window starts now, counted from
				// when it was first noticed. Nothing of it is lost either way.
				cur.foldStart = now
			}
		}
	}
	if failErr != nil {
		n.cooldown[account] = now.Add(RetryBackoff)
		n.recordFailureLocked(account, failErr, now)
	} else {
		delete(n.cooldown, account)
		n.clearFailureLocked(account)
	}
	n.mu.Unlock()
}

// recordFailureLocked keeps an account's failure in the Store (mu held by the caller, released around the
// Store calls since they may reach the bus): its count, its last error, and since when, so a later Flush and
// the health tool agree on how long it has been failing.
func (n *Notifier) recordFailureLocked(account string, sendErr error, now time.Time) {
	n.mu.Unlock()
	defer n.mu.Lock()
	since := now
	count := 1
	if prev, err := n.Store.Failures(); err == nil {
		if p, ok := prev[account]; ok {
			since = p.Since
			count = p.Count + 1
		}
	}
	if err := n.Store.PutFailure(account, FailureStatus{Count: count, LastError: sendErr.Error(), Since: since}); err != nil {
		n.Logf("[mesh-mail-notifier] a failure could not be kept (%v): the health tool may not see it", err)
	}
}

func (n *Notifier) clearFailureLocked(account string) {
	n.mu.Unlock()
	defer n.mu.Lock()
	if err := n.Store.ClearFailure(account); err != nil {
		n.Logf("[mesh-mail-notifier] a cleared failure could not be kept cleared: %v", err)
	}
}

// singleNotice is one mail shown on its own: one of an account's first FoldOver mails of its window.
func (n *Notifier) singleNotice(account string, e EventBody) Notice {
	return Notice{
		Account: account,
		Args: map[string]any{
			"summary":  e.Subject,
			"body":     e.sender() + " — " + account,
			"app_name": "mesh-mail",
		},
		Handles: []string{e.Handle()},
		Latest:  e,
	}
}

// foldNotice is one notice for everything the account's current fold holds — its qualifying mail from the
// fourth within some trailing 60 seconds onward — naming how many mail that fold holds (not the account's
// whole history of arrivals) and the latest sender (novox/hq ADR 0378 §9, review round 3 of issue 800, which
// replaces the round 1 and round 2 readings: "<k> more new mail(s) in <account>, latest from <sender of the
// last>", "mail" for one and "mails" for more).
func (n *Notifier) foldNotice(account string, events []EventBody, count int) Notice {
	latest := events[len(events)-1]
	notice := Notice{
		Account: account,
		Args: map[string]any{
			"summary":  strconv.Itoa(count) + " more new " + mailWord(count) + " in " + account,
			"body":     "latest from " + latest.sender(),
			"app_name": "mesh-mail",
		},
		Latest: latest,
	}
	for _, e := range events {
		notice.Handles = append(notice.Handles, e.Handle())
	}
	return notice
}

// mailWord says "mail" for a count of one, "mails" for anything else (review round 2 of issue 800, bullet 3;
// kept by review round 3).
func mailWord(count int) string {
	if count == 1 {
		return "mail"
	}
	return "mails"
}

// show sends one notice, marks every message it accounts for seen, keeps the account's status, and emits
// `shown`. Nothing is marked, kept or emitted when Send fails: the caller keeps the notice pending instead.
func (n *Notifier) show(notice Notice) error {
	if err := n.Send(notice.Args); err != nil {
		n.Logf("[mesh-mail-notifier] %s's notice could not be shown: %v", notice.Account, err)
		return err
	}
	n.mu.Lock()
	for _, h := range notice.Handles {
		delete(n.buffered, h)
	}
	n.mu.Unlock()
	for _, h := range notice.Handles {
		if err := n.Store.MarkHandle(h); err != nil {
			n.Logf("[mesh-mail-notifier] shown, but not kept as shown (%v): a redelivery may show it again", err)
		}
	}
	summary, _ := notice.Args["summary"].(string)
	if err := n.Store.PutStatus(notice.Account, AccountStatus{At: n.Now(), Summary: summary}); err != nil {
		n.Logf("[mesh-mail-notifier] shown, but the status could not be kept: %v", err)
	}
	return n.EmitShown(notice.Account, notice.Latest.MessageID)
}

// Status is the notifier's account of itself, for mail-notifier.status (novox/hq ADR 0378 §7: "accounts it
// announces there, when it last showed a notice"; review round 1 of issue 800: the accounts
// mail-store.accounts gives it, and the failures it is retrying).
func (n *Notifier) Status() (map[string]any, error) {
	statuses, err := n.Store.Statuses()
	if err != nil {
		return nil, err
	}
	failures, err := n.Store.Failures()
	if err != nil {
		return nil, err
	}
	known, knownErr := n.readAccounts()
	accounts := map[string]any{}
	for account := range known {
		accounts[account] = map[string]any{}
	}
	for account, s := range statuses {
		accounts[account] = map[string]any{"at": s.At.UTC().Format(time.RFC3339), "shown": s.Summary}
	}
	out := map[string]any{
		"accounts": accounts,
		"rules": []string{
			"it shows each new inbox mail at most once, through node-notifier.send on this machine (novox/hq ADR 0378 §7)",
			"quiet during a first sync, for mail from the account's own address, for mail already marked read, and for anything outside the inbox",
			"of one account's qualifying mail, one arriving when fewer than three of the account's mails fell in the 60 seconds before it is shown at once; the rest is folded and shown, as one notice naming how many it holds and the latest sender, 60 seconds after the fold's own first mail",
		},
	}
	if knownErr != nil {
		out["accounts-error"] = knownErr.Error()
	}
	if len(failures) > 0 {
		failing := map[string]any{}
		for account, f := range failures {
			failing[account] = map[string]any{
				"count":      f.Count,
				"last-error": f.LastError,
				"since":      f.Since.UTC().Format(time.RFC3339),
			}
		}
		out["failing"] = failing
	}
	return out, nil
}