vaultik/internal/database/database.go

// Package database provides the local SQLite index for Vaultik backup operations.
// The database tracks files, chunks, and their associations with blobs.
//
// Blobs in Vaultik are the final storage units uploaded to S3. Each blob is a
// large (up to 10GB) file containing many compressed and encrypted chunks from
// multiple source files. Blobs are content-addressed: the filename in S3 is
// hex(SHA256(SHA256(uncompressed blob contents))), computed from the chunk data
// before compression and encryption (not from the stored bytes). See
// blobgen.DoubleSHA256 and docs/REPOSTRUCTURE.md.
//
// Schema is managed via numbered SQL migrations embedded in the schema/
// directory. Migration 000.sql bootstraps the schema_migrations tracking
// table; subsequent migrations (001, 002, …) are applied in order.
package database

import (
	"context"
	"database/sql"
	"embed"
	"errors"
	"fmt"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"

	// Register the pure-Go sqlite driver.
	_ "modernc.org/sqlite"
	"sneak.berlin/go/vaultik/internal/log"
)

// errInvalidMigrationFilename is returned when an embedded migration file
// does not follow the "<version>[_<description>].sql" naming pattern.
var errInvalidMigrationFilename = errors.New("invalid migration filename")

//go:embed schema/*.sql
var schemaFS embed.FS

// bootstrapVersion is the migration that creates the schema_migrations
// table itself. It is applied before the normal migration loop.
const bootstrapVersion = 0

// DB represents the Vaultik local index database connection.
// It uses SQLite to track file metadata, content-defined chunks, and blob associations.
// The database enables incremental backups by detecting changed files and
// supports deduplication by tracking which chunks are already stored in blobs.
// Write operations are synchronized through a mutex to ensure thread safety.
type DB struct {
	conn *sql.DB
	path string
}

// ParseMigrationVersion extracts the numeric version prefix from a migration
// filename. Filenames must follow the pattern "<version>.sql" or
// "<version>_<description>.sql", where version is a zero-padded numeric
// string (e.g. "001", "002"). Returns the version as an integer and an
// error if the filename does not match the expected pattern.
func ParseMigrationVersion(filename string) (int, error) {
	name := strings.TrimSuffix(filename, filepath.Ext(filename))
	if name == "" {
		return 0, fmt.Errorf("%w %q: empty name", errInvalidMigrationFilename, filename)
	}

	// Split on underscore to separate version from description.
	// If there's no underscore, the entire stem is the version.
	versionStr := name
	if before, _, ok := strings.Cut(name, "_"); ok {
		versionStr = before
	}

	if versionStr == "" {
		return 0, fmt.Errorf(
			"%w %q: empty version prefix", errInvalidMigrationFilename, filename,
		)
	}

	// Validate the version is purely numeric.
	for _, ch := range versionStr {
		if ch < '0' || ch > '9' {
			return 0, fmt.Errorf(
				"%w %q: version %q contains non-numeric character %q",
				errInvalidMigrationFilename, filename, versionStr, string(ch),
			)
		}
	}

	version, err := strconv.Atoi(versionStr)
	if err != nil {
		return 0, fmt.Errorf("invalid migration filename %q: %w", filename, err)
	}

	return version, nil
}

// New creates a new database connection at the specified path.
// It creates the schema if needed and configures SQLite with WAL mode for
// better concurrency. SQLite handles crash recovery automatically when
// opening a database with journal/WAL files present.
// The path parameter can be a file path for persistent storage or ":memory:"
// for an in-memory database (useful for testing).
func New(ctx context.Context, path string) (*DB, error) {
	log.Debug("Opening database connection", "path", path)

	// Note: We do NOT delete journal/WAL files before opening.
	// SQLite handles crash recovery automatically when the database is opened.
	// Deleting these files would corrupt the database after an unclean shutdown.

	// First attempt with standard WAL mode
	log.Debug("Attempting to open database with WAL mode", "path", path)

	conn, err := sql.Open(
		"sqlite",
		path+"?_journal_mode=WAL&_synchronous=NORMAL&_busy_timeout=10000"+
			"&_locking_mode=NORMAL&_foreign_keys=ON",
	)
	if err == nil {
		configureConnPool(conn)

		err = conn.PingContext(ctx)
		if err == nil {
			// Success on first try
			log.Debug("Database opened successfully with WAL mode", "path", path)

			return finishOpen(ctx, conn, path)
		}

		log.Debug(
			"Failed to ping database, closing connection",
			"path", path, "error", err,
		)

		_ = conn.Close()
	}

	// If first attempt failed, try with TRUNCATE mode to clear any locks
	return openWithRecovery(ctx, path)
}

