yellowjacket/cmd/indexexport/main.go

// Command indexexport turns a fully built explore index into the
// compact "core" artifact that ships to users.
//
// The full dump-built index is far too large to distribute (~900MB by
// current budget estimates). The core artifact keeps the most-listened
// artists and their discography slice — enough for Explore to be useful
// on a fresh install — and leaves the long tail to the existing lazy
// per-artist fetch paths.
//
// The artifact deliberately contains no FTS table. The importing client
// inserts these rows into its own explore_index, whose AFTER INSERT
// trigger populates explore_index_fts as a side effect, so shipping a
// search index would be redundant weight.
//
// Usage:
//
//	YJ_HOME=/var/cache/yellowjacket-index indexexport -o core-index.db
package main

import (
	"database/sql"
	"errors"
	"flag"
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	_ "modernc.org/sqlite"

	"yellowjacket/backend/system"
)

// Columns copied into the artifact: the global catalog only.
//
// Deliberately excluded are the per-user columns — in_library,
// is_similar, local_artist_id, local_release_group_id,
// local_recording_id — which describe one person's library and are
// recomputed locally by PopulateLocalCrossReferences after import.
const catalogColumns = `entity_type, mbid, title, artist_name, artist_mbid,
	aliases, popularity, listener_count, duration, caa_release_mbid,
	release_name, primary_type, secondary_types, release_date, total_tracks,
	artist_type, country, disambiguation, sort_name, discog_fetched`

// sourceColumns is catalogColumns as read *from* the built index,
// which is not always shaped like the one this binary was compiled
// against.
//
// The index job's /cache volume is a real YJ_HOME that survives
// between runs and holds ~205 GB nobody can re-download casually, so
// its explore_index is classified Cache and is deliberately **not**
// dropped and recreated by cmd/indexbuild's schema repair. A column
// added to the schema after that database was built is therefore
// absent from it, and selecting it fails the whole export with
// "no such column: total_tracks" -- which is what happened the first
// time the job ran after the completeness work.
//
// So the source list is asked for rather than assumed, exactly as
// artifactHasTotals does on the importing side. Zero is what the
// column means by "the catalog does not say", and the app already
// renders that as unknown rather than as incomplete.
func sourceColumns(db *sql.DB) string {
	var n int

	err := db.QueryRow(
		`SELECT COUNT(*) FROM pragma_table_info('explore_index', 'main')
		 WHERE name = 'total_tracks'`,
	).Scan(&n)
	if err == nil && n > 0 {
		return catalogColumns
	}

	fmt.Println(
		"  note: this index predates total_tracks; exporting 0 for it",
	)

	return strings.Replace(catalogColumns, "total_tracks", "0", 1)
}

var errNoHome = errors.New(
	"YJ_HOME must be set to the directory holding the built index",
)

var errEmptyIndex = errors.New(
	"source index has no rows — run indexbuild to completion first",
)

func main() {
	out := flag.String("o", "core-index.db", "output artifact path")
	artists := flag.Int("artists", 50_000,
		"number of top artists (by listen count) to include")
	perArtistRGs := flag.Int("rgs-per-artist", 15,
		"max release groups per included artist")
	perArtistRecs := flag.Int("recs-per-artist", 30,
		"max recordings per included artist")

	flag.Parse()

	if err := run(*out, *artists, *perArtistRGs, *perArtistRecs); err != nil {
		fmt.Fprintln(os.Stderr, "indexexport:", err)
		os.Exit(1)
	}
}

func run(out string, artists, perArtistRGs, perArtistRecs int) error {
	if os.Getenv("YJ_HOME") == "" {
		return errNoHome
	}

	dataDir, err := system.GetUserDataDirPath()
	if err != nil {
		return fmt.Errorf("resolve data dir: %w", err)
	}

	srcPath := filepath.Join(dataDir, "yj.db")

	// Read-only so an export can never disturb a build that is still
	// running against the same working directory.
	db, err := sql.Open("sqlite",
		"file:"+srcPath+"?_pragma=busy_timeout(10000)&mode=ro")
	if err != nil {
		return fmt.Errorf("open source index: %w", err)
	}

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

	var srcRows int
	if err := db.QueryRow(
		"SELECT COUNT(*) FROM explore_index",
	).Scan(&srcRows); err != nil {
		return fmt.Errorf("count source rows: %w", err)
	}

	if srcRows == 0 {
		return errEmptyIndex
	}

	fmt.Printf("source: %s (%d rows)\n", srcPath, srcRows)

	if err := os.Remove(out); err != nil && !os.IsNotExist(err) {
		return fmt.Errorf("clear output: %w", err)
	}

	if _, err := db.Exec(`ATTACH DATABASE ? AS core`, out); err != nil {
		return fmt.Errorf("attach output: %w", err)
	}

	if err := createSchema(db); err != nil {
		return err
	}

	if err := copyRows(db, artists, perArtistRGs, perArtistRecs); err != nil {
		return err
	}

	if err := stampMeta(db, srcRows); err != nil {
		return err
	}

	// DETACH before VACUUM: sqlite cannot vacuum an attached database.
	if _, err := db.Exec(`DETACH DATABASE core`); err != nil {
		return fmt.Errorf("detach output: %w", err)
	}

	if err := vacuum(out); err != nil {
		return err
	}

	return report(out)
}

