homelab-frontend/internal/serviceadapter/workflow_adapter.go

package serviceadapter

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"

	"forgejo.riotpiao.com/rock/homelab-frontend/internal/temporal"
)

// WorkflowAdapter handles X-Service: workflow requests.
// It forwards workflow operations to the Temporal gRPC service.
// Users can specify namespace via the request payload.
type WorkflowAdapter struct {
	temporalHandler *temporal.Handler
}

// NewWorkflowAdapter creates a new WorkflowAdapter.
func NewWorkflowAdapter(handler *temporal.Handler) *WorkflowAdapter {
	return &WorkflowAdapter{
		temporalHandler: handler,
	}
}

// HandleStart handles workflow start requests.
// Expects payload: { "namespace": "default", "workflow_id": "...", "workflow_type": "...", "task_queue": "...", "input": {...} }
func (wa *WorkflowAdapter) HandleStart(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleDescribe handles workflow describe requests.
// Expects payload: { "namespace": "default", "workflow_id": "..." }
func (wa *WorkflowAdapter) HandleDescribe(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleList handles workflow list requests.
// Expects payload: { "namespace": "default", "query": "..." (optional) }
func (wa *WorkflowAdapter) HandleList(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleHistory handles workflow history requests.
// Expects payload: { "namespace": "default", "workflow_id": "..." }
func (wa *WorkflowAdapter) HandleHistory(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleTerminate handles workflow termination.
// Expects payload: { "namespace": "default", "workflow_id": "...", "reason": "..." }
func (wa *WorkflowAdapter) HandleTerminate(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleCancel handles workflow cancellation.
// Expects payload: { "namespace": "default", "workflow_id": "..." }
func (wa *WorkflowAdapter) HandleCancel(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleSignal handles workflow signal.
// Expects payload: { "namespace": "default", "workflow_id": "...", "signal_name": "...", "signal_data": {...} }
func (wa *WorkflowAdapter) HandleSignal(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleQuery handles workflow query.
// Expects payload: { "namespace": "default", "workflow_id": "...", "query_type": "...", "query_data": {...} }
func (wa *WorkflowAdapter) HandleQuery(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleReset handles workflow reset.
// Expects payload: { "namespace": "default", "workflow_id": "...", "reset_type": "..." }
func (wa *WorkflowAdapter) HandleReset(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// HandleUpdate handles workflow update.
// Expects payload: { "namespace": "default", "workflow_id": "...", "update_data": {...} }
func (wa *WorkflowAdapter) HandleUpdate(w http.ResponseWriter, r *http.Request) {
	wa.forwardToTemporal(w, r)
}

// resourceToAction maps X-Resource names to Temporal action names.
var resourceToAction = map[string]string{
	"execute":    "START_WORKFLOW",
	"describe":   "DESCRIBE_WORKFLOW",
	"list":       "LIST_WORKFLOWS",
	"history":    "GET_WORKFLOW_HISTORY",
	"terminate":  "TERMINATE_WORKFLOW",
	"cancel":     "CANCEL_WORKFLOW",
	"signal":     "SIGNAL_WORKFLOW",
	"query":      "QUERY_WORKFLOW",
	"reset":      "RESET_WORKFLOW",
	"update":     "UPDATE_WORKFLOW",
}

// ServeHTTP implements http.Handler for X-Service: workflow routing.
// Maps X-Resource header to Temporal action, injects action into body,
// and forwards to the temporal handler.
func (wa *WorkflowAdapter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	resource := r.Header.Get("X-Resource")
	action, ok := resourceToAction[resource]
	if !ok {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusNotFound)
		fmt.Fprintf(w, `{"error":"unknown workflow resource: %s"}`, resource)
		return
	}

	// Read body, inject action, forward
	body, err := io.ReadAll(r.Body)
	if err != nil {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusBadRequest)
		fmt.Fprintf(w, `{"error":"failed to read body: %s"}`, err.Error())
		return
	}

	var payload map[string]interface{}
	if len(body) > 0 {
		if err := json.Unmarshal(body, &payload); err != nil {
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusBadRequest)
			fmt.Fprintf(w, `{"error":"invalid JSON: %s"}`, err.Error())
			return
		}
	} else {
		payload = make(map[string]interface{})
	}

	// Inject action into body for temporal handler
	payload["action"] = action

	// namespace is required for all workflow operations
	if ns, ok := payload["namespace"].(string); !ok || ns == "" {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusBadRequest)
		fmt.Fprintf(w, `{"error":"namespace is required"}`)
		return
	}

	newBody, _ := json.Marshal(payload)
	r.Body = io.NopCloser(bytes.NewReader(newBody))
	r.ContentLength = int64(len(newBody))
	r.URL.Path = "/workflow"

	wa.temporalHandler.ServeHTTP(w, r)
}

// forwardToTemporal reads the request body, ensures namespace is specified,
// and forwards to the temporal handler.
func (wa *WorkflowAdapter) forwardToTemporal(w http.ResponseWriter, r *http.Request) {
	// Read request body
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to read request body: %v", err), http.StatusBadRequest)
		return
	}
	defer r.Body.Close()

	// Parse JSON to check for namespace
	var payload map[string]interface{}
	if err := json.Unmarshal(body, &payload); err != nil {
		http.Error(w, fmt.Sprintf("invalid JSON payload: %v", err), http.StatusBadRequest)
		return
	}

	// Ensure namespace is specified (required for Temporal routing)
	namespace, ok := payload["namespace"].(string)
	if !ok || namespace == "" {
		http.Error(w, `"namespace" field required in payload`, http.StatusBadRequest)
		return
	}

	// Forward to temporal handler by calling it with the request
	// Restore body for temporal handler
	r.Body = io.NopCloser(bytes.NewReader(body))
	r.ContentLength = int64(len(body))

	// Call temporal handler
	wa.temporalHandler.ServeHTTP(w, r)
}

// GetSpec returns the ServiceAdapter spec for workflow service.
// This defines the available resources and methods.
func GetWorkflowSpec() *Spec {
	return &Spec{
		ServiceName: "workflow",
		Upstream: Upstream{
			URL:            "grpc://temporal:7233", // gRPC endpoint
			TimeoutSeconds: 30,
		},
		Auth: Auth{
			Required:   false,
			Capability: "workflow:execute",
		},
		Retryable: true,
		Resources: []Resource{
			{
				Name: "execute",
				Methods: []Method{
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/StartWorkflowExecution",
						RequestSchema:  "workflow_start_request",
						ResponseSchema: "workflow_start_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:execute",
						},
					},
				},
			},
			{
				Name: "describe",
				Methods: []Method{
					{
						Verb:           "GET",
					},
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/DescribeWorkflowExecution",
						RequestSchema:  "workflow_describe_request",
						ResponseSchema: "workflow_describe_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:read",
						},
					},
				},
			},
			{
				Name: "list",
				Methods: []Method{
					{
						Verb:           "GET",
					},
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/ListWorkflowExecutions",
						RequestSchema:  "workflow_list_request",
						ResponseSchema: "workflow_list_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:read",
						},
					},
				},
			},
			{
				Name: "history",
				Methods: []Method{
					{
						Verb:           "GET",
					},
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/GetWorkflowExecutionHistory",
						RequestSchema:  "workflow_history_request",
						ResponseSchema: "workflow_history_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:read",
						},
					},
				},
			},
			{
				Name: "terminate",
				Methods: []Method{
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/TerminateWorkflowExecution",
						RequestSchema:  "workflow_terminate_request",
						ResponseSchema: "workflow_terminate_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:execute",
						},
					},
				},
			},
			{
				Name: "cancel",
				Methods: []Method{
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/RequestCancelWorkflowExecution",
						RequestSchema:  "workflow_cancel_request",
						ResponseSchema: "workflow_cancel_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:execute",
						},
					},
				},
			},
			{
				Name: "signal",
				Methods: []Method{
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/SignalWorkflowExecution",
						RequestSchema:  "workflow_signal_request",
						ResponseSchema: "workflow_signal_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:signal",
						},
					},
				},
			},
			{
				Name: "query",
				Methods: []Method{
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/QueryWorkflow",
						RequestSchema:  "workflow_query_request",
						ResponseSchema: "workflow_query_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:query",
						},
					},
				},
			},
			{
				Name: "reset",
				Methods: []Method{
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/ResetWorkflowExecution",
						RequestSchema:  "workflow_reset_request",
						ResponseSchema: "workflow_reset_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:execute",
						},
					},
				},
			},
			{
				Name: "update",
				Methods: []Method{
					{
						Verb:           "POST",
						UpstreamPath:   "/temporal.workflowservice.v1.WorkflowService/UpdateWorkflowExecution",
						RequestSchema:  "workflow_update_request",
						ResponseSchema: "workflow_update_response",
						Auth: &Auth{
							Required:   true,
							Capability: "workflow:execute",
						},
					},
				},
			},
		},
	}
}