sfdupes/db.go

package main

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"errors"
	"fmt"
	"io/fs"
	"net/url"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"time"

	"golang.org/x/sys/unix"
	// 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

// dbFilePerm is the mode for a database file created by scan, the mode
// SQLite gives a database file it creates.
const dbFilePerm = 0o644

// lockFilePerm is the mode for the scan lock file. Anyone who can open
// the file can hold the lock and keep every scan from running, so it
// is open to its owner only.
const lockFilePerm = 0o600

// createTableSQL is the schema applied to a fresh database. Paths are
// BLOBs because Unix paths are raw bytes, not guaranteed UTF-8. mtime
// holds whole Unix seconds and mtime_nsec the nanoseconds within that
// second. head, tail and content each hold the raw 32 bytes of a
// SHA-256.
const createTableSQL = `
CREATE TABLE files (
    path       BLOB PRIMARY KEY,
    size       INTEGER NOT NULL,
    mtime      INTEGER NOT NULL,
    mtime_nsec INTEGER NOT NULL,
    head       BLOB NOT NULL,
    tail       BLOB NOT NULL,
    content    BLOB NOT NULL
) WITHOUT ROWID
`

// createIndexSQL indexes the records by signature, so report can read
// them in signature order without sorting the table.
const createIndexSQL = `
CREATE INDEX files_signature ON files (size, head, tail, content)
`

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

// errNoDatabase reports a missing database file for report, trees and
// info.
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")

// errScanRunning reports that another scan holds the lock on the
// database.
var errScanRunning = errors.New("another scan is running")

// errHashSize reports a record whose head, tail or content is not the
// 32 bytes of a SHA-256.
var errHashSize = errors.New("hash is not 32 bytes")

// databasePath resolves the database location: SFDUPES_DATABASE when
// set and non-empty, the compiled-in default otherwise. It returns the
// path, with the working directory, its symlinks resolved, put in front
// of it when it is relative, and where the path came from, "default" or
// "from SFDUPES_DATABASE", for the line each subcommand prints first on
// stderr.
func databasePath() (string, string, error) {
	path, source := defaultDatabasePath, "default"
	if p := os.Getenv(databaseEnv); p != "" {
		path, source = p, "from "+databaseEnv
	}

	if filepath.IsAbs(path) {
		return path, source, nil
	}

	// The working directory as the kernel reports it, with no symlink
	// in it, as README "Database" says. os.Getwd may return PWD
	// instead, which can name it through a symlink.
	wd, err := unix.Getwd()
	if err != nil {
		return "", "", fmt.Errorf("resolve database path %s: %w", path, err)
	}

	// Not filepath.Abs or filepath.Join: they clean the path, which
	// drops each ".." together with the directory name before it. When
	// that directory is a symlink, the cleaned path names another file.
	// The working directory ends in "/" only when it is the root.
	return strings.TrimSuffix(wd, "/") + "/" + path, source, nil
}

// createDatabaseDirectory creates the directory the database at path
// goes in, and any directory above it, if missing. path is absolute, as
// databasePath returns it. The directory is path with its last element
// cut off as text, by filepath.Split. filepath.Dir would clean it,
// dropping each ".." together with the directory name before it; when
// that name is a symlink, the cleaned path names another directory than
// the one the system reaches when it opens path.
func createDatabaseDirectory(path string) error {
	dir, _ := filepath.Split(path)

	err := os.MkdirAll(dir, dbDirPerm)
	if err != nil {
		return fmt.Errorf("create database directory: %w", err)
	}

	return nil
}

// scanParams are the connection parameters for scan: read-write, with
// WAL journaling and a busy timeout, so a report can run while a cron
// scan is in progress. closeScanDatabase leaves WAL mode again.
const scanParams = "_pragma=busy_timeout(10000)" +
	"&_pragma=journal_mode(WAL)" +
	"&_pragma=synchronous(NORMAL)"

// reportParams are the connection parameters for report, trees and
// info: read-only, with the same busy timeout. They set no journal
// mode, because setting one is a write.
const reportParams = "mode=ro" +
	"&_pragma=busy_timeout(10000)" +
	"&_pragma=query_only(1)"

