yellowjacket/backend/database/database.go

// Package database provides SQLite database access.
package database

import (
	"context"
	"database/sql"
	"embed"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"path"
	"sort"
	"strconv"
	"strings"

	_ "modernc.org/sqlite" // Register sqlite driver.

	"yellowjacket/backend/database/sql/sqlcgen"
	"yellowjacket/backend/profiling"
	"yellowjacket/backend/system"
)

//go:generate go tool sqlc generate

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

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

// DB wraps the SQLite database connection and queries.
//
// Two handles back a single database file.  db is the single-writer
// connection (MaxOpenConns 1) used for every write and every
// transaction.  readDB is a small multi-connection, query-only pool
// used for standalone reads.  Under WAL, readers run concurrently
// with the writer, so a long background write (index build, dump
// patch) no longer blocks interactive searches — the reason searches
// stalled for seconds was that the file was in rollback-journal mode
// with a single shared connection, so any writer locked out readers.
type DB struct {
	db     *sql.DB
	readDB *sql.DB
	Ctx    context.Context
	// Queries runs on the single-writer connection.  Use it for every
	// write and for any read that must observe an uncommitted write made
	// earlier in the same logical operation.
	Queries *sqlcgen.Queries
	// ReadQueries runs on the query-only WAL read pool, so standalone
	// reads proceed concurrently with a long background write instead of
	// queueing behind it on the single writer.  It observes only
	// committed data.  In tests (no read pool) it aliases Queries.
	ReadQueries *sqlcgen.Queries
	logger      *slog.Logger
}

// Data-source names.  modernc.org/sqlite only honours PRAGMAs passed
// as `_pragma=name(value)` — the mattn-style `_journal_mode=WAL`
// form is silently ignored, which is why WAL was never actually on.
const (
	writeDSNParams = "?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)"
	readDSNParams  = "?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)" +
		"&_pragma=query_only(true)&_pragma=synchronous(NORMAL)" +
		"&_pragma=cache_size(-8000)&_pragma=mmap_size(67108864)"
	// readPoolConns bounds concurrent read connections.  A handful is
	// plenty for interactive search + art/lookup fan-out and keeps WAL
	// reader overhead small.
	readPoolConns = 4
)

// NewDB opens the database and creates the schema if it is not there.
func NewDB(logger *slog.Logger) (*DB, error) {
	defer profiling.TimeOp(logger, "database.NewDB")()

	dbCtx := context.Background()

	userDataDir, err := system.GetUserDataDirPath()
	if err != nil {
		return nil, fmt.Errorf("could not get user data directory: %w", err)
	}

	sqliteDBFilePath := path.Join(userDataDir, "yj.db")

	logger.Debug("opening sqlite database", "filepath", sqliteDBFilePath)

	db, err := sql.Open("sqlite", sqliteDBFilePath+writeDSNParams)
	if err != nil {
		return nil, fmt.Errorf("could not connect to sqlite database: %w", err)
	}

	db.SetMaxOpenConns(1) // SQLite only supports one writer at a time

	if err := applyPRAGMAs(dbCtx, db); err != nil {
		return nil, fmt.Errorf("could not apply PRAGMAs: %w", err)
	}

	if err := applySchema(dbCtx, db); err != nil {
		return nil, err
	}

	// Get generated queries
	queries := sqlcgen.New(db)

	// Open a separate query-only read pool.  The write handle above
	// has already converted the file to WAL, so these connections read
	// a consistent snapshot concurrently with in-flight writes.
	readDB, err := sql.Open("sqlite", sqliteDBFilePath+readDSNParams)
	if err != nil {
		return nil, fmt.Errorf("could not open read pool: %w", err)
	}

	readDB.SetMaxOpenConns(readPoolConns)

	return &DB{
		db:          db,
		readDB:      readDB,
		Ctx:         dbCtx,
		Queries:     queries,
		ReadQueries: sqlcgen.New(readDB),
		logger:      logger,
	}, err
}

