music-hub/server/src/queue.ts

// 下载队列：任务持久化在 SQLite，进度只在内存中并通过事件总线推送
import { config } from './config.ts';
import { downloadTrack, type Stage } from './core/pipeline.ts';
import { getProvider } from './core/providers/index.ts';
import { isQualityKey, type QualityKey } from './core/quality.ts';
import { FatalError, type Source } from './core/types.ts';
import { errorMessage, sleep } from './core/util.ts';
import { addHistory, db, getSetting, setSetting } from './db.ts';
import { bus } from './events.ts';
import { getSettings } from './settings.ts';

export type TaskStatus = 'queued' | 'running' | 'paused' | 'done' | 'failed' | 'canceled';

export interface Task {
  id: number;
  source: Source;
  trackId: string;
  name: string;
  artists: string;
  album: string;
  cover: string | null;
  quality: QualityKey;
  groupTitle: string | null;
  status: TaskStatus;
  stage: Stage | null;
  error: string | null;
  actualQuality: string | null;
  bitrate: number | null;
  size: number | null;
  file: string | null;
  skipped: boolean;
  warnings: string[];
  createdAt: string;
  updatedAt: string;
}

export interface TaskProgress {
  id: number;
  stage?: Stage;
  received: number;
  total: number;
  speed: number;
}

export interface EnqueueItem {
  source: Source;
  id: string;
  name?: string;
  artists?: string;
  album?: string;
  cover?: string;
}

const COLUMNS = `id, source, track_id AS trackId, name, artists, album, cover, quality, group_title AS groupTitle,
  status, stage, error, actual_quality AS actualQuality, bitrate, size, file, skipped, warnings,
  created_at AS createdAt, updated_at AS updatedAt`;

function toTask(row: any): Task {
  return { ...row, skipped: !!row.skipped, warnings: row.warnings ? JSON.parse(row.warnings) : [] };
}

function getTask(id: number): Task | null {
  const row = db.prepare(`SELECT ${COLUMNS} FROM tasks WHERE id = ?`).get(id);
  return row ? toTask(row) : null;
}

function update(id: number, fields: Record<string, unknown>): Task | null {
  const keys = Object.keys(fields);
  db.prepare(`UPDATE tasks SET ${keys.map((k) => `${k} = ?`).join(', ')}, updated_at = ? WHERE id = ?`).run(
    ...(keys.map((k) => fields[k]) as any[]),
    new Date().toISOString(),
    id,
  );
  const task = getTask(id);
  if (task) bus.emit('task', task);
  return task;
}

type AbortReason = 'pause' | 'cancel';

class DownloadQueue {
  private running = new Map<number, AbortController>();
  private progress = new Map<number, TaskProgress>();
  private lastStart = 0;
  paused = getSetting('queue.paused', false);

  constructor() {
    // 上次进程退出时还在跑的任务重新排队
    db.prepare("UPDATE tasks SET status = 'queued', stage = NULL WHERE status = 'running'").run();
  }

  list(scope: 'active' | 'finished' | 'all' = 'all', limit = 500): Task[] {
    const where =
      scope === 'active'
        ? "WHERE status IN ('queued','running','paused')"
        : scope === 'finished'
          ? "WHERE status IN ('done','failed','canceled')"
          : '';
    const order = scope === 'finished' ? 'ORDER BY updated_at DESC, id DESC' : 'ORDER BY id';
    return (db.prepare(`SELECT ${COLUMNS} FROM tasks ${where} ${order} LIMIT ?`).all(limit) as any[]).map(toTask);
  }

  progressSnapshot(): TaskProgress[] {
    return [...this.progress.values()];
  }

  summary() {
    const rows = db.prepare('SELECT status, COUNT(*) AS n FROM tasks GROUP BY status').all() as Array<{ status: TaskStatus; n: number }>;
    const counts = Object.fromEntries(rows.map((r) => [r.status, r.n])) as Partial<Record<TaskStatus, number>>;
    return { paused: this.paused, concurrency: getSettings().concurrency, counts };
  }

  private emitSummary() {
    bus.emit('queue', this.summary());
  }

  enqueue(items: EnqueueItem[], opts: { quality?: string; group?: string } = {}): { added: Task[]; duplicates: number } {
    const quality = isQualityKey(opts.quality) ? opts.quality : getSettings().quality;
    const active = db.prepare("SELECT id FROM tasks WHERE source = ? AND track_id = ? AND status IN ('queued','running','paused')");
    const insert = db.prepare(`
      INSERT INTO tasks (source, track_id, name, artists, album, cover, quality, group_title, status, created_at, updated_at)
      VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?)
    `);
    const added: Task[] = [];
    let duplicates = 0;
    const now = new Date().toISOString();
    db.exec('BEGIN');
    try {
      for (const item of items) {
        getProvider(item.source);
        if (active.get(item.source, String(item.id))) {
          duplicates++;
          continue;
        }
        const { lastInsertRowid } = insert.run(
          item.source,
          String(item.id),
          item.name || String(item.id),
          item.artists || '',
          item.album || '',
          item.cover || null,
          quality,
          opts.group || null,
          now,
          now,
        );
        added.push(getTask(Number(lastInsertRowid))!);
      }
      db.exec('COMMIT');
    } catch (e) {
      db.exec('ROLLBACK');
      throw e;
    }
    for (const task of added) bus.emit('task', task);
    this.emitSummary();
    this.pump();
    return { added, duplicates };
  }

