email_agent/internal/store/store.go

// Package store owns the SQLite database: opening it, applying the schema, and
// exposing a small query surface to the rest of the service.
package store

import (
	"context"
	"database/sql"
	_ "embed"
	"fmt"
	"net/url"
	"os"
	"path/filepath"
	"time"

	// Pure-Go SQLite driver. Keeps the container CGO-free.
	_ "modernc.org/sqlite"
)

//go:embed schema.sql
var schemaSQL string

// Store wraps the SQLite connection pool.
type Store struct {
	db   *sql.DB
	path string
}

// Open creates (if needed) and connects to the SQLite database at path, then
// applies the schema. The parent directory is created when missing.
func Open(ctx context.Context, path string) (*Store, error) {
	if dir := filepath.Dir(path); dir != "" && dir != "." {
		if err := os.MkdirAll(dir, 0o750); err != nil {
			return nil, fmt.Errorf("create database directory %q: %w", dir, err)
		}
	}

	dsn := buildDSN(path)
	db, err := sql.Open("sqlite", dsn)
	if err != nil {
		return nil, fmt.Errorf("open sqlite database: %w", err)
	}

	// SQLite handles one writer at a time; keep the pool small and give the
	// busy handler room. Reads and writes share the connection here.
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	db.SetConnMaxLifetime(0)

	pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	if err := db.PingContext(pingCtx); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("ping sqlite database: %w", err)
	}

	s := &Store{db: db, path: path}
	if err := s.migrate(ctx); err != nil {
		_ = db.Close()
		return nil, err
	}
	return s, nil
}

// buildDSN returns a modernc.org/sqlite DSN with safe pragmas applied on every
// new connection.
func buildDSN(path string) string {
	abs, err := filepath.Abs(path)
	if err != nil {
		abs = path
	}
	q := url.Values{}
	q.Add("_pragma", "busy_timeout(5000)")
	q.Add("_pragma", "journal_mode(WAL)")
	q.Add("_pragma", "foreign_keys(1)")
	q.Add("_pragma", "synchronous(NORMAL)")
	return "file:" + abs + "?" + q.Encode()
}

// migrate applies the embedded schema, then any incremental migrations that
// add columns to a database created by an earlier release. It is safe to run on
// every start.
func (s *Store) migrate(ctx context.Context) error {
	if _, err := s.db.ExecContext(ctx, schemaSQL); err != nil {
		return fmt.Errorf("apply schema: %w", err)
	}
	return s.applyMigrations(ctx)
}

// applyMigrations brings an existing database up to the latest schema version.
// Version 1 is the base schema (schema.sql); version 2 adds IMAP bookkeeping
// columns (AGENTS.md §9 / Phase 2). Each step is idempotent.
func (s *Store) applyMigrations(ctx context.Context) error {
	version, err := s.SchemaVersion(ctx)
	if err != nil {
		return fmt.Errorf("read schema version: %w", err)
	}
	if version < 2 {
		for _, col := range []struct{ table, name, typ string }{
			{"messages", "imap_uid", "INTEGER"},
			{"messages", "imap_uidvalidity", "INTEGER"},
		} {
			if err := s.ensureColumn(ctx, col.table, col.name, col.typ); err != nil {
				return err
			}
		}
		if _, err := s.db.ExecContext(ctx,
			`INSERT OR IGNORE INTO schema_migrations (version) VALUES (2)`); err != nil {
			return fmt.Errorf("record schema version 2: %w", err)
		}
	}
	if version < 3 {
		// FR-14: seed the default category list and a starter (comment-only)
		// rules document. Seeding happens only inside this migration so that a
		// category the user removes is not silently re-added on the next start.
		if err := s.seedCategories(ctx); err != nil {
			return err
		}
		if err := s.seedRules(ctx); err != nil {
			return err
		}
		if _, err := s.db.ExecContext(ctx,
			`INSERT OR IGNORE INTO schema_migrations (version) VALUES (3)`); err != nil {
			return fmt.Errorf("record schema version 3: %w", err)
		}
	}
	if version < 4 {
		// Phase 4 (FR-6/FR-9): triage actions file a message under an app folder
		// and remember the Gmail label it currently lives in, so a delete can be
		// undone by moving it back to the inbox.
		for _, col := range []struct{ table, name, typ string }{
			{"messages", "folder", "TEXT NOT NULL DEFAULT 'inbox'"},
			{"messages", "mailbox", "TEXT NOT NULL DEFAULT 'INBOX'"},
		} {
			if err := s.ensureColumn(ctx, col.table, col.name, col.typ); err != nil {
				return err
			}
		}
		if _, err := s.db.ExecContext(ctx,
			`CREATE INDEX IF NOT EXISTS idx_messages_folder ON messages (folder, received_at)`); err != nil {
			return fmt.Errorf("index messages.folder: %w", err)
		}
		if _, err := s.db.ExecContext(ctx,
			`INSERT OR IGNORE INTO schema_migrations (version) VALUES (4)`); err != nil {
			return fmt.Errorf("record schema version 4: %w", err)
		}
	}
	if version < 5 {
		// Phase 5 (FR-8): store the RFC822 Message-ID so a reply can set
		// In-Reply-To/References and stay in the same Gmail thread, and record
		// when a draft was sent.
		if err := s.ensureColumn(ctx, "messages", "message_id_header", "TEXT"); err != nil {
			return err
		}
		if err := s.ensureColumn(ctx, "drafts", "sent_at", "TEXT"); err != nil {
			return err
		}
		if _, err := s.db.ExecContext(ctx,
			`INSERT OR IGNORE INTO schema_migrations (version) VALUES (5)`); err != nil {
			return fmt.Errorf("record schema version 5: %w", err)
		}
	}
	return nil
}