// openDB opens the SQLite database at path with the connection
// parameters params. It does not create or verify the schema.
func openDB(path, params string) (*sql.DB, error) {
	// The path is escaped into a file: URI, so ?, # and % in it stay
	// part of the file name. SQLite reads what follows file:// up to
	// the next / as a host name, so an absolute path goes after an
	// empty host (file:///abs) and a relative path goes without one
	// (file:rel).
	uri := url.URL{
		Scheme:   "file",
		OmitHost: !filepath.IsAbs(path),
		Path:     path,
		RawQuery: params,
	}

	db, err := sql.Open("sqlite", uri.String())
	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
}

// lockScanDatabase takes the lock that keeps a second scan off the
// database at path: an exclusive flock(2) on the file beside it named
// by the database's path with symlinks resolved and ".lock" appended,
// so every path to one database names the same lock file. If missing,
// the database's parent directory and an empty database file are
// created first, so the lock file is named the same way before the
// database is set up as after. A lock held by another scan fails at
// once instead of waiting. The lock lasts until the returned file is
// closed or the process ends. The lock file is never deleted: a scan
// that deleted it would let the next scan lock a new file while
// another still holds the old one.
func lockScanDatabase(path string) (*os.File, error) {
	err := createDatabaseDirectory(path)
	if err != nil {
		return nil, err
	}

	// Like SQLite, this follows a symlink at path and creates its
	// target. SQLite opens an empty file as a new database.
	//nolint:gosec // the operator chooses the database path
	dbFile, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, dbFilePerm)
	if err != nil {
		return nil, fmt.Errorf("create database: %w", err)
	}

	_ = dbFile.Close()

	resolved, err := filepath.EvalSymlinks(path)
	if err != nil {
		return nil, fmt.Errorf("resolve database path: %w", err)
	}

	lockPath := resolved + ".lock"

	//nolint:gosec // the operator chooses the database path
	f, err := os.OpenFile(lockPath, os.O_RDWR|os.O_CREATE, lockFilePerm)
	if err != nil {
		return nil, err
	}

	err = unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB)
	if err != nil {
		_ = f.Close()

		if errors.Is(err, unix.EWOULDBLOCK) {
			return nil, fmt.Errorf("%w (lock held on %s)",
				errScanRunning, lockPath)
		}

		return nil, fmt.Errorf("lock %s: %w", lockPath, err)
	}

	return f, nil
}

// openScanDatabase opens the database for the scan subcommand, creating
// the file, its parent directory, and the schema as needed.
func openScanDatabase(ctx context.Context, path string) (*sql.DB, error) {
	err := createDatabaseDirectory(path)
	if err != nil {
		return nil, err
	}

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

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

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

	return db, nil
}

// databaseFiles returns, by device and inode, the files of the database
// at path that the scan keeps open: the database file, its -wal and
// -shm files and its .lock file, each named from the database's path
// with symlinks resolved, as SQLite and lockScanDatabase name them. The
// scan calls it once the database and the lock file are open, so that
// its walk passes over them however it reaches them, through a
// symlinked directory included. A file that does not exist then is left
// out, since no walk can reach it. There is no -journal file while the
// database is in WAL mode.
func databaseFiles(path string) ([]fileRec, error) {
	resolved, err := filepath.EvalSymlinks(path)
	if err != nil {
		return nil, fmt.Errorf("resolve database path: %w", err)
	}

	var files []fileRec

	for _, p := range []string{
		resolved, resolved + "-wal", resolved + "-shm", resolved + ".lock",
	} {
		fi, err := os.Lstat(p)
		if errors.Is(err, fs.ErrNotExist) {
			continue
		}

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

		dev, ino := inodeOfInfo(fi)
		files = append(files, fileRec{path: p, dev: dev, ino: ino})
	}

	return files, nil
}

// closeScanDatabase switches the database at path from WAL back to
// rollback-journal mode and closes it. Out of WAL mode the database
// file alone holds the whole database, so a reader needs no -wal or
// -shm file beside it, nor write access to create them. The switch
// fails while a report has the database open; the database then stays
// in WAL mode, still readable, until a later scan closes it.
func closeScanDatabase(ctx context.Context, db *sql.DB, path string) {
	// Runs on the way out of a cancelled scan too.
	_, err := db.ExecContext(context.WithoutCancel(ctx),
		"PRAGMA journal_mode = DELETE")
	if err != nil {
		fmt.Fprintf(os.Stderr, "scan: database %s left in WAL mode: %v\n",
			path, err)
	}

	_ = db.Close()
}