// reader returns the handle standalone reads should use: the
// query-only read pool when present, else the write handle (tests
// share one in-memory connection, which cannot be reopened).
func (d *DB) reader() *sql.DB {
	if d.readDB != nil {
		return d.readDB
	}

	return d.db
}

// BeginTx starts a new database transaction.
func (d *DB) BeginTx() (*sql.Tx, error) {
	return d.db.BeginTx(d.Ctx, nil)
}

// ExecContext executes a query without returning any rows.
func (d *DB) ExecContext(query string, args ...any) (sql.Result, error) {
	return d.db.ExecContext(d.Ctx, query, args...)
}

// QueryContext executes a query that returns rows.  Reads run on the
// query-only read pool so they proceed concurrently with writes under
// WAL instead of queueing behind the single writer connection.
func (d *DB) QueryContext(query string, args ...any) (*sql.Rows, error) {
	return d.reader().QueryContext(d.Ctx, query, args...)
}

// QueryContextWith executes a query that returns rows using a
// caller-supplied context instead of the DB's lifecycle context.
// This lets an individual query (e.g. a superseded search) be
// cancelled independently.  Like QueryContext it runs on the read
// pool.
func (d *DB) QueryContextWith(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
	return d.reader().QueryContext(ctx, query, args...)
}

// QueryRowWriter runs a single-row query on the writer connection rather
// than the read pool.
//
// Almost every read should use QueryContext instead.  This exists for
// the one case that cannot: statements referencing a database ATTACHed
// to the writer.  The read pool is a separate sql.DB over the same file,
// so an attachment made on the writer is invisible there and the query
// would fail with "no such table".
func (d *DB) QueryRowWriter(query string, args ...any) *sql.Row {
	return d.db.QueryRowContext(d.Ctx, query, args...)
}

// Logger returns the structured logger bound to this DB. Callers can
// use it to emit timing or diagnostic logs from query-adjacent code.
func (d *DB) Logger() *slog.Logger {
	return d.logger
}

// exploreIndexFTSTriggers is the sole definition of the explore_index →
// FTS5 sync triggers.
//
// They live here rather than in sql/schemas/explore_index.sql because a
// bulk load drops and recreates them (see SuspendExploreIndexFTS), so
// the runtime needs them as statements either way.  Defining them in
// both places would be two copies free to drift.
var exploreIndexFTSTriggers = []string{
	`CREATE TRIGGER explore_index_ai AFTER INSERT ON explore_index BEGIN
		INSERT INTO explore_index_fts(rowid, title, artist_name, aliases)
		VALUES (new.id, new.title, new.artist_name, new.aliases);
	END`,
	`CREATE TRIGGER explore_index_ad AFTER DELETE ON explore_index BEGIN
		INSERT INTO explore_index_fts(explore_index_fts, rowid, title, artist_name, aliases)
		VALUES ('delete', old.id, old.title, old.artist_name, old.aliases);
	END`,
	// Narrowed to the three columns the FTS table actually indexes, and
	// guarded on them having changed.  An UPDATE that leaves all three
	// alone has nothing to re-index, and re-indexing it is not free: an
	// FTS5 delete has to find the old row's postings in a multi-million
	// row index, which is the ~31 rows/s figure below.
	//
	// This is not a micro-optimisation.  Every writer here upserts, and
	// the merge rules keep existing values (`CASE WHEN excluded.title
	// != '' ...`), so the common write is a row arriving unchanged: the
	// discography backfill re-browsing a known artist, the incremental
	// dump refreshing popularity.  Each of those used to pay a full
	// delete + insert against the FTS index while holding the single
	// writer connection — measured at 91% of the app's CPU, with the
	// play path queued behind it.
	`CREATE TRIGGER explore_index_au AFTER UPDATE OF title, artist_name, aliases
	ON explore_index
	WHEN old.title IS NOT new.title
		OR old.artist_name IS NOT new.artist_name
		OR old.aliases IS NOT new.aliases
	BEGIN
		INSERT INTO explore_index_fts(explore_index_fts, rowid, title, artist_name, aliases)
		VALUES ('delete', old.id, old.title, old.artist_name, old.aliases);
		INSERT INTO explore_index_fts(rowid, title, artist_name, aliases)
		VALUES (new.id, new.title, new.artist_name, new.aliases);
	END`,
}

