sqlite.go

package store

import (
	"bytes"
	"context"
	"database/sql"
	"embed"
	"errors"
	"fmt"
	"runtime"
	"time"

	"github.com/evstack/apex/pkg/metrics"
	"github.com/evstack/apex/pkg/types"

	_ "modernc.org/sqlite" // registers sqlite driver
)

//go:embed migrations/*.sql
var migrations embed.FS

// SQLiteStore implements Store using modernc.org/sqlite (CGo-free).
// It maintains separate read and write connection pools to the same database.
// The writer is limited to a single connection (WAL single-writer constraint),
// while the reader pool allows concurrent API reads.
type SQLiteStore struct {
	writer  *sql.DB
	reader  *sql.DB
	metrics metrics.Recorder
}

// maxReadConns is the upper bound for the read connection pool.
// Beyond ~8 readers, SQLite WAL contention outweighs parallelism gains.
const maxReadConns = 8

// Open creates or opens a SQLite database at the given path.
// The read pool is sized to min(NumCPU, 8).
// The database is configured with WAL journal mode and a 5-second busy timeout.
func Open(path string) (*SQLiteStore, error) {
	poolSize := min(runtime.NumCPU(), maxReadConns)

	writer, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, fmt.Errorf("open sqlite writer: %w", err)
	}
	writer.SetMaxOpenConns(1)

	ctx := context.Background()

	if err := configureSQLite(ctx, writer); err != nil {
		_ = writer.Close()
		return nil, fmt.Errorf("configure writer: %w", err)
	}

	reader, err := sql.Open("sqlite", path)
	if err != nil {
		_ = writer.Close()
		return nil, fmt.Errorf("open sqlite reader: %w", err)
	}
	reader.SetMaxOpenConns(poolSize)

	if err := configureSQLite(ctx, reader); err != nil {
		_ = writer.Close()
		_ = reader.Close()
		return nil, fmt.Errorf("configure reader: %w", err)
	}

	s := &SQLiteStore{writer: writer, reader: reader, metrics: metrics.Nop()}
	if err := s.migrate(); err != nil {
		_ = writer.Close()
		_ = reader.Close()
		return nil, fmt.Errorf("migrate: %w", err)
	}
	return s, nil
}

// SetMetrics sets the metrics recorder for the store.
func (s *SQLiteStore) SetMetrics(m metrics.Recorder) {
	s.metrics = m
}

func configureSQLite(ctx context.Context, db *sql.DB) error {
	if _, err := db.ExecContext(ctx, "PRAGMA journal_mode=WAL"); err != nil {
		return fmt.Errorf("set WAL mode: %w", err)
	}
	if _, err := db.ExecContext(ctx, "PRAGMA busy_timeout=5000"); err != nil {
		return fmt.Errorf("set busy_timeout: %w", err)
	}
	if _, err := db.ExecContext(ctx, "PRAGMA foreign_keys=ON"); err != nil {
		return fmt.Errorf("set foreign_keys: %w", err)
	}
	// NORMAL is crash-safe with WAL and avoids an extra fsync per commit.
	if _, err := db.ExecContext(ctx, "PRAGMA synchronous=NORMAL"); err != nil {
		return fmt.Errorf("set synchronous: %w", err)
	}
	// 64 MB page cache (negative value = KiB).
	if _, err := db.ExecContext(ctx, "PRAGMA cache_size=-65536"); err != nil {
		return fmt.Errorf("set cache_size: %w", err)
	}
	// Keep temp tables and sort spills in memory.
	if _, err := db.ExecContext(ctx, "PRAGMA temp_store=MEMORY"); err != nil {
		return fmt.Errorf("set temp_store: %w", err)
	}
	return nil
}

// migrationStep defines a single schema migration.
type migrationStep struct {
	version int
	file    string
}

// allMigrations lists every migration in order. Add new entries here.
var allMigrations = []migrationStep{
	{version: 1, file: "migrations/001_init.sql"},
	{version: 2, file: "migrations/002_commitment_index.sql"},
	{version: 3, file: "migrations/003_blob_index_unique.sql"},
}

