/** * Source indexing orchestration (Pinchflat-style media manager; see * ROADMAP-PINCHFLAT.md). Walks a channel (its /playlists + /videos tabs) or a * single playlist with `yt-dlp --flat-playlist`, merges the result into a deduped * MediaItem list, and persists it as a Source. The download step (Phase G+) then * pulls from that persisted list rather than the live queue holding it all. * * The pure URL-classification + merge logic lives in indexerCore.ts (unit-tested); * this module is the impure shell that spawns yt-dlp and writes to disk. */ import { existsSync } from 'fs' import { getYtdlpPath, YTDLP_MISSING_MSG } from './binaries' import { execFileAsync } from './lib/exec' import { INDEX_MAX_BUFFER, INDEX_TIMEOUT_MS } from './constants' import { cleanError } from './log' import { assertHttpUrl } from './url' import { classifySource, buildMediaItems, buildFeedUrl, entryUrl, stripTabSuffix, stableSourceId, type RawEntry, type NamedPlaylist } from './indexerCore' import { getSource, upsertSource, mergeMediaItems } from './sources' import type { IndexProgress, IndexSourceResult, Source, SourceKind } from '@shared/ipc' /** The slice of `yt-dlp -J --flat-playlist` output the indexer reads. */ interface FlatInfo { _type?: string id?: string title?: string uploader?: string channel?: string /** the UC… channel id, used to build the RSS feed URL */ channel_id?: string entries?: RawEntry[] } /** * Run `yt-dlp -J --flat-playlist` on a URL and parse the JSON. Rejects on a * non-zero exit or unparseable output. maxBuffer is large because a big channel's * flat upload list can be a few MB of JSON; the timeout is generous for the same * reason. `--` terminates option parsing so the URL can't be read as a flag. */ async function probeFlat(url: string, signal?: AbortSignal): Promise { const r = await execFileAsync( getYtdlpPath(), ['-J', '--flat-playlist', '--no-warnings', '--', url], // `signal` lets a Cancel abort the in-flight probe child mid-walk (B2); Node // kills the process and the call rejects, which the caller treats as canceled. { maxBuffer: INDEX_MAX_BUFFER, timeout: INDEX_TIMEOUT_MS, signal } ) if (!r.ok) { throw new Error( r.timedOut ? 'Timed out indexing source. Check the link or your connection.' : cleanError(r.stderr) || r.error?.message || '' ) } try { return JSON.parse(r.stdout) as FlatInfo } catch { throw new Error('Could not parse source info from yt-dlp.') } } /** Probe a channel tab (e.g. /videos, /playlists); never throws — returns null. */ function probeTab(url: string, signal?: AbortSignal): Promise { return probeFlat(url, signal).catch(() => null) } // Active user-initiated index walks. `cancelIndexing()` aborts them so an // `index:cancel` IPC can stop the walk and kill the in-flight probe child (B2). // Background sync (sync.ts) calls indexSource without a signal, so it isn't // user-cancelable — it's throttled and unattended, not a blocking UI action. const activeIndexAborts = new Set() /** Abort every in-flight user index/reindex walk (the add-source "Cancel" button). */ export function cancelIndexing(): void { for (const c of activeIndexAborts) c.abort() } /** * indexSource wrapped with a fresh AbortController registered for cancelIndexing(). * The IPC layer calls this; sync.ts calls the raw indexSource (not cancelable). */ export async function indexSourceCancelable( url: string, onProgress: (p: IndexProgress) => void ): Promise { const controller = new AbortController() activeIndexAborts.add(controller) try { return await indexSource(url, onProgress, controller.signal) } finally { activeIndexAborts.delete(controller) } } /** * Index (or re-index) a Source. Resolves to the persisted Source + item count, or * an error. `onProgress` is invoked throughout so the renderer can show a live * status line — it's best-effort and never affects the result. */ export async function indexSource( url: string, onProgress: (p: IndexProgress) => void, signal?: AbortSignal ): Promise { // Returned whenever a Cancel arrives mid-walk. It deliberately does NOT push a // progress event: the renderer drives the cancel and resets its own indexing UI, // so a late 'error' event can't re-surface as a failure (B2). const canceled: IndexSourceResult = { ok: false, error: 'Indexing canceled.' } let normalizedUrl: string try { normalizedUrl = assertHttpUrl(url) // normalised form for spawning (audit F5) } catch (e) { return { ok: false, error: (e as Error).message } } if (!existsSync(getYtdlpPath())) { return { ok: false, error: YTDLP_MISSING_MSG } } const cls = classifySource(url) onProgress({ url, phase: 'start', message: 'Reading source…' }) try { let kind: SourceKind let title: string let channel: string | undefined let feedId: string | undefined const playlists: NamedPlaylist[] = [] let uploads: RawEntry[] = [] if (cls?.kind === 'channel') { kind = 'channel' // 1. Enumerate the channel's playlists, then each playlist's videos. onProgress({ url, phase: 'playlists', message: 'Finding playlists…' }) const playlistTab = await probeTab(`${cls.base}/playlists`, signal) if (signal?.aborted) return canceled const playlistEntries = playlistTab?.entries ?? [] const total = playlistEntries.length let done = 0 for (const pe of playlistEntries) { if (signal?.aborted) return canceled // stop before the next probe done++ const purl = entryUrl(pe) if (!purl) continue onProgress({ url, phase: 'playlist', message: `Indexing playlist ${done}/${total}: ${pe.title ?? ''}`.trim(), current: done, total }) const pdata = await probeTab(purl, signal) if (pdata?.entries?.length) { playlists.push({ title: pe.title || 'Playlist', entries: pdata.entries }) } } // 2. The full uploads feed — the catch-all for videos in no playlist. onProgress({ url, phase: 'uploads', message: 'Indexing channel uploads…' }) const videoTab = await probeTab(`${cls.base}/videos`, signal) if (signal?.aborted) return canceled uploads = videoTab?.entries ?? [] title = stripTabSuffix(videoTab?.channel) || stripTabSuffix(videoTab?.title) || stripTabSuffix(playlistTab?.title) || cls.base channel = stripTabSuffix(videoTab?.channel || videoTab?.uploader) || title feedId = videoTab?.channel_id || playlistTab?.channel_id } else { // A single playlist (classified) — or an unclassified URL that might still // resolve to a playlist when probed. A lone video has no entries → error. kind = cls?.kind ?? 'playlist' onProgress({ url, phase: 'uploads', message: 'Indexing playlist…' }) const data = await probeFlat(cls?.base ?? normalizedUrl, signal) if (signal?.aborted) return canceled const entries = data.entries ?? [] if (entries.length === 0) { return { ok: false, error: 'That link is a single video, not a channel or playlist. Use the download bar for one video.' } } title = data.title || 'Playlist' channel = data.uploader || data.channel // A playlist feed keys off the playlist id (PL…), carried as `id` here. feedId = data.id // File everything under the playlist's own name (not the 'Uploads' fallback). playlists.push({ title, entries }) } const sourceId = stableSourceId(cls?.base ?? url.trim()) const fresh = buildMediaItems(sourceId, playlists, uploads) if (fresh.length === 0) { return { ok: false, error: 'No downloadable videos found for this source.' } } // A cancel that landed after the final probe must not persist the source — // otherwise a "canceled" add still pops into the library later (B2). if (signal?.aborted) return canceled // Incremental merge: preserve the downloaded state of anything already on // disk, and report how many videos are new since the last index. const { items, newCount } = mergeMediaItems(sourceId, fresh) const prev = getSource(sourceId) const source: Source = { id: sourceId, url: url.trim(), kind, title, channel, addedAt: prev?.addedAt ?? Date.now(), lastIndexedAt: Date.now(), itemCount: items.length, // Preserve the watched flag across a re-index; refresh the RSS feed URL. watched: prev?.watched, feedUrl: buildFeedUrl(kind, feedId) ?? prev?.feedUrl } upsertSource(source) const newNote = newCount === items.length ? '' : ` (${newCount} new)` onProgress({ url, phase: 'done', message: `Indexed ${items.length} videos${newNote}.`, total: items.length }) return { ok: true, source, itemCount: items.length, newCount } } catch (e) { if (signal?.aborted) return canceled // the throw was our own abort — not an error const msg = e instanceof Error ? e.message : String(e) onProgress({ url, phase: 'error', message: msg }) return { ok: false, error: msg } } }