// createSchema builds the artifact's tables. No FTS and no triggers —
// the importing client's own trigger rebuilds its FTS on insert.
func createSchema(db *sql.DB) error {
	stmts := []string{
		// Column types mirror the app's own explore_index, because the
		// export is a straight copy: MBIDs as 16 raw bytes, entity
		// types as codes.  An importer that meets the older text form
		// converts it (artifactSelectColumns), so this is a size change
		// and not a compatibility break.
		`CREATE TABLE core.explore_index (
			entity_type      INTEGER NOT NULL,
			mbid             BLOB NOT NULL,
			title            TEXT NOT NULL,
			artist_name      TEXT NOT NULL,
			artist_mbid      BLOB NOT NULL,
			aliases          TEXT NOT NULL DEFAULT '',
			popularity       INTEGER NOT NULL DEFAULT 0,
			listener_count   INTEGER NOT NULL DEFAULT 0,
			duration         INTEGER NOT NULL DEFAULT 0,
			caa_release_mbid BLOB NOT NULL DEFAULT x'',
			release_name     TEXT NOT NULL DEFAULT '',
			primary_type     TEXT NOT NULL DEFAULT '',
			secondary_types  TEXT NOT NULL DEFAULT '',
			release_date     TEXT NOT NULL DEFAULT '',
			total_tracks     INTEGER NOT NULL DEFAULT 0,
			artist_type      TEXT NOT NULL DEFAULT '',
			country          TEXT NOT NULL DEFAULT '',
			disambiguation   TEXT NOT NULL DEFAULT '',
			sort_name        TEXT NOT NULL DEFAULT '',
			discog_fetched   INTEGER NOT NULL DEFAULT 0,
			PRIMARY KEY (mbid)
		) WITHOUT ROWID`,
		`CREATE TABLE core.artifact_meta (
			key   TEXT PRIMARY KEY,
			value TEXT NOT NULL
		)`,
		// Multi-artist credits.  Shipped as their own tables rather than
		// as an explore_index column because a credit is a variable
		// number of ordered parts, and because credits are *shared* --
		// an album's tracks by one artist reference one credit, which is
		// what keeps this to a few hundred thousand rows.
		//
		// An importer that predates these reads an artifact without
		// them; artifactHasCredits is what asks.
		`CREATE TABLE core.artist_credit_part (
			credit_id     INTEGER NOT NULL,
			position      INTEGER NOT NULL,
			artist_mbid   BLOB NOT NULL,
			credited_name TEXT NOT NULL,
			join_phrase   TEXT NOT NULL DEFAULT '',
			PRIMARY KEY (credit_id, position)
		) WITHOUT ROWID`,
		`CREATE TABLE core.artist_credit_ref (
			mbid      BLOB NOT NULL PRIMARY KEY,
			credit_id INTEGER NOT NULL
		) WITHOUT ROWID`,
	}

	for _, stmt := range stmts {
		if _, err := db.Exec(stmt); err != nil {
			return fmt.Errorf("create artifact schema: %w", err)
		}
	}

	return nil
}

// copyRows selects the core subset: the top artists by listen count,
// then a bounded slice of each one's release groups and recordings.
//
// The per-artist window mirrors the S2 coverage already in
// dumpcatalog.go — a flat global top-N would give a handful of
// superstars everything and everyone else nothing.
func copyRows(db *sql.DB, artists, perArtistRGs, perArtistRecs int) error {
	// The destination is created by this binary and always has every
	// column; only the source may be older.
	srcColumns := sourceColumns(db)

	if _, err := db.Exec(`
		CREATE TEMP TABLE core_artists AS
		SELECT mbid FROM main.explore_index
		WHERE entity_type = 1 /* artist */
		ORDER BY popularity DESC
		LIMIT ?`, artists,
	); err != nil {
		return fmt.Errorf("select core artists: %w", err)
	}

	copied, err := insertSelect(db, `
		INSERT INTO core.explore_index (`+catalogColumns+`)
		SELECT `+srcColumns+`
		FROM main.explore_index
		WHERE entity_type = 1 /* artist */
		  AND mbid IN (SELECT mbid FROM core_artists)`)
	if err != nil {
		return err
	}

	fmt.Printf("  artists:        %d\n", copied)

	// The entity codes are the catalog's storage form; see
	// backend/explore/mbid.go.  The exporter copies the local index's
	// encoding through unchanged, so the artifact carries it too - and
	// the importer accepts either, so an artifact built before this
	// still imports.
	for _, sel := range []struct {
		label  string
		entity int
		limit  int
	}{
		{"release groups", 2 /* release_group */, perArtistRGs},
		{"recordings", 3 /* recording */, perArtistRecs},
	} {
		// The window is over artist_mbid so each artist contributes at
		// most `limit` rows, ranked by their own listen counts.
		n, err := insertSelect(db, `
			INSERT INTO core.explore_index (`+catalogColumns+`)
			SELECT `+srcColumns+` FROM (
				SELECT *, ROW_NUMBER() OVER (
					PARTITION BY artist_mbid ORDER BY popularity DESC
				) AS rn
				FROM main.explore_index
				WHERE entity_type = ?
				  AND artist_mbid IN (SELECT mbid FROM core_artists)
			) WHERE rn <= ?`, sel.entity, sel.limit)
		if err != nil {
			return err
		}

		fmt.Printf("  %-15s %d\n", sel.label+":", n)
	}

	return copyCredits(db)
}

