email_agent/internal/mailfetch/fetch.go

package mailfetch

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"gitea.lab.audasmedia.com.au/sam/email_agent/internal/config"
	"gitea.lab.audasmedia.com.au/sam/email_agent/internal/store"
)

// ErrNoCredentials means IMAP login cannot be attempted.
var ErrNoCredentials = errors.New("GMAIL_USER and GMAIL_APP_PASSWORD are not configured")

// ErrAlreadyRunning means a scan is already in progress.
var ErrAlreadyRunning = errors.New("a mail scan is already running")

// Fetcher performs an IMAP scan and stores the results.
type Fetcher struct {
	cfg    *config.Config
	store  *store.Store
	logger *slog.Logger

	mu sync.Mutex
}

// NewFetcher builds a Fetcher.
func NewFetcher(cfg *config.Config, st *store.Store, logger *slog.Logger) *Fetcher {
	return &Fetcher{cfg: cfg, store: st, logger: logger}
}

// Summary describes the outcome of one scan.
type Summary struct {
	UIDValidity uint32
	Since       time.Time
	Fetched     int
	Stored      int
	Duration    time.Duration
}

// Fetch performs one bounded IMAP scan of INBOX and upserts the results.
func (f *Fetcher) Fetch(ctx context.Context) (Summary, error) {
	if !f.cfg.GmailConfigured() {
		return Summary{}, ErrNoCredentials
	}
	if !f.mu.TryLock() {
		return Summary{}, ErrAlreadyRunning
	}
	defer f.mu.Unlock()

	start := time.Now()
	client, err := Dial(ctx, f.cfg.IMAPHost, f.cfg.IMAPPort, DefaultTimeout)
	if err != nil {
		return Summary{}, err
	}
	defer func() { _ = client.Logout() }()

	if err := client.Login(f.cfg.GmailUser, f.cfg.GmailAppPassword); err != nil {
		return Summary{}, err
	}

	uidValidity, err := client.Select("INBOX")
	if err != nil {
		return Summary{}, err
	}

	since := time.Now().UTC().AddDate(0, 0, -f.cfg.IMAPLookbackDays)
	uids, err := client.SearchSince(since)
	if err != nil {
		return Summary{}, err
	}
	if max := f.cfg.IMAPMaxMessages; max > 0 && len(uids) > max {
		// Keep the newest; SEARCH returns ascending UIDs.
		uids = uids[len(uids)-max:]
	}

	results, err := client.Fetch(uids)
	if err != nil {
		return Summary{}, err
	}

	sum := Summary{
		UIDValidity: uidValidity,
		Since:       since,
		Fetched:     len(results),
	}
	now := time.Now().UTC()
	headersSeen := map[string]struct{}{}
	for _, r := range results {
		parsed, perr := ParseMessage(r.Raw)
		if perr != nil {
			f.logger.Warn("skip unparsable message", "uid", r.UID, "error", perr)
			continue
		}
		if parsed.SourceHeader != "" {
			headersSeen[parsed.SourceHeader] = struct{}{}
		}
		msg := &store.Message{
			GmailID:             gmailID(r, uidValidity),
			ThreadID:            r.GmailThreadID,
			MessageIDHeader:     parsed.MessageID,
			FromAddr:            parsed.From,
			ToAddr:              parsed.To,
			Subject:             parsed.Subject,
			BodyText:            parsed.BodyText,
			SourceAddress:       parsed.SourceAddress,
			ListUnsubscribe:     parsed.ListUnsubscribe,
			ListUnsubscribePost: parsed.ListUnsubscribePost,
			DateSent:            parsed.Date,
			ReceivedAt:          now,
			IMAPUID:             int64(r.UID),
			IMAPUIDValidity:     int64(uidValidity),
		}
		if _, err := f.store.UpsertMessage(ctx, msg); err != nil {
			return sum, err
		}
		sum.Stored++
	}
	sum.Duration = time.Since(start)

	if err := f.store.LogAction(ctx, "fetch", "agent",
		fmt.Sprintf("uidvalidity=%d since=%s fetched=%d stored=%d source_headers=%s",
			uidValidity, since.Format("2006-01-02"), sum.Fetched, sum.Stored, joinKeys(headersSeen))); err != nil {
		f.logger.Warn("record fetch action", "error", err)
	}

	f.logger.Info("mail scan complete",
		"uidvalidity", uidValidity,
		"since", since.Format("2006-01-02"),
		"fetched", sum.Fetched,
		"stored", sum.Stored,
		"duration_ms", sum.Duration.Milliseconds(),
	)
	return sum, nil
}

// gmailID prefers Gmail's stable X-GM-MSGID and falls back to a UID key that is
// scoped to the UIDVALIDITY it was issued under.
func gmailID(r FetchResult, uidValidity uint32) string {
	if r.GmailMsgID != "" {
		return r.GmailMsgID
	}
	return "uid:" + strconv.FormatUint(uint64(uidValidity), 10) + ":" + strconv.FormatUint(uint64(r.UID), 10)
}

// joinKeys returns the sorted keys of a string set, for a log line.
func joinKeys(set map[string]struct{}) string {
	if len(set) == 0 {
		return "none"
	}
	parts := make([]string, 0, len(set))
	for k := range set {
		parts = append(parts, k)
	}
	slices.Sort(parts)
	return strings.Join(parts, ",")
}