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" "mesh-mail/cmd/mesh-mail/internal/secret" ) // 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 := secret.Read(k.SecretsDir, a.Secret()) 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, }} }