func (s *SQLiteStore) migrate() error {
	ctx := context.Background()

	var version int
	if err := s.writer.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil {
		return fmt.Errorf("read user_version: %w", err)
	}

	for _, m := range allMigrations {
		if version >= m.version {
			continue
		}
		if err := s.applyMigration(ctx, m); err != nil {
			return err
		}
		version = m.version
	}

	return nil
}

func (s *SQLiteStore) applyMigration(ctx context.Context, m migrationStep) error {
	ddl, err := migrations.ReadFile(m.file)
	if err != nil {
		return fmt.Errorf("read migration %d: %w", m.version, err)
	}

	tx, err := s.writer.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin migration %d tx: %w", m.version, err)
	}
	defer tx.Rollback() //nolint:errcheck

	if _, err := tx.ExecContext(ctx, string(ddl)); err != nil {
		return fmt.Errorf("exec migration %d: %w", m.version, err)
	}
	if _, err := tx.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version = %d", m.version)); err != nil {
		return fmt.Errorf("set user_version to %d: %w", m.version, err)
	}

	return tx.Commit()
}

func (s *SQLiteStore) PutBlobs(ctx context.Context, blobs []types.Blob) error {
	if len(blobs) == 0 {
		return nil
	}
	start := time.Now()
	defer func() { s.metrics.ObserveStoreQueryDuration("PutBlobs", time.Since(start)) }()

	tx, err := s.writer.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	stmt, err := tx.PrepareContext(ctx,
		`INSERT OR IGNORE INTO blobs (height, namespace, commitment, data, share_version, signer, blob_index)
		 VALUES (?, ?, ?, ?, ?, ?, ?)`)
	if err != nil {
		return fmt.Errorf("prepare insert blob: %w", err)
	}
	defer stmt.Close() //nolint:errcheck

	for i := range blobs {
		b := &blobs[i]
		res, err := stmt.ExecContext(ctx,
			b.Height, b.Namespace[:], b.Commitment, b.Data, b.ShareVersion, b.Signer, b.Index,
		)
		if err != nil {
			return fmt.Errorf("insert blob at height %d index %d: %w", b.Height, b.Index, err)
		}
		n, err := res.RowsAffected()
		if err != nil {
			return fmt.Errorf("rows affected at height %d index %d: %w", b.Height, b.Index, err)
		}
		if n == 0 {
			// INSERT OR IGNORE skipped this row — a unique constraint matched.
			// Verify it is an idempotent re-insert, not a data conflict.
			if err := verifyBlobNotConflicting(ctx, tx, b); err != nil {
				return err
			}
		}
	}

	return tx.Commit()
}

func (s *SQLiteStore) GetBlob(ctx context.Context, ns types.Namespace, height uint64, index int) (*types.Blob, error) {
	row := s.reader.QueryRowContext(ctx,
		`SELECT height, namespace, commitment, data, share_version, signer, blob_index
		 FROM blobs WHERE namespace = ? AND height = ? AND blob_index = ?`,
		ns[:], height, index)

	return scanBlob(row)
}

func (s *SQLiteStore) GetBlobs(ctx context.Context, ns types.Namespace, startHeight, endHeight uint64, limit, offset int) ([]types.Blob, error) {
	start := time.Now()
	defer func() { s.metrics.ObserveStoreQueryDuration("GetBlobs", time.Since(start)) }()

	query := `SELECT height, namespace, commitment, data, share_version, signer, blob_index
		 FROM blobs WHERE namespace = ? AND height >= ? AND height <= ?
		 ORDER BY height, blob_index`
	args := []any{ns[:], startHeight, endHeight}

	if limit > 0 {
		query += ` LIMIT ? OFFSET ?`
		args = append(args, limit, offset)
	}

	rows, err := s.reader.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("query blobs: %w", err)
	}
	defer rows.Close() //nolint:errcheck

	blobs := make([]types.Blob, 0, 64) // preallocate to reduce append reallocations
	for rows.Next() {
		b, err := scanBlobRow(rows)
		if err != nil {
			return nil, err
		}
		blobs = append(blobs, b)
	}
	return blobs, rows.Err()
}

