yellowjacket/backend/explore/indexjob.go

package explore

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	"yellowjacket/backend/jobs"
)

// errIndexStageFailed wraps the per-stage error text reported by the
// dump importer so the job carries a typed failure.
var errIndexStageFailed = errors.New("index build stage failed")

// indexJobID is the stable registry ID for the search index build.
// There is only ever one, so the ID is a constant.
const indexJobID = "index:build"

// indexPausedKey marks a build the user paused.  It lives in
// explore_index_meta alongside the import's other checkpoints so a
// paused build stays paused across a restart instead of resuming on
// the next launch.
const indexPausedKey = "index_build_paused"

// SetJobRegistry wires the background job registry so index builds
// report progress, stage state, logs, and pause/cancel controls.
func (si *SearchIndex) SetJobRegistry(reg *jobs.Registry) {
	si.mu.Lock()
	si.jobs = reg
	si.mu.Unlock()
}

// jobRegistry returns the registry, or nil when none is wired.
func (si *SearchIndex) jobRegistry() *jobs.Registry {
	si.mu.RLock()
	defer si.mu.RUnlock()

	return si.jobs
}

// logIndexJob appends a line to the index build's job log, if a build
// job is currently registered.
func (si *SearchIndex) logIndexJob(level jobs.Level, message string) {
	reg := si.jobRegistry()
	if reg == nil {
		return
	}

	if h := reg.Get(indexJobID); h != nil {
		h.Logf(level, message)
	}
}

// indexJobSpec builds the registry spec for the index build.
//
// The build is genuinely pausable rather than merely cancellable: the
// dump importer checkpoints its listen-count offset to counts.bin and
// its stage to state.json, so stopping and restarting picks up where it
// left off instead of re-downloading multiple gigabytes.
func (si *SearchIndex) indexJobSpec(state jobs.State) jobs.Spec {
	return jobs.Spec{
		ID:       indexJobID,
		Kind:     jobs.KindIndexBuild,
		Title:    "Building search index",
		Subtitle: "MusicBrainz catalog + ListenBrainz popularity",
		State:    state,
		Caps: jobs.Caps{
			Pausable:    true,
			Cancellable: true,
		},
		Durable: true,
		Controls: jobs.Controls{
			Pause:  si.PauseBuild,
			Resume: si.ResumeBuild,
			Cancel: si.CancelBuild,
		},
	}
}

// syncIndexJob mirrors an IndexStatus snapshot into the job registry.
// It is driven from emitStatus, which every status mutation already
// funnels through, so there is no path that updates one view and not
// the other.
func (si *SearchIndex) syncIndexJob(status IndexStatus) {
	reg := si.jobRegistry()
	if reg == nil {
		return
	}

	si.mu.RLock()
	paused := si.buildPaused
	si.mu.RUnlock()

	h := reg.Get(indexJobID)

	// A build with no stages is the early-return path in runDumpBuild
	// (the catalog import is already done).  Nothing to show.
	if h == nil {
		if !status.Building || len(status.Tiers) == 0 {
			return
		}

		h = reg.Start(si.indexJobSpec(jobs.StateRunning))
		h.Logf(jobs.LevelInfo, "Index build started")
	}

	// A finished job is immutable.  Without this guard the 3-second
	// status ticker would keep touching it forever, re-emitting
	// JobsChanged long after the build ended.
	if h.State().IsTerminal() {
		return
	}

	si.applyStagesToJob(h, status)

	if status.Building {
		// Don't stomp a pause or cancel that is still settling; those
		// transitions are confirmed by their own control paths.
		if h.State() == jobs.StateQueued {
			h.SetState(jobs.StateRunning)
		}

		return
	}

	si.finishIndexJob(h, paused)
}

// applyStagesToJob maps index tiers onto job stages and derives the
// headline progress bar from whichever tier is currently running.
func (si *SearchIndex) applyStagesToJob(h *jobs.Handle, status IndexStatus) {
	stages := make([]jobs.Stage, 0, len(status.Tiers))

	var (
		phase            string
		current, total   int64
		foundRunningTier bool
	)

	for _, t := range status.Tiers {
		stages = append(stages, jobs.Stage{
			Name:    t.Name,
			State:   t.State,
			Current: int64(t.Completed),
			Total:   int64(t.Total),
			Error:   t.Error,
		})

		if t.State == "running" && !foundRunningTier {
			foundRunningTier = true
			phase = t.Name
			current = int64(t.Completed)
			total = int64(t.Total)

			// The listens stream's raw completed/total is a bare
			// percentage; the detail line is what makes it legible.
			if t.Detail != "" {
				phase = t.Name + " — " + t.Detail
			}
		}
	}

	h.SetStages(stages)

	if foundRunningTier {
		h.SetPhase(phase)
		h.SetProgress(current, total)
	}

	h.SetStats([]jobs.Stat{
		{Label: "Artists", Value: strconv.Itoa(status.Artists)},
		{Label: "Release groups", Value: strconv.Itoa(status.ReleaseGroups)},
		{Label: "Recordings", Value: strconv.Itoa(status.Recordings)},
		{Label: "Total rows", Value: strconv.Itoa(status.TotalRows)},
	})
}

