redo

2025-11-22 05:40:31 -06:00
parent 87cb54a17d
commit fb2e318eaa
12 changed files with 1891 additions and 353 deletions
--- a/internal/api/jobs.go
+++ b/internal/api/jobs.go
@@ -5,10 +5,14 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"net/http"
+	"strconv"
 	"time"

 	"fuego/pkg/types"
+
+	"github.com/go-chi/chi/v5"
 )

 // handleCreateJob creates a new job
@@ -39,10 +43,19 @@ func (s *Server) handleCreateJob(w http.ResponseWriter, r *http.Request) {
 		req.OutputFormat = "PNG"
 	}

+	// Default allow_parallel_runners to true if not provided
+	allowParallelRunners := true
+	if req.AllowParallelRunners != nil {
+		allowParallelRunners = *req.AllowParallelRunners
+	}
+
+	// Set job timeout to 24 hours (86400 seconds)
+	jobTimeout := 86400
+
 	result, err := s.db.Exec(
-		`INSERT INTO jobs (user_id, name, status, progress, frame_start, frame_end, output_format) 
-		 VALUES (?, ?, ?, ?, ?, ?, ?)`,
-		userID, req.Name, types.JobStatusPending, 0.0, req.FrameStart, req.FrameEnd, req.OutputFormat,
+		`INSERT INTO jobs (user_id, name, status, progress, frame_start, frame_end, output_format, allow_parallel_runners, timeout_seconds) 
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		userID, req.Name, types.JobStatusPending, 0.0, req.FrameStart, req.FrameEnd, req.OutputFormat, allowParallelRunners, jobTimeout,
 	)
 	if err != nil {
 		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to create job: %v", err))
@@ -51,11 +64,21 @@ func (s *Server) handleCreateJob(w http.ResponseWriter, r *http.Request) {

 	jobID, _ := result.LastInsertId()

+	// Determine task timeout based on output format
+	// 5 minutes (300 seconds) for frame tasks, 24 hours (86400 seconds) for FFmpeg video generation
+	taskTimeout := 300 // Default: 5 minutes for frame rendering
+	if req.OutputFormat == "MP4" {
+		// For MP4, we'll create frame tasks with 5 min timeout
+		// Video generation tasks will be created later with 24h timeout
+		taskTimeout = 300
+	}
+
 	// Create tasks for the job (one task per frame for simplicity, could be batched)
 	for frame := req.FrameStart; frame <= req.FrameEnd; frame++ {
 		_, err = s.db.Exec(
-			`INSERT INTO tasks (job_id, frame_start, frame_end, status) VALUES (?, ?, ?, ?)`,
-			jobID, frame, frame, types.TaskStatusPending,
+			`INSERT INTO tasks (job_id, frame_start, frame_end, status, timeout_seconds, max_retries) 
+			 VALUES (?, ?, ?, ?, ?, ?)`,
+			jobID, frame, frame, types.TaskStatusPending, taskTimeout, 3,
 		)
 		if err != nil {
 			s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to create tasks: %v", err))
@@ -64,17 +87,22 @@ func (s *Server) handleCreateJob(w http.ResponseWriter, r *http.Request) {
 	}

 	job := types.Job{
-		ID:           jobID,
-		UserID:       userID,
-		Name:         req.Name,
-		Status:       types.JobStatusPending,
-		Progress:     0.0,
-		FrameStart:   req.FrameStart,
-		FrameEnd:     req.FrameEnd,
-		OutputFormat: req.OutputFormat,
-		CreatedAt:    time.Now(),
+		ID:                   jobID,
+		UserID:               userID,
+		Name:                 req.Name,
+		Status:               types.JobStatusPending,
+		Progress:             0.0,
+		FrameStart:           req.FrameStart,
+		FrameEnd:             req.FrameEnd,
+		OutputFormat:         req.OutputFormat,
+		AllowParallelRunners: allowParallelRunners,
+		TimeoutSeconds:       jobTimeout,
+		CreatedAt:            time.Now(),
 	}

+	// Immediately try to distribute tasks to connected runners
+	go s.distributeTasksToRunners()
+
 	s.respondJSON(w, http.StatusCreated, job)
 }

@@ -88,7 +116,7 @@ func (s *Server) handleListJobs(w http.ResponseWriter, r *http.Request) {

 	rows, err := s.db.Query(
 		`SELECT id, user_id, name, status, progress, frame_start, frame_end, output_format, 
-		 created_at, started_at, completed_at, error_message 
+		 allow_parallel_runners, timeout_seconds, created_at, started_at, completed_at, error_message 
 		 FROM jobs WHERE user_id = ? ORDER BY created_at DESC`,
 		userID,
 	)
@@ -105,7 +133,7 @@ func (s *Server) handleListJobs(w http.ResponseWriter, r *http.Request) {

 		err := rows.Scan(
 			&job.ID, &job.UserID, &job.Name, &job.Status, &job.Progress,
-			&job.FrameStart, &job.FrameEnd, &job.OutputFormat,
+			&job.FrameStart, &job.FrameEnd, &job.OutputFormat, &job.AllowParallelRunners, &job.TimeoutSeconds,
 			&job.CreatedAt, &startedAt, &completedAt, &job.ErrorMessage,
 		)
 		if err != nil {
@@ -145,12 +173,12 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {

 	err = s.db.QueryRow(
 		`SELECT id, user_id, name, status, progress, frame_start, frame_end, output_format,
-		 created_at, started_at, completed_at, error_message
+		 allow_parallel_runners, timeout_seconds, created_at, started_at, completed_at, error_message
 		 FROM jobs WHERE id = ? AND user_id = ?`,
 		jobID, userID,
 	).Scan(
 		&job.ID, &job.UserID, &job.Name, &job.Status, &job.Progress,
-		&job.FrameStart, &job.FrameEnd, &job.OutputFormat,
+		&job.FrameStart, &job.FrameEnd, &job.OutputFormat, &job.AllowParallelRunners, &job.TimeoutSeconds,
 		&job.CreatedAt, &startedAt, &completedAt, &job.ErrorMessage,
 	)

@@ -280,7 +308,7 @@ func (s *Server) handleUploadJobFile(w http.ResponseWriter, r *http.Request) {
 	fileID, _ := result.LastInsertId()

 	s.respondJSON(w, http.StatusCreated, map[string]interface{}{
-		"id":       fileID,
+		"id":        fileID,
 		"file_name": header.Filename,
 		"file_path": filePath,
 		"file_size": header.Size,
@@ -496,3 +524,454 @@ func (s *Server) handleStreamVideo(w http.ResponseWriter, r *http.Request) {
 	}
 }

+// handleGetTaskLogs retrieves logs for a specific task
+func (s *Server) handleGetTaskLogs(w http.ResponseWriter, r *http.Request) {
+	userID, err := getUserID(r)
+	if err != nil {
+		s.respondError(w, http.StatusUnauthorized, err.Error())
+		return
+	}
+
+	jobID, err := parseID(r, "id")
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	taskIDStr := chi.URLParam(r, "taskId")
+	taskID, err := strconv.ParseInt(taskIDStr, 10, 64)
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, "Invalid task ID")
+		return
+	}
+
+	// Verify job belongs to user
+	var jobUserID int64
+	err = s.db.QueryRow("SELECT user_id FROM jobs WHERE id = ?", jobID).Scan(&jobUserID)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Job not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify job: %v", err))
+		return
+	}
+	if jobUserID != userID {
+		s.respondError(w, http.StatusForbidden, "Access denied")
+		return
+	}
+
+	// Verify task belongs to job
+	var taskJobID int64
+	err = s.db.QueryRow("SELECT job_id FROM tasks WHERE id = ?", taskID).Scan(&taskJobID)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Task not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify task: %v", err))
+		return
+	}
+	if taskJobID != jobID {
+		s.respondError(w, http.StatusBadRequest, "Task does not belong to this job")
+		return
+	}
+
+	// Get query parameters for filtering
+	stepName := r.URL.Query().Get("step_name")
+	logLevel := r.URL.Query().Get("log_level")
+	limitStr := r.URL.Query().Get("limit")
+	limit := 1000 // default
+	if limitStr != "" {
+		if l, err := strconv.Atoi(limitStr); err == nil && l > 0 {
+			limit = l
+		}
+	}
+
+	// Build query
+	query := `SELECT id, task_id, runner_id, log_level, message, step_name, created_at 
+			  FROM task_logs WHERE task_id = ?`
+	args := []interface{}{taskID}
+	if stepName != "" {
+		query += " AND step_name = ?"
+		args = append(args, stepName)
+	}
+	if logLevel != "" {
+		query += " AND log_level = ?"
+		args = append(args, logLevel)
+	}
+	query += " ORDER BY created_at ASC LIMIT ?"
+	args = append(args, limit)
+
+	rows, err := s.db.Query(query, args...)
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to query logs: %v", err))
+		return
+	}
+	defer rows.Close()
+
+	logs := []types.TaskLog{}
+	for rows.Next() {
+		var log types.TaskLog
+		var runnerID sql.NullInt64
+		err := rows.Scan(
+			&log.ID, &log.TaskID, &runnerID, &log.LogLevel, &log.Message,
+			&log.StepName, &log.CreatedAt,
+		)
+		if err != nil {
+			s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to scan log: %v", err))
+			return
+		}
+		if runnerID.Valid {
+			log.RunnerID = &runnerID.Int64
+		}
+		logs = append(logs, log)
+	}
+
+	s.respondJSON(w, http.StatusOK, logs)
+}
+
+// handleGetTaskSteps retrieves step timeline for a specific task
+func (s *Server) handleGetTaskSteps(w http.ResponseWriter, r *http.Request) {
+	userID, err := getUserID(r)
+	if err != nil {
+		s.respondError(w, http.StatusUnauthorized, err.Error())
+		return
+	}
+
+	jobID, err := parseID(r, "id")
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	taskIDStr := chi.URLParam(r, "taskId")
+	taskID, err := strconv.ParseInt(taskIDStr, 10, 64)
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, "Invalid task ID")
+		return
+	}
+
+	// Verify job belongs to user
+	var jobUserID int64
+	err = s.db.QueryRow("SELECT user_id FROM jobs WHERE id = ?", jobID).Scan(&jobUserID)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Job not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify job: %v", err))
+		return
+	}
+	if jobUserID != userID {
+		s.respondError(w, http.StatusForbidden, "Access denied")
+		return
+	}
+
+	// Verify task belongs to job
+	var taskJobID int64
+	err = s.db.QueryRow("SELECT job_id FROM tasks WHERE id = ?", taskID).Scan(&taskJobID)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Task not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify task: %v", err))
+		return
+	}
+	if taskJobID != jobID {
+		s.respondError(w, http.StatusBadRequest, "Task does not belong to this job")
+		return
+	}
+
+	rows, err := s.db.Query(
+		`SELECT id, task_id, step_name, status, started_at, completed_at, duration_ms, error_message
+		 FROM task_steps WHERE task_id = ? ORDER BY started_at ASC`,
+		taskID,
+	)
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to query steps: %v", err))
+		return
+	}
+	defer rows.Close()
+
+	steps := []types.TaskStep{}
+	for rows.Next() {
+		var step types.TaskStep
+		var startedAt, completedAt sql.NullTime
+		var durationMs sql.NullInt64
+		err := rows.Scan(
+			&step.ID, &step.TaskID, &step.StepName, &step.Status,
+			&startedAt, &completedAt, &durationMs, &step.ErrorMessage,
+		)
+		if err != nil {
+			s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to scan step: %v", err))
+			return
+		}
+		if startedAt.Valid {
+			step.StartedAt = &startedAt.Time
+		}
+		if completedAt.Valid {
+			step.CompletedAt = &completedAt.Time
+		}
+		if durationMs.Valid {
+			duration := int(durationMs.Int64)
+			step.DurationMs = &duration
+		}
+		steps = append(steps, step)
+	}
+
+	s.respondJSON(w, http.StatusOK, steps)
+}
+
+// handleRetryTask retries a failed task
+func (s *Server) handleRetryTask(w http.ResponseWriter, r *http.Request) {
+	userID, err := getUserID(r)
+	if err != nil {
+		s.respondError(w, http.StatusUnauthorized, err.Error())
+		return
+	}
+
+	jobID, err := parseID(r, "id")
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	taskIDStr := chi.URLParam(r, "taskId")
+	taskID, err := strconv.ParseInt(taskIDStr, 10, 64)
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, "Invalid task ID")
+		return
+	}
+
+	// Verify job belongs to user
+	var jobUserID int64
+	err = s.db.QueryRow("SELECT user_id FROM jobs WHERE id = ?", jobID).Scan(&jobUserID)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Job not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify job: %v", err))
+		return
+	}
+	if jobUserID != userID {
+		s.respondError(w, http.StatusForbidden, "Access denied")
+		return
+	}
+
+	// Verify task belongs to job and is in a retryable state
+	var taskJobID int64
+	var taskStatus string
+	var retryCount, maxRetries int
+	err = s.db.QueryRow(
+		"SELECT job_id, status, retry_count, max_retries FROM tasks WHERE id = ?",
+		taskID,
+	).Scan(&taskJobID, &taskStatus, &retryCount, &maxRetries)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Task not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify task: %v", err))
+		return
+	}
+	if taskJobID != jobID {
+		s.respondError(w, http.StatusBadRequest, "Task does not belong to this job")
+		return
+	}
+
+	if taskStatus != string(types.TaskStatusFailed) {
+		s.respondError(w, http.StatusBadRequest, "Task is not in failed state")
+		return
+	}
+
+	if retryCount >= maxRetries {
+		s.respondError(w, http.StatusBadRequest, "Maximum retries exceeded")
+		return
+	}
+
+	// Reset task to pending
+	_, err = s.db.Exec(
+		`UPDATE tasks SET status = ?, runner_id = NULL, current_step = NULL, 
+		 error_message = NULL, started_at = NULL, completed_at = NULL 
+		 WHERE id = ?`,
+		types.TaskStatusPending, taskID,
+	)
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to retry task: %v", err))
+		return
+	}
+
+	s.respondJSON(w, http.StatusOK, map[string]string{"message": "Task queued for retry"})
+}
+
+// handleStreamTaskLogsWebSocket streams task logs via WebSocket
+// Note: This is called after auth middleware, so userID is already verified
+func (s *Server) handleStreamTaskLogsWebSocket(w http.ResponseWriter, r *http.Request) {
+	userID, err := getUserID(r)
+	if err != nil {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return
+	}
+
+	jobID, err := parseID(r, "id")
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	taskIDStr := chi.URLParam(r, "taskId")
+	taskID, err := strconv.ParseInt(taskIDStr, 10, 64)
+	if err != nil {
+		s.respondError(w, http.StatusBadRequest, "Invalid task ID")
+		return
+	}
+
+	// Verify job belongs to user
+	var jobUserID int64
+	err = s.db.QueryRow("SELECT user_id FROM jobs WHERE id = ?", jobID).Scan(&jobUserID)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Job not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify job: %v", err))
+		return
+	}
+	if jobUserID != userID {
+		s.respondError(w, http.StatusForbidden, "Access denied")
+		return
+	}
+
+	// Verify task belongs to job
+	var taskJobID int64
+	err = s.db.QueryRow("SELECT job_id FROM tasks WHERE id = ?", taskID).Scan(&taskJobID)
+	if err == sql.ErrNoRows {
+		s.respondError(w, http.StatusNotFound, "Task not found")
+		return
+	}
+	if err != nil {
+		s.respondError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to verify task: %v", err))
+		return
+	}
+	if taskJobID != jobID {
+		s.respondError(w, http.StatusBadRequest, "Task does not belong to this job")
+		return
+	}
+
+	// Upgrade to WebSocket
+	conn, err := s.wsUpgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Printf("Failed to upgrade WebSocket: %v", err)
+		return
+	}
+	defer conn.Close()
+
+	key := fmt.Sprintf("%d:%d", jobID, taskID)
+	s.frontendConnsMu.Lock()
+	s.frontendConns[key] = conn
+	s.frontendConnsMu.Unlock()
+
+	defer func() {
+		s.frontendConnsMu.Lock()
+		delete(s.frontendConns, key)
+		s.frontendConnsMu.Unlock()
+	}()
+
+	// Send initial connection message
+	conn.WriteJSON(map[string]interface{}{
+		"type":      "connected",
+		"timestamp": time.Now().Unix(),
+	})
+
+	// Get last log ID to start streaming from
+	lastIDStr := r.URL.Query().Get("last_id")
+	lastID := int64(0)
+	if lastIDStr != "" {
+		if id, err := strconv.ParseInt(lastIDStr, 10, 64); err == nil {
+			lastID = id
+		}
+	}
+
+	// Send existing logs
+	rows, err := s.db.Query(
+		`SELECT id, task_id, runner_id, log_level, message, step_name, created_at
+		 FROM task_logs WHERE task_id = ? AND id > ? ORDER BY created_at ASC LIMIT 100`,
+		taskID, lastID,
+	)
+	if err == nil {
+		defer rows.Close()
+		for rows.Next() {
+			var log types.TaskLog
+			var runnerID sql.NullInt64
+			err := rows.Scan(
+				&log.ID, &log.TaskID, &runnerID, &log.LogLevel, &log.Message,
+				&log.StepName, &log.CreatedAt,
+			)
+			if err != nil {
+				continue
+			}
+			if runnerID.Valid {
+				log.RunnerID = &runnerID.Int64
+			}
+			if log.ID > lastID {
+				lastID = log.ID
+			}
+
+			conn.WriteJSON(map[string]interface{}{
+				"type":      "log",
+				"data":      log,
+				"timestamp": time.Now().Unix(),
+			})
+		}
+	}
+
+	// Poll for new logs and send them
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	ctx := r.Context()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			rows, err := s.db.Query(
+				`SELECT id, task_id, runner_id, log_level, message, step_name, created_at
+				 FROM task_logs WHERE task_id = ? AND id > ? ORDER BY created_at ASC LIMIT 100`,
+				taskID, lastID,
+			)
+			if err != nil {
+				continue
+			}
+
+			for rows.Next() {
+				var log types.TaskLog
+				var runnerID sql.NullInt64
+				err := rows.Scan(
+					&log.ID, &log.TaskID, &runnerID, &log.LogLevel, &log.Message,
+					&log.StepName, &log.CreatedAt,
+				)
+				if err != nil {
+					rows.Close()
+					continue
+				}
+				if runnerID.Valid {
+					log.RunnerID = &runnerID.Int64
+				}
+				if log.ID > lastID {
+					lastID = log.ID
+				}
+
+				conn.WriteJSON(map[string]interface{}{
+					"type":      "log",
+					"data":      log,
+					"timestamp": time.Now().Unix(),
+				})
+			}
+			rows.Close()
+		}
+	}
+}