mbrts/jmap.go

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))
	}
}