/** * πŸ”₯ μ„±λŠ₯ μ΅œμ ν™”: λ°±κ·ΈλΌμš΄λ“œ μž‘μ—… 큐 μ‹œμŠ€ν…œ * * μ œμ–΄κ΄€λ¦¬ μž‘μ—…μ„ λ°±κ·ΈλΌμš΄λ“œμ—μ„œ μ²˜λ¦¬ν•˜μ—¬ * μ‚¬μš©μžμ—κ²Œ μ¦‰μ‹œ 응닡을 μ œκ³΅ν•©λ‹ˆλ‹€. */ import { ButtonActionType, ButtonTypeConfig, DataflowExecutionResult } from "@/types/screen"; import { apiClient } from "@/lib/api/client"; export type JobPriority = "high" | "normal" | "low"; export type JobStatus = "pending" | "processing" | "completed" | "failed"; export interface DataflowJob { id: string; buttonId: string; actionType: ButtonActionType; config: ButtonTypeConfig; contextData: Record; companyCode: string; priority: JobPriority; status: JobStatus; createdAt: number; startedAt?: number; completedAt?: number; result?: DataflowExecutionResult; error?: string; retryCount: number; maxRetries: number; } export interface QueueMetrics { totalJobs: number; pendingJobs: number; processingJobs: number; completedJobs: number; failedJobs: number; averageProcessingTime: number; // 평균 처리 μ‹œκ°„ (ms) throughput: number; // μ²˜λ¦¬λŸ‰ (jobs/min) } /** * πŸ”₯ λ°±κ·ΈλΌμš΄λ“œ μž‘μ—… 큐 * * - μš°μ„ μˆœμœ„ 기반 처리 * - 배치 처리 (μ΅œλŒ€ 3개 λ™μ‹œ) * - μžλ™ μž¬μ‹œλ„ * - μ‹€μ‹œκ°„ μƒνƒœ 좔적 */ export class DataflowJobQueue { private queue: DataflowJob[] = []; private processing = false; private readonly maxConcurrentJobs = 3; private activeJobs = new Map(); private completedJobs: DataflowJob[] = []; private maxCompletedJobs = 100; // μ΅œλŒ€ μ™„λ£Œλœ μž‘μ—… 보관 개수 private metrics: QueueMetrics = { totalJobs: 0, pendingJobs: 0, processingJobs: 0, completedJobs: 0, failedJobs: 0, averageProcessingTime: 0, throughput: 0, }; // μƒνƒœ λ³€κ²½ 이벀트 λ¦¬μŠ€λ„ˆ private statusChangeListeners = new Map void>(); /** * πŸ”₯ μž‘μ—… 큐에 μΆ”κ°€ (μ¦‰μ‹œ λ°˜ν™˜) */ enqueue( buttonId: string, actionType: ButtonActionType, config: ButtonTypeConfig, contextData: Record, companyCode: string, priority: JobPriority = "normal", maxRetries: number = 3, ): string { const jobId = this.generateJobId(); const now = Date.now(); const job: DataflowJob = { id: jobId, buttonId, actionType, config, contextData, companyCode, priority, status: "pending", createdAt: now, retryCount: 0, maxRetries, }; // 큐에 μΆ”κ°€ this.queue.push(job); this.metrics.totalJobs++; this.metrics.pendingJobs++; // μš°μ„ μˆœμœ„ μ •λ ¬ this.sortQueueByPriority(); // 비동기 처리 μ‹œμž‘ setTimeout(() => this.processQueue(), 0); console.log(`πŸ“‹ Job enqueued: ${jobId} (priority: ${priority})`); return jobId; } /** * πŸ”₯ μž‘μ—… μƒνƒœ 쑰회 */ getJobStatus(jobId: string): { status: JobStatus; result?: any; progress?: number } { // ν™œμ„± μž‘μ—…μ—μ„œ μ°ΎκΈ° const activeJob = this.activeJobs.get(jobId); if (activeJob) { return { status: activeJob.status, result: activeJob.result, progress: this.calculateProgress(activeJob), }; } // μ™„λ£Œλœ μž‘μ—…μ—μ„œ μ°ΎκΈ° const completedJob = this.completedJobs.find((job) => job.id === jobId); if (completedJob) { return { status: completedJob.status, result: completedJob.result, progress: 100, }; } // λŒ€κΈ° 쀑인 μž‘μ—…μ—μ„œ μ°ΎκΈ° const pendingJob = this.queue.find((job) => job.id === jobId); if (pendingJob) { const queuePosition = this.queue.indexOf(pendingJob) + 1; return { status: "pending", progress: 0, }; } throw new Error(`Job not found: ${jobId}`); } /** * πŸ”₯ μž‘μ—… μƒνƒœ λ³€κ²½ λ¦¬μŠ€λ„ˆ 등둝 */ onStatusChange(jobId: string, callback: (job: DataflowJob) => void): () => void { this.statusChangeListeners.set(jobId, callback); // ν•΄μ œ ν•¨μˆ˜ λ°˜ν™˜ return () => { this.statusChangeListeners.delete(jobId); }; } /** * πŸ”₯ 큐 처리 (배치 처리) */ private async processQueue(): Promise { if (this.processing || this.queue.length === 0) return; if (this.activeJobs.size >= this.maxConcurrentJobs) return; this.processing = true; try { // μ²˜λ¦¬ν•  수 μžˆλŠ” 만큼 μž‘μ—… 선택 const availableSlots = this.maxConcurrentJobs - this.activeJobs.size; const jobsToProcess = this.queue.splice(0, availableSlots); if (jobsToProcess.length > 0) { console.log(`πŸ”„ Processing ${jobsToProcess.length} jobs (${this.activeJobs.size} active)`); // 병렬 처리 const promises = jobsToProcess.map((job) => this.executeJob(job)); await Promise.allSettled(promises); } } finally { this.processing = false; // 큐에 더 λ§Žμ€ μž‘μ—…μ΄ 있으면 계속 처리 if (this.queue.length > 0 && this.activeJobs.size < this.maxConcurrentJobs) { setTimeout(() => this.processQueue(), 10); } } } /** * πŸ”₯ κ°œλ³„ μž‘μ—… μ‹€ν–‰ */ private async executeJob(job: DataflowJob): Promise { const startTime = performance.now(); // ν™œμ„± μž‘μ—…μœΌλ‘œ 이동 this.activeJobs.set(job.id, job); this.updateJobStatus(job, "processing"); this.metrics.pendingJobs--; this.metrics.processingJobs++; job.startedAt = Date.now(); try { console.log(`⚑ Starting job: ${job.id}`); // μ‹€μ œ μ œμ–΄κ΄€λ¦¬ μ‹€ν–‰ const result = await this.executeDataflowLogic(job); // 성곡 처리 job.result = result; job.completedAt = Date.now(); this.updateJobStatus(job, "completed"); const executionTime = performance.now() - startTime; this.updateProcessingTimeMetrics(executionTime); console.log(`βœ… Job completed: ${job.id} (${executionTime.toFixed(2)}ms)`); } catch (error) { console.error(`❌ Job failed: ${job.id}`, error); job.error = error.message || "Unknown error"; job.retryCount++; // μž¬μ‹œλ„ 둜직 if (job.retryCount < job.maxRetries) { console.log(`πŸ”„ Retrying job: ${job.id} (${job.retryCount}/${job.maxRetries})`); // μ§€μˆ˜ λ°±μ˜€ν”„λ‘œ μž¬μ‹œλ„ μ§€μ—° const retryDelay = Math.pow(2, job.retryCount) * 1000; // 2^n 초 setTimeout(() => { job.status = "pending"; this.queue.unshift(job); // μš°μ„ μˆœμœ„λ‘œ λ‹€μ‹œ 큐에 μΆ”κ°€ this.processQueue(); }, retryDelay); return; } // μ΅œλŒ€ μž¬μ‹œλ„ 횟수 초과 μ‹œ μ‹€νŒ¨ 처리 job.completedAt = Date.now(); this.updateJobStatus(job, "failed"); this.metrics.failedJobs++; } finally { // ν™œμ„± μž‘μ—…μ—μ„œ 제거 this.activeJobs.delete(job.id); this.metrics.processingJobs--; // μ™„λ£Œλœ μž‘μ—… λͺ©λ‘μ— μΆ”κ°€ this.addToCompletedJobs(job); } } /** * πŸ”₯ μ‹€μ œ λ°μ΄ν„°ν”Œλ‘œμš° 둜직 μ‹€ν–‰ */ private async executeDataflowLogic(job: DataflowJob): Promise { const { config, contextData, companyCode } = job; try { const response = await apiClient.post("/api/button-dataflow/execute-background", { buttonId: job.buttonId, actionType: job.actionType, buttonConfig: config, contextData, companyCode, timing: config.dataflowTiming || "after", }); if (response.data.success) { return response.data.data as DataflowExecutionResult; } else { throw new Error(response.data.message || "Dataflow execution failed"); } } catch (error) { if (error.response?.data?.message) { throw new Error(error.response.data.message); } throw error; } } /** * πŸ”₯ μž‘μ—… μƒνƒœ μ—…λ°μ΄νŠΈ */ private updateJobStatus(job: DataflowJob, status: JobStatus): void { job.status = status; // λ¦¬μŠ€λ„ˆμ—κ²Œ μ•Œλ¦Ό const listener = this.statusChangeListeners.get(job.id); if (listener) { listener(job); } } /** * μš°μ„ μˆœμœ„λ³„ 큐 μ •λ ¬ */ private sortQueueByPriority(): void { const priorityWeights = { high: 3, normal: 2, low: 1 }; this.queue.sort((a, b) => { // μš°μ„ μˆœμœ„ μš°μ„  const priorityDiff = priorityWeights[b.priority] - priorityWeights[a.priority]; if (priorityDiff !== 0) return priorityDiff; // 같은 μš°μ„ μˆœμœ„λ©΄ 생성 μ‹œκ°„ 순 return a.createdAt - b.createdAt; }); } /** * μž‘μ—… ID 생성 */ private generateJobId(): string { return `job_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; } /** * μ§„ν–‰λ₯  계산 (μΆ”μ •) */ private calculateProgress(job: DataflowJob): number { if (job.status === "completed") return 100; if (job.status === "failed") return 0; if (job.status === "pending") return 0; if (job.status === "processing") { // 처리 쀑인 경우 κ²½κ³Ό μ‹œκ°„ 기반으둜 μΆ”μ • const elapsed = Date.now() - (job.startedAt || job.createdAt); const estimatedDuration = 5000; // 5초둜 μΆ”μ • return Math.min(90, (elapsed / estimatedDuration) * 100); } return 0; } /** * μ™„λ£Œλœ μž‘μ—… λͺ©λ‘μ— μΆ”κ°€ */ private addToCompletedJobs(job: DataflowJob): void { this.completedJobs.push(job); if (job.status === "completed") { this.metrics.completedJobs++; } // 였래된 μ™„λ£Œ μž‘μ—… 제거 if (this.completedJobs.length > this.maxCompletedJobs) { this.completedJobs.shift(); } } /** * 처리 μ‹œκ°„ λ©”νŠΈλ¦­ μ—…λ°μ΄νŠΈ */ private updateProcessingTimeMetrics(processingTime: number): void { if (this.metrics.averageProcessingTime === 0) { this.metrics.averageProcessingTime = processingTime; } else { // 이동 평균 this.metrics.averageProcessingTime = this.metrics.averageProcessingTime * 0.9 + processingTime * 0.1; } // μ²˜λ¦¬λŸ‰ 계산 (κ°„λ‹¨ν•œ μΆ”μ •) this.metrics.throughput = 60000 / this.metrics.averageProcessingTime; // jobs/min } /** * πŸ”₯ 큐 톡계 쑰회 */ getMetrics(): QueueMetrics { this.metrics.pendingJobs = this.queue.length; this.metrics.processingJobs = this.activeJobs.size; return { ...this.metrics }; } /** * πŸ”₯ 상세 큐 정보 쑰회 (λ””λ²„κΉ…μš©) */ getQueueInfo(): { pending: DataflowJob[]; active: DataflowJob[]; recentCompleted: DataflowJob[]; } { return { pending: [...this.queue], active: Array.from(this.activeJobs.values()), recentCompleted: this.completedJobs.slice(-10), // 졜근 10개 }; } /** * πŸ”₯ νŠΉμ • μž‘μ—… μ·¨μ†Œ */ cancelJob(jobId: string): boolean { // λŒ€κΈ° 쀑인 μž‘μ—…μ—μ„œ 제거 const queueIndex = this.queue.findIndex((job) => job.id === jobId); if (queueIndex !== -1) { this.queue.splice(queueIndex, 1); this.metrics.pendingJobs--; console.log(`❌ Job cancelled: ${jobId}`); return true; } // ν™œμ„± μž‘μ—…μ€ μ·¨μ†Œν•  수 μ—†μŒ (이미 μ‹€ν–‰ 쀑) return false; } /** * πŸ”₯ λͺ¨λ“  λŒ€κΈ° μž‘μ—… μ·¨μ†Œ */ clearQueue(): number { const cancelledCount = this.queue.length; this.queue = []; this.metrics.pendingJobs = 0; console.log(`πŸ—‘οΈ Cleared ${cancelledCount} pending jobs`); return cancelledCount; } } // πŸ”₯ μ „μ—­ 싱글톀 μΈμŠ€ν„΄μŠ€ export const dataflowJobQueue = new DataflowJobQueue(); // πŸ”₯ 개발 λͺ¨λ“œμ—μ„œ 큐 정보λ₯Ό μ „μ—­ 객체에 λ…ΈμΆœ if (typeof window !== "undefined" && process.env.NODE_ENV === "development") { (window as any).dataflowQueue = { getMetrics: () => dataflowJobQueue.getMetrics(), getQueueInfo: () => dataflowJobQueue.getQueueInfo(), clearQueue: () => dataflowJobQueue.clearQueue(), }; }