// defaultCategories are the starter categories from AGENTS.md §13.2. The
// descriptions are shown in the UI and sent to JEV as the Choice rubrics.
var defaultCategories = []struct{ name, description string }{
	{"Action", "Needs Sam to do something"},
	{"Important", "Worth Sam's attention, but no specific task"},
	{"Info", "Informational, no action required"},
	{"Spam", "Unwanted, deceptive, or scam mail"},
	{"Newsletter", "Bulk mail, digests, and marketing"},
	{"Auto-reply", "Automated notifications and receipts"},
}

// seedCategories inserts the built-in categories when the table is empty.
func (s *Store) seedCategories(ctx context.Context) error {
	var n int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM categories`).Scan(&n); err != nil {
		return fmt.Errorf("count categories: %w", err)
	}
	if n > 0 {
		return nil
	}
	for i, c := range defaultCategories {
		if _, err := s.db.ExecContext(ctx,
			`INSERT INTO categories (name, description, position) VALUES (?, ?, ?)`,
			c.name, c.description, i+1); err != nil {
			return fmt.Errorf("seed category %q: %w", c.name, err)
		}
	}
	return nil
}

// starterRules documents the rule syntax (FR-4/FR-11). The editor that writes
// real rules lands in Phase 6; the classifier already reads whatever is stored.
const starterRules = `# Email Agent rules (FR-4). One rule per line.
#
#   <condition> => <effects>
#
# condition: <field> <operator> "<value>", joined with " and ".
#   fields:    from, to, subject, source, body
#   operators: contains, equals, starts_with, ends_with
# a single "*" matches every message (use it as the last line for a default).
#
# effects:   comma-separated category=<name>, importance=<low|medium|high>,
#            action=<delete|unsubscribe|reply|file|keep>
#
# The first matching rule wins and overrides the JEV classification for the
# fields it names. Category names must exist in the category list.
#
# Examples (remove the leading # to enable):
# from contains "amazon.com" => category=Newsletter, action=unsubscribe
# subject contains "invoice" => category=Action, importance=high, action=file
`

// seedRules stores the starter rules document when no rules exist yet.
func (s *Store) seedRules(ctx context.Context) error {
	var n int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM rules`).Scan(&n); err != nil {
		return fmt.Errorf("count rules: %w", err)
	}
	if n > 0 {
		return nil
	}
	if _, err := s.db.ExecContext(ctx,
		`INSERT INTO rules (text, version) VALUES (?, ?)`, starterRules, 1); err != nil {
		return fmt.Errorf("seed rules: %w", err)
	}
	return nil
}

// ensureColumn adds a column when it is missing. SQLite has no
// "ADD COLUMN IF NOT EXISTS", so the current shape is inspected first.
func (s *Store) ensureColumn(ctx context.Context, table, name, typ string) error {
	rows, err := s.db.QueryContext(ctx, fmt.Sprintf("PRAGMA table_info(%s)", table))
	if err != nil {
		return fmt.Errorf("inspect %s: %w", table, err)
	}
	defer rows.Close()
	for rows.Next() {
		var (
			cid, notnull, pk int
			dflt             sql.NullString
			colName, colType string
		)
		if err := rows.Scan(&cid, &colName, &colType, &notnull, &dflt, &pk); err != nil {
			return fmt.Errorf("scan %s columns: %w", table, err)
		}
		if colName == name {
			return nil
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("inspect %s columns: %w", table, err)
	}
	if _, err := s.db.ExecContext(ctx,
		fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", table, name, typ)); err != nil {
		return fmt.Errorf("add %s.%s: %w", table, name, err)
	}
	return nil
}

// DB exposes the underlying pool for callers that need it.
func (s *Store) DB() *sql.DB { return s.db }

// Path returns the database file path.
func (s *Store) Path() string { return s.path }

// Ping verifies the database is reachable.
func (s *Store) Ping(ctx context.Context) error {
	return s.db.PingContext(ctx)
}

// Close releases the connection pool.
func (s *Store) Close() error {
	if s == nil || s.db == nil {
		return nil
	}
	return s.db.Close()
}

// SchemaVersion returns the highest applied migration version.
func (s *Store) SchemaVersion(ctx context.Context) (int, error) {
	var v sql.NullInt64
	err := s.db.QueryRowContext(ctx, `SELECT MAX(version) FROM schema_migrations`).Scan(&v)
	if err != nil {
		return 0, err
	}
	if !v.Valid {
		return 0, nil
	}
	return int(v.Int64), nil
}