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)) 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 { 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 } return resp } func jobWithStatsToResp(j models.JobWithStats) JobResponse { resp := jobToResp(j.Job) resp.DurationSeconds = j.DurationSeconds resp.LogLineCount = &j.LogLineCount return resp }