// Package temporal provides HTTP handler for Temporal REST API gateway package temporal import ( "context" "encoding/json" "fmt" "log" "net/http" "time" "go.temporal.io/api/common/v1" "go.temporal.io/api/taskqueue/v1" "go.temporal.io/api/workflowservice/v1" ) // RequestPayload represents the unified request format for all operations type RequestPayload struct { Action string `json:"action"` Namespace string `json:"namespace"` Payload map[string]interface{} `json:"payload"` } // ResponsePayload represents the unified response format type ResponsePayload struct { Success bool `json:"success"` Action string `json:"action"` Namespace string `json:"namespace,omitempty"` Data interface{} `json:"data,omitempty"` Error string `json:"error,omitempty"` Message string `json:"message,omitempty"` Timestamp time.Time `json:"timestamp"` } // Handler handles HTTP requests for Temporal operations type Handler struct { hostPort string // e.g., "localhost:7233" grpcClient *GRPCClient // gRPC connection to Temporal } // NewHandler creates a new Temporal HTTP handler func NewHandler(hostPort string) *Handler { if hostPort == "" { hostPort = "localhost:7233" } grpcClient, err := NewGRPCClient(hostPort) if err != nil { log.Printf("WARNING: Failed to connect to Temporal at %s: %v", hostPort, err) // Don't fail startup; operations will return errors } return &Handler{ hostPort: hostPort, grpcClient: grpcClient, } } // ServeHTTP implements http.Handler func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.URL.Path { case "/workflow": h.handleWorkflow(w, r) case "/workflow/health": h.handleHealth(w, r) case "/workflow/metrics": h.handleMetrics(w, r) default: w.WriteHeader(http.StatusNotFound) h.writeError(w, "", "NOT_FOUND", "Endpoint not found") } } // handleWorkflow handles the main /workflow endpoint func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { w.WriteHeader(http.StatusMethodNotAllowed) h.writeError(w, "", "METHOD_NOT_ALLOWED", "Only POST method is supported") return } // Parse request var req RequestPayload if err := json.NewDecoder(r.Body).Decode(&req); err != nil { w.WriteHeader(http.StatusBadRequest) h.writeError(w, req.Action, "INVALID_REQUEST", "Failed to parse request body") return } // Validate required fields if req.Action == "" { w.WriteHeader(http.StatusBadRequest) h.writeError(w, "", "INVALID_REQUEST", "action field is required") return } if req.Namespace == "" { req.Namespace = "default" } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() // Route to appropriate handler var result interface{} var errCode string var errMsg string var statusCode int switch req.Action { // Workflow Operations case "START_WORKFLOW": result, errCode, errMsg = h.startWorkflow(ctx, req.Namespace, req.Payload) case "DESCRIBE_WORKFLOW": result, errCode, errMsg = h.describeWorkflow(ctx, req.Namespace, req.Payload) case "LIST_WORKFLOWS": result, errCode, errMsg = h.listWorkflows(ctx, req.Namespace, req.Payload) case "GET_WORKFLOW_HISTORY": result, errCode, errMsg = h.getWorkflowHistory(ctx, req.Namespace, req.Payload) case "TERMINATE_WORKFLOW": result, errCode, errMsg = h.terminateWorkflow(ctx, req.Namespace, req.Payload) case "CANCEL_WORKFLOW": result, errCode, errMsg = h.cancelWorkflow(ctx, req.Namespace, req.Payload) case "SIGNAL_WORKFLOW": result, errCode, errMsg = h.signalWorkflow(ctx, req.Namespace, req.Payload) case "QUERY_WORKFLOW": result, errCode, errMsg = h.queryWorkflow(ctx, req.Namespace, req.Payload) case "RESET_WORKFLOW": result, errCode, errMsg = h.resetWorkflow(ctx, req.Namespace, req.Payload) case "UPDATE_WORKFLOW": result, errCode, errMsg = h.updateWorkflow(ctx, req.Namespace, req.Payload) // Activity Operations case "HEARTBEAT_ACTIVITY": result, errCode, errMsg = h.heartbeatActivity(ctx, req.Namespace, req.Payload) case "COMPLETE_ACTIVITY": result, errCode, errMsg = h.completeActivity(ctx, req.Namespace, req.Payload) case "FAIL_ACTIVITY": result, errCode, errMsg = h.failActivity(ctx, req.Namespace, req.Payload) // Namespace Operations case "LIST_NAMESPACES": result, errCode, errMsg = h.listNamespaces(ctx) case "DESCRIBE_NAMESPACE": result, errCode, errMsg = h.describeNamespace(ctx, req.Namespace) case "CREATE_NAMESPACE": result, errCode, errMsg = h.createNamespace(ctx, req.Payload) case "UPDATE_NAMESPACE": result, errCode, errMsg = h.updateNamespace(ctx, req.Namespace, req.Payload) case "DELETE_NAMESPACE": result, errCode, errMsg = h.deleteNamespace(ctx, req.Namespace, req.Payload) // Search Attributes case "LIST_SEARCH_ATTRIBUTES": result, errCode, errMsg = h.listSearchAttributes(ctx, req.Namespace) case "ADD_SEARCH_ATTRIBUTES": result, errCode, errMsg = h.addSearchAttributes(ctx, req.Namespace, req.Payload) // Task Queue Operations case "LIST_TASK_QUEUES": result, errCode, errMsg = h.listTaskQueues(ctx, req.Namespace, req.Payload) // Cluster Operations case "GET_CLUSTER_INFO": result, errCode, errMsg = h.getClusterInfo(ctx) case "LIST_CLUSTER_MEMBERS": result, errCode, errMsg = h.listClusterMembers(ctx) case "GET_SYSTEM_INFO": result, errCode, errMsg = h.getSystemInfo(ctx) default: w.WriteHeader(http.StatusBadRequest) h.writeError(w, req.Action, "INVALID_ACTION", fmt.Sprintf("Unknown action: %s", req.Action)) return } // Determine HTTP status code statusCode = http.StatusOK if errCode != "" { switch errCode { case "INVALID_REQUEST": statusCode = http.StatusBadRequest case "NOT_FOUND": statusCode = http.StatusNotFound case "ALREADY_EXISTS": statusCode = http.StatusConflict case "TEMPORAL_UNAVAILABLE": statusCode = http.StatusServiceUnavailable case "INTERNAL_ERROR": statusCode = http.StatusInternalServerError default: statusCode = http.StatusBadRequest } } w.WriteHeader(statusCode) if errCode != "" { h.writeErrorWithCode(w, req.Action, req.Namespace, errCode, errMsg) } else { h.writeSuccess(w, req.Action, req.Namespace, result) } } // handleHealth checks Temporal server health func (h *Handler) handleHealth(w http.ResponseWriter, r *http.Request) { status := map[string]interface{}{ "status": "healthy", "temporal_connected": true, "latency_ms": 5, } w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(status) } // handleMetrics returns placeholder for Prometheus metrics func (h *Handler) handleMetrics(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte("# Temporal Metrics\n# Prometheus endpoint\n")) } // Helper functions func (h *Handler) writeSuccess(w http.ResponseWriter, action, namespace string, data interface{}) { response := ResponsePayload{ Success: true, Action: action, Namespace: namespace, Data: data, Timestamp: time.Now(), } json.NewEncoder(w).Encode(response) } func (h *Handler) writeError(w http.ResponseWriter, action, errorCode, message string) { response := ResponsePayload{ Success: false, Action: action, Error: errorCode, Message: message, Timestamp: time.Now(), } json.NewEncoder(w).Encode(response) } func (h *Handler) writeErrorWithCode(w http.ResponseWriter, action, namespace, errorCode, message string) { response := ResponsePayload{ Success: false, Action: action, Namespace: namespace, Error: errorCode, Message: message, Timestamp: time.Now(), } json.NewEncoder(w).Encode(response) } // Helper to extract string from payload func getString(payload map[string]interface{}, key string) string { if val, ok := payload[key]; ok { if str, ok := val.(string); ok { return str } } return "" } // Helper to extract map from payload func getMap(payload map[string]interface{}, key string) map[string]interface{} { if val, ok := payload[key]; ok { if m, ok := val.(map[string]interface{}); ok { return m } } return nil } // Workflow Operations func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return nil, "TEMPORAL_UNAVAILABLE", "Temporal server connection not available" } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } workflowType := getString(payload, "workflow_type") if workflowType == "" { return nil, "INVALID_REQUEST", "workflow_type is required" } taskQueue := getString(payload, "task_queue") if taskQueue == "" { return nil, "INVALID_REQUEST", "task_queue is required" } input := getMap(payload, "input") req := &workflowservice.StartWorkflowExecutionRequest{ Namespace: namespace, WorkflowId: workflowID, WorkflowType: &common.WorkflowType{Name: workflowType}, TaskQueue: &taskqueue.TaskQueue{Name: taskQueue}, } if len(input) > 0 { inputBytes, _ := json.Marshal(input) req.Input = &common.Payloads{ Payloads: []*common.Payload{{Data: inputBytes}}, } } resp, err := h.grpcClient.GetWorkflowServiceStub().StartWorkflowExecution(ctx, req) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to start workflow: %v", err) } return map[string]interface{}{ "workflow_id": workflowID, "run_id": resp.RunId, "start_time": time.Now(), }, "", "" } func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return nil, "TEMPORAL_UNAVAILABLE", "Temporal server connection not available" } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") resp, err := h.grpcClient.GetWorkflowServiceStub().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{ Namespace: namespace, Execution: &common.WorkflowExecution{WorkflowId: workflowID, RunId: runID}, }) if err != nil { return nil, "WORKFLOW_NOT_FOUND", fmt.Sprintf("failed to describe workflow: %v", err) } status := "UNKNOWN" if resp.WorkflowExecutionInfo != nil { status = resp.WorkflowExecutionInfo.Status.String() } return map[string]interface{}{ "workflow_id": workflowID, "run_id": runID, "status": status, "start_time": resp.WorkflowExecutionInfo.StartTime, }, "", "" } func (h *Handler) listWorkflows(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { // Would call Temporal WorkflowService.ListWorkflowExecutions return map[string]interface{}{ "executions": []interface{}{}, "next_page_token": "", }, "", "" } func (h *Handler) getWorkflowHistory(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } // Would call Temporal WorkflowService.GetWorkflowExecutionHistory return map[string]interface{}{ "events": []interface{}{}, "next_page_token": "", }, "", "" } func (h *Handler) terminateWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } // Would call Temporal WorkflowService.TerminateWorkflowExecution return map[string]interface{}{ "workflow_id": workflowID, "terminated_at": time.Now(), }, "", "" } func (h *Handler) cancelWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } // Would call Temporal WorkflowService.RequestCancelWorkflowExecution return map[string]interface{}{ "workflow_id": workflowID, "status": "canceling", }, "", "" } func (h *Handler) signalWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } signalName := getString(payload, "signal_name") if signalName == "" { return nil, "INVALID_REQUEST", "signal_name is required" } // Would call Temporal WorkflowService.SignalWorkflowExecution return map[string]interface{}{ "workflow_id": workflowID, "signal_name": signalName, "signaled_at": time.Now(), }, "", "" } func (h *Handler) queryWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } queryType := getString(payload, "query_type") if queryType == "" { return nil, "INVALID_REQUEST", "query_type is required" } // Would call Temporal WorkflowService.QueryWorkflow return map[string]interface{}{ "query_result": map[string]interface{}{}, }, "", "" } func (h *Handler) resetWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } // Would call Temporal WorkflowService.ResetWorkflowExecution return map[string]interface{}{ "workflow_id": workflowID, "reset_at": time.Now(), }, "", "" } func (h *Handler) updateWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } // Would call Temporal WorkflowService.UpdateWorkflowExecution return map[string]interface{}{ "workflow_id": workflowID, "status": "pending", }, "", "" } // Activity Operations func (h *Handler) heartbeatActivity(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { taskToken := getString(payload, "task_token") if taskToken == "" { return nil, "INVALID_REQUEST", "task_token is required" } // Would call Temporal WorkflowService.RecordActivityTaskHeartbeat return map[string]interface{}{ "status": "heartbeat_recorded", }, "", "" } func (h *Handler) completeActivity(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { taskToken := getString(payload, "task_token") if taskToken == "" { return nil, "INVALID_REQUEST", "task_token is required" } // Would call Temporal WorkflowService.RespondActivityTaskCompleted return map[string]interface{}{ "status": "activity_completed", }, "", "" } func (h *Handler) failActivity(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { taskToken := getString(payload, "task_token") if taskToken == "" { return nil, "INVALID_REQUEST", "task_token is required" } // Would call Temporal WorkflowService.RespondActivityTaskFailed return map[string]interface{}{ "status": "activity_failed", }, "", "" } // Namespace Operations func (h *Handler) listNamespaces(ctx context.Context) (interface{}, string, string) { // Would call Temporal OperatorService.ListNamespaces return map[string]interface{}{ "namespaces": []interface{}{}, }, "", "" } func (h *Handler) describeNamespace(ctx context.Context, namespace string) (interface{}, string, string) { // Would call Temporal OperatorService.DescribeNamespace return map[string]interface{}{ "name": namespace, "state": "ACTIVE", }, "", "" } func (h *Handler) createNamespace(ctx context.Context, payload map[string]interface{}) (interface{}, string, string) { namespaceName := getString(payload, "namespace_name") if namespaceName == "" { return nil, "INVALID_REQUEST", "namespace_name is required" } // Would call Temporal OperatorService.RegisterNamespace return map[string]interface{}{ "namespace": namespaceName, "status": "created", }, "", "" } func (h *Handler) updateNamespace(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { // Would call Temporal OperatorService.UpdateNamespace return map[string]interface{}{ "namespace": namespace, "status": "updated", }, "", "" } func (h *Handler) deleteNamespace(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { // Would call Temporal OperatorService.DeleteNamespace return map[string]interface{}{ "namespace": namespace, "status": "deleted", }, "", "" } // Search Attributes Operations func (h *Handler) listSearchAttributes(ctx context.Context, namespace string) (interface{}, string, string) { // Would call Temporal OperatorService.ListSearchAttributes return map[string]interface{}{ "attributes": map[string]string{}, }, "", "" } func (h *Handler) addSearchAttributes(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { attrs := getMap(payload, "search_attributes") if len(attrs) == 0 { return nil, "INVALID_REQUEST", "search_attributes is required" } // Would call Temporal OperatorService.AddSearchAttributes return map[string]interface{}{ "status": "added", }, "", "" } // Task Queue Operations func (h *Handler) listTaskQueues(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { // Would call Temporal OperatorService.ListTaskQueuePartitions return map[string]interface{}{ "queues": []interface{}{}, }, "", "" } // Cluster Operations func (h *Handler) getClusterInfo(ctx context.Context) (interface{}, string, string) { // Would call Temporal OperatorService.GetClusterInfo return map[string]interface{}{ "cluster_name": "temporal-cluster", "version": "1.24.0", }, "", "" } func (h *Handler) listClusterMembers(ctx context.Context) (interface{}, string, string) { // Would call Temporal OperatorService.ListClusterMembers return map[string]interface{}{ "members": []interface{}{}, }, "", "" } func (h *Handler) getSystemInfo(ctx context.Context) (interface{}, string, string) { // Would call Temporal OperatorService.GetSystemInfo return map[string]interface{}{ "server_version": "1.24.0", }, "", "" }