email_agent/internal/store/messages.go

// Message persistence for the Email Agent (AGENTS.md §7).
//
// Phase 2 stores the metadata, body text, and — critically for FR-2 — the
// source address the mail was forwarded to (X-Pm-Original-To and its
// fallbacks). Classification columns arrive with Phase 3.
package store

import (
	"context"
	"fmt"
	"strings"
	"time"
)

// Message is one row of the messages table.
type Message struct {
	ID       int64
	GmailID  string
	ThreadID string
	// MessageIDHeader is the RFC822 Message-ID header (Phase 5, FR-8). A reply
	// uses it for In-Reply-To and References so Gmail keeps the thread.
	MessageIDHeader string
	// ListUnsubscribe and ListUnsubscribePost are the raw RFC 2369/8058 headers
	// (Phase 6, FR-7). Detection runs over them on demand.
	ListUnsubscribe     string
	ListUnsubscribePost string
	FromAddr            string
	ToAddr              string
	Subject             string
	BodyText            string
	SourceAddress       string
	DateSent            time.Time
	ReceivedAt          time.Time
	IMAPUID             int64
	IMAPUIDValidity     int64
	// Folder is the app folder the message is filed under (Phase 4, FR-6/FR-9):
	// inbox, delete, saved, or replied.
	Folder string
	// Mailbox is the exact IMAP label the message currently lives in on Gmail.
	Mailbox string
}

// MessageListItem is a message plus its latest classification, when present.
type MessageListItem struct {
	Message
	Category          string
	Importance        string
	Action            string
	Confidence        float64
	HasClassification bool
}

// App folders (Phase 4). A folder is a low-cardinality navigation concept; the
// exact Gmail label is tracked separately on each message.
const (
	FolderInbox   = "inbox"
	FolderDelete  = "delete"
	FolderSaved   = "saved"
	FolderReplied = "replied"
)

// ValidFolder reports whether folder is an app folder the UI knows how to show.
func ValidFolder(folder string) bool {
	switch folder {
	case FolderInbox, FolderDelete, FolderSaved, FolderReplied:
		return true
	}
	return false
}

// UpsertMessage inserts a message, or refreshes the mutable fields when the
// same Gmail id is seen again. received_at is preserved on update so the
// original arrival time is stable.
func (s *Store) UpsertMessage(ctx context.Context, m *Message) (int64, error) {
	const q = `
INSERT INTO messages (
    gmail_id, thread_id, message_id_header, from_addr, to_addr, subject, body_text,
    source_address, list_unsubscribe, list_unsubscribe_post, date_sent, received_at,
    imap_uid, imap_uidvalidity
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (gmail_id) DO UPDATE SET
    thread_id             = excluded.thread_id,
    message_id_header     = excluded.message_id_header,
    from_addr             = excluded.from_addr,
    to_addr               = excluded.to_addr,
    subject               = excluded.subject,
    body_text             = excluded.body_text,
    source_address        = excluded.source_address,
    list_unsubscribe      = excluded.list_unsubscribe,
    list_unsubscribe_post = excluded.list_unsubscribe_post,
    date_sent             = excluded.date_sent,
    imap_uid              = excluded.imap_uid,
    imap_uidvalidity      = excluded.imap_uidvalidity`

	receivedAt := m.ReceivedAt
	if receivedAt.IsZero() {
		receivedAt = time.Now().UTC()
	}
	_, err := s.db.ExecContext(ctx, q,
		m.GmailID,
		nullString(m.ThreadID),
		nullString(m.MessageIDHeader),
		nullString(m.FromAddr),
		nullString(m.ToAddr),
		nullString(m.Subject),
		nullString(m.BodyText),
		nullString(m.SourceAddress),
		nullString(m.ListUnsubscribe),
		nullString(m.ListUnsubscribePost),
		nullTime(m.DateSent),
		receivedAt.UTC().Format(sqlTimeLayout),
		m.IMAPUID,
		m.IMAPUIDValidity,
	)
	if err != nil {
		return 0, fmt.Errorf("upsert message %q: %w", m.GmailID, err)
	}

	var id int64
	if err := s.db.QueryRowContext(ctx,
		`SELECT id FROM messages WHERE gmail_id = ?`, m.GmailID).Scan(&id); err != nil {
		return 0, fmt.Errorf("read message id %q: %w", m.GmailID, err)
	}
	return id, nil
}

