oto/services/core/internal/httpserver/dto.go

package httpserver

import (
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/toki/oto/services/core/internal/cicdstate"
	"github.com/toki/oto/services/core/internal/runnerregistry"
	otopb "github.com/toki/oto/services/core/oto"
)

func jobToJSON(j *cicdstate.Job) map[string]interface{} {
	out := map[string]interface{}{
		"id":           j.ID,
		"name":         j.Name,
		"state":        j.State,
		"created_at":   j.CreatedAt.Format(time.RFC3339),
		"updated_at":   j.UpdatedAt.Format(time.RFC3339),
		"execution_id": j.ExecutionID,
	}
	if j.RunInput != nil {
		out["run_request"] = runInputToJSON(j.RunInput)
	}
	return out
}

// validateRunInput enforces the remote dispatch contract: exactly one of
// pipeline_yaml or pipeline_yaml_path must be provided.
func validateRunInput(in *cicdstate.RunInput) error {
	hasYAML := strings.TrimSpace(in.PipelineYAML) != ""
	hasPath := strings.TrimSpace(in.PipelineYAMLPath) != ""
	if hasYAML == hasPath {
		return fmt.Errorf("run request must set exactly one of pipeline_yaml or pipeline_yaml_path")
	}
	return nil
}

// runRequestFromJSON converts an incoming otopb.RunRequest decoded from the job
// create body into a store-level RunInput. It returns nil when no run request
// was supplied, keeping job creation backward compatible.
func runRequestFromJSON(rr *otopb.RunRequest) *cicdstate.RunInput {
	if rr == nil {
		return nil
	}
	in := &cicdstate.RunInput{
		PipelineYAMLPath: rr.GetPipelineYamlPath(),
		PipelineYAML:     rr.GetPipelineYaml(),
		CommandTypes:     append([]string(nil), rr.GetCommandTypes()...),
	}
	if len(rr.GetVariables()) > 0 {
		in.Variables = make(map[string]string, len(rr.GetVariables()))
		for k, v := range rr.GetVariables() {
			in.Variables[k] = v
		}
	}
	return in
}

// runRequestToProto builds the runner-facing RunRequest from the stored input,
// filling in the runner/job/execution identifiers resolved at dispatch time.
func runRequestToProto(in *cicdstate.RunInput, runnerID, jobID, execID string) *otopb.RunRequest {
	if in == nil {
		return nil
	}
	rr := &otopb.RunRequest{
		RunnerId:         runnerID,
		JobId:            jobID,
		ExecutionId:      execID,
		PipelineYamlPath: in.PipelineYAMLPath,
		PipelineYaml:     in.PipelineYAML,
		CommandTypes:     append([]string(nil), in.CommandTypes...),
	}
	if len(in.Variables) > 0 {
		rr.Variables = make(map[string]string, len(in.Variables))
		for k, v := range in.Variables {
			rr.Variables[k] = v
		}
	}
	return rr
}

func runInputToJSON(in *cicdstate.RunInput) map[string]interface{} {
	return map[string]interface{}{
		"pipeline_yaml_path": in.PipelineYAMLPath,
		"pipeline_yaml":      in.PipelineYAML,
		"variables":          in.Variables,
		"command_types":      in.CommandTypes,
	}
}

func execToJSON(e *cicdstate.Execution) map[string]interface{} {
	return map[string]interface{}{
		"id":           e.ID,
		"job_id":       e.JobID,
		"state":        e.State,
		"created_at":   e.CreatedAt.Format(time.RFC3339),
		"updated_at":   e.UpdatedAt.Format(time.RFC3339),
		"execution_id": e.ID,
	}
}

func logsToJSON(logs []cicdstate.LogEntry) map[string]interface{} {
	return map[string]interface{}{
		"logs": logs,
	}
}

func artifactsToJSON(artifacts []cicdstate.ArtifactEntry) map[string]interface{} {
	return map[string]interface{}{
		"artifacts": artifacts,
	}
}

func errorToJSON(message string) map[string]interface{} {
	return map[string]interface{}{
		"error": message,
	}
}

func successToJSON(success bool) map[string]interface{} {
	return map[string]interface{}{
		"success": success,
	}
}

func runnerDisconnectToJSON(success bool, err string) map[string]interface{} {
	out := successToJSON(success)
	if err != "" {
		out["error"] = err
	}
	return out
}

func runnerRecordToJSON(record runnerregistry.RunnerRecord) map[string]interface{} {
	return map[string]interface{}{
		"runner_id":          record.RunnerID,
		"alias":              record.Alias,
		"protocol_version":   record.ProtocolVersion,
		"status":             record.Status,
		"accepted_at":        record.AcceptedAt.Format(time.RFC3339),
		"first_heartbeat_at": record.FirstHeartbeatAt.Format(time.RFC3339),
		"last_heartbeat_at":  record.LastHeartbeatAt.Format(time.RFC3339),
		"failure_reason":     record.FailureReason,
	}
}

func runnerStatusToJSON(record runnerregistry.RunnerRecord, runnerID string, lastExec *cicdstate.Execution) map[string]interface{} {
	currentJobID := ""
	currentExecID := ""
	if lastExec != nil {
		currentJobID = lastExec.JobID
		currentExecID = lastExec.ID
	}
	return map[string]interface{}{
		"accepted":             true,
		"runner_id":            runnerID,
		"status":               record.Status,
		"current_job_id":       currentJobID,
		"current_execution_id": currentExecID,
		"message":              record.FailureReason,
	}
}

func runnerStatusRejectedToJSON(message string) map[string]interface{} {
	return map[string]interface{}{
		"accepted": false,
		"message":  message,
	}
}

func selfUpdateToJSON(success bool, accepted bool, deferred bool, targetVersion string, message string) map[string]interface{} {
	out := map[string]interface{}{
		"success":  success,
		"accepted": accepted,
		"deferred": deferred,
		"message":  message,
	}
	if !success && !deferred {
		out["error_message"] = message
	}
	if targetVersion != "" {
		out["target_version"] = targetVersion
	}
	return out
}

func writeResponse(w http.ResponseWriter, status int, data interface{}) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(data)
}