import { Request, Response } from 'express'; import { IStatisticsService } from '../services/interfaces/IStatisticsService'; import { IWorkerService } from '../services/interfaces/IWorkerService'; import { ITaskQueueService } from '../services/interfaces/ITaskQueueService'; import { IDatasetService } from '../services/interfaces/IDatasetService'; import { TaskBrokerWebSocketServer } from '../websocket/wsServer'; import { TaskStatus } from '../types/task.types'; import { CONFIG } from '../config/app.config'; export class AdminController { private statisticsService: IStatisticsService; private workerService: IWorkerService; private taskQueueService: ITaskQueueService; private datasetService: IDatasetService; private wsServer?: TaskBrokerWebSocketServer; constructor( statisticsService: IStatisticsService, workerService: IWorkerService, taskQueueService: ITaskQueueService, datasetService: IDatasetService, wsServer?: TaskBrokerWebSocketServer ) { this.statisticsService = statisticsService; this.workerService = workerService; this.taskQueueService = taskQueueService; this.datasetService = datasetService; this.wsServer = wsServer; } public setWsServer(wsServer: TaskBrokerWebSocketServer): void { this.wsServer = wsServer; } public getStatistics = async (req: Request, res: Response): Promise => { try { const stats = await this.statisticsService.getSystemStatistics(); res.status(200).json(stats); } catch (err: any) { res.status(500).json({ error: err.message }); } }; public getWorkers = async (req: Request, res: Response): Promise => { try { const status = req.query.status as any; const workers = await this.workerService.getAllWorkers(status); res.status(200).json({ count: workers.length, workers }); } catch (err: any) { res.status(500).json({ error: err.message }); } }; public getTasks = async (req: Request, res: Response): Promise => { try { const status = req.query.status as TaskStatus; const workerId = req.query.workerId as string; const limit = req.query.limit ? parseInt(req.query.limit as string, 10) : 100; const offset = req.query.offset ? parseInt(req.query.offset as string, 10) : 0; const tasks = await this.taskQueueService.getTasks({ status, workerId, limit, offset }); const counts = await this.taskQueueService.getTaskCounts(); res.status(200).json({ summary: counts, limit, offset, count: tasks.length, tasks, }); } catch (err: any) { res.status(500).json({ error: err.message }); } }; public createTask = async (req: Request, res: Response): Promise => { try { const { startRow, endRow, totalRows, chunkSize } = req.body; if (totalRows !== undefined) { const count = await this.taskQueueService.generateBatchTasks( Number(totalRows), chunkSize ? Number(chunkSize) : undefined ); res.status(201).json({ message: `Successfully queued batch tasks. Total tasks: ${count}` }); return; } if (startRow === undefined || endRow === undefined) { res.status(400).json({ error: 'Parameters startRow and endRow (or totalRows) are required.' }); return; } const task = await this.taskQueueService.createTask({ startRow: Number(startRow), endRow: Number(endRow), }); res.status(201).json({ message: 'Task created successfully', task }); } catch (err: any) { res.status(500).json({ error: err.message }); } }; public resetTasks = async (req: Request, res: Response): Promise => { try { const clearedCount = await this.taskQueueService.resetAllTasks(); res.status(200).json({ message: `Cleared ${clearedCount} existing tasks.` }); } catch (err: any) { res.status(500).json({ error: err.message }); } }; public shutdownWorker = async (req: Request, res: Response): Promise => { try { const { workerId, reason } = req.body; if (!workerId) { res.status(400).json({ error: 'workerId is required' }); return; } if (this.wsServer) { const sent = this.wsServer.sendToWorker(workerId, { type: 'shutdown', payload: { reason: reason || 'Administrative shutdown order' }, }); if (sent) { await this.workerService.setWorkerStatus(workerId, 'offline', null); res.status(200).json({ message: `Shutdown message dispatched to worker ${workerId}` }); return; } } await this.workerService.setWorkerStatus(workerId, 'offline', null); res.status(200).json({ message: `Worker ${workerId} marked offline (no active WS connection)` }); } catch (err: any) { res.status(500).json({ error: err.message }); } }; public getDatasetRows = async (req: Request, res: Response): Promise => { try { const startRow = req.query.startRow ? parseInt(req.query.startRow as string, 10) : 0; const limit = req.query.limit ? parseInt(req.query.limit as string, 10) : 5; const safeStartRow = Math.max(0, isNaN(startRow) ? 0 : startRow); const safeLimit = Math.min(100, Math.max(1, isNaN(limit) ? 5 : limit)); const rawRows = await this.datasetService.readRowContentSnippet(safeStartRow, safeLimit); const datasetStats = await this.datasetService.getDatasetStats(); res.status(200).json({ startRow: safeStartRow, limit: safeLimit, count: rawRows.length, totalRows: datasetStats.totalIndexedRows, attributes: CONFIG.DATASET_ATTRIBUTES, header: CONFIG.DATASET_HEADER, rows: rawRows.map((content, idx) => ({ rowIndex: safeStartRow + idx, content, })), }); } catch (err: any) { res.status(500).json({ error: err.message }); } }; public getTaskData = async (req: Request, res: Response): Promise => { try { const taskId = (req.params.taskId || req.query.taskId || req.body?.taskId) as string | undefined; let startRow = req.query.startRow !== undefined ? parseInt(req.query.startRow as string, 10) : req.body?.startRow; let endRow = req.query.endRow !== undefined ? parseInt(req.query.endRow as string, 10) : req.body?.endRow; let startOffset = req.query.startOffset !== undefined ? parseInt(req.query.startOffset as string, 10) : req.body?.startOffset; let endOffset = req.query.endOffset !== undefined ? parseInt(req.query.endOffset as string, 10) : req.body?.endOffset; const format = (req.query.format || req.body?.format) as string | undefined; if (taskId) { const task = await this.taskQueueService.getTask(taskId); if (task) { if (startRow === undefined) startRow = task.startRow; if (endRow === undefined) endRow = task.endRow; if (startOffset === undefined && task.startOffset !== undefined) startOffset = task.startOffset; if (endOffset === undefined && task.endOffset !== undefined) endOffset = task.endOffset; } } if ( taskId === undefined && (startRow === undefined || endRow === undefined) && (startOffset === undefined || endOffset === undefined) ) { res.status(400).json({ error: 'Missing required parameters. Please specify taskId, or startRow & endRow, or startOffset & endOffset.', }); return; } const result = await this.datasetService.getTaskData({ taskId, startRow: startRow !== undefined ? Number(startRow) : undefined, endRow: endRow !== undefined ? Number(endRow) : undefined, startOffset: startOffset !== undefined ? Number(startOffset) : undefined, endOffset: endOffset !== undefined ? Number(endOffset) : undefined, }); const wantsJson = format === 'json' || (req.headers.accept && req.headers.accept.includes('application/json')); if (wantsJson) { const contentStr = result.buffer.toString('utf-8'); res.status(200).json({ taskId: taskId || null, startRow: result.startRow, endRow: result.endRow, startOffset: result.startOffset, endOffset: result.endOffset, byteLength: result.byteLength, attributes: result.attributes, header: result.header, content: contentStr, }); return; } res.setHeader('Content-Type', 'text/csv; charset=utf-8'); res.setHeader('Content-Length', result.buffer.length); if (taskId) { res.setHeader('Content-Disposition', `attachment; filename="${taskId}.csv"`); } res.status(200).send(result.buffer); } catch (err: any) { res.status(500).json({ error: err.message }); } }; }