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

package api

import (
	"database/sql"
	"fmt"
	"net/http"
	"os"
	"strconv"
	"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 {
		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 {
		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 {
		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
	}

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

	repo.UpdateStatus(id, "cancelled")
	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 {
		writeError(w, http.StatusInternalServerError, "failed to create job")
		return
	}

	go func() {
		h.engine.Run(r.Context(), jobID, pairID)
	}()

	jobRepo := models.NewJobRepository(h.db)
	j, _ := jobRepo.GetByID(jobID)
	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 {
		writeError(w, http.StatusInternalServerError, "failed to fetch logs")
		return
	}

	count, _ := logRepo.CountByJobID(id)
	w.Header().Set("X-Total-Count", fmt.Sprintf("%d", count))
	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 {
		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,
	}
	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
	}
	return resp
}

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