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

package api

import (
	"database/sql"
	"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
	}

	repo := models.NewJobRepository(h.db)
	jobs, err := repo.GetAll(limit, offset)
	if err != nil {
		writeError(w, http.StatusInternalServerError, "failed to fetch jobs")
		return
	}

	out := make([]JobResponse, len(jobs))
	for i, j := range jobs {
		out[i] = jobToResp(j)
	}
	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) StreamLog(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
	}

	flusher, ok := w.(http.Flusher)
	if !ok {
		writeError(w, http.StatusInternalServerError, "streaming not supported")
		return
	}

	w.Header().Set("Content-Type", "text/plain")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	flusher.Flush()

	jobRepo := models.NewJobRepository(h.db)
	j, err := jobRepo.GetByID(id)
	if err == nil && j.LogFile != nil {
		data, _ := os.ReadFile(*j.LogFile)
		w.Write(data)
		flusher.Flush()
	}
}

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
}