  pause(id: number) {
    const task = getTask(id);
    if (task?.status === 'queued') update(id, { status: 'paused' });
    else if (task?.status === 'running') this.running.get(id)?.abort('pause' satisfies AbortReason);
    this.emitSummary();
  }

  resume(id: number) {
    const task = getTask(id);
    if (task?.status === 'paused') update(id, { status: 'queued', stage: null });
    this.emitSummary();
    this.pump();
  }

  retry(id: number) {
    const task = getTask(id);
    if (task && (task.status === 'failed' || task.status === 'canceled')) {
      update(id, { status: 'queued', stage: null, error: null });
    }
    this.emitSummary();
    this.pump();
  }

  retryFailed(): number {
    const ids = (db.prepare("SELECT id FROM tasks WHERE status = 'failed'").all() as Array<{ id: number }>).map((r) => r.id);
    for (const id of ids) update(id, { status: 'queued', stage: null, error: null });
    this.emitSummary();
    this.pump();
    return ids.length;
  }

  cancel(id: number) {
    const task = getTask(id);
    if (task?.status === 'queued' || task?.status === 'paused') update(id, { status: 'canceled', stage: null });
    else if (task?.status === 'running') this.running.get(id)?.abort('cancel' satisfies AbortReason);
    this.emitSummary();
  }

  remove(id: number) {
    if (this.running.has(id)) return;
    db.prepare('DELETE FROM tasks WHERE id = ?').run(id);
    bus.emit('task-removed', { ids: [id] });
    this.emitSummary();
  }

  clearFinished(includeFailed = false) {
    const statuses = includeFailed ? "('done','canceled','failed')" : "('done','canceled')";
    const ids = (db.prepare(`SELECT id FROM tasks WHERE status IN ${statuses}`).all() as Array<{ id: number }>).map((r) => r.id);
    db.prepare(`DELETE FROM tasks WHERE status IN ${statuses}`).run();
    bus.emit('task-removed', { ids });
    this.emitSummary();
    return ids.length;
  }

  setPaused(paused: boolean) {
    this.paused = paused;
    setSetting('queue.paused', paused);
    this.emitSummary();
    if (!paused) this.pump();
  }

  /** 按并发数启动排队中的任务。全局暂停只阻止新任务开始，进行中的任务会跑完 */
  pump() {
    if (this.paused) return;
    const { concurrency } = getSettings();
    while (this.running.size < concurrency) {
      const exclude = [...this.running.keys()];
      const row = db
        .prepare(
          `SELECT id FROM tasks WHERE status = 'queued' ${exclude.length ? `AND id NOT IN (${exclude.join(',')})` : ''} ORDER BY id LIMIT 1`,
        )
        .get() as { id: number } | undefined;
      if (!row) break;
      const controller = new AbortController();
      this.running.set(row.id, controller);
      update(row.id, { status: 'running', stage: null, error: null });
      void this.run(row.id, controller);
    }
  }

  private async run(id: number, controller: AbortController) {
    const task = getTask(id)!;
    const settings = getSettings();
    const live: TaskProgress = { id, received: 0, total: 0, speed: 0 };
    this.progress.set(id, live);
    try {
      // 任务之间保持间隔，避免短时间内大量请求上游接口
      const wait = this.lastStart + settings.gapMs - Date.now();
      this.lastStart = Math.max(Date.now(), this.lastStart + settings.gapMs);
      if (wait > 0) await sleep(wait, controller.signal);

      const result = await downloadTrack(
        getProvider(task.source),
        task.trackId,
        { ...settings, quality: task.quality },
        {
          musicDir: config.musicDir,
          signal: controller.signal,
          reporter: {
            stage: (stage) => {
              live.stage = stage;
              update(id, { stage });
            },
            progress: (p) => {
              Object.assign(live, p);
              bus.emit('progress', live);
            },
          },
        },
      );
      update(id, {
        status: 'done',
        stage: null,
        name: result.name,
        artists: result.artists,
        album: result.album,
        actual_quality: result.quality,
        bitrate: result.bitrate ?? null,
        size: result.size,
        file: result.relFile,
        skipped: result.skipped ? 1 : 0,
        warnings: JSON.stringify(result.warnings),
      });
      addHistory({
        source: task.source,
        trackId: task.trackId,
        name: result.name,
        artists: result.artists,
        album: result.album,
        quality: result.quality,
        bitrate: result.bitrate ?? null,
        size: result.size,
        file: result.relFile,
      });
      bus.emit('history');
    } catch (e) {
      const reason = controller.signal.aborted ? (controller.signal.reason as AbortReason) : null;
      if (reason === 'pause') update(id, { status: 'paused', stage: null });
      else if (reason === 'cancel') update(id, { status: 'canceled', stage: null });
      else {
        update(id, { status: 'failed', stage: null, error: errorMessage(e) });
        if (!(e instanceof FatalError)) console.warn(`✖ 任务 ${id} 失败: ${errorMessage(e)}`);
      }
    } finally {
      this.running.delete(id);
      this.progress.delete(id);
      this.emitSummary();
      this.pump();
    }
  }
}

export const queue = new DownloadQueue();

bus.on('settings', () => queue.pump());