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