Maven/cmd/mavmaild/main.go

// mavmaild — the mail reader module (Vikunja #246,
// docs/plans/01-email-reader.md).
//
// Every so often it opens one IMAP mailbox read-only, fetches the messages it
// has not read yet, and hands each one to core over ipc.MethodIngestMail. Core
// runs the extraction on the resident model and writes what comes back as task
// CANDIDATES he reviews on /tasks. Nothing here writes to the store, nothing
// here can create a reminder, and nothing here speaks.
//
// Why a separate daemon rather than a loop inside mavend, when extraction has
// to happen in mavend anyway: the credential. mavpoll set the precedent with the
// zenmoney token (#125) — the module that talks to a third party holds the
// secret, reads it from a FILE so it never appears in `ps`, in
// docker-compose.yml or in shell history, and core never sees it. Core learns
// that mail exists only as message text on one IPC method; it cannot connect to
// the mailbox even if it wanted to, and a compromised core yields no mail
// password.
//
// Off unless configured: without -password-file there is nothing to run, and
// the daemon says so and exits. If core has no `email` block the very first
// ingest comes back ErrUnknownMethod and this daemon stops polling instead of
// hammering a socket that will keep refusing.
//
// Mail is personal, so the log is counts and UIDs: how many messages were
// fetched, how many were bulk, how many candidates came back. No subject, no
// sender, no body, ever — reviewing a candidate is what /tasks is for.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"os"
	"os/signal"
	"path/filepath"
	"sort"
	"strings"
	"syscall"
	"time"

	"github.com/kami/maven/internal/email"
	"github.com/kami/maven/internal/ipc"
)

func main() {
	if err := run(os.Args[1:]); err != nil {
		fmt.Fprintln(os.Stderr, "mavmaild:", err)
		os.Exit(1)
	}
}

func run(args []string) error {
	fs := flag.NewFlagSet("mavmaild", flag.ContinueOnError)
	socket := fs.String("socket", "", "core IPC socket path (required)")
	server := fs.String("imap", "", "IMAP server, host or host:993 (required)")
	user := fs.String("user", "", "IMAP username (required)")
	passFile := fs.String("password-file", "", "file holding the IMAP password (required — never passed as a flag value)")
	mailbox := fs.String("mailbox", "INBOX", "mailbox to read, read-only")
	interval := fs.Duration("interval", 15*time.Minute, "how often to read the mailbox")
	lookback := fs.Duration("lookback", 72*time.Hour, "how far back to search on each poll")
	max := fs.Int("max", 25, "most messages to fetch in one poll")
	timeout := fs.Duration("timeout", 30*time.Second, "IMAP network timeout")
	statePath := fs.String("state", "", "file remembering which UIDs were read (default: none — every poll re-reads the window)")
	if err := fs.Parse(args); err != nil {
		return err
	}
	if *socket == "" {
		return fmt.Errorf("-socket is required")
	}
	if *server == "" || *user == "" || *passFile == "" {
		return fmt.Errorf("mail reading is off unless configured: set -imap, -user and -password-file")
	}

	// The password is read from a file, never taken as a flag value: an argv
	// secret is visible in `ps` to every user on the box and lands in the compose
	// file and the shell history. Read once at start — a rotated password means a
	// restart, which is cheaper than re-reading his credential every quarter hour.
	raw, err := os.ReadFile(*passFile)
	if err != nil {
		return fmt.Errorf("read password file: %w", err)
	}
	password := strings.TrimSpace(string(raw))
	if password == "" {
		return fmt.Errorf("password file %s is empty", *passFile)
	}

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	core, err := ipc.DialWait(*socket, 60*time.Second)
	if err != nil {
		return err
	}
	defer core.Close()

	r := &reader{
		core:     core,
		addr:     *server,
		user:     *user,
		mailbox:  *mailbox,
		lookback: *lookback,
		max:      *max,
		timeout:  *timeout,
		state:    newSeenState(*statePath),
	}
	if err := r.state.load(); err != nil {
		// A missing or corrupt state file must not stop mail from being read: the
		// worst case is re-reading the window, and capture dedupes on text.
		log.Printf("mavmaild: state: %v (starting from an empty seen-set)", err)
	}

	// The password is never logged, not even its length.
	log.Printf("mavmaild: reading %s on %s every %s (lookback %s, max %d/poll)",
		*mailbox, *server, *interval, *lookback, *max)

	r.pollOnce(ctx, password) // don't idle a full interval on start
	t := time.NewTicker(*interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			log.Printf("mavmaild: bye")
			return nil
		case <-t.C:
			if r.disabled {
				// Core told us mail ingestion is not configured. Nothing will change
				// without a core restart, and a restart restarts us too.
				log.Printf("mavmaild: core does not accept mail — idling")
				return nil
			}
			r.pollOnce(ctx, password)
		}
	}
}

// mailIngester — the slice of core this daemon uses. One method: hand over a
// message. It cannot write a fact, create a reminder or read the store, and the
// interface says so.
type mailIngester interface {
	IngestMail(ctx context.Context, req ipc.IngestMailReq) (ipc.IngestMailResp, error)
}

type reader struct {
	core     mailIngester
	addr     string
	user     string
	mailbox  string
	lookback time.Duration
	max      int
	timeout  time.Duration
	state    *seenState

	// dial — connection seam for the tests; nil ⇒ implicit TLS.
	dial func(addr string, timeout time.Duration) (*email.Conn, error)

	// disabled — core answered ErrUnknownMethod, i.e. it has no email block.
	disabled bool
}