// finishIndexJob resolves a build that is no longer running into the
// right terminal (or paused) state.  A stopped build that never wrote
// the done marker is reported as cancelled rather than complete —
// claiming success for a half-finished import would be a lie.
func (si *SearchIndex) finishIndexJob(h *jobs.Handle, paused bool) {
	if h.State().IsTerminal() || h.State() == jobs.StatePaused {
		return
	}

	// A stage that errored means the build failed; reporting that as
	// "stopped" would hide a real failure behind a neutral word.
	if !paused {
		for _, stage := range h.Snapshot().Stages {
			if stage.State == "error" {
				h.Fail(fmt.Errorf("%w: %s: %s",
					errIndexStageFailed, stage.Name, stage.Error))

				return
			}
		}
	}

	if paused {
		h.SetPhase("Paused")
		h.SetState(jobs.StatePaused)
		h.Logf(jobs.LevelInfo,
			"Build paused — progress is checkpointed and will resume "+
				"from here")

		return
	}

	if si.hasMeta(dumpImportDoneKey) {
		h.Logf(jobs.LevelInfo, "Index build complete")
		h.Complete()

		return
	}

	h.Logf(jobs.LevelInfo, "Build stopped before finishing")
	h.Cancelled()
}

// PauseBuild stops the in-flight index build and remembers that the
// user asked for it, so it is not restarted automatically — including
// on the next launch.  Blocks until the build goroutine exits; the job
// registry invokes controls on their own goroutine.
func (si *SearchIndex) PauseBuild() {
	si.mu.Lock()
	si.buildPaused = true
	si.mu.Unlock()

	si.setMeta(indexPausedKey, "1")
	si.StopBuild()
	si.emitStatus()
}

// ResumeBuild clears the pause and restarts the build, which picks up
// from the importer's last checkpoint.
func (si *SearchIndex) ResumeBuild() {
	si.mu.Lock()
	si.buildPaused = false
	ctx := si.runtimeCtx
	si.mu.Unlock()

	si.deleteMeta(indexPausedKey)

	if reg := si.jobRegistry(); reg != nil {
		if h := reg.Get(indexJobID); h != nil {
			h.SetState(jobs.StateRunning)
			h.Logf(jobs.LevelInfo, "Resuming from last checkpoint")
		}
	}

	if ctx != nil {
		si.StartBuild(ctx)
	}
}

// CancelBuild stops the build without marking it paused.  The importer's
// on-disk checkpoints are left in place, so starting a new build later
// still resumes rather than re-downloading — cancel here means "stop
// working now", not "throw away the progress".
func (si *SearchIndex) CancelBuild() {
	si.mu.Lock()
	si.buildPaused = false
	si.mu.Unlock()

	si.deleteMeta(indexPausedKey)
	si.StopBuild()
	si.emitStatus()
}

// ImportComplete reports whether the dump import wrote its done marker,
// meaning every stage finished.  A resumable import that was interrupted
// leaves this false even though the index may already be queryable.
func (si *SearchIndex) ImportComplete() bool {
	return si.hasMeta(dumpImportDoneKey)
}

// BaselineSeries returns the incremental listens series the index's
// popularity numbers are currently caught up to, or 0 when no baseline
// import has completed.  A change in this value between two runs is the
// signal that a refresh actually folded in new data.
func (si *SearchIndex) BaselineSeries() int {
	series, _ := si.metaInt(listensAppliedSeriesKey)

	return series
}

// LastImported returns when the dump import last completed, or the zero
// time if it never has.  Drives the rebuild cadence.
func (si *SearchIndex) LastImported() time.Time {
	rows, err := si.db.QueryContext(
		"SELECT value FROM explore_index_meta WHERE key = ?", dumpImportDoneKey,
	)
	if err != nil {
		return time.Time{}
	}

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

	if !rows.Next() {
		return time.Time{}
	}

	var raw string
	if err := rows.Scan(&raw); err != nil {
		return time.Time{}
	}

	parsed, err := time.Parse(time.RFC3339, raw)
	if err != nil {
		return time.Time{}
	}

	return parsed
}

// PrepareRebuild clears the completion marker so the next StartBuild
// re-imports from the newest published dump instead of short-circuiting.
//
// The importer deletes its staging directory on completion, so there is
// no stale checkpoint to clear as well — a rebuild rediscovers the
// current dump and starts from offset zero.  Existing rows are left in
// place: assembly upserts by MBID, so the index stays queryable
// throughout rather than going empty for the length of a re-import.
func (si *SearchIndex) PrepareRebuild() {
	si.deleteMeta(dumpImportDoneKey)
	si.logger.Info("search index: cleared completion marker for rebuild")
}

// RefreshNow folds any newly published incremental listens dumps into
// the index's popularity numbers, synchronously.  Pass 0 to bypass the
// cadence gate.
func (si *SearchIndex) RefreshNow(ctx context.Context, minInterval time.Duration) {
	si.RefreshListenCounts(ctx, minInterval)
}

// buildPausedByUser reports whether a build was paused and not resumed,
// including by a previous session.
func (si *SearchIndex) buildPausedByUser() bool {
	si.mu.RLock()
	paused := si.buildPaused
	si.mu.RUnlock()

	if paused {
		return true
	}

	return si.hasMeta(indexPausedKey)
}

// AdoptPausedBuild re-registers a build that was paused when the app
// last shut down, so it shows up in the jobs panel with a resume button
// instead of silently not running.  Called during startup.
func (si *SearchIndex) AdoptPausedBuild() {
	reg := si.jobRegistry()
	if reg == nil {
		return
	}

	// A completed import cannot be meaningfully paused; clear a stale
	// marker rather than showing a job that would never do anything.
	if si.hasMeta(dumpImportDoneKey) {
		si.deleteMeta(indexPausedKey)
		reg.Remove(indexJobID)

		return
	}

	if !si.hasMeta(indexPausedKey) {
		return
	}

	si.mu.Lock()
	si.buildPaused = true
	si.mu.Unlock()

	h := reg.Start(si.indexJobSpec(jobs.StatePaused))
	h.SetPhase("Paused")
	h.Logf(jobs.LevelInfo,
		"Paused in a previous session — resume to continue from the "+
			"last checkpoint")

	si.logger.Info("search index: restored paused build")
}