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

package main

// The adapter for an account of the kind `imap` (novox/hq ADR 0359 §1, §2, §5). For each account it keeps one
// connection: IDLE on the inbox, and every PollEvery a look at every other folder. Each look compares what the server
// holds with the index and says each difference as an event: a message added, its flags changed, or removed. The first
// look at a folder indexes what is there without saying it — what the bus is told is what changed since the mesh
// started watching, never the history of a mailbox.
//
// **It only reads.** Every folder is opened read-only (EXAMINE), and every message is fetched with BODY.PEEK, so
// watching never marks a message seen. The client is used through reader alone, which has no command that flags,
// moves, copies, deletes, appends or sends (§5); a test holds the code to that.

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/emersion/go-imap/v2"
	"github.com/emersion/go-imap/v2/imapclient"
)

// How the keeper paces itself. IdleCycle is how long one IDLE lasts before it is ended and begun again, which proves
// the server still answers; SilentAfter is how long without an answer makes the account unhealthy (§6).
const (
	IdleCycle   = 2 * time.Minute
	PollEvery   = 5 * time.Minute
	SilentAfter = 5 * time.Minute
	RetryFirst  = 30 * time.Second
	RetryMost   = 5 * time.Minute
	FetchBatch  = 25
	// BatchBytes bounds the mail one fetch of bodies holds; SizeBatch is how many sizes one fetch asks.
	BatchBytes = 64 << 20
	SizeBatch  = 500
)

// Batches cuts the uids, in order, into batches of at most most messages whose sizes add up to at most limit bytes. A
// message larger than limit is a batch of its own. A size not known counts as nothing: the server listed the message,
// and the fetch says what it holds.
func Batches(uids []uint32, sizes map[uint32]int64, limit int64, most int) [][]uint32 {
	var out [][]uint32
	var cur []uint32
	var bytes int64
	for _, u := range uids {
		size := sizes[u]
		if len(cur) > 0 && (len(cur) >= most || bytes+size > limit) {
			out = append(out, cur)
			cur, bytes = nil, 0
		}
		cur = append(cur, u)
		bytes += size
	}
	if len(cur) > 0 {
		out = append(out, cur)
	}
	return out
}

func uidSet(uids []uint32) imap.UIDSet {
	set := make([]imap.UID, len(uids))
	for i, u := range uids {
		set[i] = imap.UID(u)
	}
	return imap.UIDSetNum(set...)
}

func orBytes(n, def int64) int64 {
	if n > 0 {
		return n
	}
	return def
}

// reader is the part of an IMAP client the keeper uses: nothing in it changes a mailbox.
type reader interface {
	Login(username, password string) *imapclient.Command
	List(ref, pattern string, options *imap.ListOptions) *imapclient.ListCommand
	Select(mailbox string, options *imap.SelectOptions) *imapclient.SelectCommand
	Fetch(numSet imap.NumSet, options *imap.FetchOptions) *imapclient.FetchCommand
	Idle() (*imapclient.IdleCommand, error)
	Caps() imap.CapSet
	Closed() <-chan struct{}
	Logout() *imapclient.Command
	Close() error
}

// Dialer opens a connection to an account's server, handing what the server says unasked to the handler.
type Dialer func(a Account, handler *imapclient.UnilateralDataHandler) (reader, error)

// DialServer is the dialer for the mesh: TLS from the first byte, or STARTTLS before signing in, as the account says.
func DialServer(a Account, handler *imapclient.UnilateralDataHandler) (reader, error) {
	opts := &imapclient.Options{UnilateralDataHandler: handler, TLSConfig: &tls.Config{ServerName: a.Incoming.Host, MinVersion: tls.VersionTLS12}}
	var c *imapclient.Client
	var err error
	switch a.Incoming.Security {
	case "tls":
		c, err = imapclient.DialTLS(a.Incoming.Address(), opts)
	case "starttls":
		c, err = imapclient.DialStartTLS(a.Incoming.Address(), opts)
	default:
		err = fmt.Errorf("the account %s's security %q signs in over nothing this version speaks", a.Name, a.Incoming.Security)
	}
	if err != nil {
		return nil, err
	}
	return c, nil
}

