tutti/services/tuttid/data/workspace/sqlite_store.go

package workspace

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"time"

	agentstore "github.com/tutti-os/tutti/packages/agent/store-sqlite"
	workspaceissues "github.com/tutti-os/tutti/packages/workspace/issues"
	workspacebiz "github.com/tutti-os/tutti/services/tuttid/biz/workspace"
	tuttitypes "github.com/tutti-os/tutti/services/tuttid/types"
	sqlitedriver "modernc.org/sqlite"
)

const defaultSQLiteBusyTimeoutMillisec = 5000
const defaultSQLiteReaderConnections = 4

const sqliteIOErrTruncate = 1546

var sqliteWALRetryDelays = []time.Duration{100 * time.Millisecond, 300 * time.Millisecond}

type SQLiteStore struct {
	dbPath                 string
	writeDB                *sql.DB
	readDB                 *sql.DB
	agentWriter            *agentstore.Store
	agentReader            *agentstore.Store
	sourceActivityRowsHook func(tuttiModeSourceActivityRows) tuttiModeSourceActivityRows
}

type sqliteOnlineBackuper interface {
	NewBackup(string) (*sqlitedriver.Backup, error)
}

type sqlitePragmaExecutor interface {
	Exec(string, ...any) (sql.Result, error)
}

type sqliteErrorCoder interface {
	Code() int
}

func OpenSQLiteStore(dbPath string) (*SQLiteStore, error) {
	dbPath = strings.TrimSpace(dbPath)
	if dbPath == "" {
		return nil, errors.New("workspace database path is empty")
	}
	if err := os.MkdirAll(filepath.Dir(dbPath), 0o755); err != nil {
		return nil, fmt.Errorf("create tutti database directory: %w", err)
	}

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

	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	store := &SQLiteStore{dbPath: dbPath, writeDB: db}
	store.agentWriter = newAgentStore(db)

	if err := enableSQLiteWAL(db, time.Sleep); err != nil {
		_ = store.Close()
		return nil, fmt.Errorf("enable sqlite wal mode: %w", err)
	}

	return store, nil
}

func enableSQLiteWAL(executor sqlitePragmaExecutor, sleep func(time.Duration)) error {
	for attempt := 0; ; attempt++ {
		if _, err := executor.Exec("PRAGMA journal_mode = WAL"); err != nil {
			if attempt >= len(sqliteWALRetryDelays) || !isSQLiteIOErrTruncate(err) {
				return err
			}
			delay := sqliteWALRetryDelays[attempt]
			slog.Warn("retrying sqlite wal mode after transient truncate error",
				"event", "workspace.sqlite.wal_retry",
				"attempt", attempt+1,
				"retry_in_ms", delay.Milliseconds(),
				"sqlite_error_code", sqliteIOErrTruncate,
				"error", err)
			sleep(delay)
			continue
		}
		return nil
	}
}

func isSQLiteIOErrTruncate(err error) bool {
	var coded sqliteErrorCoder
	return errors.As(err, &coded) && coded.Code() == sqliteIOErrTruncate
}

func DefaultDBPath() string {
	return tuttitypes.TuttidDBPath()
}

func (s *SQLiteStore) Close() error {
	if s == nil {
		return nil
	}
	var readErr error
	if s.readDB != nil {
		readErr = s.readDB.Close()
	}
	var writeErr error
	if s.writeDB != nil {
		writeErr = s.writeDB.Close()
	}
	return errors.Join(readErr, writeErr)
}

func (s *SQLiteStore) BackupTo(ctx context.Context, destination string) error {
	if s == nil || s.writeDB == nil {
		return errors.New("workspace database is not initialized")
	}
	destination = filepath.Clean(strings.TrimSpace(destination))
	if destination == "." || destination == "" {
		return errors.New("workspace database backup destination is empty")
	}
	if err := os.MkdirAll(filepath.Dir(destination), 0o700); err != nil {
		return fmt.Errorf("create workspace database backup directory: %w", err)
	}
	if err := os.Remove(destination); err != nil && !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("replace workspace database backup: %w", err)
	}
	connection, err := s.writeDB.Conn(ctx)
	if err != nil {
		return fmt.Errorf("open workspace database backup connection: %w", err)
	}
	defer func() { _ = connection.Close() }()
	if err := connection.Raw(func(driverConnection any) error {
		backuper, ok := driverConnection.(sqliteOnlineBackuper)
		if !ok {
			return errors.New("sqlite online backup is unavailable")
		}
		backup, err := backuper.NewBackup(destination)
		if err != nil {
			return err
		}
		more := true
		for more {
			if err := ctx.Err(); err != nil {
				_ = backup.Finish()
				return err
			}
			more, err = backup.Step(256)
			if err != nil {
				_ = backup.Finish()
				return err
			}
		}
		return backup.Finish()
	}); err != nil {
		_ = os.Remove(destination)
		return fmt.Errorf("backup workspace database: %w", err)
	}
	if err := os.Chmod(destination, 0o600); err != nil {
		return fmt.Errorf("protect workspace database backup: %w", err)
	}
	return nil
}