// openReportDatabase opens an existing database for the report, trees
// and info subcommands. A missing database file is an error directing
// the user to run scan first; the schema version must match exactly.
func openReportDatabase(ctx context.Context,
	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, reportParams)
	if err != nil {
		return nil, err
	}

	v, err := userVersion(ctx, db)
	if err == nil && v == 0 {
		// An empty database passes this check and fails the version
		// check below.
		err = checkUnversioned(ctx, 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(ctx context.Context, db *sql.DB) error {
	v, err := userVersion(ctx, db)
	if err != nil {
		return err
	}

	switch v {
	case 0:
		err = checkUnversioned(ctx, db)
		if err != nil {
			return err
		}

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

// checkUnversioned checks a database at user_version 0 before it is
// taken for an empty one. createSchema creates the files table and
// sets the version together, so a files table at version 0 was made by
// something else. Adopting it could corrupt unrelated data, so that is
// a schema-version error telling the operator to remove the file and
// rescan.
func checkUnversioned(ctx context.Context, db *sql.DB) error {
	var name string

	err := db.QueryRowContext(ctx,
		"SELECT name FROM sqlite_master "+
			"WHERE type = 'table' AND name = 'files'").Scan(&name)

	switch {
	case err == nil:
		return fmt.Errorf(
			"has a files table but no schema version; "+
				"remove the file and rescan: %w", errSchemaVersion)
	case errors.Is(err, sql.ErrNoRows):
		return nil
	default:
		return fmt.Errorf("check for files table: %w", err)
	}
}

// createSchema applies the schema to a fresh database and stamps the
// schema version in one transaction, so a creation stopped partway, by
// an interrupt or an error, leaves an empty database the next scan
// sets up, never a files table at version 0, which checkUnversioned
// refuses.
func createSchema(ctx context.Context, db *sql.DB) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("create schema: %w", err)
	}

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

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

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

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

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

	return nil
}

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

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

	return v, nil
}

// loadFileRows gives total the number of records, then streams every
// record to fn in path order: byte order, which is the order of the
// primary key, so SQLite does not sort. The count and the records are
// read in one transaction, so they agree while a scan is committing.
func loadFileRows(ctx context.Context, db *sql.DB, total func(n int64),
	fn func(r scanRec),
) error {
	// Everything goes through tx: db has one connection, so a query on
	// db would wait for tx forever.
	tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
	if err != nil {
		return fmt.Errorf("read records: %w", err)
	}

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

	var n int64

	err = tx.QueryRowContext(ctx, "SELECT COUNT(*) FROM files").Scan(&n)
	if err != nil {
		return fmt.Errorf("count records: %w", err)
	}

	total(n)

	rows, err := tx.QueryContext(ctx,
		"SELECT path, size, mtime, mtime_nsec, head, tail, content "+
			"FROM files ORDER BY path")
	if err != nil {
		return fmt.Errorf("read records: %w", err)
	}

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

	for rows.Next() {
		var (
			path, head, tail, content []byte
			sec, nsec                 int64
			r                         scanRec
		)

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

		r.path = string(path)
		r.mtime = time.Unix(sec, nsec)

		err = r.setHashes(head, tail, content)
		if err != nil {
			return fmt.Errorf("read record %s: %w", escapePath(r.path), err)
		}

		fn(r)
	}

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

	return nil
}

// signatureRowsSQL selects every record's signature and path in
// signature order, largest size first. That is the order of the
// files_signature index read backwards, so SQLite reads the index alone
// and returns the first row at once, with nothing to sort.
const signatureRowsSQL = `
SELECT size, head, tail, content, path
FROM files
ORDER BY size DESC, head DESC, tail DESC, content DESC
`

// loadSignatureRows gives total the number of records, then streams
// every record's signature and path to fn, without its mtime, in the
// order of signatureRowsSQL. The count and the records are read in one
// transaction, so they agree while a scan is committing. An error from
// fn stops the reading and is returned as it is.
func loadSignatureRows(ctx context.Context, db *sql.DB,
	total func(n int64), fn func(r scanRec) error,
) error {
	// Everything goes through tx: the report connection is the only
	// one, so a query on db would wait for tx forever.
	tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
	if err != nil {
		return fmt.Errorf("read records: %w", err)
	}

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

	var n int64

	err = tx.QueryRowContext(ctx, "SELECT COUNT(*) FROM files").Scan(&n)
	if err != nil {
		return fmt.Errorf("count records: %w", err)
	}

	total(n)

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

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

	for rows.Next() {
		var (
			path, head, tail, content []byte
			r                         scanRec
		)

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

		r.path = string(path)

		err = r.setHashes(head, tail, content)
		if err != nil {
			return fmt.Errorf("read record %s: %w", escapePath(r.path), err)
		}

		err = fn(r)
		if err != nil {
			return err
		}
	}

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

	return nil
}

// setHashes sets r's head, tail and content to the values read from
// those columns. Each must be the raw 32 bytes of a SHA-256.
func (r *scanRec) setHashes(head, tail, content []byte) error {
	for _, b := range [][]byte{head, tail, content} {
		if len(b) != sha256.Size {
			return fmt.Errorf("%w: found %d bytes", errHashSize, len(b))
		}
	}

	r.head = [sha256.Size]byte(head)
	r.tail = [sha256.Size]byte(tail)
	r.content = [sha256.Size]byte(content)

	return nil
}

// countRecords returns the number of records in the database.
func countRecords(ctx context.Context, db *sql.DB) (int64, error) {
	var n int64

	err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM files").Scan(&n)
	if err != nil {
		return 0, fmt.Errorf("count records: %w", err)
	}

	return n, nil
}

// databaseCountsSQL counts the records and adds up their sizes, in one
// pass over the table. A SUM over no rows is NULL, so on an empty table
// the SUM falls back to 0.
const databaseCountsSQL = `
SELECT COUNT(*), COALESCE(SUM(size), 0)
FROM files
`

// databaseCounts is what info reads from the database: its schema
// version and the counts of databaseCountsSQL.
type databaseCounts struct {
	schemaVersion int
	records       int64 // every record
	size          int64 // the sizes of all records added up
}

// readDatabaseCounts reads the schema version, then the counts of
// databaseCountsSQL, in one read-only transaction, so both describe the
// database at one moment while a scan is committing.
func readDatabaseCounts(ctx context.Context,
	db *sql.DB,
) (databaseCounts, error) {
	var c databaseCounts

	// Everything goes through tx: the report connection is the only
	// one, so a query on db would wait for tx forever.
	tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
	if err != nil {
		return c, fmt.Errorf("count records: %w", err)
	}

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

	err = tx.QueryRowContext(ctx, "PRAGMA user_version").
		Scan(&c.schemaVersion)
	if err != nil {
		return c, fmt.Errorf("read schema version: %w", err)
	}

	err = tx.QueryRowContext(ctx, databaseCountsSQL).
		Scan(&c.records, &c.size)
	if err != nil {
		return c, fmt.Errorf("count records: %w", err)
	}

	return c, nil
}

// loadFileMeta streams every record's path, size and mtime 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(ctx context.Context, db *sql.DB,
	fn func(path string, size int64, mtime time.Time),
) error {
	rows, err := db.QueryContext(ctx,
		"SELECT path, size, mtime, mtime_nsec FROM files")
	if err != nil {
		return fmt.Errorf("read records: %w", err)
	}

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

	for rows.Next() {
		var (
			path            []byte
			size, sec, nsec int64
		)

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

		fn(string(path), size, time.Unix(sec, nsec))
	}

	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(ctx context.Context, db *sql.DB, upserts []scanRec,
	deletes []string, prog *progress,
) error {
	for batch := range slices.Chunk(upserts, updateBatchSize) {
		err := applyBatch(ctx, db, batch, nil, prog)
		if err != nil {
			return err
		}
	}

	for batch := range slices.Chunk(deletes, updateBatchSize) {
		err := applyBatch(ctx, 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(ctx context.Context, db *sql.DB, upserts []scanRec,
	deletes []string, prog *progress,
) error {
	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.Unix(), r.mtime.Nanosecond(),
			r.head[:], r.tail[:], r.content[:])
		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
}