// createExploreIndexFTSTriggers installs the sync triggers.  Safe to
// call on a database that already has them.
//
// It drops first rather than tolerating "already exists", because a
// trigger is a definition and not a row: an install that already has
// the old one would otherwise keep it forever, and these definitions
// are exactly where this table's write cost is decided.  Three DDL
// statements against a table with no rows to rewrite, on open.
func createExploreIndexFTSTriggers(ctx context.Context, db *sql.DB) error {
	if err := dropExploreIndexFTSTriggers(ctx, db); err != nil {
		return err
	}

	for _, stmt := range exploreIndexFTSTriggers {
		if _, err := db.ExecContext(ctx, stmt); err != nil &&
			!strings.Contains(err.Error(), "already exists") {
			return fmt.Errorf("create explore FTS trigger: %w", err)
		}
	}

	return nil
}

// exploreIndexFTSTriggerNames is what both the drop paths remove.
var exploreIndexFTSTriggerNames = []string{
	"explore_index_ai", "explore_index_ad", "explore_index_au",
}

func dropExploreIndexFTSTriggers(ctx context.Context, db *sql.DB) error {
	for _, name := range exploreIndexFTSTriggerNames {
		if _, err := db.ExecContext(
			ctx, "DROP TRIGGER IF EXISTS "+name,
		); err != nil {
			return fmt.Errorf("drop explore FTS trigger %s: %w", name, err)
		}
	}

	return nil
}

// SuspendExploreIndexFTS drops the FTS sync triggers so a bulk load can
// write explore_index without paying per-row FTS maintenance.
//
// Row-at-a-time FTS upkeep is what makes a full dump import take
// nearly a day: a bulk DELETE fires the delete trigger once per row,
// which buries the FTS5 index in delete markers and stale segments, and
// every subsequent upsert then works against that debris.  Measured on
// a real import, assembly runs at ~31 rows/s with the triggers attached
// and ~4,700 rows/s without.
//
// Callers MUST pair this with ResumeExploreIndexFTS — while suspended,
// explore_index_fts stops tracking the table and search goes stale.
func (d *DB) SuspendExploreIndexFTS() error {
	if err := dropExploreIndexFTSTriggers(d.Ctx, d.db); err != nil {
		return fmt.Errorf("suspend explore FTS: %w", err)
	}

	return nil
}

// ResumeExploreIndexFTS reinstates the sync triggers and rebuilds the
// FTS index from the content table, discarding whatever accumulated
// while it was suspended.  The rebuild is a single linear pass and is
// far cheaper than the per-row maintenance it replaces.
//
// Safe to call when the triggers are already present, so it can run
// from a defer on both the success and failure paths.
func (d *DB) ResumeExploreIndexFTS() error {
	if err := createExploreIndexFTSTriggers(d.Ctx, d.db); err != nil {
		return fmt.Errorf("resume explore FTS: %w", err)
	}

	if _, err := d.db.ExecContext(
		d.Ctx, "INSERT INTO explore_index_fts(explore_index_fts) VALUES('rebuild')",
	); err != nil {
		return fmt.Errorf("resume explore FTS: rebuild: %w", err)
	}

	return nil
}