// Keeper watches the accounts and keeps the index.
type Keeper struct {
	Store      Store
	Objects    Objects
	Dial       Dialer
	SecretsDir string
	Logf       func(format string, a ...any)
	Now        func() time.Time

	// Paces, for the tests; the constants above when zero.
	IdleCycle, PollEvery, RetryFirst time.Duration
	// BatchBytes, for the tests; the constant when zero.
	BatchBytes int64
}

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

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

func orDefault(d, def time.Duration) time.Duration {
	if d > 0 {
		return d
	}
	return def
}

// Run watches every account until ctx ends.
func (k *Keeper) Run(ctx context.Context, accounts []Account) {
	var wg sync.WaitGroup
	for _, a := range accounts {
		wg.Add(1)
		go func(a Account) {
			defer wg.Done()
			k.Watch(ctx, a)
		}(a)
	}
	wg.Wait()
}

func (k *Keeper) say(ctx context.Context, a Account, state, why string, heard, synced time.Time) {
	if err := k.Store.SetState(ctx, AccountState{Account: a.Name, State: state, Why: why, Heard: heard, Synced: synced, Said: k.now()}); err != nil {
		k.logf("[mesh-mail] the account %s's state could not be kept: %v", a.Name, err)
	}
}

// Watch keeps one account until ctx ends: waits while its secret is not given, signs in, watches, and on any fault
// says it, waits a little longer each time, and begins again.
func (k *Keeper) Watch(ctx context.Context, a Account) {
	retry := orDefault(k.RetryFirst, RetryFirst)
	var heard, synced time.Time
	for ctx.Err() == nil {
		pw, given, err := ReadSecret(k.SecretsDir, a)
		switch {
		case err != nil:
			k.say(ctx, a, AccountFailing, "its password was given and cannot be read: "+err.Error(), heard, synced)
		case !given:
			k.say(ctx, a, AccountWaits, "its password ("+a.Secret()+") has not been given", heard, synced)
		default:
			s := &session{k: k, a: a, wake: make(chan struct{}, 1), heard: heard, synced: synced}
			err = s.run(ctx, pw)
			heard, synced = s.heard, s.synced
			if ctx.Err() != nil {
				return
			}
			k.say(ctx, a, AccountFailing, err.Error(), heard, synced)
			k.logf("[mesh-mail] the account %s: %v; trying again in %s", a.Name, err, retry)
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(retry):
		}
		if given && err != nil {
			retry = min(retry*2, RetryMost)
		} else {
			retry = orDefault(k.RetryFirst, RetryFirst)
		}
	}
}

// session is one connection to one account's server.
type session struct {
	k      *Keeper
	a      Account
	c      reader
	wake   chan struct{}
	heard  time.Time
	synced time.Time
	// state and why are what the session last said of the account, and told when it said it.
	state, why string
	told       time.Time
}

// answered is the server having answered: kept, and told at least once a minute, so a long first look at a large
// mailbox still shows the account alive to the health watch.
func (s *session) answered() {
	s.heard = s.k.now()
	if s.state != "" && s.heard.Sub(s.told) >= time.Minute {
		s.tell(s.state, s.why)
	}
}

// tell says the account's state.
func (s *session) tell(state, why string) {
	s.state, s.why, s.told = state, why, s.k.now()
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	s.k.say(ctx, s.a, state, why, s.heard, s.synced)
}

func (s *session) poke() {
	select {
	case s.wake <- struct{}{}:
	default:
	}
}