// configureConnPool serializes all database access through one connection.
// SQLite can handle multiple readers but only one writer at a time; setting
// MaxOpenConns to 1 ensures all writes go through a single connection,
// preventing SQLITE_BUSY errors.
func configureConnPool(conn *sql.DB) {
	conn.SetMaxOpenConns(1)
	conn.SetMaxIdleConns(1)
}

// finishOpen enables foreign keys, wraps the connection, and applies any
// pending migrations. On migration failure the connection is closed.
func finishOpen(ctx context.Context, conn *sql.DB, path string) (*DB, error) {
	// Enable foreign keys explicitly
	_, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = ON")
	if err != nil {
		log.Warn("Failed to enable foreign keys", "path", path, "error", err)
	}

	db := &DB{conn: conn, path: path}

	err = applyMigrations(ctx, conn)
	if err != nil {
		_ = conn.Close()

		return nil, fmt.Errorf("applying migrations: %w", err)
	}

	return db, nil
}

// openWithRecovery retries opening the database in TRUNCATE journal mode to
// clear stale locks, then switches back to WAL mode.
func openWithRecovery(ctx context.Context, path string) (*DB, error) {
	log.Info(
		"Database appears locked, attempting recovery with TRUNCATE mode",
		"path", path,
	)

	conn, err := sql.Open(
		"sqlite",
		path+"?_journal_mode=TRUNCATE&_synchronous=NORMAL&_busy_timeout=10000"+
			"&_foreign_keys=ON",
	)
	if err != nil {
		return nil, fmt.Errorf("opening database in recovery mode: %w", err)
	}

	configureConnPool(conn)

	err = conn.PingContext(ctx)
	if err != nil {
		log.Debug(
			"Failed to ping database in recovery mode, closing",
			"path", path, "error", err,
		)

		_ = conn.Close()

		return nil, fmt.Errorf(
			"database still locked after recovery attempt: %w",
			err,
		)
	}

	log.Debug("Database opened in TRUNCATE mode", "path", path)

	// Switch back to WAL mode
	log.Debug("Switching database back to WAL mode", "path", path)

	_, err = conn.ExecContext(ctx, "PRAGMA journal_mode=WAL")
	if err != nil {
		log.Warn("Failed to switch back to WAL mode", "path", path, "error", err)
	}

	db, err := finishOpen(ctx, conn, path)
	if err != nil {
		return nil, err
	}

	log.Debug("Database connection established successfully", "path", path)

	return db, nil
}

// errUntrustedSnapshotSchema is returned when a downloaded snapshot
// database carries schema objects the real schema never defines, or is
// missing a table the restore and deep-verify queries read.
var errUntrustedSnapshotSchema = errors.New(
	"downloaded snapshot database has an untrusted schema")

// snapshotReadOnlyDSN builds the driver DSN that opens a materialized
// snapshot database file read-only. mode=ro opens the file read-only at
// the OS level, query_only rejects any write the engine is asked to make,
// and trusted_schema=OFF refuses to run application code named in the
// schema. The file: URI form is required for the driver to honour the
// mode parameter.
func snapshotReadOnlyDSN(path string) string {
	u := url.URL{
		Scheme:   "file",
		Path:     path,
		RawQuery: "mode=ro&_pragma=query_only(true)&_pragma=trusted_schema(false)",
	}

	return u.String()
}

// OpenReadOnly opens an already-materialized SQLite file for read-only
// querying of a snapshot database downloaded from the store, used by
// restore and deep verify. Unlike New it never applies schema migrations
// and never writes: the connection is opened read-only with query_only
// and trusted_schema=OFF. It refuses any file whose schema carries a
// trigger, view or virtual table, or lacks an expected table, so a forged
// file cannot redefine what the restore queries return. The caller owns
// the file and must remove it.
func OpenReadOnly(ctx context.Context, path string) (*DB, error) {
	conn, err := sql.Open("sqlite", snapshotReadOnlyDSN(path))
	if err != nil {
		return nil, fmt.Errorf("opening read-only database: %w", err)
	}

	configureConnPool(conn)

	err = conn.PingContext(ctx)
	if err != nil {
		_ = conn.Close()

		return nil, fmt.Errorf("opening read-only database: %w", err)
	}

	err = verifySnapshotSchema(ctx, conn)
	if err != nil {
		_ = conn.Close()

		return nil, err
	}

	return &DB{conn: conn, path: path}, nil
}

