silo/internal/api/job_handlers.go

package api

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

	"github.com/go-chi/chi/v5"
	"github.com/kindredsystems/silo/internal/auth"
	"github.com/kindredsystems/silo/internal/db"
)

// HandleListJobs returns jobs filtered by status and/or item.
func (s *Server) HandleListJobs(w http.ResponseWriter, r *http.Request) {
	status := r.URL.Query().Get("status")
	itemID := r.URL.Query().Get("item_id")

	limit := 50
	if v := r.URL.Query().Get("limit"); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 200 {
			limit = n
		}
	}
	offset := 0
	if v := r.URL.Query().Get("offset"); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n >= 0 {
			offset = n
		}
	}

	jobs, err := s.jobs.ListJobs(r.Context(), status, itemID, limit, offset)
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to list jobs")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to list jobs")
		return
	}

	writeJSON(w, http.StatusOK, jobs)
}

// HandleGetJob returns a single job by ID.
func (s *Server) HandleGetJob(w http.ResponseWriter, r *http.Request) {
	jobID := chi.URLParam(r, "jobID")

	job, err := s.jobs.GetJob(r.Context(), jobID)
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to get job")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to get job")
		return
	}
	if job == nil {
		writeError(w, http.StatusNotFound, "not_found", "Job not found")
		return
	}

	writeJSON(w, http.StatusOK, job)
}

// HandleGetJobLogs returns log entries for a job.
func (s *Server) HandleGetJobLogs(w http.ResponseWriter, r *http.Request) {
	jobID := chi.URLParam(r, "jobID")

	logs, err := s.jobs.GetJobLogs(r.Context(), jobID)
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to get job logs")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to get job logs")
		return
	}

	writeJSON(w, http.StatusOK, logs)
}

// HandleCreateJob manually triggers a job.
func (s *Server) HandleCreateJob(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	user := auth.UserFromContext(ctx)

	var req struct {
		DefinitionName string         `json:"definition_name"`
		ItemID         *string        `json:"item_id,omitempty"`
		ProjectID      *string        `json:"project_id,omitempty"`
		ScopeMetadata  map[string]any `json:"scope_metadata,omitempty"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid_body", "Invalid JSON body")
		return
	}

	if req.DefinitionName == "" {
		writeError(w, http.StatusBadRequest, "missing_field", "definition_name is required")
		return
	}

	// Look up definition
	def, err := s.jobs.GetDefinition(ctx, req.DefinitionName)
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to look up job definition")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to look up definition")
		return
	}
	if def == nil {
		writeError(w, http.StatusNotFound, "not_found", "Job definition not found: "+req.DefinitionName)
		return
	}

	var createdBy *string
	if user != nil {
		createdBy = &user.Username
	}

	job := &db.Job{
		JobDefinitionID: &def.ID,
		DefinitionName:  def.Name,
		Priority:        def.Priority,
		ItemID:          req.ItemID,
		ProjectID:       req.ProjectID,
		ScopeMetadata:   req.ScopeMetadata,
		RunnerTags:      def.RunnerTags,
		TimeoutSeconds:  def.TimeoutSeconds,
		MaxRetries:      def.MaxRetries,
		CreatedBy:       createdBy,
	}

	if err := s.jobs.CreateJob(ctx, job); err != nil {
		s.logger.Error().Err(err).Msg("failed to create job")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to create job")
		return
	}

	s.broker.Publish("job.created", mustMarshal(map[string]any{
		"job_id":          job.ID,
		"definition_name": job.DefinitionName,
		"item_id":         job.ItemID,
	}))

	writeJSON(w, http.StatusCreated, job)
}

// HandleCancelJob cancels a pending or active job.
func (s *Server) HandleCancelJob(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	jobID := chi.URLParam(r, "jobID")
	user := auth.UserFromContext(ctx)

	cancelledBy := "system"
	if user != nil {
		cancelledBy = user.Username
	}

	if err := s.jobs.CancelJob(ctx, jobID, cancelledBy); err != nil {
		writeError(w, http.StatusBadRequest, "cancel_failed", err.Error())
		return
	}

	s.broker.Publish("job.cancelled", mustMarshal(map[string]any{
		"job_id":       jobID,
		"cancelled_by": cancelledBy,
	}))

	writeJSON(w, http.StatusOK, map[string]string{"status": "cancelled"})
}

// HandleListJobDefinitions returns all loaded job definitions.
func (s *Server) HandleListJobDefinitions(w http.ResponseWriter, r *http.Request) {
	defs, err := s.jobs.ListDefinitions(r.Context())
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to list job definitions")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to list definitions")
		return
	}

	writeJSON(w, http.StatusOK, defs)
}

// HandleGetJobDefinition returns a single job definition by name.
func (s *Server) HandleGetJobDefinition(w http.ResponseWriter, r *http.Request) {
	name := chi.URLParam(r, "name")

	def, err := s.jobs.GetDefinition(r.Context(), name)
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to get job definition")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to get definition")
		return
	}
	if def == nil {
		writeError(w, http.StatusNotFound, "not_found", "Job definition not found")
		return
	}

	writeJSON(w, http.StatusOK, def)
}

// HandleReloadJobDefinitions re-reads YAML files from disk and upserts them.
func (s *Server) HandleReloadJobDefinitions(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	if s.jobDefsDir == "" {
		writeError(w, http.StatusBadRequest, "no_directory", "Job definitions directory not configured")
		return
	}

	defs, err := loadAndUpsertJobDefs(ctx, s.jobDefsDir, s.jobs)
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to reload job definitions")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to reload definitions")
		return
	}

	// Update in-memory map
	s.jobDefs = defs

	writeJSON(w, http.StatusOK, map[string]any{
		"reloaded": len(defs),
	})
}

// HandleListRunners returns all registered runners (admin).
func (s *Server) HandleListRunners(w http.ResponseWriter, r *http.Request) {
	runners, err := s.jobs.ListRunners(r.Context())
	if err != nil {
		s.logger.Error().Err(err).Msg("failed to list runners")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to list runners")
		return
	}

	// Redact token hashes from response
	type runnerResponse struct {
		ID            string         `json:"id"`
		Name          string         `json:"name"`
		TokenPrefix   string         `json:"token_prefix"`
		Tags          []string       `json:"tags"`
		Status        string         `json:"status"`
		LastHeartbeat *string        `json:"last_heartbeat,omitempty"`
		LastJobID     *string        `json:"last_job_id,omitempty"`
		Metadata      map[string]any `json:"metadata,omitempty"`
		CreatedAt     string         `json:"created_at"`
	}

	resp := make([]runnerResponse, len(runners))
	for i, runner := range runners {
		var hb *string
		if runner.LastHeartbeat != nil {
			s := runner.LastHeartbeat.Format("2006-01-02T15:04:05Z07:00")
			hb = &s
		}
		resp[i] = runnerResponse{
			ID:            runner.ID,
			Name:          runner.Name,
			TokenPrefix:   runner.TokenPrefix,
			Tags:          runner.Tags,
			Status:        runner.Status,
			LastHeartbeat: hb,
			LastJobID:     runner.LastJobID,
			Metadata:      runner.Metadata,
			CreatedAt:     runner.CreatedAt.Format("2006-01-02T15:04:05Z07:00"),
		}
	}

	writeJSON(w, http.StatusOK, resp)
}

// HandleRegisterRunner creates a new runner and returns the token (admin).
func (s *Server) HandleRegisterRunner(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	var req struct {
		Name     string         `json:"name"`
		Tags     []string       `json:"tags"`
		Metadata map[string]any `json:"metadata,omitempty"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid_body", "Invalid JSON body")
		return
	}
	if req.Name == "" {
		writeError(w, http.StatusBadRequest, "missing_field", "name is required")
		return
	}
	if len(req.Tags) == 0 {
		writeError(w, http.StatusBadRequest, "missing_field", "tags is required (at least one)")
		return
	}

	rawToken, tokenHash, tokenPrefix := generateRunnerToken()

	runner := &db.Runner{
		Name:        req.Name,
		TokenHash:   tokenHash,
		TokenPrefix: tokenPrefix,
		Tags:        req.Tags,
		Metadata:    req.Metadata,
	}

	if err := s.jobs.RegisterRunner(ctx, runner); err != nil {
		s.logger.Error().Err(err).Msg("failed to register runner")
		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to register runner")
		return
	}

	s.broker.Publish("runner.online", mustMarshal(map[string]any{
		"runner_id": runner.ID,
		"name":      runner.Name,
	}))

	writeJSON(w, http.StatusCreated, map[string]any{
		"id":    runner.ID,
		"name":  runner.Name,
		"token": rawToken,
		"tags":  runner.Tags,
	})
}

