move-data-nas/internal/api/handlers_jobs.go

package api

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/syncserver/internal/models"
	"github.com/syncserver/internal/syncengine"
)

type JobHandler struct {
	db     *sql.DB
	engine *syncengine.Engine
}

func NewJobHandler(db *sql.DB, engine *syncengine.Engine) *JobHandler {
	return &JobHandler{db: db, engine: engine}
}

func (h *JobHandler) List(w http.ResponseWriter, r *http.Request) {
	limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
	offset, _ := strconv.Atoi(r.URL.Query().Get("offset"))
	if limit <= 0 || limit > 100 {
		limit = 50
	}

	var syncPairID *int64
	if spidStr := r.URL.Query().Get("sync_pair_id"); spidStr != "" {
		if spid, err := strconv.ParseInt(spidStr, 10, 64); err == nil {
			syncPairID = &spid
		}
	}

	status := r.URL.Query().Get("status")
	triggerType := r.URL.Query().Get("trigger_type")

	var from, to *time.Time
	if fromStr := r.URL.Query().Get("from"); fromStr != "" {
		if t, err := time.Parse(time.RFC3339, fromStr); err == nil {
			from = &t
		}
	}
	if toStr := r.URL.Query().Get("to"); toStr != "" {
		if t, err := time.Parse(time.RFC3339, toStr); err == nil {
			to = &t
		}
	}

	repo := models.NewJobLogRepository(h.db)
	jobs, total, err := repo.GetAllFiltered(limit, offset, syncPairID, status, triggerType, from, to)
	if err != nil {
		slog.Error("failed to fetch jobs", "error", err)
		writeError(w, http.StatusInternalServerError, "failed to fetch jobs")
		return
	}

	out := make([]JobResponse, len(jobs))
	for i, j := range jobs {
		out[i] = jobWithStatsToResp(j)
	}
	w.Header().Set("X-Total-Count", fmt.Sprintf("%d", total))
	writeJSON(w, out)
}

func (h *JobHandler) Get(w http.ResponseWriter, r *http.Request) {
	id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid id")
		return
	}

	repo := models.NewJobRepository(h.db)
	j, err := repo.GetByID(id)
	if err == sql.ErrNoRows {
		writeError(w, http.StatusNotFound, "job not found")
		return
	}
	if err != nil {
		slog.Error("failed to fetch job", "id", id, "error", err)
		writeError(w, http.StatusInternalServerError, "failed to fetch job")
		return
	}
	writeJSON(w, jobToResp(*j))
}

func (h *JobHandler) Cancel(w http.ResponseWriter, r *http.Request) {
	id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid id")
		return
	}

	repo := models.NewJobRepository(h.db)
	j, err := repo.GetByID(id)
	if err == sql.ErrNoRows {
		writeError(w, http.StatusNotFound, "job not found")
		return
	}
	if err != nil {
		slog.Error("failed to fetch job", "id", id, "error", err)
		writeError(w, http.StatusInternalServerError, "failed to fetch job")
		return
	}

	if j.Status != "queued" && j.Status != "waking_up" && j.Status != "running" {
		writeError(w, http.StatusBadRequest, "job is not cancellable")
		return
	}

	var body struct {
		Reason string `json:"reason"`
	}
	if r.Body != nil && r.ContentLength > 0 {
		_ = json.NewDecoder(r.Body).Decode(&body)
	}
	reason := strings.TrimSpace(body.Reason)
	if reason == "" {
		reason = "Job was cancelled by user"
	}

	if h.engine != nil {
		h.engine.Cancel(id, true)
	}

	repo.UpdateStatus(id, "cancelled")
	repo.SetError(id, "cancelled_user", reason)
	writeJSON(w, map[string]string{"status": "cancelled"})
}

func (h *JobHandler) TriggerRun(w http.ResponseWriter, r *http.Request) {
	pairID, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid id")
		return
	}

	if h.engine == nil {
		writeError(w, http.StatusInternalServerError, "engine not available")
		return
	}

	jobID, err := h.engine.CreateJob(pairID, "manual")
	if err != nil {
		slog.Error("failed to create job", "pair_id", pairID, "error", err)
		writeError(w, http.StatusInternalServerError, "failed to create job")
		return
	}

	go func() {
		defer func() {
			if r := recover(); r != nil {
				slog.Error("job run goroutine panicked", "job_id", jobID, "panic", r)
			}
		}()
		h.engine.Run(context.Background(), jobID, pairID)
	}()

	jobRepo := models.NewJobRepository(h.db)
	j, err := jobRepo.GetByID(jobID)
	if err != nil {
		slog.Error("failed to fetch created job", "job_id", jobID, "error", err)
		writeError(w, http.StatusInternalServerError, "failed to fetch created job")
		return
	}
	writeJSON(w, jobToResp(*j), http.StatusCreated)
}

func (h *JobHandler) GetLog(w http.ResponseWriter, r *http.Request) {
	id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid id")
		return
	}

	offset, _ := strconv.Atoi(r.URL.Query().Get("offset"))
	limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
	if limit <= 0 {
		limit = 1000
	}

	logRepo := models.NewJobLogRepository(h.db)
	logs, err := logRepo.GetByJobID(id, limit, offset)
	if err != nil {
		slog.Error("failed to fetch logs", "id", id, "error", err)
		writeError(w, http.StatusInternalServerError, "failed to fetch logs")
		return
	}

	count, _ := logRepo.CountByJobID(id)
	w.Header().Set("X-Total-Count", fmt.Sprintf("%d", count))
	if logs == nil {
		writeJSON(w, []any{})
	} else {
		writeJSON(w, logs)
	}
}

func (h *JobHandler) DownloadLog(w http.ResponseWriter, r *http.Request) {
	id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid id")
		return
	}

	jobRepo := models.NewJobRepository(h.db)
	j, err := jobRepo.GetByID(id)
	if err != nil {
		slog.Error("failed to fetch job", "id", id, "error", err)
		writeError(w, http.StatusInternalServerError, "failed to fetch job")
		return
	}

	if j.LogFile != nil {
		data, err := os.ReadFile(*j.LogFile)
		if err == nil {
			w.Header().Set("Content-Type", "text/plain")
			w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="job-%d.log"`, id))
			w.Write(data)
			return
		}
	}

	logRepo := models.NewJobLogRepository(h.db)
	logs, _ := logRepo.GetByJobID(id, 100000, 0)
	for _, l := range logs {
		fmt.Fprintf(w, "[%s] %s\n", l.Timestamp.Format(time.RFC3339), l.Content)
	}
}

func jobToResp(j models.Job) JobResponse {
	resp := JobResponse{
		ID:           j.ID,
		SyncPairID:   j.SyncPairID,
		TriggerType:  j.TriggerType,
		Status:       j.Status,
		LogFile:      j.LogFile,
		ErrorMessage: j.ErrorMessage,
		ErrorCode:    j.ErrorCode,
	}
	if j.StartedAt != nil {
		s := j.StartedAt.Format(time.RFC3339)
		resp.StartedAt = &s
	}
	if j.FinishedAt != nil {
		s := j.FinishedAt.Format(time.RFC3339)
		resp.FinishedAt = &s
	}
	if j.TotalSizeBytes > 0 {
		resp.TotalSizeBytes = &j.TotalSizeBytes
	}
	if j.SentBytes > 0 {
		resp.SentBytes = &j.SentBytes
	}
	return resp
}

func jobWithStatsToResp(j models.JobWithStats) JobResponse {
	resp := jobToResp(j.Job)
	resp.DurationSeconds = j.DurationSeconds
	resp.LogLineCount = &j.LogLineCount
	return resp
}