// MessageByID loads one message including the reply fields (body, source
// address, and Message-ID header). It is the read path for a reply draft.
func (s *Store) MessageByID(ctx context.Context, id int64) (Message, error) {
	const q = `
SELECT m.id, m.gmail_id, COALESCE(m.thread_id, ''), COALESCE(m.from_addr, ''),
       COALESCE(m.to_addr, ''), COALESCE(m.subject, ''), COALESCE(m.body_text, ''),
       COALESCE(m.source_address, ''), COALESCE(m.date_sent, ''), m.received_at,
       COALESCE(m.imap_uid, 0), COALESCE(m.imap_uidvalidity, 0),
       COALESCE(m.folder, 'inbox'), COALESCE(m.mailbox, 'INBOX'),
       COALESCE(m.message_id_header, ''),
       COALESCE(m.list_unsubscribe, ''), COALESCE(m.list_unsubscribe_post, '')
FROM messages m WHERE m.id = ?`
	var (
		m        Message
		dateSent string
		received string
	)
	err := s.db.QueryRowContext(ctx, q, id).Scan(
		&m.ID, &m.GmailID, &m.ThreadID, &m.FromAddr, &m.ToAddr, &m.Subject,
		&m.BodyText, &m.SourceAddress, &dateSent, &received,
		&m.IMAPUID, &m.IMAPUIDValidity, &m.Folder, &m.Mailbox, &m.MessageIDHeader,
		&m.ListUnsubscribe, &m.ListUnsubscribePost,
	)
	if err != nil {
		return Message{}, fmt.Errorf("load message %d: %w", id, err)
	}
	m.DateSent = parseSQLTime(dateSent)
	m.ReceivedAt = parseSQLTime(received)
	return m, nil
}

// ListMessages returns the most recent messages, newest first, joined to their
// latest classification when one exists.
func (s *Store) ListMessages(ctx context.Context, limit int) ([]MessageListItem, error) {
	if limit < 1 {
		limit = 100
	}
	const q = `
SELECT
    m.id, m.gmail_id, COALESCE(m.thread_id, ''), COALESCE(m.from_addr, ''),
    COALESCE(m.to_addr, ''), COALESCE(m.subject, ''), COALESCE(m.body_text, ''),
    COALESCE(m.source_address, ''), COALESCE(m.date_sent, ''), m.received_at,
    COALESCE(m.imap_uid, 0), COALESCE(m.imap_uidvalidity, 0),
    COALESCE(m.folder, 'inbox'),
    COALESCE(c.category, ''), COALESCE(c.importance, ''),
    COALESCE(c.action, ''), COALESCE(c.confidence, 0)
FROM messages m
LEFT JOIN classifications c ON c.id = (
    SELECT id FROM classifications WHERE message_id = m.id ORDER BY id DESC LIMIT 1
)
ORDER BY COALESCE(NULLIF(m.date_sent, ''), m.received_at) DESC, m.id DESC
LIMIT ?`

	rows, err := s.db.QueryContext(ctx, q, limit)
	if err != nil {
		return nil, fmt.Errorf("list messages: %w", err)
	}
	defer rows.Close()

	var out []MessageListItem
	for rows.Next() {
		var (
			item     MessageListItem
			dateSent string
			received string
		)
		if err := rows.Scan(
			&item.ID, &item.GmailID, &item.ThreadID, &item.FromAddr,
			&item.ToAddr, &item.Subject, &item.BodyText,
			&item.SourceAddress, &dateSent, &received,
			&item.IMAPUID, &item.IMAPUIDValidity, &item.Folder,
			&item.Category, &item.Importance, &item.Action, &item.Confidence,
		); err != nil {
			return nil, fmt.Errorf("scan message: %w", err)
		}
		item.DateSent = parseSQLTime(dateSent)
		item.ReceivedAt = parseSQLTime(received)
		item.HasClassification = item.Category != ""
		out = append(out, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list messages: %w", err)
	}
	return out, nil
}

// CountMessages returns the total number of stored messages.
func (s *Store) CountMessages(ctx context.Context) (int, error) {
	var n int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM messages`).Scan(&n); err != nil {
		return 0, fmt.Errorf("count messages: %w", err)
	}
	return n, nil
}

// LogAction appends to the actions_log (FR-13). The table is append-only.
func (s *Store) LogAction(ctx context.Context, action, actor, detail string) error {
	const q = `INSERT INTO actions_log (message_id, action, actor, detail) VALUES (NULL, ?, ?, ?)`
	if _, err := s.db.ExecContext(ctx, q, action, nullString(actor), nullString(detail)); err != nil {
		return fmt.Errorf("log action %q: %w", action, err)
	}
	return nil
}

// LogMessageAction appends a per-message action to the actions_log (FR-13).
func (s *Store) LogMessageAction(ctx context.Context, messageID int64, action, actor, detail string) error {
	const q = `INSERT INTO actions_log (message_id, action, actor, detail) VALUES (?, ?, ?, ?)`
	if _, err := s.db.ExecContext(ctx, q, messageID, action, nullString(actor), nullString(detail)); err != nil {
		return fmt.Errorf("log message action %q: %w", action, err)
	}
	return nil
}

const sqlTimeLayout = "2006-01-02 15:04:05"

func nullString(v string) any {
	if strings.TrimSpace(v) == "" {
		return nil
	}
	return v
}

func nullTime(t time.Time) any {
	if t.IsZero() {
		return nil
	}
	return t.UTC().Format(sqlTimeLayout)
}

func parseSQLTime(v string) time.Time {
	if v == "" {
		return time.Time{}
	}
	for _, layout := range []string{sqlTimeLayout, time.RFC3339, "2006-01-02T15:04:05Z"} {
		if t, err := time.Parse(layout, v); err == nil {
			return t.UTC()
		}
	}
	return time.Time{}
}