//go:build notmuch // +build notmuch package notmuch import ( "bufio" "bytes" "context" "errors" "fmt" "io" "net/url" "os" "os/exec" "path/filepath" "strconv" "strings" "time" "git.sr.ht/~rjarry/aerc/config" "git.sr.ht/~rjarry/aerc/log" "git.sr.ht/~rjarry/aerc/models" "git.sr.ht/~rjarry/aerc/worker/handlers" "git.sr.ht/~rjarry/aerc/worker/lib" notmuch "git.sr.ht/~rjarry/aerc/worker/notmuch/lib" "git.sr.ht/~rjarry/aerc/worker/types" "github.com/mitchellh/go-homedir" ) func init() { handlers.RegisterWorkerFactory("notmuch", NewWorker) } var errUnsupported = fmt.Errorf("unsupported command") const backgroundRefreshDelay = 1 * time.Minute type worker struct { w *types.Worker nmEvents chan eventType query string currentQueryName string queryMapOrder []string nameQueryMap map[string]string store *lib.MaildirStore db *notmuch.DB setupErr error currentSortCriteria []*types.SortCriterion } // NewWorker creates a new notmuch worker with the provided worker. func NewWorker(w *types.Worker) (types.Backend, error) { events := make(chan eventType, 20) return &worker{ w: w, nmEvents: events, }, nil } // Run starts the worker's message handling loop. func (w *worker) Run() { for { select { case action := <-w.w.Actions: msg := w.w.ProcessAction(action) if err := w.handleMessage(msg); errors.Is(err, errUnsupported) { w.w.PostMessage(&types.Unsupported{ Message: types.RespondTo(msg), }, nil) log.Errorf("ProcessAction(%T) unsupported: %v", msg, err) } else if err != nil { w.w.PostMessage(&types.Error{ Message: types.RespondTo(msg), Error: err, }, nil) log.Errorf("ProcessAction(%T) failure: %v", msg, err) } case nmEvent := <-w.nmEvents: err := w.handleNotmuchEvent(nmEvent) if err != nil { log.Errorf("notmuch event failure: %v", err) } } } } func (w *worker) done(msg types.WorkerMessage) { w.w.PostMessage(&types.Done{Message: types.RespondTo(msg)}, nil) } func (w *worker) err(msg types.WorkerMessage, err error) { w.w.PostMessage(&types.Error{ Message: types.RespondTo(msg), Error: err, }, nil) } func (w *worker) handleMessage(msg types.WorkerMessage) error { if w.setupErr != nil { // only configure can recover from a config error, bail for everything else _, isConfigure := msg.(*types.Configure) if !isConfigure { return w.setupErr } } switch msg := msg.(type) { case *types.Unsupported: // No-op case *types.Configure: return w.handleConfigure(msg) case *types.Connect: return w.handleConnect(msg) case *types.ListDirectories: return w.handleListDirectories(msg) case *types.OpenDirectory: return w.handleOpenDirectory(msg) case *types.FetchDirectoryContents: return w.handleFetchDirectoryContents(msg) case *types.FetchDirectoryThreaded: return w.handleFetchDirectoryThreaded(msg) case *types.FetchMessageHeaders: return w.handleFetchMessageHeaders(msg) case *types.FetchMessageBodyPart: return w.handleFetchMessageBodyPart(msg) case *types.FetchFullMessages: return w.handleFetchFullMessages(msg) case *types.FlagMessages: return w.handleFlagMessages(msg) case *types.AnsweredMessages: return w.handleAnsweredMessages(msg) case *types.SearchDirectory: return w.handleSearchDirectory(msg) case *types.ModifyLabels: return w.handleModifyLabels(msg) case *types.CheckMail: go w.handleCheckMail(msg) return nil case *types.DeleteMessages: return w.handleDeleteMessages(msg) case *types.CopyMessages: return w.handleCopyMessages(msg) case *types.MoveMessages: return w.handleMoveMessages(msg) case *types.AppendMessage: return w.handleAppendMessage(msg) case *types.CreateDirectory: return w.handleCreateDirectory(msg) case *types.RemoveDirectory: return w.handleRemoveDirectory(msg) } return errUnsupported } func (w *worker) handleConfigure(msg *types.Configure) error { var err error defer func() { if err == nil { w.setupErr = nil return } w.setupErr = fmt.Errorf("notmuch: %w", err) }() u, err := url.Parse(msg.Config.Source) if err != nil { log.Errorf("error configuring notmuch worker: %v", err) return err } home, err := homedir.Expand(u.Hostname()) if err != nil { return fmt.Errorf("could not resolve home directory: %w", err) } pathToDB := filepath.Join(home, u.Path) err = w.loadQueryMap(msg.Config) if err != nil { return fmt.Errorf("could not load query map configuration: %w", err) } excludedTags := w.loadExcludeTags(msg.Config) w.db = notmuch.NewDB(pathToDB, excludedTags) val, ok := msg.Config.Params["maildir-store"] if ok { path, err := homedir.Expand(val) if err != nil { return err } store, err := lib.NewMaildirStore(path, false) if err != nil { return fmt.Errorf("Cannot initialize maildir store: %w", err) } w.store = store } return nil } func (w *worker) handleConnect(msg *types.Connect) error { err := w.db.Connect() if err != nil { return err } w.done(msg) w.emitLabelList() go func() { defer log.PanicHandler() for { w.nmEvents <- &updateDirCounts{} time.Sleep(backgroundRefreshDelay) } }() return nil } func (w *worker) handleListDirectories(msg *types.ListDirectories) error { if w.store != nil { folders, err := w.store.FolderMap() if err != nil { log.Errorf("failed listing directories: %v", err) return err } for name := range folders { w.w.PostMessage(&types.Directory{ Message: types.RespondTo(msg), Dir: &models.Directory{ Name: name, Attributes: []string{}, }, }, nil) } } for _, name := range w.queryMapOrder { w.w.PostMessage(&types.Directory{ Message: types.RespondTo(msg), Dir: &models.Directory{ Name: name, Attributes: []string{}, }, }, nil) } w.done(msg) return nil } func (w *worker) getDirectoryInfo(name string, query string) *models.DirectoryInfo { dirInfo := &models.DirectoryInfo{ Name: name, Flags: []string{}, ReadOnly: false, // total messages Exists: 0, // new messages since mailbox was last opened Recent: 0, // total unread Unseen: 0, AccurateCounts: true, Caps: &models.Capabilities{ Sort: true, Thread: true, }, } count, err := w.db.QueryCountMessages(query) if err != nil { return dirInfo } dirInfo.Exists = count.Exists dirInfo.Unseen = count.Unread return dirInfo } func (w *worker) handleOpenDirectory(msg *types.OpenDirectory) error { log.Tracef("opening %s", msg.Directory) var isDynamicFolder bool q := "" if w.store != nil { folders, _ := w.store.FolderMap() dir, ok := folders[msg.Directory] if ok { q = fmt.Sprintf("folder:%s", strconv.Quote(msg.Directory)) if err := w.processNewMaildirFiles(string(dir)); err != nil { return err } } } if q == "" { var ok bool q, ok = w.nameQueryMap[msg.Directory] if !ok { q = msg.Directory isDynamicFolder = true } } w.query = q w.currentQueryName = msg.Directory w.w.PostMessage(&types.DirectoryInfo{ Info: w.getDirectoryInfo(msg.Directory, w.query), Message: types.RespondTo(msg), }, nil) if isDynamicFolder { w.w.PostMessage(&types.DirectoryInfo{ Info: w.getDirectoryInfo(msg.Directory, w.query), Message: types.RespondTo(msg), }, nil) } w.done(msg) return nil } func (w *worker) handleFetchDirectoryContents( msg *types.FetchDirectoryContents, ) error { w.currentSortCriteria = msg.SortCriteria err := w.emitDirectoryContents(msg) if err != nil { return err } w.done(msg) return nil } func (w *worker) handleFetchDirectoryThreaded( msg *types.FetchDirectoryThreaded, ) error { // w.currentSortCriteria = msg.SortCriteria err := w.emitDirectoryThreaded(msg) if err != nil { return err } w.done(msg) return nil } func (w *worker) handleFetchMessageHeaders( msg *types.FetchMessageHeaders, ) error { for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message: %v", err) w.w.PostMessageInfoError(msg, uid, err) continue } err = w.emitMessageInfo(m, msg) if err != nil { log.Errorf("could not emit message info: %v", err) w.w.PostMessageInfoError(msg, uid, err) continue } } w.done(msg) return nil } func (w *worker) uidsFromQuery(query string) ([]uint32, error) { msgIDs, err := w.db.MsgIDsFromQuery(query) if err != nil { return nil, err } var uids []uint32 for _, id := range msgIDs { uid := w.db.UidFromKey(id) uids = append(uids, uid) } return uids, nil } func (w *worker) msgFromUid(uid uint32) (*Message, error) { key, ok := w.db.KeyFromUid(uid) if !ok { return nil, fmt.Errorf("Invalid uid: %v", uid) } msg := &Message{ key: key, uid: uid, db: w.db, } return msg, nil } func (w *worker) handleFetchMessageBodyPart( msg *types.FetchMessageBodyPart, ) error { m, err := w.msgFromUid(msg.Uid) if err != nil { log.Errorf("could not get message %d: %v", msg.Uid, err) return err } r, err := m.NewBodyPartReader(msg.Part) if err != nil { log.Errorf( "could not get body part reader for message=%d, parts=%#v: %w", msg.Uid, msg.Part, err) return err } w.w.PostMessage(&types.MessageBodyPart{ Message: types.RespondTo(msg), Part: &models.MessageBodyPart{ Reader: r, Uid: msg.Uid, }, }, nil) w.done(msg) return nil } func (w *worker) handleFetchFullMessages(msg *types.FetchFullMessages) error { for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message %d: %v", uid, err) return err } r, err := m.NewReader() if err != nil { log.Errorf("could not get message reader: %v", err) return err } defer r.Close() b, err := io.ReadAll(r) if err != nil { return err } w.w.PostMessage(&types.FullMessage{ Message: types.RespondTo(msg), Content: &models.FullMessage{ Uid: uid, Reader: bytes.NewReader(b), }, }, nil) } w.done(msg) return nil } func (w *worker) handleAnsweredMessages(msg *types.AnsweredMessages) error { for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message: %v", err) w.err(msg, err) continue } if err := m.MarkAnswered(msg.Answered); err != nil { log.Errorf("could not mark message as answered: %v", err) w.err(msg, err) continue } err = w.emitMessageInfo(m, msg) if err != nil { log.Errorf("could not emit message info: %v", err) w.err(msg, err) continue } } w.w.PostMessage(&types.DirectoryInfo{ Info: w.getDirectoryInfo(w.currentQueryName, w.query), }, nil) w.done(msg) return nil } func (w *worker) handleFlagMessages(msg *types.FlagMessages) error { for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message: %v", err) w.err(msg, err) continue } if err := m.SetFlag(msg.Flags, msg.Enable); err != nil { log.Errorf("could not set flag %v as %t for message: %v", msg.Flags, msg.Enable, err) w.err(msg, err) continue } err = w.emitMessageInfo(m, msg) if err != nil { log.Errorf("could not emit message info: %v", err) w.err(msg, err) continue } } w.w.PostMessage(&types.DirectoryInfo{ Info: w.getDirectoryInfo(w.currentQueryName, w.query), }, nil) w.done(msg) return nil } func (w *worker) handleSearchDirectory(msg *types.SearchDirectory) error { // the first item is the command (search / filter) s := strings.Join(msg.Argv[1:], " ") // we only want to search in the current query, so merge the two together search := w.query if s != "" { search = fmt.Sprintf("(%v) and (%v)", w.query, s) } uids, err := w.uidsFromQuery(search) if err != nil { return err } w.w.PostMessage(&types.SearchResults{ Message: types.RespondTo(msg), Uids: uids, }, nil) return nil } func (w *worker) handleModifyLabels(msg *types.ModifyLabels) error { for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { return fmt.Errorf("could not get message from uid %d: %w", uid, err) } err = m.ModifyTags(msg.Add, msg.Remove) if err != nil { return fmt.Errorf("could not modify message tags: %w", err) } err = w.emitMessageInfo(m, msg) if err != nil { return err } } // tags changed, most probably some messages shifted to other folders // so we need to re-enumerate the query content err := w.emitDirectoryContents(msg) if err != nil { return err } // and update the list of possible tags w.emitLabelList() w.w.PostMessage(&types.DirectoryInfo{ Info: w.getDirectoryInfo(w.currentQueryName, w.query), }, nil) w.done(msg) return nil } func (w *worker) loadQueryMap(acctConfig *config.AccountConfig) error { raw, ok := acctConfig.Params["query-map"] if !ok { // nothing to do return nil } file, err := homedir.Expand(raw) if err != nil { return err } f, err := os.Open(file) if err != nil { return err } defer f.Close() w.nameQueryMap = make(map[string]string) scanner := bufio.NewScanner(f) for scanner.Scan() { line := scanner.Text() if line == "" || line[0] == '#' { continue } split := strings.SplitN(line, "=", 2) if len(split) != 2 { return fmt.Errorf("%v: invalid line %q, want name=query", file, line) } w.nameQueryMap[strings.TrimSpace(split[0])] = split[1] w.queryMapOrder = append(w.queryMapOrder, strings.TrimSpace(split[0])) } return nil } func (w *worker) loadExcludeTags( acctConfig *config.AccountConfig, ) []string { raw, ok := acctConfig.Params["exclude-tags"] if !ok { // nothing to do return nil } excludedTags := strings.Split(raw, ",") for idx, tag := range excludedTags { excludedTags[idx] = strings.Trim(tag, " ") } return excludedTags } func (w *worker) emitDirectoryContents(parent types.WorkerMessage) error { query := w.query if msg, ok := parent.(*types.FetchDirectoryContents); ok { s := strings.Join(msg.FilterCriteria[1:], " ") if s != "" { query = fmt.Sprintf("(%v) and (%v)", query, s) } } uids, err := w.uidsFromQuery(query) if err != nil { return fmt.Errorf("could not fetch uids: %w", err) } sortedUids, err := w.sort(uids, w.currentSortCriteria) if err != nil { log.Errorf("error sorting directory: %v", err) return err } w.w.PostMessage(&types.DirectoryContents{ Message: types.RespondTo(parent), Uids: sortedUids, }, nil) return nil } func (w *worker) emitDirectoryThreaded(parent types.WorkerMessage) error { query := w.query if msg, ok := parent.(*types.FetchDirectoryThreaded); ok { s := strings.Join(msg.FilterCriteria[1:], " ") if s != "" { query = fmt.Sprintf("(%v) and (%v)", query, s) } } threads, err := w.db.ThreadsFromQuery(query) if err != nil { return err } w.w.PostMessage(&types.DirectoryThreaded{ Threads: threads, }, nil) return nil } func (w *worker) emitMessageInfo(m *Message, parent types.WorkerMessage, ) error { info, err := m.MessageInfo() if err != nil { return fmt.Errorf("could not get MessageInfo: %w", err) } w.w.PostMessage(&types.MessageInfo{ Message: types.RespondTo(parent), Info: info, }, nil) return nil } func (w *worker) emitLabelList() { tags, err := w.db.ListTags() if err != nil { log.Errorf("could not load tags: %v", err) return } w.w.PostMessage(&types.LabelList{Labels: tags}, nil) } func (w *worker) sort(uids []uint32, criteria []*types.SortCriterion, ) ([]uint32, error) { if len(criteria) == 0 { return uids, nil } var msgInfos []*models.MessageInfo for _, uid := range uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message: %v", err) continue } info, err := m.MessageInfo() if err != nil { log.Errorf("could not get message info: %v", err) continue } msgInfos = append(msgInfos, info) } sortedUids, err := lib.Sort(msgInfos, criteria) if err != nil { log.Errorf("could not sort the messages: %v", err) return nil, err } return sortedUids, nil } func (w *worker) handleCheckMail(msg *types.CheckMail) { if msg.Command == "" { w.err(msg, fmt.Errorf("checkmail: no command specified")) return } ctx, cancel := context.WithTimeout(context.Background(), msg.Timeout) defer cancel() cmd := exec.CommandContext(ctx, "sh", "-c", msg.Command) ch := make(chan error) go func() { err := cmd.Run() ch <- err }() select { case <-ctx.Done(): w.err(msg, fmt.Errorf("checkmail: timed out")) case err := <-ch: if err != nil { w.err(msg, fmt.Errorf("checkmail: error running command: %w", err)) } else { w.w.PostMessage(&types.DirectoryInfo{ Info: w.getDirectoryInfo(w.currentQueryName, w.query), }, nil) w.done(msg) } } } func (w *worker) handleDeleteMessages(msg *types.DeleteMessages) error { if w.store == nil { return errUnsupported } var deleted []uint32 // With notmuch, two identical files can be referenced under // the same index key, even if they exist in two different // folders. So in order to remove the message from the right // maildir folder we need to pass a hint to Remove() so it // can purge the right file. folders, _ := w.store.FolderMap() path, ok := folders[w.currentQueryName] if !ok { w.err(msg, fmt.Errorf("Can only delete file from a maildir folder")) w.done(msg) return nil } for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message: %v", err) w.err(msg, err) continue } if err := m.Remove(path); err != nil { log.Errorf("could not remove message: %v", err) w.err(msg, err) continue } deleted = append(deleted, uid) } if len(deleted) > 0 { w.w.PostMessage(&types.MessagesDeleted{ Message: types.RespondTo(msg), Uids: deleted, }, nil) } w.done(msg) return nil } func (w *worker) handleCopyMessages(msg *types.CopyMessages) error { if w.store == nil { return errUnsupported } // Only allow file to be copied to a maildir folder folders, _ := w.store.FolderMap() dest, ok := folders[msg.Destination] if !ok { return fmt.Errorf("Can only copy file to a maildir folder") } for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message: %v", err) return err } if err := m.Copy(dest); err != nil { log.Errorf("could not copy message: %v", err) return err } } w.w.PostMessage(&types.MessagesCopied{ Message: types.RespondTo(msg), Destination: msg.Destination, Uids: msg.Uids, }, nil) w.done(msg) return nil } func (w *worker) handleMoveMessages(msg *types.MoveMessages) error { if w.store == nil { return errUnsupported } var moved []uint32 folders, _ := w.store.FolderMap() // Only allow file to be moved to a maildir folder dest, ok := folders[msg.Destination] if !ok { return fmt.Errorf("Can only move file to a maildir folder") } var err error for _, uid := range msg.Uids { m, err := w.msgFromUid(uid) if err != nil { log.Errorf("could not get message: %v", err) break } filenames, err := m.db.MsgFilenames(m.key) if err != nil { return err } // In the future, it'd be nice if we could overload move with // the possibility to affect some or all of the files // corresponding to a message. if len(filenames) > 1 { return fmt.Errorf("Cannot move: message %d has multiple files", m.uid) } source, key := parseFilename(filenames[0]) if key == "" { return fmt.Errorf("failed to parse message filename: %s", filenames[0]) } if err := m.Move(source, dest); err != nil { log.Errorf("could not copy message: %v", err) break } moved = append(moved, uid) } w.w.PostMessage(&types.MessagesDeleted{ Message: types.RespondTo(msg), Uids: moved, }, nil) if err == nil { w.done(msg) } return err } func (w *worker) handleAppendMessage(msg *types.AppendMessage) error { if w.store == nil { return errUnsupported } // Only allow file to be created in a maildir folder // since we are the "master" maildir process, we can modify the maildir directly folders, _ := w.store.FolderMap() dest, ok := folders[msg.Destination] if !ok { return fmt.Errorf("Can only create file in a maildir folder") } key, writer, err := dest.Create(lib.ToMaildirFlags(msg.Flags)) if err != nil { log.Errorf("could not create message at %s: %v", msg.Destination, err) return err } filename, err := dest.Filename(key) if err != nil { writer.Close() return err } if _, err := io.Copy(writer, msg.Reader); err != nil { log.Errorf("could not write message to destination: %v", err) writer.Close() os.Remove(filename) return err } writer.Close() if _, err := w.db.IndexFile(filename); err != nil { return err } w.w.PostMessage(&types.DirectoryInfo{ Info: w.getDirectoryInfo(w.currentQueryName, w.query), }, nil) w.done(msg) return nil } func (w *worker) handleCreateDirectory(msg *types.CreateDirectory) error { if w.store == nil { return errUnsupported } dir := w.store.Dir(msg.Directory) if err := dir.Init(); err != nil { log.Errorf("could not create directory %s: %v", msg.Directory, err) return err } w.done(msg) return nil } func (w *worker) handleRemoveDirectory(msg *types.RemoveDirectory) error { if w.store == nil { return errUnsupported } dir := w.store.Dir(msg.Directory) if err := os.RemoveAll(string(dir)); err != nil { log.Errorf("could not remove directory %s: %v", msg.Directory, err) return err } w.done(msg) return nil } // This is a hack that calls MsgModifyTags with an empty list of tags to // apply on new messages causing notmuch to rename files and effectively // move them into the cur/ dir. func (w *worker) processNewMaildirFiles(dir string) error { f, err := os.Open(filepath.Join(dir, "new")) if err != nil { return err } defer f.Close() names, err := f.Readdirnames(0) if err != nil { return err } for _, n := range names { if n[0] == '.' { continue } key, err := w.db.MsgIDFromFilename(filepath.Join(dir, "new", n)) if err != nil { // Message is not yet indexed, leave it alone continue } // Force message to move from new/ to cur/ err = w.db.MsgModifyTags(key, nil, nil) if err != nil { log.Errorf("MsgModifyTags failed: %v", err) } } return nil }