ratatoskr-go/internal/storage/storage.go

package storage

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"fmt"
	"strings"
	"time"

	_ "modernc.org/sqlite"
)

// SQLiteTime — time.Time с sql.Scanner и driver.Valuer для modernc.org/sqlite.
// Хранит как INTEGER (Unix-секунды).
type SQLiteTime time.Time

func (st *SQLiteTime) Scan(src any) error {
	if src == nil {
		*st = SQLiteTime(time.Time{})
		return nil
	}
	switch v := src.(type) {
	case int64:
		*st = SQLiteTime(time.Unix(v, 0).UTC())
		return nil
	case float64:
		*st = SQLiteTime(time.Unix(int64(v), 0).UTC())
		return nil
	case string:
		// fallback: RFC3339
		t, err := time.Parse(time.RFC3339, v)
		if err != nil {
			return fmt.Errorf("parse time %q: %w", v, err)
		}
		*st = SQLiteTime(t)
		return nil
	default:
		return fmt.Errorf("cannot scan %T as SQLiteTime", src)
	}
}

func (st SQLiteTime) Value() (driver.Value, error) {
	return time.Time(st).Unix(), nil
}

func (st SQLiteTime) Time() time.Time {
	return time.Time(st)
}

// NullSQLiteTime — nullable версия SQLiteTime.
type NullSQLiteTime struct {
	Time  SQLiteTime
	Valid bool
}

func (n *NullSQLiteTime) Scan(src any) error {
	if src == nil {
		n.Valid = false
		return nil
	}
	n.Valid = true
	return n.Time.Scan(src)
}

func (n NullSQLiteTime) Value() (driver.Value, error) {
	if !n.Valid {
		return nil, nil
	}
	return n.Time.Value()
}

// Storage — компонент БД для задач и трассировок.
type Storage struct {
	db *sql.DB
}

// Open открывает SQLite-БД и выполняет миграции.
// path — путь к файлу БД; ":memory:" для тестов.
func Open(ctx context.Context, path string) (*Storage, error) {
	db, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, fmt.Errorf("%w: open: %w", ErrDB, err)
	}
	// Один connection — сериализует доступ, исключает SQLITE_BUSY
	// при конкурентной записи из нескольких горутин (worker).
	db.SetMaxOpenConns(1)
	// Прагмы: WAL + синхронность
	pragmas := []string{
		"PRAGMA journal_mode=WAL",
		"PRAGMA synchronous=NORMAL",
		"PRAGMA busy_timeout=5000",
		"PRAGMA foreign_keys=ON",
	}
	for _, p := range pragmas {
		if _, err := db.ExecContext(ctx, p); err != nil {
			db.Close()
			return nil, fmt.Errorf("%w: pragma %q: %w", ErrDB, p, err)
		}
	}
	s := &Storage{db: db}
	if err := s.migrate(ctx); err != nil {
		db.Close()
		return nil, err
	}
	return s, nil
}

// Close закрывает БД.
func (s *Storage) Close() error {
	return s.db.Close()
}

// DB возвращает сырой *sql.DB для использования в транзакциях.
func (s *Storage) DB() *sql.DB {
	return s.db
}

// migrate создаёт таблицы при первом запуске.
func (s *Storage) migrate(ctx context.Context) error {
	schema := `
	CREATE TABLE IF NOT EXISTS tasks (
		id          INTEGER PRIMARY KEY AUTOINCREMENT,
		chat_id     TEXT    NOT NULL DEFAULT '',
		title       TEXT    NOT NULL DEFAULT '',
		goal        TEXT    NOT NULL DEFAULT '',
		repo        TEXT    NOT NULL DEFAULT '',
		why         TEXT    NOT NULL DEFAULT '',
		ac          TEXT    NOT NULL DEFAULT '',
		task_tag    TEXT    NOT NULL DEFAULT '',
		repos       TEXT    NOT NULL DEFAULT '',
		status      TEXT    NOT NULL DEFAULT 'draft',
		created_at  INTEGER NOT NULL DEFAULT (unixepoch()),
		updated_at  INTEGER NOT NULL DEFAULT (unixepoch())
	);

	CREATE INDEX IF NOT EXISTS idx_tasks_chat    ON tasks(chat_id);
	CREATE INDEX IF NOT EXISTS idx_tasks_status  ON tasks(status);

	CREATE TABLE IF NOT EXISTS task_history (
		id          INTEGER PRIMARY KEY AUTOINCREMENT,
		task_id     INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
		role        TEXT    NOT NULL,
		content     TEXT    NOT NULL DEFAULT '',
		created_at  INTEGER NOT NULL DEFAULT (unixepoch())
	);

	CREATE INDEX IF NOT EXISTS idx_history_task ON task_history(task_id);

	CREATE TABLE IF NOT EXISTS traces (
		id          INTEGER PRIMARY KEY AUTOINCREMENT,
		task_id     INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
		agent       TEXT    NOT NULL DEFAULT '',
		session_id  TEXT    NOT NULL DEFAULT '',
		prompt      TEXT    NOT NULL DEFAULT '',
		output      TEXT    NOT NULL DEFAULT '',
		status      TEXT    NOT NULL DEFAULT 'running',
		started_at  INTEGER NOT NULL DEFAULT (unixepoch()),
		finished_at INTEGER
	);

	CREATE INDEX IF NOT EXISTS idx_traces_task ON traces(task_id);
	`
	if _, err := s.db.ExecContext(ctx, schema); err != nil {
		return fmt.Errorf("%w: migrate: %w", ErrDB, err)
	}
	// Доп. колонка repos (множественные репозитории). Idempotent: если колонка
	// уже есть, ALTER вернёт ошибку duplicate column — её игнорируем.
	if _, err := s.db.ExecContext(ctx,
		`ALTER TABLE tasks ADD COLUMN repos TEXT NOT NULL DEFAULT ''`); err != nil {
		// SQLite 3.35+ выдаёт "duplicate column name"; ранние версии — "duplicate column".
		if !isDuplicateColumn(err) {
			return fmt.Errorf("%w: migrate add repos: %w", ErrDB, err)
		}
	}
	// Доп. колонка steps (этапы задачи). Idempotent.
	if _, err := s.db.ExecContext(ctx,
		`ALTER TABLE tasks ADD COLUMN steps TEXT NOT NULL DEFAULT ''`); err != nil {
		if !isDuplicateColumn(err) {
			return fmt.Errorf("%w: migrate add steps: %w", ErrDB, err)
		}
	}
	return nil
}

// isDuplicateColumn распознаёт ошибку SQLite «duplicate column name».
func isDuplicateColumn(err error) bool {
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "duplicate column")
}

// Now возвращает текущее время UTC как SQLiteTime.
func Now() SQLiteTime {
	return SQLiteTime(time.Now().UTC())
}