// pollOnce — one read of the mailbox, then one ingest per message.
//
// A fetch error aborts this poll and nothing else; the next tick tries again.
// An ingest error for one message does not skip the rest — one mail the model
// choked on should not hide the four behind it.
func (r *reader) pollOnce(ctx context.Context, password string) {
	msgs, err := r.fetch(password)
	if err != nil {
		// The error may name a UID; it never names a subject or a sender.
		log.Printf("mavmaild: fetch: %v", err)
		if len(msgs) == 0 {
			return
		}
	}
	var junk, candidates, created int
	for _, m := range msgs {
		if ctx.Err() != nil {
			return
		}
		if m.Junk {
			junk++
			// Marked seen without a model call: the header filter already decided,
			// and re-classifying it every quarter hour would be pure waste.
			r.state.mark(m.UID)
			continue
		}
		resp, err := r.core.IngestMail(ctx, ipc.IngestMailReq{
			Mailbox: r.mailbox,
			UID:     m.UID,
			From:    m.From,
			Subject: m.Subject,
			Date:    m.Date,
			Body:    m.Body,
		})
		if errors.Is(err, ipc.ErrUnknownMethod) {
			log.Printf("mavmaild: core has no email block configured — mail ingestion is off; stopping")
			r.disabled = true
			return
		}
		if err != nil {
			// Not marked seen: an ingest that failed should be retried next poll.
			log.Printf("mavmaild: ingest uid %d: %v", m.UID, err)
			continue
		}
		r.state.mark(m.UID)
		candidates += len(resp.TaskIDs)
		created += resp.Created
	}
	if err := r.state.save(); err != nil {
		log.Printf("mavmaild: state: %v", err)
	}
	log.Printf("mavmaild: %s: %d read, %d bulk, %d candidate(s), %d new", r.mailbox, len(msgs), junk, candidates, created)
}

// fetch reads the mailbox. Messages already in the seen-set are not fetched at
// all, so a steady mailbox costs one SEARCH per poll and nothing else.
func (r *reader) fetch(password string) ([]email.Message, error) {
	f := email.FetchSince{
		Addr:    r.addr,
		User:    r.user,
		Mailbox: r.mailbox,
		Timeout: r.timeout,
		Since:   time.Now().Add(-r.lookback),
		Max:     r.max,
		Skip:    r.state.seen,
	}
	return f.RunWith(password, r.dial)
}

// ---- seen state ------------------------------------------------------------

// seenState — the UIDs already handed to core, persisted so a restart does not
// re-read (and re-extract, at multi-second LLM cost) the whole lookback window.
//
// Correctness does not depend on it: ipc.CaptureTask dedupes on normalised text
// among live tasks, so a re-read produces no duplicate rows. This exists to save
// the model's time, which is why a broken state file is a log line rather than a
// failure.
//
// UIDs are per-mailbox and monotonic, so the set is kept as a high-water mark
// plus the stragglers above it. If the server ever changes UIDVALIDITY, UIDs
// reset and the window is simply re-read once — dedupe absorbs it.
type seenState struct {
	path  string
	high  uint32
	set   map[uint32]bool
	dirty bool
}

func newSeenState(path string) *seenState {
	return &seenState{path: path, set: map[uint32]bool{}}
}

type seenFile struct {
	High uint32   `json:"high"`
	UIDs []uint32 `json:"uids,omitempty"`
}

func (s *seenState) seen(uid uint32) bool {
	return uid <= s.high || s.set[uid]
}

func (s *seenState) mark(uid uint32) {
	if s.seen(uid) {
		return
	}
	s.set[uid] = true
	s.dirty = true
	// Advance the high-water mark through any contiguous run, so the explicit set
	// stays small on a mailbox read in order.
	for {
		next := s.high + 1
		if !s.set[next] {
			break
		}
		delete(s.set, next)
		s.high = next
	}
}

func (s *seenState) load() error {
	if s.path == "" {
		return nil
	}
	b, err := os.ReadFile(s.path)
	if errors.Is(err, os.ErrNotExist) {
		return nil // first run
	}
	if err != nil {
		return err
	}
	var f seenFile
	if err := json.Unmarshal(b, &f); err != nil {
		return fmt.Errorf("parse %s: %w", s.path, err)
	}
	s.high = f.High
	for _, u := range f.UIDs {
		s.set[u] = true
	}
	return nil
}

// save writes the state atomically (temp file + rename), 0600: it is a list of
// message ids from his mailbox, which is metadata about his mail.
func (s *seenState) save() error {
	if s.path == "" || !s.dirty {
		return nil
	}
	uids := make([]uint32, 0, len(s.set))
	for u := range s.set {
		uids = append(uids, u)
	}
	sort.Slice(uids, func(i, j int) bool { return uids[i] < uids[j] })
	b, err := json.Marshal(seenFile{High: s.high, UIDs: uids})
	if err != nil {
		return err
	}
	tmp := s.path + ".tmp"
	if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil {
		return err
	}
	if err := os.WriteFile(tmp, b, 0o600); err != nil {
		return err
	}
	if err := os.Rename(tmp, s.path); err != nil {
		return err
	}
	s.dirty = false
	return nil
}