|
@@ -5,16 +5,25 @@ import {
|
|
|
BatchJob,
|
|
BatchJob,
|
|
|
BatchJobCreateInput,
|
|
BatchJobCreateInput,
|
|
|
BatchJobItem,
|
|
BatchJobItem,
|
|
|
|
|
+ BatchJobItemLog,
|
|
|
BatchJobRunResult,
|
|
BatchJobRunResult,
|
|
|
} from '../models/batch-job.model';
|
|
} from '../models/batch-job.model';
|
|
|
|
|
+import { IndexedDbJsonStorage } from './indexed-db-json-storage';
|
|
|
|
|
+import { userFriendlyError } from './user-message.util';
|
|
|
|
|
|
|
|
const JOBS_STORAGE_KEY = 'videoWorkflow.batchJobs.v1';
|
|
const JOBS_STORAGE_KEY = 'videoWorkflow.batchJobs.v1';
|
|
|
const ITEMS_STORAGE_KEY = 'videoWorkflow.batchJobItems.v1';
|
|
const ITEMS_STORAGE_KEY = 'videoWorkflow.batchJobItems.v1';
|
|
|
|
|
|
|
|
-type BatchItemRunner = (item: BatchJobItem, job: BatchJob) => Promise<BatchJobRunResult | void>;
|
|
|
|
|
|
|
+interface BatchItemRunContext {
|
|
|
|
|
+ markRunning: (stepId?: string) => void;
|
|
|
|
|
+ markWaitingExternal: (externalTaskIds: Record<string, string>, stepId?: string) => void;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+export type BatchItemRunner = (item: BatchJobItem, job: BatchJob, context: BatchItemRunContext) => Promise<BatchJobRunResult | void>;
|
|
|
|
|
|
|
|
@Injectable({ providedIn: 'root' })
|
|
@Injectable({ providedIn: 'root' })
|
|
|
export class BatchJobService {
|
|
export class BatchJobService {
|
|
|
|
|
+ private readonly idb = new IndexedDbJsonStorage('batchJobs');
|
|
|
private readonly jobsSubject = new BehaviorSubject<BatchJob[]>(this.readJobs());
|
|
private readonly jobsSubject = new BehaviorSubject<BatchJob[]>(this.readJobs());
|
|
|
private readonly itemsSubject = new BehaviorSubject<BatchJobItem[]>(this.readItems());
|
|
private readonly itemsSubject = new BehaviorSubject<BatchJobItem[]>(this.readItems());
|
|
|
readonly jobs$ = this.jobsSubject.asObservable();
|
|
readonly jobs$ = this.jobsSubject.asObservable();
|
|
@@ -23,7 +32,9 @@ export class BatchJobService {
|
|
|
private runningJobId = '';
|
|
private runningJobId = '';
|
|
|
private stopRequested = false;
|
|
private stopRequested = false;
|
|
|
|
|
|
|
|
- constructor(private auth: AuthCreditService) {}
|
|
|
|
|
|
|
+ constructor(private auth: AuthCreditService) {
|
|
|
|
|
+ void this.restoreFromIndexedDb();
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
get jobs(): BatchJob[] {
|
|
get jobs(): BatchJob[] {
|
|
|
return this.jobsSubject.value;
|
|
return this.jobsSubject.value;
|
|
@@ -70,6 +81,10 @@ export class BatchJobService {
|
|
|
completed: 0,
|
|
completed: 0,
|
|
|
failed: 0,
|
|
failed: 0,
|
|
|
estimatedCredits: Math.max(0, Math.ceil(Number(input.estimatedCredits || 0))),
|
|
estimatedCredits: Math.max(0, Math.ceil(Number(input.estimatedCredits || 0))),
|
|
|
|
|
+ concurrency: 1,
|
|
|
|
|
+ intervalMs: 2000,
|
|
|
|
|
+ runnerVersion: 1,
|
|
|
|
|
+ failureSummary: {},
|
|
|
createdAt: now,
|
|
createdAt: now,
|
|
|
updatedAt: now,
|
|
updatedAt: now,
|
|
|
};
|
|
};
|
|
@@ -99,8 +114,10 @@ export class BatchJobService {
|
|
|
this.patchItem(itemId, {
|
|
this.patchItem(itemId, {
|
|
|
status: 'queued',
|
|
status: 'queued',
|
|
|
errorMessage: undefined,
|
|
errorMessage: undefined,
|
|
|
|
|
+ lastErrorCode: undefined,
|
|
|
resultUrl: undefined,
|
|
resultUrl: undefined,
|
|
|
retryCount: Number(item.retryCount || 0) + 1,
|
|
retryCount: Number(item.retryCount || 0) + 1,
|
|
|
|
|
+ logs: this.appendItemLog(item, 'info', 'retry', '失败子任务已重新排队'),
|
|
|
});
|
|
});
|
|
|
this.recalculateJob(item.batchJobId, 'queued');
|
|
this.recalculateJob(item.batchJobId, 'queued');
|
|
|
}
|
|
}
|
|
@@ -114,7 +131,11 @@ export class BatchJobService {
|
|
|
|
|
|
|
|
this.runningJobId = jobId;
|
|
this.runningJobId = jobId;
|
|
|
this.stopRequested = false;
|
|
this.stopRequested = false;
|
|
|
- this.patchJob(jobId, { status: 'running' });
|
|
|
|
|
|
|
+ this.patchJob(jobId, {
|
|
|
|
|
+ status: 'running',
|
|
|
|
|
+ startedAt: job.startedAt || new Date().toISOString(),
|
|
|
|
|
+ lastRunAt: new Date().toISOString(),
|
|
|
|
|
+ });
|
|
|
|
|
|
|
|
try {
|
|
try {
|
|
|
while (!this.stopRequested) {
|
|
while (!this.stopRequested) {
|
|
@@ -124,18 +145,34 @@ export class BatchJobService {
|
|
|
const nextItem = this.listItems(jobId).find((item) => item.status === 'queued');
|
|
const nextItem = this.listItems(jobId).find((item) => item.status === 'queued');
|
|
|
if (!nextItem) break;
|
|
if (!nextItem) break;
|
|
|
|
|
|
|
|
- this.patchItem(nextItem.id, { status: 'running', errorMessage: undefined });
|
|
|
|
|
|
|
+ this.patchItem(nextItem.id, {
|
|
|
|
|
+ status: 'running',
|
|
|
|
|
+ errorMessage: undefined,
|
|
|
|
|
+ lastErrorCode: undefined,
|
|
|
|
|
+ startedAt: nextItem.startedAt || new Date().toISOString(),
|
|
|
|
|
+ logs: this.appendItemLog(nextItem, 'info', 'queue', '开始执行子任务'),
|
|
|
|
|
+ });
|
|
|
try {
|
|
try {
|
|
|
- const result = await runner(nextItem, currentJob);
|
|
|
|
|
|
|
+ const result = await runner(nextItem, currentJob, this.createRunContext(nextItem.id));
|
|
|
|
|
+ const latestItem = this.items.find((item) => item.id === nextItem.id) || nextItem;
|
|
|
this.patchItem(nextItem.id, {
|
|
this.patchItem(nextItem.id, {
|
|
|
status: 'completed',
|
|
status: 'completed',
|
|
|
resultUrl: result?.resultUrl,
|
|
resultUrl: result?.resultUrl,
|
|
|
generationTaskId: result?.generationTaskId,
|
|
generationTaskId: result?.generationTaskId,
|
|
|
|
|
+ externalTaskIds: result?.externalTaskIds,
|
|
|
|
|
+ finishedAt: new Date().toISOString(),
|
|
|
|
|
+ logs: this.appendItemLog(latestItem, 'info', 'complete', '子任务执行完成'),
|
|
|
});
|
|
});
|
|
|
} catch (err: any) {
|
|
} catch (err: any) {
|
|
|
|
|
+ const latestItem = this.items.find((item) => item.id === nextItem.id) || nextItem;
|
|
|
|
|
+ const message = userFriendlyError(err, '子任务执行失败,请稍后重试');
|
|
|
|
|
+ const errorCode = this.classifyErrorCode(err);
|
|
|
this.patchItem(nextItem.id, {
|
|
this.patchItem(nextItem.id, {
|
|
|
status: 'failed',
|
|
status: 'failed',
|
|
|
- errorMessage: err?.message || '子任务执行失败',
|
|
|
|
|
|
|
+ errorMessage: message,
|
|
|
|
|
+ lastErrorCode: errorCode,
|
|
|
|
|
+ finishedAt: new Date().toISOString(),
|
|
|
|
|
+ logs: this.appendItemLog(latestItem, 'error', errorCode, message),
|
|
|
});
|
|
});
|
|
|
}
|
|
}
|
|
|
this.recalculateJob(jobId);
|
|
this.recalculateJob(jobId);
|
|
@@ -163,7 +200,7 @@ export class BatchJobService {
|
|
|
const failed = items.filter((item) => item.status === 'failed').length;
|
|
const failed = items.filter((item) => item.status === 'failed').length;
|
|
|
const cancelled = items.filter((item) => item.status === 'cancelled').length;
|
|
const cancelled = items.filter((item) => item.status === 'cancelled').length;
|
|
|
const queued = items.filter((item) => item.status === 'queued').length;
|
|
const queued = items.filter((item) => item.status === 'queued').length;
|
|
|
- const running = items.filter((item) => item.status === 'running').length;
|
|
|
|
|
|
|
+ const running = items.filter((item) => item.status === 'running' || item.status === 'waiting_external').length;
|
|
|
const status = (() => {
|
|
const status = (() => {
|
|
|
if (job.status === 'cancelled') return 'cancelled';
|
|
if (job.status === 'cancelled') return 'cancelled';
|
|
|
if (fallbackStatus) return fallbackStatus;
|
|
if (fallbackStatus) return fallbackStatus;
|
|
@@ -173,7 +210,16 @@ export class BatchJobService {
|
|
|
if (failed > 0 && completed + failed + cancelled === items.length) return 'failed';
|
|
if (failed > 0 && completed + failed + cancelled === items.length) return 'failed';
|
|
|
return job.status;
|
|
return job.status;
|
|
|
})();
|
|
})();
|
|
|
- this.patchJob(jobId, { completed, failed, status });
|
|
|
|
|
|
|
+ const patch: Partial<BatchJob> = {
|
|
|
|
|
+ completed,
|
|
|
|
|
+ failed,
|
|
|
|
|
+ status,
|
|
|
|
|
+ failureSummary: this.buildFailureSummary(items),
|
|
|
|
|
+ };
|
|
|
|
|
+ if (status === 'completed' || status === 'failed' || status === 'cancelled') {
|
|
|
|
|
+ patch.finishedAt = job.finishedAt || new Date().toISOString();
|
|
|
|
|
+ }
|
|
|
|
|
+ this.patchJob(jobId, patch);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
private patchJob(id: string, patch: Partial<BatchJob>): void {
|
|
private patchJob(id: string, patch: Partial<BatchJob>): void {
|
|
@@ -186,6 +232,30 @@ export class BatchJobService {
|
|
|
this.setItems(this.items.map((item) => item.id === id ? { ...item, ...patch, id, updatedAt: now } : item));
|
|
this.setItems(this.items.map((item) => item.id === id ? { ...item, ...patch, id, updatedAt: now } : item));
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ private createRunContext(itemId: string): BatchItemRunContext {
|
|
|
|
|
+ return {
|
|
|
|
|
+ markRunning: (stepId?: string) => {
|
|
|
|
|
+ const item = this.items.find((row) => row.id === itemId);
|
|
|
|
|
+ this.patchItem(itemId, {
|
|
|
|
|
+ status: 'running',
|
|
|
|
|
+ lastStepId: stepId,
|
|
|
|
|
+ startedAt: item?.startedAt || new Date().toISOString(),
|
|
|
|
|
+ logs: item ? this.appendItemLog(item, 'info', stepId || 'running', '子任务进入执行步骤') : undefined,
|
|
|
|
|
+ });
|
|
|
|
|
+ },
|
|
|
|
|
+ markWaitingExternal: (externalTaskIds: Record<string, string>, stepId?: string) => {
|
|
|
|
|
+ const item = this.items.find((row) => row.id === itemId);
|
|
|
|
|
+ this.patchItem(itemId, {
|
|
|
|
|
+ status: 'waiting_external',
|
|
|
|
|
+ lastStepId: stepId,
|
|
|
|
|
+ externalTaskIds: { ...(item?.externalTaskIds || {}), ...externalTaskIds },
|
|
|
|
|
+ startedAt: item?.startedAt || new Date().toISOString(),
|
|
|
|
|
+ logs: item ? this.appendItemLog(item, 'info', stepId || 'external', '已记录外部任务 ID,等待上游结果') : undefined,
|
|
|
|
|
+ });
|
|
|
|
|
+ },
|
|
|
|
|
+ };
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
private findJob(id: string): BatchJob | undefined {
|
|
private findJob(id: string): BatchJob | undefined {
|
|
|
const userId = this.currentUserId();
|
|
const userId = this.currentUserId();
|
|
|
return this.jobs.find((job) => job.id === id && job.userId === userId);
|
|
return this.jobs.find((job) => job.id === id && job.userId === userId);
|
|
@@ -211,7 +281,7 @@ export class BatchJobService {
|
|
|
private readItems(): BatchJobItem[] {
|
|
private readItems(): BatchJobItem[] {
|
|
|
try {
|
|
try {
|
|
|
const parsed = JSON.parse(localStorage.getItem(ITEMS_STORAGE_KEY) || '[]');
|
|
const parsed = JSON.parse(localStorage.getItem(ITEMS_STORAGE_KEY) || '[]');
|
|
|
- return Array.isArray(parsed) ? parsed : [];
|
|
|
|
|
|
|
+ return this.normalizeRestoredItems(parsed);
|
|
|
} catch {
|
|
} catch {
|
|
|
return [];
|
|
return [];
|
|
|
}
|
|
}
|
|
@@ -220,12 +290,18 @@ export class BatchJobService {
|
|
|
private setJobs(jobs: BatchJob[]): void {
|
|
private setJobs(jobs: BatchJob[]): void {
|
|
|
const sorted = [...jobs].sort((a, b) => Date.parse(b.updatedAt) - Date.parse(a.updatedAt)).slice(0, 100);
|
|
const sorted = [...jobs].sort((a, b) => Date.parse(b.updatedAt) - Date.parse(a.updatedAt)).slice(0, 100);
|
|
|
this.jobsSubject.next(sorted);
|
|
this.jobsSubject.next(sorted);
|
|
|
- localStorage.setItem(JOBS_STORAGE_KEY, JSON.stringify(sorted));
|
|
|
|
|
|
|
+ try {
|
|
|
|
|
+ localStorage.setItem(JOBS_STORAGE_KEY, JSON.stringify(sorted));
|
|
|
|
|
+ } catch {}
|
|
|
|
|
+ void this.idb.writeArray(JOBS_STORAGE_KEY, sorted);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
private setItems(items: BatchJobItem[]): void {
|
|
private setItems(items: BatchJobItem[]): void {
|
|
|
this.itemsSubject.next(items);
|
|
this.itemsSubject.next(items);
|
|
|
- localStorage.setItem(ITEMS_STORAGE_KEY, JSON.stringify(items));
|
|
|
|
|
|
|
+ try {
|
|
|
|
|
+ localStorage.setItem(ITEMS_STORAGE_KEY, JSON.stringify(items));
|
|
|
|
|
+ } catch {}
|
|
|
|
|
+ void this.idb.writeArray(ITEMS_STORAGE_KEY, items);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
private createId(prefix: string): string {
|
|
private createId(prefix: string): string {
|
|
@@ -235,4 +311,56 @@ export class BatchJobService {
|
|
|
private cloneSerializable<T>(value: T): T {
|
|
private cloneSerializable<T>(value: T): T {
|
|
|
return JSON.parse(JSON.stringify(value || {}));
|
|
return JSON.parse(JSON.stringify(value || {}));
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ private appendItemLog(item: BatchJobItem, level: BatchJobItemLog['level'], step: string, message: string): BatchJobItemLog[] {
|
|
|
|
|
+ const logs = Array.isArray(item.logs) ? item.logs : [];
|
|
|
|
|
+ return [
|
|
|
|
|
+ ...logs,
|
|
|
|
|
+ {
|
|
|
|
|
+ at: new Date().toISOString(),
|
|
|
|
|
+ level,
|
|
|
|
|
+ step,
|
|
|
|
|
+ message,
|
|
|
|
|
+ },
|
|
|
|
|
+ ].slice(-30);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private buildFailureSummary(items: BatchJobItem[]): Record<string, number> {
|
|
|
|
|
+ return items.reduce((acc, item) => {
|
|
|
|
|
+ if (item.status !== 'failed') return acc;
|
|
|
|
|
+ const code = item.lastErrorCode || 'unknown';
|
|
|
|
|
+ acc[code] = (acc[code] || 0) + 1;
|
|
|
|
|
+ return acc;
|
|
|
|
|
+ }, {} as Record<string, number>);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private classifyErrorCode(error: any): string {
|
|
|
|
|
+ const text = `${error?.code || ''} ${error?.status || ''} ${error?.message || error || ''}`.toLowerCase();
|
|
|
|
|
+ if (text.includes('429') || text.includes('rate') || text.includes('limit') || text.includes('限流')) return 'rate_limit';
|
|
|
|
|
+ if (text.includes('timeout') || text.includes('超时')) return 'timeout';
|
|
|
|
|
+ if (text.includes('network') || text.includes('fetch') || text.includes('网络')) return 'network';
|
|
|
|
|
+ if (text.includes('credential') || text.includes('unauthorized') || text.includes('401') || text.includes('403')) return 'credential';
|
|
|
|
|
+ if (text.includes('invalid') || text.includes('参数')) return 'invalid_input';
|
|
|
|
|
+ return 'unknown';
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private normalizeRestoredItems(value: any): BatchJobItem[] {
|
|
|
|
|
+ if (!Array.isArray(value)) return [];
|
|
|
|
|
+ return value.map((item) => item?.status === 'running'
|
|
|
|
|
+ ? { ...item, status: 'queued', errorMessage: item.errorMessage || '页面刷新后已自动回到队列,可继续执行。' }
|
|
|
|
|
+ : item);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private async restoreFromIndexedDb(): Promise<void> {
|
|
|
|
|
+ const [jobs, items] = await Promise.all([
|
|
|
|
|
+ this.idb.readArray<BatchJob>(JOBS_STORAGE_KEY),
|
|
|
|
|
+ this.idb.readArray<BatchJobItem>(ITEMS_STORAGE_KEY),
|
|
|
|
|
+ ]);
|
|
|
|
|
+ if (jobs.length) {
|
|
|
|
|
+ this.jobsSubject.next([...jobs].sort((a, b) => Date.parse(b.updatedAt) - Date.parse(a.updatedAt)).slice(0, 100));
|
|
|
|
|
+ }
|
|
|
|
|
+ if (items.length) {
|
|
|
|
|
+ this.itemsSubject.next(this.normalizeRestoredItems(items));
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|