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
	FromAddr        string
	ToAddr          string
	Subject         string
	BodyText        string
	SourceAddress   string
	DateSent        time.Time
	ReceivedAt      time.Time
	IMAPUID         int64
	IMAPUIDValidity int64
}

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

// 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, from_addr, to_addr, subject, body_text,
    source_address, date_sent, received_at, imap_uid, imap_uidvalidity
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (gmail_id) DO UPDATE SET
    thread_id         = excluded.thread_id,
    from_addr         = excluded.from_addr,
    to_addr           = excluded.to_addr,
    subject           = excluded.subject,
    body_text         = excluded.body_text,
    source_address    = excluded.source_address,
    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.FromAddr),
		nullString(m.ToAddr),
		nullString(m.Subject),
		nullString(m.BodyText),
		nullString(m.SourceAddress),
		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
}

// 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(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.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
}

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