// HandleDeleteRunner removes a runner (admin).
func (s *Server) HandleDeleteRunner(w http.ResponseWriter, r *http.Request) {
	runnerID := chi.URLParam(r, "runnerID")

	if err := s.jobs.DeleteRunner(r.Context(), runnerID); err != nil {
		writeError(w, http.StatusNotFound, "not_found", err.Error())
		return
	}

	w.WriteHeader(http.StatusNoContent)
}

// triggerJobs creates jobs for all enabled definitions matching the trigger type.
// It applies trigger filters (e.g. item_type) before creating each job.
func (s *Server) triggerJobs(ctx context.Context, triggerType string, itemID string, item *db.Item) {
	if !s.modules.IsEnabled("jobs") {
		return
	}

	defs, err := s.jobs.GetDefinitionsByTrigger(ctx, triggerType)
	if err != nil {
		s.logger.Error().Err(err).Str("trigger", triggerType).Msg("failed to get job definitions for trigger")
		return
	}

	for _, def := range defs {
		// Apply trigger filter (e.g. item_type == "assembly")
		if def.Definition != nil {
			if triggerCfg, ok := def.Definition["trigger"].(map[string]any); ok {
				if filterCfg, ok := triggerCfg["filter"].(map[string]any); ok {
					if reqType, ok := filterCfg["item_type"].(string); ok && item != nil {
						if item.ItemType != reqType {
							continue
						}
					}
				}
			}
		}

		job := &db.Job{
			JobDefinitionID: &def.ID,
			DefinitionName:  def.Name,
			Priority:        def.Priority,
			ItemID:          &itemID,
			RunnerTags:      def.RunnerTags,
			TimeoutSeconds:  def.TimeoutSeconds,
			MaxRetries:      def.MaxRetries,
		}

		if err := s.jobs.CreateJob(ctx, job); err != nil {
			s.logger.Error().Err(err).Str("definition", def.Name).Msg("failed to create triggered job")
			continue
		}

		s.broker.Publish("job.created", mustMarshal(map[string]any{
			"job_id":          job.ID,
			"definition_name": def.Name,
			"trigger":         triggerType,
			"item_id":         itemID,
		}))

		s.logger.Info().
			Str("job_id", job.ID).
			Str("definition", def.Name).
			Str("trigger", triggerType).
			Str("item_id", itemID).
			Msg("triggered job")
	}
}