func (s *session) run(ctx context.Context, password string) error {
	handler := &imapclient.UnilateralDataHandler{
		Expunge: func(uint32) { s.poke() },
		Mailbox: func(*imapclient.UnilateralDataMailbox) { s.poke() },
		Fetch:   func(*imapclient.FetchMessageData) { s.poke() },
	}
	c, err := s.k.Dial(s.a, handler)
	if err != nil {
		return fmt.Errorf("its server %s cannot be reached: %v", s.a.Incoming.Host, err)
	}
	s.c = c
	defer c.Close()
	if err := c.Login(s.a.SignInAs(), password).Wait(); err != nil {
		var said *imap.Error
		if errors.As(err, &said) {
			return fmt.Errorf("its server refused the sign-in (%s): its password may be wrong, and is given again at the desk", strings.TrimSpace(said.Text))
		}
		return fmt.Errorf("signing in failed: %v", err)
	}
	s.answered()
	s.tell(AccountSyncing, "signed in; looking at its folders")
	folders, err := s.folders()
	if err != nil {
		return err
	}
	idles := c.Caps().Has(imap.CapIdle) || c.Caps().Has(imap.CapIMAP4rev2)
	var polled time.Time
	for ctx.Err() == nil {
		if polled.IsZero() || s.k.now().Sub(polled) >= orDefault(s.k.PollEvery, PollEvery) {
			for _, f := range folders {
				if err := s.k.syncFolder(ctx, c, s.a, f, s.answered); err != nil {
					return err
				}
			}
			polled = s.k.now()
		}
		// The inbox last, so it is the folder selected while the session idles.
		if err := s.k.syncFolder(ctx, c, s.a, "INBOX", s.answered); err != nil {
			return err
		}
		s.synced = s.k.now()
		s.tell(AccountCurrent, fmt.Sprintf("watching the inbox and %d other folder(s)", len(folders)))
		if err := s.wait(ctx, idles); err != nil {
			return err
		}
	}
	_ = c.Logout().Wait()
	return ctx.Err()
}

// wait is one IDLE on the inbox, ended by what the server says unasked, by the cycle's end or by ctx. Ending it waits
// for the server's answer, which is what proves the connection alive.
func (s *session) wait(ctx context.Context, idles bool) error {
	cycle := time.NewTimer(orDefault(s.k.IdleCycle, IdleCycle))
	defer cycle.Stop()
	if !idles {
		select {
		case <-ctx.Done():
		case <-cycle.C:
		case <-s.c.Closed():
			return errors.New("its server closed the connection")
		}
		return nil
	}
	idle, err := s.c.Idle()
	if err != nil {
		return fmt.Errorf("its server refused to IDLE on the inbox: %v", err)
	}
	select {
	case <-ctx.Done():
	case <-s.wake:
	case <-cycle.C:
	case <-s.c.Closed():
	}
	if err := idle.Close(); err != nil {
		return fmt.Errorf("its IDLE on the inbox ended badly: %v", err)
	}
	if err := idle.Wait(); err != nil {
		return fmt.Errorf("its IDLE on the inbox ended badly: %v", err)
	}
	s.answered()
	return nil
}

// folders is what is watched besides the inbox: the account's list, or every folder the server can open.
func (s *session) folders() ([]string, error) {
	if len(s.a.Folders) > 0 {
		return append([]string(nil), s.a.Folders...), nil
	}
	listed, err := s.c.List("", "*", nil).Collect()
	if err != nil {
		return nil, fmt.Errorf("its folders cannot be listed: %v", err)
	}
	s.answered()
	var out []string
	for _, l := range listed {
		if strings.EqualFold(l.Mailbox, "INBOX") || hasAttr(l.Attrs, imap.MailboxAttrNoSelect) || hasAttr(l.Attrs, imap.MailboxAttrNonExistent) {
			continue
		}
		out = append(out, l.Mailbox)
	}
	sort.Strings(out)
	return out, nil
}

func hasAttr(attrs []imap.MailboxAttr, want imap.MailboxAttr) bool {
	for _, a := range attrs {
		if strings.EqualFold(string(a), string(want)) {
			return true
		}
	}
	return false
}

