darroyo
7a024cec3b
r.Context() is cancelled when the HTTP handler returns (after 201 is sent), causing the job to be immediately marked as cancelled_shutdown before WoL even runs. Use context.Background() so the job goroutine runs independently of the HTTP request lifecycle.
266 lines
6.7 KiB
Go
266 lines
6.7 KiB
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, j.SyncPairID, 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
|
|
}
|
|
return resp
|
|
}
|
|
|
|
func jobWithStatsToResp(j models.JobWithStats) JobResponse {
|
|
resp := jobToResp(j.Job)
|
|
resp.DurationSeconds = j.DurationSeconds
|
|
resp.LogLineCount = &j.LogLineCount
|
|
return resp
|
|
}
|