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, } } // 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 if _, ok := payload["namespace"]; !ok { payload["namespace"] = "default" } 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) } // GetWorkflowSpec 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", TimeoutSeconds: 30, }, Auth: Auth{ Required: true, 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", }, }, }, }, }, } }