// verifySnapshotSchema rejects a downloaded database whose schema is not
// the plain table set the real schema defines. Any trigger, view or
// virtual table, or a missing expected table, fails the open.
func verifySnapshotSchema(ctx context.Context, conn *sql.DB) error {
	// expectedSnapshotTables are the tables the restore and deep-verify
	// queries read. A downloaded database missing any of them is not a
	// genuine snapshot database and is refused.
	expectedSnapshotTables := []string{
		"blob_chunks",
		"blobs",
		"chunks",
		"file_chunks",
		"files",
	}

	rows, err := conn.QueryContext(
		ctx, "SELECT type, name, sql FROM sqlite_master")
	if err != nil {
		return fmt.Errorf("reading snapshot schema: %w", err)
	}

	defer func() { _ = rows.Close() }()

	present := make(map[string]struct{})

	for rows.Next() {
		var objType, name string

		var objSQL sql.NullString

		err = rows.Scan(&objType, &name, &objSQL)
		if err != nil {
			return fmt.Errorf("reading snapshot schema: %w", err)
		}

		switch objType {
		case "trigger", "view":
			return fmt.Errorf(
				"%w: unexpected %s %q", errUntrustedSnapshotSchema, objType, name)
		case "table":
			if isVirtualTableSQL(objSQL.String) {
				return fmt.Errorf(
					"%w: unexpected virtual table %q",
					errUntrustedSnapshotSchema, name)
			}

			present[name] = struct{}{}
		}
	}

	err = rows.Err()
	if err != nil {
		return fmt.Errorf("reading snapshot schema: %w", err)
	}

	for _, table := range expectedSnapshotTables {
		if _, ok := present[table]; !ok {
			return fmt.Errorf(
				"%w: missing table %q", errUntrustedSnapshotSchema, table)
		}
	}

	return nil
}

// isVirtualTableSQL reports whether a sqlite_master row's SQL defines a
// virtual table. Virtual tables are recorded with type 'table' but a
// "CREATE VIRTUAL TABLE" definition and can run module code, so they are
// refused alongside triggers and views.
func isVirtualTableSQL(createSQL string) bool {
	return strings.HasPrefix(
		strings.ToUpper(strings.TrimSpace(createSQL)), "CREATE VIRTUAL TABLE")
}

// NewTestDB creates an in-memory SQLite database for testing purposes.
// The database is automatically initialized with the schema and is ready
// for use. Each call creates a new independent database instance.
func NewTestDB() (*DB, error) {
	return New(context.Background(), ":memory:")
}

// Close closes the database connection.
// It ensures all pending operations are completed before closing.
// Returns an error if the database connection cannot be closed properly.
func (db *DB) Close() error {
	log.Debug("Closing database connection", "path", db.path)

	err := db.conn.Close()
	if err != nil {
		log.Error("Failed to close database", "path", db.path, "error", err)

		return fmt.Errorf("failed to close database: %w", err)
	}

	log.Debug("Database connection closed successfully", "path", db.path)

	return nil
}

// Conn returns the underlying *sql.DB connection.
// This should be used sparingly and primarily for read operations.
// For write operations, prefer using the ExecWithLog method.
func (db *DB) Conn() *sql.DB {
	return db.conn
}

// Path returns the path to the database file.
func (db *DB) Path() string {
	return db.path
}

// BeginTx starts a new database transaction with the given options.
// The caller is responsible for committing or rolling back the transaction.
// For write transactions, consider using the Repositories.WithTx method instead,
// which handles locking and rollback automatically.
func (db *DB) BeginTx(
	ctx context.Context,
	opts *sql.TxOptions,
) (*sql.Tx, error) {
	return db.conn.BeginTx(ctx, opts)
}

// Note: LockForWrite and UnlockWrite methods have been removed.
// SQLite handles its own locking internally, so explicit locking is not needed.

// ExecWithLog executes a write query with SQL logging.
// SQLite handles its own locking internally, so we just pass through to ExecContext.
// The query and args parameters follow the same format as sql.DB.ExecContext.
func (db *DB) ExecWithLog(
	ctx context.Context,
	query string,
	args ...any,
) (sql.Result, error) {
	LogSQL("Execute", query, args...)

	return db.conn.ExecContext(ctx, query, args...)
}

