llmgw/internal/control/adminapi/jobs.go

package adminapi

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"strconv"
	"time"

	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)

// Jobs — 3D models asked for and asked after — as an operator sees them:
// every owner's, which a client sees only of its own.

// jobStore is what jobs need of the store; asked of the usage source by
// type assertion, as requestFinder is.
type jobStore interface {
	AllJobs(ctx context.Context, status string, limit int) ([]store.Job, error)
	JobByID(ctx context.Context, id string) (store.Job, error)
	CancelJob(ctx context.Context, id string, now time.Time) (store.Job, error)
}

// JobView is one job.
type JobView struct {
	ID           string          `json:"id"`
	Owner        string          `json:"owner"`
	KeyID        string          `json:"key_id"`
	Endpoint     string          `json:"endpoint"`
	Model        string          `json:"model"`
	Status       string          `json:"status"`
	Progress     json.RawMessage `json:"progress,omitempty"`
	OutputFileID string          `json:"output_file_id,omitempty"`
	ContentType  string          `json:"content_type,omitempty"`
	Error        string          `json:"error,omitempty"`
	CreatedAt    time.Time       `json:"created_at"`
	StartedAt    *time.Time      `json:"started_at,omitempty"`
	FinishedAt   *time.Time      `json:"finished_at,omitempty"`
}

func jobView(j store.Job) JobView {
	v := JobView{ID: j.ID, Owner: j.Owner, KeyID: j.KeyID, Endpoint: j.Endpoint, Model: j.Model, Status: j.Status,
		OutputFileID: j.OutputFileID, ContentType: j.ContentType, Error: j.Error, CreatedAt: j.CreatedAt}
	if j.Progress != "" && json.Valid([]byte(j.Progress)) {
		v.Progress = json.RawMessage(j.Progress)
	}
	if !j.StartedAt.IsZero() {
		v.StartedAt = &j.StartedAt
	}
	if !j.FinishedAt.IsZero() {
		v.FinishedAt = &j.FinishedAt
	}
	return v
}

func (a *API) jobs() (jobStore, bool) {
	s, ok := a.usage.(jobStore)
	return s, ok
}

func (a *API) listJobs(w http.ResponseWriter, r *http.Request) {
	s, ok := a.jobs()
	if !ok {
		writeError(w, http.StatusNotImplemented, "jobs need the store, which this control plane runs without")
		return
	}
	limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
	js, err := s.AllJobs(r.Context(), r.URL.Query().Get("status"), limit)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	out := make([]JobView, 0, len(js))
	for _, j := range js {
		out = append(out, jobView(j))
	}
	writeJSON(w, http.StatusOK, out)
}

func (a *API) getJob(w http.ResponseWriter, r *http.Request) {
	s, ok := a.jobs()
	if !ok {
		writeError(w, http.StatusNotImplemented, "jobs need the store, which this control plane runs without")
		return
	}
	j, err := s.JobByID(r.Context(), r.PathValue("id"))
	if errors.Is(err, store.ErrNotFound) {
		writeError(w, http.StatusNotFound, "no job "+r.PathValue("id"))
		return
	}
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, jobView(j))
}

func (a *API) cancelJob(w http.ResponseWriter, r *http.Request) {
	s, ok := a.jobs()
	if !ok {
		writeError(w, http.StatusNotImplemented, "jobs need the store, which this control plane runs without")
		return
	}
	if _, err := s.JobByID(r.Context(), r.PathValue("id")); errors.Is(err, store.ErrNotFound) {
		writeError(w, http.StatusNotFound, "no job "+r.PathValue("id"))
		return
	}
	j, err := s.CancelJob(r.Context(), r.PathValue("id"), time.Now())
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, jobView(j))
}