func (s *SQLiteStore) GetBlobByCommitment(ctx context.Context, commitment []byte) (*types.Blob, error) {
	start := time.Now()
	defer func() { s.metrics.ObserveStoreQueryDuration("GetBlobByCommitment", time.Since(start)) }()

	row := s.reader.QueryRowContext(ctx,
		`SELECT height, namespace, commitment, data, share_version, signer, blob_index
		 FROM blobs WHERE commitment = ? LIMIT 1`, commitment)
	return scanBlob(row)
}

func (s *SQLiteStore) PutHeader(ctx context.Context, header *types.Header) error {
	start := time.Now()
	defer func() { s.metrics.ObserveStoreQueryDuration("PutHeader", time.Since(start)) }()

	_, err := s.writer.ExecContext(ctx,
		`INSERT OR IGNORE INTO headers (height, hash, data_hash, time_ns, raw_header)
		 VALUES (?, ?, ?, ?, ?)`,
		header.Height, header.Hash, header.DataHash, header.Time.UnixNano(), header.RawHeader)
	if err != nil {
		return fmt.Errorf("insert header at height %d: %w", header.Height, err)
	}
	return nil
}

func (s *SQLiteStore) GetHeader(ctx context.Context, height uint64) (*types.Header, error) {
	start := time.Now()
	defer func() { s.metrics.ObserveStoreQueryDuration("GetHeader", time.Since(start)) }()

	var h types.Header
	var timeNs int64
	err := s.reader.QueryRowContext(ctx,
		`SELECT height, hash, data_hash, time_ns, raw_header FROM headers WHERE height = ?`,
		height).Scan(&h.Height, &h.Hash, &h.DataHash, &timeNs, &h.RawHeader)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("query header at height %d: %w", height, err)
	}
	h.Time = time.Unix(0, timeNs)
	return &h, nil
}

func (s *SQLiteStore) PutNamespace(ctx context.Context, ns types.Namespace) error {
	_, err := s.writer.ExecContext(ctx,
		`INSERT OR IGNORE INTO namespaces (namespace) VALUES (?)`, ns[:])
	if err != nil {
		return fmt.Errorf("insert namespace: %w", err)
	}
	return nil
}

func (s *SQLiteStore) GetNamespaces(ctx context.Context) ([]types.Namespace, error) {
	rows, err := s.reader.QueryContext(ctx, `SELECT namespace FROM namespaces`)
	if err != nil {
		return nil, fmt.Errorf("query namespaces: %w", err)
	}
	defer rows.Close() //nolint:errcheck

	var namespaces []types.Namespace
	for rows.Next() {
		var nsBytes []byte
		if err := rows.Scan(&nsBytes); err != nil {
			return nil, fmt.Errorf("scan namespace: %w", err)
		}
		if len(nsBytes) != types.NamespaceSize {
			return nil, fmt.Errorf("invalid namespace size: got %d, want %d", len(nsBytes), types.NamespaceSize)
		}
		var ns types.Namespace
		copy(ns[:], nsBytes)
		namespaces = append(namespaces, ns)
	}
	return namespaces, rows.Err()
}

func (s *SQLiteStore) GetSyncState(ctx context.Context) (*types.SyncStatus, error) {
	start := time.Now()
	defer func() { s.metrics.ObserveStoreQueryDuration("GetSyncState", time.Since(start)) }()

	var state int
	var latestHeight, networkHeight uint64
	err := s.reader.QueryRowContext(ctx,
		`SELECT state, latest_height, network_height FROM sync_state WHERE id = 1`).
		Scan(&state, &latestHeight, &networkHeight)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("query sync_state: %w", err)
	}
	return &types.SyncStatus{
		State:         types.SyncState(state),
		LatestHeight:  latestHeight,
		NetworkHeight: networkHeight,
	}, nil
}