// QueryRowWithLog executes a query that returns at most one row with SQL
// logging. This is useful for queries that modify data and return values
// (e.g., INSERT ... RETURNING). SQLite handles its own locking internally.
// The query and args parameters follow the same format as
// sql.DB.QueryRowContext.
func (db *DB) QueryRowWithLog(
	ctx context.Context,
	query string,
	args ...any,
) *sql.Row {
	LogSQL("QueryRow", query, args...)

	return db.conn.QueryRowContext(ctx, query, args...)
}

// collectMigrations reads the embedded schema directory and returns
// migration filenames sorted lexicographically.
func collectMigrations() ([]string, error) {
	entries, err := schemaFS.ReadDir("schema")
	if err != nil {
		return nil, fmt.Errorf("failed to read schema directory: %w", err)
	}

	var migrations []string

	for _, entry := range entries {
		if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".sql") {
			migrations = append(migrations, entry.Name())
		}
	}

	sort.Strings(migrations)

	return migrations, nil
}

// bootstrapMigrationsTable ensures the schema_migrations table exists
// by applying 000.sql if the table is missing.
func bootstrapMigrationsTable(ctx context.Context, db *sql.DB) error {
	var tableExists int

	err := db.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='schema_migrations'",
	).Scan(&tableExists)
	if err != nil {
		return fmt.Errorf("failed to check for migrations table: %w", err)
	}

	if tableExists > 0 {
		return nil
	}

	content, err := schemaFS.ReadFile("schema/000.sql")
	if err != nil {
		return fmt.Errorf("failed to read bootstrap migration 000.sql: %w", err)
	}

	log.Info("applying bootstrap migration", "version", bootstrapVersion)

	_, err = db.ExecContext(ctx, string(content))
	if err != nil {
		return fmt.Errorf("failed to apply bootstrap migration: %w", err)
	}

	return nil
}

// applyMigrations applies all pending migrations to db. It first bootstraps
// the schema_migrations table via 000.sql, then iterates through remaining
// migration files in order.
func applyMigrations(ctx context.Context, db *sql.DB) error {
	err := bootstrapMigrationsTable(ctx, db)
	if err != nil {
		return err
	}

	migrations, err := collectMigrations()
	if err != nil {
		return err
	}

	for _, migration := range migrations {
		version, parseErr := ParseMigrationVersion(migration)
		if parseErr != nil {
			return parseErr
		}

		// Check if already applied.
		var count int

		err := db.QueryRowContext(ctx,
			"SELECT COUNT(*) FROM schema_migrations WHERE version = ?",
			version,
		).Scan(&count)
		if err != nil {
			return fmt.Errorf("failed to check migration status: %w", err)
		}

		if count > 0 {
			log.Debug("migration already applied", "version", version)

			continue
		}

		// Read and apply migration.
		content, readErr := schemaFS.ReadFile(filepath.Join("schema", migration))
		if readErr != nil {
			return fmt.Errorf("failed to read migration %s: %w", migration, readErr)
		}

		log.Info("applying migration", "version", version)

		_, execErr := db.ExecContext(ctx, string(content))
		if execErr != nil {
			return fmt.Errorf("failed to apply migration %s: %w", migration, execErr)
		}

		// Record migration as applied.
		_, recErr := db.ExecContext(ctx,
			"INSERT INTO schema_migrations (version) VALUES (?)",
			version,
		)
		if recErr != nil {
			return fmt.Errorf("failed to record migration %s: %w", migration, recErr)
		}

		log.Info("migration applied successfully", "version", version)
	}

	return nil
}

// repeatPlaceholder generates a string of ", ?" repeated n times for IN
// clause construction. For example, repeatPlaceholder(2) returns ", ?, ?".
func repeatPlaceholder(n int) string {
	if n <= 0 {
		return ""
	}

	return strings.Repeat(", ?", n)
}

// LogSQL logs SQL queries and their arguments when debug mode is enabled.
// Debug mode is activated by setting the GODEBUG environment variable to
// include "vaultik". This is useful for troubleshooting database operations
// and understanding query patterns.
//
// The operation parameter describes the type of SQL operation (e.g.,
// "Execute", "Query"). The query parameter is the SQL statement being
// executed. The args parameter contains the query arguments that will be
// interpolated.
func LogSQL(operation, query string, args ...any) {
	if strings.Contains(os.Getenv("GODEBUG"), "vaultik") {
		log.Debug(
			"SQL "+operation,
			"query",
			strings.TrimSpace(query),
			"args",
			fmt.Sprintf("%v", args),
		)
	}
}