func (s *SQLiteStore) DatabaseSizeBytes() (int64, error) {
	if s == nil || strings.TrimSpace(s.dbPath) == "" {
		return 0, errors.New("workspace database is not initialized")
	}
	info, err := os.Stat(s.dbPath)
	if err != nil {
		return 0, fmt.Errorf("stat workspace database: %w", err)
	}
	return info.Size(), nil
}

func sqliteWriterDSN(dbPath string) string {
	return sqliteDSN(dbPath, false)
}

func sqliteReaderDSN(dbPath string) string {
	return sqliteDSN(dbPath, true)
}

func sqliteDSN(dbPath string, readOnly bool) string {
	query := url.Values{}
	query.Add("_pragma", fmt.Sprintf("busy_timeout(%d)", defaultSQLiteBusyTimeoutMillisec))
	query.Add("_pragma", "foreign_keys(1)")
	if readOnly {
		query.Set("mode", "ro")
		query.Add("_pragma", "query_only(1)")
	}
	return (&url.URL{Scheme: "file", Path: sqliteDSNPath(dbPath, runtime.GOOS), RawQuery: query.Encode()}).String()
}

func sqliteDSNPath(dbPath string, goos string) string {
	if goos != "windows" {
		return dbPath
	}
	path := strings.ReplaceAll(dbPath, `\`, "/")
	if len(path) >= 2 && path[1] == ':' && !strings.HasPrefix(path, "/") {
		path = "/" + path
	}
	return path
}

func (s *SQLiteStore) openReadPool(ctx context.Context) error {
	if s == nil || s.writeDB == nil {
		return errors.New("workspace database is not initialized")
	}
	if s.readDB != nil {
		return nil
	}

	db, err := sql.Open("sqlite", sqliteReaderDSN(s.dbPath))
	if err != nil {
		return fmt.Errorf("open tutti database read pool: %w", err)
	}
	db.SetMaxOpenConns(defaultSQLiteReaderConnections)
	db.SetMaxIdleConns(defaultSQLiteReaderConnections)
	if err := db.PingContext(ctx); err != nil {
		_ = db.Close()
		return fmt.Errorf("ping tutti database read pool: %w", err)
	}

	s.readDB = db
	s.agentReader = newAgentStore(db)
	return nil
}

func (s *SQLiteStore) Create(ctx context.Context, item workspacebiz.Summary) error {
	if s == nil || s.writeDB == nil {
		return errors.New("workspace database is not initialized")
	}

	id := strings.TrimSpace(item.ID)
	name := strings.TrimSpace(item.Name)
	if id == "" || name == "" {
		return errors.New("workspace id and name are required")
	}

	now := unixMs(time.Now().UTC())
	tx, err := s.writeDB.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin create workspace: %w", err)
	}
	defer func() {
		_ = tx.Rollback()
	}()

	_, err = tx.ExecContext(ctx, `
INSERT INTO workspaces (id, name, created_at_unix_ms, updated_at_unix_ms, last_opened_at_unix_ms)
VALUES (?, ?, ?, ?, NULL)
`, id, name, now, now)
	if err != nil {
		return fmt.Errorf("create workspace: %w", err)
	}

	_, err = tx.ExecContext(ctx, `
INSERT INTO workspace_issue_topics (
  topic_id, workspace_id, title, summary, is_default, pinned_at_unix_ms,
  last_activity_at_unix_ms, created_at_unix_ms, updated_at_unix_ms
) VALUES (?, ?, ?, '', 1, 0, ?, ?, ?)
`, workspaceissues.DefaultTopicID, id, workspaceissues.DefaultTopicID, now, now, now)
	if err != nil {
		return fmt.Errorf("create default workspace issue topic: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit create workspace: %w", err)
	}

	return nil
}

func (s *SQLiteStore) Delete(ctx context.Context, workspaceID string) error {
	if s == nil || s.writeDB == nil {
		return errors.New("workspace database is not initialized")
	}

	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return errors.New("workspace id is required")
	}

	tx, err := s.writeDB.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin delete workspace: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()

	result, err := tx.ExecContext(ctx, `
DELETE FROM workspaces
WHERE id = ?
`, workspaceID)
	if err != nil {
		return fmt.Errorf("delete workspace: %w", err)
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("delete workspace rows affected: %w", err)
	}
	if rowsAffected == 0 {
		return ErrWorkspaceNotFound
	}

	// Agent activity tables no longer carry a foreign key into workspaces on
	// fresh schemas; cascade the deletion explicitly through the agent store
	// inside the same transaction so a failure leaves no orphaned agent rows.
	if _, err := s.agentStore().ClearSessionsTx(ctx, tx, workspaceID); err != nil {
		return fmt.Errorf("clear agent sessions for deleted workspace: %w", err)
	}
	// Older Tutti-mode schemas predate the workspace foreign key on immutable
	// turn snapshots. Keep deletion explicit so upgraded databases cannot retain
	// orphaned reservations.
	if _, err := tx.ExecContext(ctx, `DELETE FROM tutti_mode_turn_snapshots WHERE workspace_id = ?`, workspaceID); err != nil {
		return fmt.Errorf("clear Tutti mode turn snapshots for deleted workspace: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit delete workspace: %w", err)
	}
	committed = true

	return nil
}

func (s *SQLiteStore) Get(ctx context.Context, workspaceID string) (workspacebiz.Summary, error) {
	if s == nil || s.writeDB == nil {
		return workspacebiz.Summary{}, errors.New("workspace database is not initialized")
	}

	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return workspacebiz.Summary{}, errors.New("workspace id is required")
	}

	row := s.readDB.QueryRowContext(ctx, `
SELECT id, name
     , last_opened_at_unix_ms
FROM workspaces
WHERE id = ?
`, workspaceID)

	var item workspacebiz.Summary
	var lastOpenedAt sql.NullInt64
	if err := row.Scan(&item.ID, &item.Name, &lastOpenedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return workspacebiz.Summary{}, ErrWorkspaceNotFound
		}
		return workspacebiz.Summary{}, fmt.Errorf("get workspace: %w", err)
	}
	item.LastOpenedAt = nullableUnixMs(lastOpenedAt)

	return item, nil
}

func (s *SQLiteStore) List(ctx context.Context) ([]workspacebiz.Summary, error) {
	if s == nil || s.writeDB == nil {
		return nil, errors.New("workspace database is not initialized")
	}

	rows, err := s.readDB.QueryContext(ctx, `
SELECT id, name, last_opened_at_unix_ms
FROM workspaces
ORDER BY COALESCE(last_opened_at_unix_ms, 0) DESC, updated_at_unix_ms DESC, id ASC`)
	if err != nil {
		return nil, fmt.Errorf("list workspaces: %w", err)
	}
	defer rows.Close()

	items := make([]workspacebiz.Summary, 0)
	for rows.Next() {
		var item workspacebiz.Summary
		var lastOpenedAt sql.NullInt64
		if err := rows.Scan(&item.ID, &item.Name, &lastOpenedAt); err != nil {
			return nil, fmt.Errorf("scan workspace: %w", err)
		}
		item.LastOpenedAt = nullableUnixMs(lastOpenedAt)
		items = append(items, item)
	}

	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workspaces: %w", err)
	}

	return items, nil
}

func (s *SQLiteStore) GetStartup(ctx context.Context) (*workspacebiz.Summary, error) {
	if s == nil || s.writeDB == nil {
		return nil, errors.New("workspace database is not initialized")
	}

	row := s.readDB.QueryRowContext(ctx, `
SELECT id, name, last_opened_at_unix_ms
FROM workspaces
WHERE last_opened_at_unix_ms IS NOT NULL
ORDER BY last_opened_at_unix_ms DESC, updated_at_unix_ms DESC, id ASC
LIMIT 1`)

	var item workspacebiz.Summary
	var lastOpenedAt sql.NullInt64
	if err := row.Scan(&item.ID, &item.Name, &lastOpenedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, nil
		}
		return nil, fmt.Errorf("get startup workspace: %w", err)
	}

	item.LastOpenedAt = nullableUnixMs(lastOpenedAt)
	return &item, nil
}

func (s *SQLiteStore) Update(ctx context.Context, item workspacebiz.Summary) error {
	if s == nil || s.writeDB == nil {
		return errors.New("workspace database is not initialized")
	}

	id := strings.TrimSpace(item.ID)
	name := strings.TrimSpace(item.Name)
	if id == "" || name == "" {
		return errors.New("workspace id and name are required")
	}

	result, err := s.writeDB.ExecContext(ctx, `
UPDATE workspaces
SET name = ?, updated_at_unix_ms = ?
WHERE id = ?
`, name, unixMs(time.Now().UTC()), id)
	if err != nil {
		return fmt.Errorf("update workspace: %w", err)
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("update workspace rows affected: %w", err)
	}
	if rowsAffected == 0 {
		return ErrWorkspaceNotFound
	}

	return nil
}

func (s *SQLiteStore) Open(ctx context.Context, workspaceID string) (workspacebiz.Summary, error) {
	if s == nil || s.writeDB == nil {
		return workspacebiz.Summary{}, errors.New("workspace database is not initialized")
	}

	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return workspacebiz.Summary{}, errors.New("workspace id is required")
	}

	now := unixMs(time.Now().UTC())
	result, err := s.writeDB.ExecContext(ctx, `
UPDATE workspaces
SET last_opened_at_unix_ms = ?, updated_at_unix_ms = ?
WHERE id = ?
`, now, now, workspaceID)
	if err != nil {
		return workspacebiz.Summary{}, fmt.Errorf("open workspace: %w", err)
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return workspacebiz.Summary{}, fmt.Errorf("open workspace rows affected: %w", err)
	}
	if rowsAffected == 0 {
		return workspacebiz.Summary{}, ErrWorkspaceNotFound
	}

	return s.Get(ctx, workspaceID)
}

func nullableUnixMs(value sql.NullInt64) *time.Time {
	if !value.Valid {
		return nil
	}

	converted := time.UnixMilli(value.Int64).UTC()
	return &converted
}

func unixMs(value time.Time) int64 {
	return value.UnixMilli()
}