// copyCredits ships the credit decomposition for the entities that made
// it into the artifact, and only those.
//
// The refs go first and the parts follow *from* the refs, so a credit is
// carried only if something in the artifact points at it.  The source
// index holds credits for every catalog entity, while the artifact is a
// windowed subset -- copying all of them would carry a large table most
// of which nothing in the artifact can reach.
//
// A source index built before the credit pass simply has no rows here,
// which is not an error: the artifact then carries the tables empty, and
// every credit falls back to its single artist exactly as before.
func copyCredits(db *sql.DB) error {
	// Asked, not assumed.  A source index built before the credit pass
	// has no such table, and "no such table" would fail an export whose
	// catalog is otherwise complete.
	for _, table := range []string{"artist_credit_ref", "artist_credit_part"} {
		var n int

		if err := db.QueryRow(
			`SELECT COUNT(*) FROM main.sqlite_master
			 WHERE type = 'table' AND name = ?`, table,
		).Scan(&n); err != nil {
			return fmt.Errorf("probe %s: %w", table, err)
		}

		if n == 0 {
			fmt.Printf("  %-15s none in source\n", "credits:")

			return nil
		}
	}

	refs, err := insertSelect(db, `
		INSERT INTO core.artist_credit_ref (mbid, credit_id)
		SELECT r.mbid, r.credit_id
		FROM main.artist_credit_ref r
		WHERE r.mbid IN (SELECT mbid FROM core.explore_index)`)
	if err != nil {
		return err
	}

	parts, err := insertSelect(db, `
		INSERT INTO core.artist_credit_part
			(credit_id, position, artist_mbid, credited_name, join_phrase)
		SELECT p.credit_id, p.position, p.artist_mbid, p.credited_name, p.join_phrase
		FROM main.artist_credit_part p
		WHERE p.credit_id IN (SELECT credit_id FROM core.artist_credit_ref)`)
	if err != nil {
		return err
	}

	fmt.Printf("  %-15s %d refs, %d parts\n", "credits:", refs, parts)

	return nil
}

func insertSelect(db *sql.DB, query string, args ...any) (int64, error) {
	res, err := db.Exec(query, args...)
	if err != nil {
		return 0, fmt.Errorf("copy rows: %w", err)
	}

	n, err := res.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("rows affected: %w", err)
	}

	return n, nil
}

// stampMeta records what the importing client needs to know: that the
// catalog half is already populated, and which incremental listens
// series the popularity numbers are baselined on, so the incremental
// refresh resumes from the right point instead of reapplying deltas.
func stampMeta(db *sql.DB, srcRows int) error {
	series := lookupMeta(db, "listens_applied_series")
	built := lookupMeta(db, "dump_import_done")

	if built == "" {
		built = time.Now().UTC().Format(time.RFC3339)
	}

	entries := map[string]string{
		"artifact_version":       "1",
		"built_at":               built,
		"source_rows":            strconv.Itoa(srcRows),
		"listens_applied_series": series,
	}

	for k, v := range entries {
		if _, err := db.Exec(
			`INSERT OR REPLACE INTO core.artifact_meta (key, value) VALUES (?, ?)`,
			k, v,
		); err != nil {
			return fmt.Errorf("stamp %s: %w", k, err)
		}
	}

	return nil
}

func lookupMeta(db *sql.DB, key string) string {
	var value string

	row := db.QueryRow(
		`SELECT value FROM main.explore_index_meta WHERE key = ?`, key)
	if err := row.Scan(&value); err != nil {
		return ""
	}

	return value
}

func vacuum(path string) error {
	db, err := sql.Open("sqlite", "file:"+path)
	if err != nil {
		return fmt.Errorf("reopen artifact: %w", err)
	}

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

	if _, err := db.Exec("VACUUM"); err != nil {
		return fmt.Errorf("vacuum artifact: %w", err)
	}

	return nil
}

func report(path string) error {
	fi, err := os.Stat(path)
	if err != nil {
		return fmt.Errorf("stat artifact: %w", err)
	}

	fmt.Printf("\nartifact: %s (%.1f MB)\n",
		path, float64(fi.Size())/(1<<20))
	fmt.Println("compress with: zstd -19 -T0", path)

	return nil
}