yellowjacket/backend/database/database.go

// Package database provides SQLite database access.
package database

import (
	"context"
	"database/sql"
	"embed"
	"fmt"
	"io/fs"
	"log/slog"
	"path"
	"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

// 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)
	}

	// Before the schema is applied, not after: applySchema is
	// CREATE ... IF NOT EXISTS, which no-ops against a table that
	// already exists in an older shape.  Retiring the stale one first is
	// what turns that no-op into a create.
	if err := retireStaleTables(dbCtx, db, logger); err != nil {
		return nil, 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.
//
// The schema files under sql/schemas are CREATE ... IF NOT EXISTS and
// declare the current, latest shape of every table — so running them
// against a fresh database produces exactly that shape, and running
// them against a database already at that shape does nothing.  That is
// the whole mechanism; there is no migration chain and no
// schema_migrations table.
//
// There was one, and it was squashed (see .planning/plans/013): a chain
// only earns its keep once real user databases exist in the wild, and
// until then it is a second description of the schema that can drift
// from the first — which this project has already been bitten by once.
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)
	}

	return 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
}