func (s *SQLiteStore) SetSyncState(ctx context.Context, status types.SyncStatus) error {
	_, err := s.writer.ExecContext(ctx,
		`INSERT INTO sync_state (id, state, latest_height, network_height, updated_at)
		 VALUES (1, ?, ?, ?, ?)
		 ON CONFLICT(id) DO UPDATE SET
		   state = excluded.state,
		   latest_height = excluded.latest_height,
		   network_height = excluded.network_height,
		   updated_at = excluded.updated_at`,
		int(status.State), status.LatestHeight, status.NetworkHeight, time.Now().UnixNano())
	if err != nil {
		return fmt.Errorf("upsert sync_state: %w", err)
	}
	return nil
}

func (s *SQLiteStore) Close() error {
	return errors.Join(s.reader.Close(), s.writer.Close())
}

// scanBlob scans a single blob from a *sql.Row.
func scanBlob(row *sql.Row) (*types.Blob, error) {
	var b types.Blob
	var nsBytes []byte
	err := row.Scan(&b.Height, &nsBytes, &b.Commitment, &b.Data, &b.ShareVersion, &b.Signer, &b.Index)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("scan blob: %w", err)
	}
	copy(b.Namespace[:], nsBytes)
	return &b, nil
}

// scanBlobRow scans a single blob from *sql.Rows.
func scanBlobRow(rows *sql.Rows) (types.Blob, error) {
	var b types.Blob
	var nsBytes []byte
	err := rows.Scan(&b.Height, &nsBytes, &b.Commitment, &b.Data, &b.ShareVersion, &b.Signer, &b.Index)
	if err != nil {
		return types.Blob{}, fmt.Errorf("scan blob row: %w", err)
	}
	copy(b.Namespace[:], nsBytes)
	return b, nil
}

// verifyBlobNotConflicting is called only when INSERT OR IGNORE skipped a row.
// It distinguishes an idempotent re-insert (same data) from a true conflict
// (different data at the same position or commitment).
func verifyBlobNotConflicting(ctx context.Context, tx *sql.Tx, b *types.Blob) error {
	existing, err := queryBlobByIndex(ctx, tx, b.Namespace, b.Height, b.Index)
	if err != nil {
		return err
	}
	if existing != nil && !sameBlob(existing, b) {
		return fmt.Errorf("blob conflict at height %d namespace %s index %d", b.Height, b.Namespace, b.Index)
	}

	existingByCommitment, err := queryBlobByCommitment(ctx, tx, b.Commitment)
	if err != nil {
		return err
	}
	if existingByCommitment != nil && !sameBlob(existingByCommitment, b) {
		return fmt.Errorf("blob commitment conflict for %x", b.Commitment)
	}

	return nil
}

func queryBlobByIndex(ctx context.Context, tx *sql.Tx, ns types.Namespace, height uint64, index int) (*types.Blob, error) {
	row := tx.QueryRowContext(ctx,
		`SELECT height, namespace, commitment, data, share_version, signer, blob_index
		 FROM blobs WHERE namespace = ? AND height = ? AND blob_index = ?`,
		ns[:], height, index)
	b, err := scanBlob(row)
	if errors.Is(err, ErrNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("query blob by index: %w", err)
	}
	return b, nil
}

func queryBlobByCommitment(ctx context.Context, tx *sql.Tx, commitment []byte) (*types.Blob, error) {
	row := tx.QueryRowContext(ctx,
		`SELECT height, namespace, commitment, data, share_version, signer, blob_index
		 FROM blobs WHERE commitment = ? LIMIT 1`, commitment)
	b, err := scanBlob(row)
	if errors.Is(err, ErrNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("query blob by commitment: %w", err)
	}
	return b, nil
}

func sameBlob(a, b *types.Blob) bool {
	return a.Height == b.Height &&
		a.Namespace == b.Namespace &&
		a.Index == b.Index &&
		a.ShareVersion == b.ShareVersion &&
		bytes.Equal(a.Commitment, b.Commitment) &&
		bytes.Equal(a.Data, b.Data) &&
		bytes.Equal(a.Signer, b.Signer)
}