// applySchema creates the full schema on a fresh database and brings
// an existing one up to date via sql/migrations.
//
// The schema files under sql/schemas are CREATE ... IF NOT EXISTS,
// so on a genuinely new database they create every table already at
// its current, latest shape — that's the fast path new installs
// take.  A database that already has an older shape (e.g. a
// tagging_items missing a column a later build added) needs the gap
// closed, which IF NOT EXISTS can't do: it silently no-ops on a
// table that already exists, columns and all.  sql/migrations holds
// small, additive, numbered files (ALTER TABLE, CREATE INDEX, etc.)
// for exactly that gap, tracked in schema_migrations so each applies
// at most once — see applyMigrations for how a fresh database's
// already-current tables tolerate replaying them anyway.
func applySchema(ctx context.Context, db *sql.DB) error {
	dirEntries, err := schemas.ReadDir("sql/schemas")
	if err != nil {
		return fmt.Errorf("could not read schemas directory: %w", err)
	}

	for _, dirEntry := range dirEntries {
		if dirEntry.IsDir() {
			continue
		}

		filePath := path.Join("sql/schemas", dirEntry.Name())

		sqlContent, err := fs.ReadFile(schemas, filePath)
		if err != nil {
			return fmt.Errorf("could not read file %s: %w", filePath, err)
		}

		if _, err := db.ExecContext(ctx, string(sqlContent)); err != nil {
			return fmt.Errorf("error executing sql from file %s: %w", filePath, err)
		}
	}

	// The FTS sync triggers are defined in Go, not in the schema files,
	// because the bulk-load path drops and recreates them.
	if err := createExploreIndexFTSTriggers(ctx, db); err != nil {
		return fmt.Errorf("could not create explore FTS triggers: %w", err)
	}

	if err := applyMigrations(ctx, db); err != nil {
		return fmt.Errorf("could not apply migrations: %w", err)
	}

	// The download subsystem's Want/Request rename reuses table names
	// (download_requests names a different table before and after), so
	// it cannot be a plain sql/migrations file the way an ADD COLUMN
	// migration can; see download_rename_migration.go for why.
	if err := migrateDownloadRename(ctx, db); err != nil {
		return fmt.Errorf("could not migrate download rename: %w", err)
	}

	return nil
}

// schemaMigrationsTable tracks which sql/migrations files have run,
// by their leading numeric prefix.
const schemaMigrationsTable = `
CREATE TABLE IF NOT EXISTS schema_migrations (
  version    INTEGER PRIMARY KEY,
  applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
)`

// applyMigrations runs every sql/migrations file not yet recorded in
// schema_migrations, in filename order (numeric prefix), one
// statement at a time.
//
// Every migration runs on EVERY database, fresh or old — there is no
// "skip on fresh install" branch. A fresh database's tables already
// carry a migration's effect (sql/schemas declares the target shape
// directly), so its statements are expected to sometimes be no-ops
// there: "duplicate column name" from an ALTER TABLE ADD COLUMN is
// tolerated and treated as "already applied", the same way
// createExploreIndexFTSTriggers tolerates "already exists". Any
// other error is fatal. This is deliberately simpler than detecting
// "is this database fresh" — every migration converges both a fresh
// and an upgraded database to the identical final schema (including
// column order — ALTER TABLE ADD COLUMN always appends at the end,
// so sql/schemas must declare a migrated column last too; see the
// comment on tagging_items.sql and the regression test in
// migrations_test.go).
func applyMigrations(ctx context.Context, db *sql.DB) error {
	if _, err := db.ExecContext(ctx, schemaMigrationsTable); err != nil {
		return fmt.Errorf("create schema_migrations: %w", err)
	}

	dirEntries, err := migrations.ReadDir("sql/migrations")
	if err != nil {
		return fmt.Errorf("could not read migrations directory: %w", err)
	}

	sort.Slice(dirEntries, func(i, j int) bool {
		return dirEntries[i].Name() < dirEntries[j].Name()
	})

	for _, dirEntry := range dirEntries {
		if dirEntry.IsDir() {
			continue
		}

		version, err := migrationVersion(dirEntry.Name())
		if err != nil {
			return err
		}

		applied, err := migrationApplied(ctx, db, version)
		if err != nil {
			return err
		}

		if applied {
			continue
		}

		filePath := path.Join("sql/migrations", dirEntry.Name())

		sqlContent, err := fs.ReadFile(migrations, filePath)
		if err != nil {
			return fmt.Errorf("could not read file %s: %w", filePath, err)
		}

		if err := execMigrationStatements(ctx, db, string(sqlContent)); err != nil {
			return fmt.Errorf("error executing migration %s: %w", dirEntry.Name(), err)
		}

		if _, err := db.ExecContext(
			ctx, `INSERT INTO schema_migrations (version) VALUES (?)`, version,
		); err != nil {
			return fmt.Errorf("record migration %d applied: %w", version, err)
		}
	}

	return nil
}

