AeroFetch/src/main/indexer.ts

/**
 * 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 './core/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<FlatInfo> {
  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<FlatInfo | null> {
  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<AbortController>()

/** 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(
  onProgress: (p: IndexProgress) => void,
  url: string
): Promise<IndexSourceResult> {
  const controller = new AbortController()
  activeIndexAborts.add(controller)
  try {
    return await indexSource(onProgress, url, 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(
  onProgress: (p: IndexProgress) => void,
  url: string,
  signal?: AbortSignal
): Promise<IndexSourceResult> {
  // 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 }
  }
}