package main import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "log/slog" "net/http" "os" "os/exec" "path/filepath" "strings" ) const ( jmapSessionURL = "https://api.fastmail.com/jmap/session" capMail = "urn:ietf:params:jmap:mail" capCore = "urn:ietf:params:jmap:core" ) type jmapSession struct { PrimaryAccounts map[string]string `json:"primaryAccounts"` EventSourceUrl string `json:"eventSourceUrl"` DownloadUrl string `json:"downloadUrl"` APIUrl string `json:"apiUrl"` } type stateChange struct { Changed map[string]map[string]string `json:"changed"` } type jmapRequest struct { Using []string `json:"using"` MethodCalls [][]interface{} `json:"methodCalls"` } type jmapResponse struct { MethodResponses [][]json.RawMessage `json:"methodResponses"` } type emailGetResult struct { State string `json:"state"` List []email `json:"list"` } type email struct { ID string `json:"id"` BlobID string `json:"blobId"` MailboxIDs map[string]bool `json:"mailboxIds"` } type mailboxGetResult struct { List []mailbox `json:"list"` } type mailbox struct { ID string `json:"id"` Name string `json:"name"` Role string `json:"role"` } func watch(ctx context.Context, a AccountConfig) error { auth, err := buildAuth(a) if err != nil { return fmt.Errorf("auth: %w", err) } sess, err := fetchSession(ctx, auth) if err != nil { return fmt.Errorf("session: %w", err) } accountID := sess.PrimaryAccounts[capMail] if accountID == "" { return fmt.Errorf("no primary mail account in JMAP session") } statePath := accountStatePath(a.Name) saved, err := loadAccountState(statePath) if err != nil { slog.Warn("could not load state, starting fresh", "account", a.Name, "err", err) } var emailState string if saved.EmailState != "" && saved.AccountID == accountID { slog.Info("catching up from saved state", "account", a.Name, "since", saved.EmailState) catchupState, n, catchupErr := fetchAndDeliver(a, sess, auth, accountID, saved.EmailState) if catchupErr != nil { slog.Warn("state too stale (cannotCalculateChanges), falling back to mbsync", "account", a.Name, "err", catchupErr) runMbsync(a) emailState, err = getEmailState(sess, auth, accountID) if err != nil { return fmt.Errorf("email state after mbsync fallback: %w", err) } } else { emailState = catchupState if n > 0 { slog.Info("caught up", "account", a.Name, "count", n) runNotmuchNew() } } } else { slog.Info("no saved state, running mbsync for initial sync", "account", a.Name) runMbsync(a) emailState, err = getEmailState(sess, auth, accountID) if err != nil { return fmt.Errorf("initial email state: %w", err) } } if err := saveAccountState(statePath, accountState{AccountID: accountID, EmailState: emailState}); err != nil { slog.Warn("save state", "account", a.Name, "err", err) } esURL := strings.NewReplacer( "{types}", "Email", "{closeafter}", "no", "{ping}", "60", ).Replace(sess.EventSourceUrl) req, err := http.NewRequestWithContext(ctx, "GET", esURL, nil) if err != nil { return err } req.Header.Set("Authorization", auth) req.Header.Set("Accept", "text/event-stream") req.Header.Set("Cache-Control", "no-cache") resp, err := http.DefaultClient.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return fmt.Errorf("EventSource HTTP %d: %s", resp.StatusCode, b) } slog.Info("connected to push stream", "account", a.Name, "state", emailState) scanner := bufio.NewScanner(resp.Body) var evType, evData string for scanner.Scan() { line := scanner.Text() switch { case strings.HasPrefix(line, "event:"): evType = strings.TrimSpace(strings.TrimPrefix(line, "event:")) case strings.HasPrefix(line, "data:"): evData = strings.TrimSpace(strings.TrimPrefix(line, "data:")) case line == "" && evData != "": if evType == "state" { newState, n, err := handleStateEvent(a, sess, auth, accountID, emailState, evData) if err != nil { slog.Error("state event", "account", a.Name, "err", err) } else { emailState = newState if n > 0 { runNotmuchNew() } if err := saveAccountState(statePath, accountState{AccountID: accountID, EmailState: emailState}); err != nil { slog.Warn("save state", "account", a.Name, "err", err) } } } evType, evData = "", "" } } return scanner.Err() } func handleStateEvent(a AccountConfig, sess *jmapSession, auth, accountID, currentState, rawData string) (newState string, delivered int, err error) { var sc stateChange if err := json.Unmarshal([]byte(rawData), &sc); err != nil { return currentState, 0, fmt.Errorf("parse StateChange: %w", err) } changes, ok := sc.Changed[accountID] if !ok { return currentState, 0, nil } newEmailState, ok := changes["Email"] if !ok || newEmailState == currentState { return currentState, 0, nil } slog.Info("email state changed", "account", a.Name, "from", currentState, "to", newEmailState) deliveredState, n, err := fetchAndDeliver(a, sess, auth, accountID, currentState) if err != nil { return currentState, 0, err } return deliveredState, n, nil } func fetchAndDeliver(a AccountConfig, sess *jmapSession, auth, accountID, sinceState string) (newState string, delivered int, err error) { resp, err := callJMAP(sess.APIUrl, auth, jmapRequest{ Using: []string{capCore, capMail}, MethodCalls: [][]interface{}{ {"Email/changes", map[string]interface{}{ "accountId": accountID, "sinceState": sinceState, }, "a"}, {"Email/get", map[string]interface{}{ "accountId": accountID, "#ids": map[string]interface{}{ "resultOf": "a", "name": "Email/changes", "path": "/created", }, "properties": []string{"id", "blobId", "mailboxIds"}, }, "b"}, {"Mailbox/get", map[string]interface{}{ "accountId": accountID, "ids": nil, "properties": []string{"id", "name", "role"}, }, "c"}, }, }) if err != nil { return sinceState, 0, err } if len(resp.MethodResponses) < 3 { return sinceState, 0, nil } // Surface Email/changes errors (e.g. cannotCalculateChanges). var methodName string if err := json.Unmarshal(resp.MethodResponses[0][0], &methodName); err == nil && methodName == "error" { var apiErr struct{ Type string `json:"type"` } json.Unmarshal(resp.MethodResponses[0][1], &apiErr) return sinceState, 0, fmt.Errorf("Email/changes: %s", apiErr.Type) } var changesResult struct { NewState string `json:"newState"` } if err := json.Unmarshal(resp.MethodResponses[0][1], &changesResult); err != nil { return sinceState, 0, fmt.Errorf("parse Email/changes response: %w", err) } newState = changesResult.NewState var egr emailGetResult if err := json.Unmarshal(resp.MethodResponses[1][1], &egr); err != nil { return newState, 0, fmt.Errorf("parse Email/get response: %w", err) } var mbr mailboxGetResult if err := json.Unmarshal(resp.MethodResponses[2][1], &mbr); err != nil { return newState, 0, fmt.Errorf("parse Mailbox/get response: %w", err) } folderName := make(map[string]string, len(mbr.List)) for _, mb := range mbr.List { if mb.Role == "all" { continue } folderName[mb.ID] = mb.Name } seen := make(map[string]bool, len(egr.List)) for _, em := range egr.List { if seen[em.ID] { continue } seen[em.ID] = true msg, err := downloadBlob(sess, auth, accountID, em.BlobID) if err != nil { slog.Error("download blob", "id", em.ID, "err", err) continue } msgID := extractMessageID(msg) for mbID := range em.MailboxIDs { name, ok := folderName[mbID] if !ok { slog.Warn("unknown mailbox", "mailboxId", mbID) continue } folderPath := filepath.Join(a.MaildirPath, name) if msgID != "" && maildirHasMessageID(folderPath, msgID) { slog.Debug("skipping duplicate", "account", a.Name, "folder", name, "messageID", msgID) continue } if err := writeMaildir(folderPath, msg); err != nil { slog.Error("write maildir", "id", em.ID, "folder", name, "err", err) continue } slog.Info("delivered", "account", a.Name, "folder", name, "messageID", msgID) delivered++ } } return newState, delivered, nil } func getEmailState(sess *jmapSession, auth, accountID string) (string, error) { resp, err := callJMAP(sess.APIUrl, auth, jmapRequest{ Using: []string{capCore, capMail}, MethodCalls: [][]interface{}{ {"Email/get", map[string]interface{}{ "accountId": accountID, "ids": []string{}, }, "0"}, }, }) if err != nil { return "", err } var result emailGetResult if err := json.Unmarshal(resp.MethodResponses[0][1], &result); err != nil { return "", err } return result.State, nil } func downloadBlob(sess *jmapSession, auth, accountID, blobID string) ([]byte, error) { dlURL := strings.NewReplacer( "{accountId}", accountID, "{blobId}", blobID, "{type}", "message%2Frfc822", "{name}", "email.eml", ).Replace(sess.DownloadUrl) req, _ := http.NewRequest("GET", dlURL, nil) req.Header.Set("Authorization", auth) resp, err := http.DefaultClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return nil, fmt.Errorf("blob download HTTP %d: %s", resp.StatusCode, b) } return io.ReadAll(resp.Body) } func fetchSession(ctx context.Context, auth string) (*jmapSession, error) { req, err := http.NewRequestWithContext(ctx, "GET", jmapSessionURL, nil) if err != nil { return nil, err } req.Header.Set("Authorization", auth) resp, err := http.DefaultClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return nil, fmt.Errorf("session HTTP %d: %s", resp.StatusCode, b) } var s jmapSession return &s, json.NewDecoder(resp.Body).Decode(&s) } func callJMAP(apiURL, auth string, request jmapRequest) (*jmapResponse, error) { body, _ := json.Marshal(request) req, _ := http.NewRequest("POST", apiURL, bytes.NewReader(body)) req.Header.Set("Authorization", auth) req.Header.Set("Content-Type", "application/json") resp, err := http.DefaultClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return nil, fmt.Errorf("JMAP API HTTP %d: %s", resp.StatusCode, b) } var jr jmapResponse return &jr, json.NewDecoder(resp.Body).Decode(&jr) } func buildAuth(a AccountConfig) (string, error) { var token string var err error switch { case a.PasswordCommand != "": var out []byte out, err = exec.Command("sh", "-c", a.PasswordCommand).Output() token = strings.TrimSpace(string(out)) case a.TokenFile != "": var b []byte b, err = os.ReadFile(a.TokenFile) token = strings.TrimSpace(string(b)) case a.TokenEnv != "": token = os.Getenv(a.TokenEnv) if token == "" { err = fmt.Errorf("env var %s is empty or unset", a.TokenEnv) } } if err != nil { return "", err } return "Bearer " + token, nil } // runMbsync backfills messages that arrived while mbrts was not running. It // pulls only: mbrts delivers to the Maildir out of band, so a full sync would // see those messages as new local mail and push them back up to the server as // duplicates. --pull propagates new messages far-to-near without any push. func runMbsync(a AccountConfig) { ch := a.mbsyncChannel() slog.Info("running mbsync", "account", a.Name, "channel", ch) if out, err := exec.Command("mbsync", "--pull", ch).CombinedOutput(); err != nil { slog.Error("mbsync failed", "channel", ch, "err", err, "output", string(out)) } } func runNotmuchNew() { if out, err := exec.Command("notmuch", "new").CombinedOutput(); err != nil { slog.Error("notmuch new", "err", err, "output", string(out)) } }