// syncFolder brings the index of one folder to what the server holds, and says each difference, unless this is the
// first look at the folder.
func (k *Keeper) syncFolder(ctx context.Context, c reader, a Account, folder string, answered func()) error {
	sel, err := c.Select(folder, &imap.SelectOptions{ReadOnly: true}).Wait()
	if err != nil {
		return fmt.Errorf("the folder %s cannot be opened: %v", folder, err)
	}
	answered()
	known, seen, err := k.Store.Folder(ctx, a.Name, folder)
	if err != nil {
		return err
	}
	if seen && known.UIDValidity != sel.UIDValidity {
		k.logf("[mesh-mail] the account %s's folder %s was renumbered by its server; it is indexed afresh", a.Name, folder)
		if err := k.Store.DropFolder(ctx, a.Name, folder); err != nil {
			return err
		}
		seen = false
	}
	server := map[uint32][]string{}
	if sel.NumMessages > 0 {
		all, err := c.Fetch(imap.UIDSet{{Start: 1, Stop: 0}}, &imap.FetchOptions{UID: true, Flags: true}).Collect()
		if err != nil {
			return fmt.Errorf("the folder %s's messages cannot be listed: %v", folder, err)
		}
		answered()
		for _, m := range all {
			server[uint32(m.UID)] = flagStrings(m.Flags)
		}
	}
	index, err := k.Store.Flags(ctx, a.Name, folder)
	if err != nil {
		return err
	}
	say := func(e *Event) *Event {
		if seen {
			return e
		}
		return nil
	}
	// Removed: in the index, no longer on the server.
	for uid := range index {
		if _, still := server[uid]; still {
			continue
		}
		m, _, err := k.Store.Get(ctx, a.Name, folder, uid)
		if err != nil {
			return err
		}
		if m.UID == 0 {
			m = Message{Account: a.Name, Folder: folder, UID: uid, UIDValidity: sel.UIDValidity}
		}
		if err := k.Store.Remove(ctx, a.Name, folder, uid, say(removedEvent(m))); err != nil {
			return err
		}
	}
	// Changed: on both, with other flags.
	var added []uint32
	for uid, flags := range server {
		before, indexed := index[uid]
		if !indexed {
			added = append(added, uid)
			continue
		}
		if sameStrings(before, flags) {
			continue
		}
		m, _, err := k.Store.Get(ctx, a.Name, folder, uid)
		if err != nil {
			return err
		}
		m.Flags = flags
		if err := k.Store.SetFlags(ctx, a.Name, folder, uid, flags, say(changedEvent(m, before, flags))); err != nil {
			return err
		}
	}
	// Added: on the server, not yet in the index. Their sizes first, then their bodies in batches bounded by bytes as
	// well as by count, so one batch never holds more than BatchBytes of mail in memory; a message larger than that
	// is fetched alone, and said. Fetched without marking them seen.
	sort.Slice(added, func(i, j int) bool { return added[i] < added[j] })
	sizes := map[uint32]int64{}
	for start := 0; start < len(added); start += SizeBatch {
		part := added[start:min(start+SizeBatch, len(added))]
		got, err := c.Fetch(uidSet(part), &imap.FetchOptions{UID: true, RFC822Size: true}).Collect()
		if err != nil {
			return fmt.Errorf("the folder %s's new messages cannot be measured: %v", folder, err)
		}
		answered()
		for _, m := range got {
			sizes[uint32(m.UID)] = m.RFC822Size
		}
	}
	cap := orBytes(k.BatchBytes, BatchBytes)
	for _, batch := range Batches(added, sizes, cap, FetchBatch) {
		if len(batch) == 1 && sizes[batch[0]] > cap {
			k.logf("[mesh-mail] the account %s's message %d in %s is %d bytes, more than a batch holds; it is fetched alone",
				a.Name, batch[0], folder, sizes[batch[0]])
		}
		msgs, err := c.Fetch(uidSet(batch), &imap.FetchOptions{
			UID: true, Flags: true, Envelope: true, RFC822Size: true, InternalDate: true,
			BodySection: []*imap.FetchItemBodySection{{Peek: true}},
		}).Collect()
		if err != nil {
			return fmt.Errorf("the folder %s's new messages cannot be fetched: %v", folder, err)
		}
		answered()
		for _, fm := range msgs {
			m, err := k.index(ctx, a, folder, sel.UIDValidity, fm)
			if err != nil {
				return err
			}
			if err := k.Store.Add(ctx, m, say(addedEvent(m))); err != nil {
				return err
			}
		}
	}
	return k.Store.SetFolder(ctx, Folder{Account: a.Name, Name: folder, UIDValidity: sel.UIDValidity, Messages: len(server), Synced: k.now()})
}

