email_agent/internal/store/classify.go

// Classification persistence and the rules read/write paths (FR-3, FR-4, FR-11).
package store

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
)

// Classification is one classifier output for a message.
type Classification struct {
	ID         int64
	MessageID  int64
	Category   string
	Importance string
	Action     string
	Confidence float64
	Model      string
}

// SaveClassification appends a classification row. The triage list reads the
// newest row per message, so re-classifying keeps a history.
func (s *Store) SaveClassification(ctx context.Context, c Classification) (int64, error) {
	const q = `INSERT INTO classifications
        (message_id, category, importance, action, confidence, model)
        VALUES (?, ?, ?, ?, ?, ?)`
	res, err := s.db.ExecContext(ctx, q,
		c.MessageID, c.Category, nullString(c.Importance), nullString(c.Action),
		c.Confidence, nullString(c.Model))
	if err != nil {
		return 0, fmt.Errorf("save classification for message %d: %w", c.MessageID, err)
	}
	id, err := res.LastInsertId()
	if err != nil {
		return 0, fmt.Errorf("read classification id: %w", err)
	}
	return id, nil
}

// PendingClassifications returns messages whose latest classification is
// missing or points at a category that no longer exists. Oldest first so a
// bounded run makes steady progress.
func (s *Store) PendingClassifications(ctx context.Context, limit int) ([]Message, error) {
	return s.messagesForClassification(ctx, limit, false)
}

// AllMessagesForClassification returns every message, for a full re-classify
// after the category list or the rules change.
func (s *Store) AllMessagesForClassification(ctx context.Context, limit int) ([]Message, error) {
	return s.messagesForClassification(ctx, limit, true)
}

func (s *Store) messagesForClassification(ctx context.Context, limit int, all bool) ([]Message, error) {
	if limit < 1 {
		limit = 100
	}
	filter := "WHERE c.id IS NULL OR c.category NOT IN (SELECT name FROM categories)"
	if all {
		filter = ""
	}
	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)
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
)
` + filter + `
ORDER BY m.id ASC
LIMIT ?`
	rows, err := s.db.QueryContext(ctx, q, limit)
	if err != nil {
		return nil, fmt.Errorf("list messages for classification: %w", err)
	}
	defer rows.Close()

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

// ClassificationCoverage reports how many messages have a usable classification.
type ClassificationCoverage struct {
	Total      int
	Classified int
}

// Missing reports how many messages still need a classification.
func (c ClassificationCoverage) Missing() int { return c.Total - c.Classified }

// ClassificationCoverage returns the total message count and how many have a
// classification whose category still exists.
func (s *Store) ClassificationCoverage(ctx context.Context) (ClassificationCoverage, error) {
	var cov ClassificationCoverage
	const q = `
SELECT
    (SELECT COUNT(*) FROM messages),
    (SELECT COUNT(*) FROM messages m
      WHERE (SELECT cc.category FROM classifications cc
              WHERE cc.message_id = m.id ORDER BY cc.id DESC LIMIT 1)
            IN (SELECT name FROM categories))`
	if err := s.db.QueryRowContext(ctx, q).Scan(&cov.Total, &cov.Classified); err != nil {
		return cov, fmt.Errorf("classification coverage: %w", err)
	}
	return cov, nil
}

// CurrentRules returns the newest rules text and its version. An empty string
// and version 0 mean no rules are stored yet.
func (s *Store) CurrentRules(ctx context.Context) (string, int, error) {
	var (
		text    string
		version int
	)
	err := s.db.QueryRowContext(ctx,
		`SELECT text, version FROM rules ORDER BY version DESC LIMIT 1`).Scan(&text, &version)
	if errors.Is(err, sql.ErrNoRows) {
		return "", 0, nil
	}
	if err != nil {
		return "", 0, fmt.Errorf("read rules: %w", err)
	}
	return text, version, nil
}

// SaveRules appends a new rules version and returns the new version number
// (FR-11 versioning). The rules editor UI lands in Phase 6.
func (s *Store) SaveRules(ctx context.Context, text string) (int, error) {
	var next int
	if err := s.db.QueryRowContext(ctx,
		`SELECT COALESCE(MAX(version), 0) + 1 FROM rules`).Scan(&next); err != nil {
		return 0, fmt.Errorf("next rules version: %w", err)
	}
	if _, err := s.db.ExecContext(ctx,
		`INSERT INTO rules (text, version) VALUES (?, ?)`, text, next); err != nil {
		return 0, fmt.Errorf("save rules version %d: %w", next, err)
	}
	return next, nil
}

// RuleVersion is one saved rules revision, for the version list in the editor
// (FR-11).
type RuleVersion struct {
	ID        int64
	Version   int
	Text      string
	CreatedAt string
}

// RulesVersions returns saved rules revisions, newest first.
func (s *Store) RulesVersions(ctx context.Context, limit int) ([]RuleVersion, error) {
	if limit < 1 {
		limit = 20
	}
	rows, err := s.db.QueryContext(ctx,
		`SELECT id, version, text, created_at FROM rules ORDER BY version DESC LIMIT ?`, limit)
	if err != nil {
		return nil, fmt.Errorf("list rules versions: %w", err)
	}
	defer rows.Close()
	var out []RuleVersion
	for rows.Next() {
		var v RuleVersion
		if err := rows.Scan(&v.ID, &v.Version, &v.Text, &v.CreatedAt); err != nil {
			return nil, fmt.Errorf("scan rules version: %w", err)
		}
		out = append(out, v)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list rules versions: %w", err)
	}
	return out, nil
}

// RuleByVersion returns the text of one saved rules revision. ok is false when
// the version does not exist.
func (s *Store) RuleByVersion(ctx context.Context, version int) (string, bool, error) {
	var text string
	err := s.db.QueryRowContext(ctx, `SELECT text FROM rules WHERE version = ?`, version).Scan(&text)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	if err != nil {
		return "", false, fmt.Errorf("read rules version %d: %w", version, err)
	}
	return text, true, nil
}