video-workflow/src/app/services/batch-job.service.ts

import { Injectable } from '@angular/core';
import { BehaviorSubject } from 'rxjs';
import { AuthCreditService } from './auth-credit.service';
import {
  BatchJob,
  BatchJobCreateInput,
  BatchJobItem,
  BatchJobRunResult,
} from '../models/batch-job.model';

const JOBS_STORAGE_KEY = 'videoWorkflow.batchJobs.v1';
const ITEMS_STORAGE_KEY = 'videoWorkflow.batchJobItems.v1';

type BatchItemRunner = (item: BatchJobItem, job: BatchJob) => Promise<BatchJobRunResult | void>;

@Injectable({ providedIn: 'root' })
export class BatchJobService {
  private readonly jobsSubject = new BehaviorSubject<BatchJob[]>(this.readJobs());
  private readonly itemsSubject = new BehaviorSubject<BatchJobItem[]>(this.readItems());
  readonly jobs$ = this.jobsSubject.asObservable();
  readonly items$ = this.itemsSubject.asObservable();

  private runningJobId = '';
  private stopRequested = false;

  constructor(private auth: AuthCreditService) {}

  get jobs(): BatchJob[] {
    return this.jobsSubject.value;
  }

  get items(): BatchJobItem[] {
    return this.itemsSubject.value;
  }

  listJobs(): BatchJob[] {
    const userId = this.currentUserId();
    if (!userId) return [];
    return this.jobs.filter((job) => job.userId === userId);
  }

  listItems(batchJobId: string): BatchJobItem[] {
    return this.items.filter((item) => item.batchJobId === batchJobId);
  }

  create(input: BatchJobCreateInput): BatchJob {
    const userId = this.requireUserId();
    const now = new Date().toISOString();
    const jobId = this.createId('batch');
    const items = input.items
      .map((item, index) => ({
        id: this.createId('batch_item'),
        batchJobId: jobId,
        topicId: item.topicId,
        title: item.title?.trim() || `子任务 ${index + 1}`,
        input: this.cloneSerializable(item.input || {}),
        status: 'queued' as const,
        retryCount: 0,
        createdAt: now,
        updatedAt: now,
      }));
    const job: BatchJob = {
      id: jobId,
      userId,
      title: input.title?.trim() || '未命名批量任务',
      pipelineId: input.pipelineId,
      templateId: input.templateId,
      status: 'queued',
      total: items.length,
      completed: 0,
      failed: 0,
      estimatedCredits: Math.max(0, Math.ceil(Number(input.estimatedCredits || 0))),
      createdAt: now,
      updatedAt: now,
    };
    this.setJobs([job, ...this.jobs]);
    this.setItems([...items, ...this.items]);
    return job;
  }

  pause(jobId: string): void {
    const job = this.findJob(jobId);
    if (!job || job.status !== 'running') return;
    this.stopRequested = true;
    this.patchJob(jobId, { status: 'paused' });
  }

  cancel(jobId: string): void {
    this.stopRequested = this.runningJobId === jobId;
    this.patchJob(jobId, { status: 'cancelled' });
    this.setItems(this.items.map((item) => item.batchJobId === jobId && (item.status === 'queued' || item.status === 'running')
      ? { ...item, status: 'cancelled', updatedAt: new Date().toISOString() }
      : item));
  }

  retryItem(itemId: string): void {
    const item = this.items.find((row) => row.id === itemId);
    if (!item || item.status !== 'failed') return;
    this.patchItem(itemId, {
      status: 'queued',
      errorMessage: undefined,
      resultUrl: undefined,
      retryCount: Number(item.retryCount || 0) + 1,
    });
    this.recalculateJob(item.batchJobId, 'queued');
  }

