netfelix-audio-fix/server/services/rescan.ts

import type { Database } from "bun:sqlite";
import type { JellyfinItem, MediaStream } from "../types";
import { analyzeItem } from "./analyzer";
import { extractOriginalLanguage, mapStream } from "./jellyfin";

export interface RescanConfig {
	audioLanguages: string[];
}

export interface RescanResult {
	itemId: number;
	origLang: string | null;
	origLangSource: string | null;
	needsReview: number;
	isNoop: boolean;
}

/**
 * Upsert a single Jellyfin item (metadata + streams + review_plan + decisions)
 * in one transaction. Shared by the full scan loop and the post-execute refresh.
 *
 * Returns the internal item id and a summary of what happened so callers can
 * aggregate counters or emit SSE events.
 */
export async function upsertJellyfinItem(
	db: Database,
	jellyfinItem: JellyfinItem,
	cfg: RescanConfig,
	opts: { executed?: boolean; source?: "scan" | "webhook" } = {},
): Promise<RescanResult> {
	const source = opts.source ?? "scan";
	if (!jellyfinItem.Name || !jellyfinItem.Path) {
		throw new Error(`Jellyfin item ${jellyfinItem.Id} missing Name or Path`);
	}
	const itemName: string = jellyfinItem.Name;
	const itemPath: string = jellyfinItem.Path;

	const providerIds = jellyfinItem.ProviderIds ?? {};
	const seriesProviderIds = jellyfinItem.SeriesProviderIds ?? {};
	const imdbId = providerIds.Imdb ?? null;
	const tmdbId = providerIds.Tmdb ?? null;
	const tvdbId =
		jellyfinItem.Type === "Episode"
			? (seriesProviderIds.Tvdb ?? providerIds.Tvdb ?? null)
			: (providerIds.Tvdb ?? null);

	// Preserve manual overrides: if the user has already pinned a language via
	// /series/:key/language or /:id/language, scans must not blast it back to
	// the audio-track guess. Read the prior row BEFORE running the guesses.
	const existing = db
		.prepare("SELECT original_language, orig_lang_source FROM media_items WHERE jellyfin_id = ?")
		.get(jellyfinItem.Id) as { original_language: string | null; orig_lang_source: string | null } | undefined;
	const hasManualOverride = existing?.orig_lang_source === "manual";

	// See scan.ts for the "8 Mile got labelled Turkish" rationale. Jellyfin's
	// first-audio-track guess is an unverified starting point.
	const jellyfinGuess = extractOriginalLanguage(jellyfinItem);
	let origLang: string | null = hasManualOverride ? (existing?.original_language ?? null) : jellyfinGuess;
	let origLangSource: "jellyfin" | "manual" | null = hasManualOverride
		? "manual"
		: jellyfinGuess
			? "jellyfin"
			: null;
	let needsReview = origLang ? 0 : 1;
	const authoritative = hasManualOverride;

	if (origLang && !authoritative && !needsReview) needsReview = 1;

	const jellyfinRaw = JSON.stringify(jellyfinItem);

	const result: RescanResult = {
		itemId: -1,
		origLang: null,
		origLangSource: null,
		needsReview: 1,
		isNoop: false,
	};

	// One transaction per item keeps scan throughput high on SQLite — every
	// INSERT/UPDATE would otherwise hit WAL independently.
	db.transaction(() => {
		const upsertItem = db.prepare(`
			INSERT INTO media_items (
				jellyfin_id, type, name, original_title, series_name, series_jellyfin_id,
				season_number, episode_number, year, file_path, file_size, container,
				runtime_ticks, date_last_refreshed,
				original_language, orig_lang_source, needs_review,
					imdb_id, tmdb_id, tvdb_id,
					jellyfin_raw, external_raw,
					scan_status, last_scanned_at, ingest_source${opts.executed ? ", last_executed_at" : ""}
				) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'scanned', datetime('now'), ?${opts.executed ? ", datetime('now')" : ""})
				ON CONFLICT(jellyfin_id) DO UPDATE SET
				type = excluded.type, name = excluded.name, original_title = excluded.original_title,
				series_name = excluded.series_name, series_jellyfin_id = excluded.series_jellyfin_id,
				season_number = excluded.season_number, episode_number = excluded.episode_number,
				year = excluded.year, file_path = excluded.file_path,
				file_size = excluded.file_size, container = excluded.container,
				runtime_ticks = excluded.runtime_ticks, date_last_refreshed = excluded.date_last_refreshed,
				original_language = excluded.original_language, orig_lang_source = excluded.orig_lang_source,
					needs_review = excluded.needs_review, imdb_id = excluded.imdb_id,
					tmdb_id = excluded.tmdb_id, tvdb_id = excluded.tvdb_id,
					jellyfin_raw = excluded.jellyfin_raw, external_raw = excluded.external_raw,
					scan_status = 'scanned', last_scanned_at = datetime('now'),
					ingest_source = excluded.ingest_source
					${opts.executed ? ", last_executed_at = datetime('now')" : ""}
			`);
		upsertItem.run(
			jellyfinItem.Id,
			jellyfinItem.Type === "Episode" ? "Episode" : "Movie",
			itemName,
			jellyfinItem.OriginalTitle ?? null,
			jellyfinItem.SeriesName ?? null,
			jellyfinItem.SeriesId ?? null,
			jellyfinItem.ParentIndexNumber ?? null,
			jellyfinItem.IndexNumber ?? null,
			jellyfinItem.ProductionYear ?? null,
			itemPath,
			jellyfinItem.Size ?? null,
			jellyfinItem.Container ?? null,
			jellyfinItem.RunTimeTicks ?? null,
			jellyfinItem.DateLastRefreshed ?? null,
			origLang,
			origLangSource,
			needsReview,
			imdbId,
			tmdbId,
			tvdbId,
			jellyfinRaw,
			null, // external_raw — no longer populated during scan
			source,
		);

		const itemRow = db.prepare("SELECT id FROM media_items WHERE jellyfin_id = ?").get(jellyfinItem.Id) as {
			id: number;
		};
		const itemId = itemRow.id;
		result.itemId = itemId;

		db.prepare("DELETE FROM media_streams WHERE item_id = ?").run(itemId);
		const insertStream = db.prepare(`
			INSERT INTO media_streams (
				item_id, stream_index, type, codec, profile, language, language_display,
				title, is_default, is_forced, is_hearing_impaired,
				channels, channel_layout, bit_rate, sample_rate, bit_depth
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		`);
		for (const jStream of jellyfinItem.MediaStreams ?? []) {
			if (jStream.IsExternal) continue;
			const s = mapStream(jStream);
			insertStream.run(
				itemId,
				s.stream_index,
				s.type,
				s.codec,
				s.profile,
				s.language,
				s.language_display,
				s.title,
				s.is_default,
				s.is_forced,
				s.is_hearing_impaired,
				s.channels,
				s.channel_layout,
				s.bit_rate,
				s.sample_rate,
				s.bit_depth,
			);
		}

		const streams = db.prepare("SELECT * FROM media_streams WHERE item_id = ?").all(itemId) as MediaStream[];
		const analysis = analyzeItem(
			{
				original_language: origLang,
				orig_lang_source: origLangSource,
				needs_review: needsReview,
				container: jellyfinItem.Container ?? null,
			},
			streams,
			{ audioLanguages: cfg.audioLanguages },
		);

		// Status transition rules:
		//   is_noop=1 → done (all sources)
		//   done + is_noop=0 + source='webhook' → pending (Jellyfin says the
		//     on-disk file doesn't match the plan, re-open for review)
		//   done + is_noop=0 + source='scan' → done (safety net: a plain
		//     rescan must not reopen plans and risk duplicate jobs; see the
		//     commit that made done terminal)
		//   error → pending (retry loop)
		//   else keep current status
		db
			.prepare(`
			INSERT INTO review_plans (item_id, status, is_noop, auto_class, apple_compat, job_type, notes)
			VALUES (?, 'pending', ?, ?, ?, ?, ?)
			ON CONFLICT(item_id) DO UPDATE SET
				status = CASE
					WHEN excluded.is_noop = 1 THEN 'done'
					WHEN review_plans.status = 'done' AND ? = 'webhook' THEN 'pending'
					WHEN review_plans.status = 'done' THEN 'done'
					WHEN review_plans.status = 'error' THEN 'pending'
					ELSE review_plans.status
				END,
				is_noop = excluded.is_noop,
				auto_class = excluded.auto_class,
				apple_compat = excluded.apple_compat,
				job_type = excluded.job_type,
				notes = excluded.notes
		`)
			.run(
				itemId,
				analysis.is_noop ? 1 : 0,
				analysis.auto_class,
				analysis.apple_compat,
				analysis.job_type,
				analysis.notes.length > 0 ? analysis.notes.join("\n") : null,
				source, // for the CASE WHEN ? = 'webhook' branch
			);

		const planRow = db.prepare("SELECT id FROM review_plans WHERE item_id = ?").get(itemId) as { id: number };
		const upsertDecision = db.prepare(`
			INSERT INTO stream_decisions (plan_id, stream_id, action, target_index, transcode_codec)
			VALUES (?, ?, ?, ?, ?)
			ON CONFLICT(plan_id, stream_id) DO UPDATE SET
				action = excluded.action,
				target_index = excluded.target_index,
				transcode_codec = excluded.transcode_codec
		`);
		for (const dec of analysis.decisions) {
			upsertDecision.run(planRow.id, dec.stream_id, dec.action, dec.target_index, dec.transcode_codec);
		}

		result.origLang = origLang;
		result.origLangSource = origLangSource;
		result.needsReview = needsReview;
		result.isNoop = analysis.is_noop;
	})();

	return result;
}