sfdupes/db.go

package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"slices"
	"strconv"

	// The pure-Go SQLite driver, registered as "sqlite"; keeps cgo
	// disabled.
	_ "modernc.org/sqlite"
)

// defaultDatabasePath is where the persistent scan database lives when
// SFDUPES_DATABASE is not set.
const defaultDatabasePath = "/var/lib/sfdupes/db.sqlite"

// databaseEnv is the environment variable that overrides the database
// path.
const databaseEnv = "SFDUPES_DATABASE"

// schemaVersion is the database schema version this build reads and
// writes, stored in PRAGMA user_version.
const schemaVersion = 1

// dbDirPerm is the mode for a database parent directory created by
// scan.
const dbDirPerm = 0o755

// createTableSQL is the schema applied to a fresh database. Paths are
// BLOBs because Unix paths are raw bytes, not guaranteed UTF-8.
const createTableSQL = `
CREATE TABLE files (
    path  BLOB PRIMARY KEY,
    size  INTEGER NOT NULL,
    mtime INTEGER NOT NULL,
    head  TEXT NOT NULL,
    tail  TEXT NOT NULL
) WITHOUT ROWID
`

// upsertSQL inserts one file record, replacing any existing record for
// the same path.
const upsertSQL = `
INSERT INTO files (path, size, mtime, head, tail)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT (path) DO UPDATE SET
    size = excluded.size, mtime = excluded.mtime,
    head = excluded.head, tail = excluded.tail
`

// errNoDatabase reports a missing database file for report/trees.
var errNoDatabase = errors.New(
	"no database (run \"sfdupes scan\" first, or set " + databaseEnv + ")")

// errSchemaVersion reports a database whose schema version this build
// does not understand.
var errSchemaVersion = errors.New("unsupported database schema version")

// databasePath resolves the database location: SFDUPES_DATABASE when
// set and non-empty, the compiled-in default otherwise.
func databasePath() string {
	if p := os.Getenv(databaseEnv); p != "" {
		return p
	}

	return defaultDatabasePath
}

// openDB opens the SQLite database at path with WAL journaling and a
// busy timeout, so a report can run while a cron scan is in progress.
// It does not create or verify the schema.
func openDB(path string) (*sql.DB, error) {
	dsn := "file:" + path +
		"?_pragma=busy_timeout(10000)" +
		"&_pragma=journal_mode(WAL)" +
		"&_pragma=synchronous(NORMAL)"

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

	// A single connection avoids SQLITE_BUSY between this process's
	// own connections; concurrency lives in the worker pools, not in
	// parallel database access.
	db.SetMaxOpenConns(1)

	return db, nil
}

// openScanDatabase opens the database for the scan subcommand, creating
// the file, its parent directory, and the schema as needed.
func openScanDatabase(path string) (*sql.DB, error) {
	err := os.MkdirAll(filepath.Dir(path), dbDirPerm)
	if err != nil {
		return nil, fmt.Errorf("create database directory: %w", err)
	}

	db, err := openDB(path)
	if err != nil {
		return nil, err
	}

	err = initSchema(db)
	if err != nil {
		_ = db.Close()

		return nil, fmt.Errorf("database %s: %w", path, err)
	}

	return db, nil
}

// openReportDatabase opens an existing database for the report and
// trees subcommands. A missing database file is an error directing the
// user to run scan first; the schema version must match exactly.
func openReportDatabase(path string) (*sql.DB, error) {
	_, err := os.Stat(path)
	if errors.Is(err, fs.ErrNotExist) {
		return nil, fmt.Errorf("%s: %w", path, errNoDatabase)
	}

	if err != nil {
		return nil, fmt.Errorf("database: %w", err)
	}

	db, err := openDB(path)
	if err != nil {
		return nil, err
	}

	v, err := userVersion(db)
	if err != nil {
		_ = db.Close()

		return nil, fmt.Errorf("database %s: %w", path, err)
	}

	if v != schemaVersion {
		_ = db.Close()

		return nil, fmt.Errorf("database %s: version %d, want %d: %w",
			path, v, schemaVersion, errSchemaVersion)
	}

	return db, nil
}

// initSchema creates the schema on a fresh database and verifies the
// schema version on an existing one.
func initSchema(db *sql.DB) error {
	v, err := userVersion(db)
	if err != nil {
		return err
	}

	switch v {
	case 0:
		return createSchema(db)
	case schemaVersion:
		return nil
	default:
		return fmt.Errorf("version %d, want %d: %w",
			v, schemaVersion, errSchemaVersion)
	}
}