// execMigrationStatements runs a migration file one statement at a
// time — NOT as one multi-statement Exec — so that one statement
// being a tolerable no-op (ALTER TABLE ADD COLUMN on a fresh
// database) doesn't abort the statements after it in the same file
// (e.g. a trailing CREATE INDEX that a fresh database still needs,
// since sql/schemas deliberately doesn't declare an index on a
// migrated column — see the comment on tagging_items.sql).
//
// Splitting on ";" is safe for the simple ALTER/CREATE TABLE/CREATE
// INDEX statements migrations are expected to contain; it is NOT
// safe for statements embedding a literal semicolon (e.g. a CREATE
// TRIGGER body) — write those with executeContext calls in Go
// instead of a sql/migrations file, the same way the explore FTS
// triggers already are.
func execMigrationStatements(ctx context.Context, db *sql.DB, script string) error {
	for stmt := range strings.SplitSeq(script, ";") {
		stmt = strings.TrimSpace(stmt)
		if stmt == "" {
			continue
		}

		if _, err := db.ExecContext(ctx, stmt); err != nil {
			if strings.Contains(err.Error(), "duplicate column name") {
				continue
			}

			return fmt.Errorf("statement %q: %w", stmt, err)
		}
	}

	return nil
}

// migrationVersion extracts the leading integer prefix from a
// migration filename, e.g. "0001_tagging_items_synthetic.sql" -> 1.
func migrationVersion(filename string) (int, error) {
	prefix, _, ok := strings.Cut(filename, "_")
	if !ok {
		return 0, fmt.Errorf("%w: %s", errMigrationFilename, filename)
	}

	version, err := strconv.Atoi(prefix)
	if err != nil {
		return 0, fmt.Errorf("%w: %s", errMigrationFilename, filename)
	}

	return version, nil
}

var errMigrationFilename = errors.New(
	"migration filename must start with a numeric prefix followed by '_' (e.g. 0001_description.sql)",
)

func migrationApplied(ctx context.Context, db *sql.DB, version int) (bool, error) {
	var v int

	err := db.QueryRowContext(
		ctx, `SELECT version FROM schema_migrations WHERE version = ?`, version,
	).Scan(&v)

	switch {
	case errors.Is(err, sql.ErrNoRows):
		return false, nil
	case err != nil:
		return false, fmt.Errorf("check migration %d: %w", version, err)
	default:
		return true, nil
	}
}

// applyPRAGMAs configures SQLite connection settings. Called by both
// NewDB and NewTestDB to ensure identical behavior.
func applyPRAGMAs(ctx context.Context, db *sql.DB) error {
	pragmas := []string{
		"PRAGMA foreign_keys = ON",
		"PRAGMA synchronous = NORMAL",
		"PRAGMA cache_size = -8000",
		"PRAGMA mmap_size = 67108864",
	}

	for _, pragma := range pragmas {
		if _, err := db.ExecContext(ctx, pragma); err != nil {
			return fmt.Errorf(
				"could not apply PRAGMA %q: %w", pragma, err,
			)
		}
	}

	return nil
}