// index is a fetched message as the index keeps it: its metadata, its text, and its attachments kept in the store.
func (k *Keeper) index(ctx context.Context, a Account, folder string, validity uint32, fm *imapclient.FetchMessageBuffer) (Message, error) {
	m := Message{Account: a.Name, Folder: folder, UIDValidity: validity, UID: uint32(fm.UID), Size: fm.RFC822Size,
		Flags: flagStrings(fm.Flags), Date: fm.InternalDate.UTC(), From: []Person{}, To: []Person{}, Attachments: []AttachmentRef{}}
	if env := fm.Envelope; env != nil {
		m.MessageID, m.Subject = env.MessageID, env.Subject
		if len(env.InReplyTo) > 0 {
			m.InReplyTo = env.InReplyTo[0]
		}
		if !env.Date.IsZero() {
			m.Date = env.Date.UTC()
		}
		m.From, m.To, m.Cc = people(env.From), people(env.To), people(env.Cc)
	}
	var raw []byte
	if len(fm.BodySection) > 0 {
		raw = fm.BodySection[0].Bytes
	}
	parsed, err := ParseMessage(raw)
	if err != nil {
		k.logf("[mesh-mail] the account %s's message %d in %s is read in part: %v", a.Name, m.UID, folder, err)
	}
	m.Body = parsed.Text
	for _, p := range parsed.Attachments {
		ref, err := Keep(ctx, k.Objects, p.Data, p.Type, p.Name)
		if err != nil {
			return m, err
		}
		m.Attachments = append(m.Attachments, ref)
	}
	return m, nil
}

func people(as []imap.Address) []Person {
	out := []Person{}
	for _, a := range as {
		if addr := a.Addr(); addr != "" {
			out = append(out, Person{Name: a.Name, Address: addr})
		}
	}
	return out
}

func flagStrings(fs []imap.Flag) []string {
	out := make([]string, 0, len(fs))
	for _, f := range fs {
		if strings.EqualFold(string(f), `\Recent`) {
			continue
		}
		out = append(out, string(f))
	}
	sort.Strings(out)
	return out
}

func sameStrings(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

// ---- the events ---------------------------------------------------------------------------------------------------
// EventKey and Changes, the event's shape, now live in mailevent.go of their own (issue 800 part A).

// metadata is what an event says of a message: never its body, never an attachment's bytes (ADR 0359 §2).
func metadata(m Message) map[string]any {
	atts := make([]map[string]any, 0, len(m.Attachments))
	for _, r := range m.Attachments {
		atts = append(atts, map[string]any{"bucket": r.Bucket, "key": r.Key, "size": r.Size, "type": r.Type, "name": r.Name})
	}
	return map[string]any{
		"account": m.Account, "folder": m.Folder, "uid": m.UID, "uidvalidity": m.UIDValidity,
		"message-id": m.MessageID, "in-reply-to": m.InReplyTo, "date": m.Date.UTC().Format(time.RFC3339),
		"from": m.From, "to": m.To, "cc": m.Cc, "subject": m.Subject, "size": m.Size, "flags": m.Flags,
		"attachments": atts,
	}
}

func addedEvent(m Message) *Event {
	return &Event{Key: EventKey(m.Account, "added"), Body: metadata(m)}
}

func changedEvent(m Message, before, after []string) *Event {
	body := metadata(m)
	body["changes"] = map[string]any{"flags": map[string]any{"before": before, "after": after}}
	return &Event{Key: EventKey(m.Account, "changed"), Body: body}
}

func removedEvent(m Message) *Event {
	return &Event{Key: EventKey(m.Account, "removed"), Body: map[string]any{
		"account": m.Account, "folder": m.Folder, "uid": m.UID, "uidvalidity": m.UIDValidity, "message-id": m.MessageID,
	}}
}