  async run(jobId: string, runner: BatchItemRunner = this.mockRunner): Promise<void> {
    const job = this.findJob(jobId);
    if (!job || job.status === 'cancelled') return;
    if (this.runningJobId && this.runningJobId !== jobId) {
      throw new Error('已有批量任务正在执行，请稍后再试');
    }

    this.runningJobId = jobId;
    this.stopRequested = false;
    this.patchJob(jobId, { status: 'running' });

    try {
      while (!this.stopRequested) {
        const currentJob = this.findJob(jobId);
        if (!currentJob || currentJob.status === 'cancelled') break;

        const nextItem = this.listItems(jobId).find((item) => item.status === 'queued');
        if (!nextItem) break;

        this.patchItem(nextItem.id, { status: 'running', errorMessage: undefined });
        try {
          const result = await runner(nextItem, currentJob);
          this.patchItem(nextItem.id, {
            status: 'completed',
            resultUrl: result?.resultUrl,
            generationTaskId: result?.generationTaskId,
          });
        } catch (err: any) {
          this.patchItem(nextItem.id, {
            status: 'failed',
            errorMessage: err?.message || '子任务执行失败',
          });
        }
        this.recalculateJob(jobId);
      }
    } finally {
      const finalJob = this.findJob(jobId);
      if (finalJob && finalJob.status === 'running') {
        this.recalculateJob(jobId);
      }
      this.runningJobId = '';
    }
  }

  private mockRunner(item: BatchJobItem): Promise<BatchJobRunResult> {
    return new Promise((resolve) => {
      window.setTimeout(() => resolve({ generationTaskId: `mock-${item.id}` }), 300);
    });
  }

  private recalculateJob(jobId: string, fallbackStatus?: BatchJob['status']): void {
    const job = this.findJob(jobId);
    if (!job) return;
    const items = this.listItems(jobId);
    const completed = items.filter((item) => item.status === 'completed').length;
    const failed = items.filter((item) => item.status === 'failed').length;
    const cancelled = items.filter((item) => item.status === 'cancelled').length;
    const queued = items.filter((item) => item.status === 'queued').length;
    const running = items.filter((item) => item.status === 'running').length;
    const status = (() => {
      if (job.status === 'cancelled') return 'cancelled';
      if (fallbackStatus) return fallbackStatus;
      if (running > 0) return 'running';
      if (queued > 0) return job.status === 'paused' ? 'paused' : 'running';
      if (completed === items.length && items.length > 0) return 'completed';
      if (failed > 0 && completed + failed + cancelled === items.length) return 'failed';
      return job.status;
    })();
    this.patchJob(jobId, { completed, failed, status });
  }

  private patchJob(id: string, patch: Partial<BatchJob>): void {
    const now = new Date().toISOString();
    this.setJobs(this.jobs.map((job) => job.id === id ? { ...job, ...patch, id, updatedAt: now } : job));
  }

  private patchItem(id: string, patch: Partial<BatchJobItem>): void {
    const now = new Date().toISOString();
    this.setItems(this.items.map((item) => item.id === id ? { ...item, ...patch, id, updatedAt: now } : item));
  }

  private findJob(id: string): BatchJob | undefined {
    const userId = this.currentUserId();
    return this.jobs.find((job) => job.id === id && job.userId === userId);
  }

  private currentUserId(): string {
    return this.auth.currentUser?.objectId || 'local-guest';
  }

  private requireUserId(): string {
    return this.currentUserId();
  }

  private readJobs(): BatchJob[] {
    try {
      const parsed = JSON.parse(localStorage.getItem(JOBS_STORAGE_KEY) || '[]');
      return Array.isArray(parsed) ? parsed : [];
    } catch {
      return [];
    }
  }

  private readItems(): BatchJobItem[] {
    try {
      const parsed = JSON.parse(localStorage.getItem(ITEMS_STORAGE_KEY) || '[]');
      return Array.isArray(parsed) ? parsed : [];
    } catch {
      return [];
    }
  }

  private setJobs(jobs: BatchJob[]): void {
    const sorted = [...jobs].sort((a, b) => Date.parse(b.updatedAt) - Date.parse(a.updatedAt)).slice(0, 100);
    this.jobsSubject.next(sorted);
    localStorage.setItem(JOBS_STORAGE_KEY, JSON.stringify(sorted));
  }

  private setItems(items: BatchJobItem[]): void {
    this.itemsSubject.next(items);
    localStorage.setItem(ITEMS_STORAGE_KEY, JSON.stringify(items));
  }

  private createId(prefix: string): string {
    return `${prefix}_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
  }

  private cloneSerializable<T>(value: T): T {
    return JSON.parse(JSON.stringify(value || {}));
  }
}