// createSchema applies the schema to a fresh database and stamps the
// schema version.
func createSchema(db *sql.DB) error {
	ctx := context.Background()

	_, err := db.ExecContext(ctx, createTableSQL)
	if err != nil {
		return fmt.Errorf("create schema: %w", err)
	}

	_, err = db.ExecContext(ctx,
		"PRAGMA user_version = "+strconv.Itoa(schemaVersion))
	if err != nil {
		return fmt.Errorf("set schema version: %w", err)
	}

	return nil
}

// userVersion reads the database's PRAGMA user_version.
func userVersion(db *sql.DB) (int, error) {
	var v int

	err := db.QueryRowContext(context.Background(),
		"PRAGMA user_version").Scan(&v)
	if err != nil {
		return 0, fmt.Errorf("read schema version: %w", err)
	}

	return v, nil
}

// loadFileRows reads every record from the files table.
func loadFileRows(db *sql.DB) ([]scanRec, error) {
	rows, err := db.QueryContext(context.Background(),
		"SELECT path, size, mtime, head, tail FROM files")
	if err != nil {
		return nil, fmt.Errorf("read records: %w", err)
	}

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

	var recs []scanRec

	for rows.Next() {
		var (
			path []byte
			r    scanRec
		)

		err = rows.Scan(&path, &r.size, &r.mtime, &r.head, &r.tail)
		if err != nil {
			return nil, fmt.Errorf("read record: %w", err)
		}

		r.path = string(path)
		recs = append(recs, r)
	}

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

	return recs, nil
}

// loadFileMeta streams every record's path, size, mtime, and whether
// it carries hashes to fn. Scan change detection needs no hash
// values, and skipping the hash columns keeps the scan's in-memory
// index small on multi-million-file databases.
func loadFileMeta(db *sql.DB,
	fn func(path string, size, mtime int64, hashed bool),
) error {
	rows, err := db.QueryContext(context.Background(),
		"SELECT path, size, mtime, head <> '' FROM files")
	if err != nil {
		return fmt.Errorf("read records: %w", err)
	}

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

	for rows.Next() {
		var (
			path        []byte
			size, mtime int64
			hashed      int64
		)

		err = rows.Scan(&path, &size, &mtime, &hashed)
		if err != nil {
			return fmt.Errorf("read record: %w", err)
		}

		fn(string(path), size, mtime, hashed != 0)
	}

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

	return nil
}

// updateBatchSize is the number of record changes committed per
// transaction during the update pass. The filesystem is authoritative
// and the database an eventually-consistent reflection of it, so
// scan-level atomicity is not required; smaller transactions keep the
// WAL small and let concurrent reports observe progress.
const updateBatchSize = 10000

// applyChanges writes one scan's database changes — upserts for new and
// changed files, deletes for vanished ones — in batched transactions.
// Progress is rendered on prog (one increment per change).
func applyChanges(db *sql.DB, upserts []scanRec, deletes []string,
	prog *progress,
) error {
	for batch := range slices.Chunk(upserts, updateBatchSize) {
		err := applyBatch(db, batch, nil, prog)
		if err != nil {
			return err
		}
	}

	for batch := range slices.Chunk(deletes, updateBatchSize) {
		err := applyBatch(db, nil, batch, prog)
		if err != nil {
			return err
		}
	}

	return nil
}

// applyBatch commits one batch of upserts and deletes in a single
// transaction.
func applyBatch(db *sql.DB, upserts []scanRec, deletes []string,
	prog *progress,
) error {
	ctx := context.Background()

	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin transaction: %w", err)
	}

	defer func() { _ = tx.Rollback() }()

	err = execUpserts(ctx, tx, upserts, prog)
	if err != nil {
		return err
	}

	err = execDeletes(ctx, tx, deletes, prog)
	if err != nil {
		return err
	}

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

	return nil
}

// execUpserts inserts or updates one record per new or changed file.
func execUpserts(ctx context.Context, tx *sql.Tx, upserts []scanRec,
	prog *progress,
) error {
	st, err := tx.PrepareContext(ctx, upsertSQL)
	if err != nil {
		return fmt.Errorf("prepare upsert: %w", err)
	}

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

	for _, r := range upserts {
		_, err = st.ExecContext(ctx,
			[]byte(r.path), r.size, r.mtime, r.head, r.tail)
		if err != nil {
			return fmt.Errorf("upsert %s: %w", r.path, err)
		}

		prog.increment()
	}

	return nil
}

// execDeletes removes the records for paths no longer present.
func execDeletes(ctx context.Context, tx *sql.Tx, deletes []string,
	prog *progress,
) error {
	st, err := tx.PrepareContext(ctx, "DELETE FROM files WHERE path = ?")
	if err != nil {
		return fmt.Errorf("prepare delete: %w", err)
	}

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

	for _, p := range deletes {
		_, err = st.ExecContext(ctx, []byte(p))
		if err != nil {
			return fmt.Errorf("delete %s: %w", p, err)
		}

		prog.increment()
	}

	return nil
}