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 and -interval are one decision, not two. Every non-bulk message in a
	// poll is one serialized llama-server call on core's side, and core gates
	// mail extraction behind voice turns (llm.Gate), so a large batch does not
	// mute Maven, it just takes a while. Raise -max only alongside whatever
	// bound core is running.
	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:
			// Core told us mail ingestion is not configured. Nothing will change
			// without a core restart, and a restart restarts us too, so the
			// daemon stays up and does nothing at all.
			//
			// It does NOT exit. The compose service inherits restart:
			// unless-stopped, which restarts a clean exit as readily as a crash,
			// so exiting here produced a loop: log in to IMAP, get refused by
			// core, exit, restart, log in again. Four IMAP logins an hour
			// against a mailbox that has nothing to give, and Gmail and Yandex
			// both rate-limit exactly that.
			if r.disabled {
				continue
			}
			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

	// fetchMail — the read seam, nil ⇒ the real IMAP read. The tests replace
	// the whole read rather than the transport: internal/email keeps its dialer
	// unexported so that no code outside that package can point the reader at a
	// cleartext socket and hand it the password, and this daemon is code
	// outside that package.
	fetchMail func(password string) ([]email.Message, error)

	// disabled — core answered ErrUnknownMethod, i.e. it has no email block.
	// Written in pollOnce and read in the ticker loop, both on the one
	// goroutine that run() drives, so it needs no atomic. If a second caller of
	// pollOnce ever appears, this becomes a race and has to change.
	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
		}
		req := ipc.IngestMailReq{
			Mailbox: r.mailbox,
			UID:     m.UID,
			From:    m.From,
			Subject: m.Subject,
			Date:    m.Date,
			Body:    m.Body,
		}
		if m.Junk {
			junk++
			// Core is TOLD, which is what its wire doc says: it counts the bulk
			// message and answers Skipped without spending the model. The header
			// filter already decided, so no content is sent with the verdict —
			// nothing will read it.
			req = ipc.IngestMailReq{Mailbox: r.mailbox, UID: m.UID, Junk: true}
		}
		resp, err := r.core.IngestMail(ctx, req)
		if errors.Is(err, ipc.ErrUnknownMethod) {
			log.Printf("mavmaild: core has no email block configured — mail ingestion is off; idling until a restart")
			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) {
	if r.fetchMail != nil {
		return r.fetchMail(password)
	}
	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,
		// Everything below the oldest searchable UID has aged out of the
		// lookback window and can never be read again. Retiring it is what keeps
		// one permanently failing message from pinning the high-water mark
		// forever. See seenState.retire.
		OnSearch: func(uids []uint32) {
			if len(uids) == 0 {
				return
			}
			low := uids[0]
			for _, u := range uids {
				if u < low {
					low = u
				}
			}
			r.state.retire(low)
		},
	}
	return f.Run(password)
}

// ---- 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.
//
// The high-water mark only advances through a CONTIGUOUS run, so a UID that
// never ingests successfully would pin it forever: everything above stays in
// the explicit set, and save rewrites all of it every poll. A year of that is
// a few hundred thousand entries written every quarter hour, which breaks
// nothing loudly and is exactly why it is worth catching. retire is the answer:
// a UID that has fallen out of the SEARCH SINCE window can never be fetched
// again, so there is nothing left to wait for.
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
	}
}

// retire records that no UID below floor is reachable any more — they have
// aged out of the lookback window, so no poll will ever fetch them. The
// high-water mark can jump past the gap they were holding open, and the
// stragglers below it leave the explicit set.
//
// It never moves backwards, so a UIDVALIDITY reset (UIDs restarting low) makes
// this a no-op rather than a way to un-see a mailbox.
func (s *seenState) retire(floor uint32) {
	if floor == 0 || floor-1 <= s.high {
		return
	}
	s.high = floor - 1
	for u := range s.set {
		if u <= s.high {
			delete(s.set, u)
		}
	}
	// The run above the new mark may now be contiguous with it.
	for s.set[s.high+1] {
		delete(s.set, s.high+1)
		s.high++
	}
	s.dirty = true
}

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
}