// 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" update "go.temporal.io/api/update/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 grpcClient *GRPCClient ops *WorkflowGRPCImpl searchOps *SearchAttributesGRPCImpl } // 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) } h := &Handler{ hostPort: hostPort, grpcClient: grpcClient, } if grpcClient != nil { h.ops = NewWorkflowGRPCImpl(grpcClient) h.searchOps = NewSearchAttributesGRPCImpl(grpcClient) } return h } // unavailable returns TEMPORAL_UNAVAILABLE when grpcClient is nil func (h *Handler) unavailable() (interface{}, string, string) { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("Temporal not reachable at %s", h.hostPort) } // 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") } } 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 } var req RequestPayload if err := json.NewDecoder(r.Body).Decode(&req); err != nil { w.WriteHeader(http.StatusBadRequest) h.writeError(w, "", "INVALID_REQUEST", "Failed to parse request body") return } 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() var result interface{} var errCode, errMsg string switch req.Action { 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) case "HEARTBEAT_ACTIVITY": result, errCode, errMsg = h.heartbeatActivity(ctx, req.Payload) case "COMPLETE_ACTIVITY": result, errCode, errMsg = h.completeActivity(ctx, req.Payload) case "FAIL_ACTIVITY": result, errCode, errMsg = h.failActivity(ctx, req.Payload) 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) case "LIST_SEARCH_ATTRIBUTES": result, errCode, errMsg = h.listSearchAttributes(ctx, req.Namespace) case "LIST_TASK_QUEUES": result, errCode, errMsg = h.listTaskQueues(ctx, req.Namespace, req.Payload) 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 } statusCode := http.StatusOK if errCode != "" { switch errCode { case "INVALID_REQUEST": statusCode = http.StatusBadRequest case "NOT_FOUND", "WORKFLOW_NOT_FOUND": statusCode = http.StatusNotFound case "ALREADY_EXISTS": statusCode = http.StatusConflict case "TEMPORAL_UNAVAILABLE": statusCode = http.StatusServiceUnavailable default: statusCode = http.StatusInternalServerError } } w.WriteHeader(statusCode) if errCode != "" { h.writeErrorWithCode(w, req.Action, req.Namespace, errCode, errMsg) } else { h.writeSuccess(w, req.Action, req.Namespace, result) } } func (h *Handler) handleHealth(w http.ResponseWriter, r *http.Request) { connected := h.grpcClient != nil status := "healthy" if !connected { status = "degraded" } w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(map[string]interface{}{ "status": status, "temporal_connected": connected, }) } func (h *Handler) handleMetrics(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte("# Temporal gateway metrics\n")) } // ── Workflow operations ────────────────────────────────────────────────────── func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.ops == nil { return h.unavailable() } 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") result, err := h.ops.StartWorkflowExecution(ctx, namespace, workflowID, workflowType, taskQueue, input) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to start workflow: %v", err) } return result, "", "" } func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") result, err := h.ops.DescribeWorkflowExecution(ctx, namespace, workflowID, runID) if err != nil { return nil, "WORKFLOW_NOT_FOUND", fmt.Sprintf("failed to describe workflow: %v", err) } return result, "", "" } func (h *Handler) listWorkflows(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } query := getString(payload, "query") var pageSize int32 = 100 resp, err := h.grpcClient.GetWorkflowServiceStub().ListWorkflowExecutions(ctx, &workflowservice.ListWorkflowExecutionsRequest{ Namespace: namespace, PageSize: pageSize, Query: query, }) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to list workflows: %v", err) } executions := make([]interface{}, 0, len(resp.Executions)) for _, e := range resp.Executions { entry := map[string]interface{}{ "workflow_id": e.Execution.GetWorkflowId(), "run_id": e.Execution.GetRunId(), "status": e.Status.String(), } if e.StartTime != nil { entry["start_time"] = e.StartTime.AsTime() } if e.CloseTime != nil { entry["close_time"] = e.CloseTime.AsTime() } if e.Type != nil { entry["workflow_type"] = e.Type.Name } executions = append(executions, entry) } return map[string]interface{}{ "executions": executions, "next_page_token": string(resp.NextPageToken), }, "", "" } func (h *Handler) getWorkflowHistory(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") result, err := h.ops.GetWorkflowExecutionHistory(ctx, namespace, workflowID, runID) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to get history: %v", err) } return result, "", "" } func (h *Handler) terminateWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") reason := getString(payload, "reason") result, err := h.ops.TerminateWorkflowExecution(ctx, namespace, workflowID, runID, reason) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to terminate workflow: %v", err) } return result, "", "" } func (h *Handler) cancelWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") result, err := h.ops.CancelWorkflowExecution(ctx, namespace, workflowID, runID) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to cancel workflow: %v", err) } return result, "", "" } func (h *Handler) signalWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.ops == nil { return h.unavailable() } 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" } runID := getString(payload, "run_id") signalData := getMap(payload, "signal_data") result, err := h.ops.SignalWorkflowExecution(ctx, namespace, workflowID, runID, signalName, signalData) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to signal workflow: %v", err) } return result, "", "" } func (h *Handler) queryWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.ops == nil { return h.unavailable() } 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" } runID := getString(payload, "run_id") result, err := h.ops.QueryWorkflowExecution(ctx, namespace, workflowID, runID, queryType) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to query workflow: %v", err) } return result, "", "" } func (h *Handler) resetWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") eventID, _ := payload["event_id"].(float64) resp, err := h.grpcClient.GetWorkflowServiceStub().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{ Namespace: namespace, WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID, RunId: runID}, WorkflowTaskFinishEventId: int64(eventID), }) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to reset workflow: %v", err) } return map[string]interface{}{ "workflow_id": workflowID, "new_run_id": resp.RunId, }, "", "" } func (h *Handler) updateWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } updateName := getString(payload, "update_name") if updateName == "" { return nil, "INVALID_REQUEST", "update_name is required" } _, err := h.grpcClient.GetWorkflowServiceStub().UpdateWorkflowExecution(ctx, &workflowservice.UpdateWorkflowExecutionRequest{ Namespace: namespace, WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID}, Request: &update.Request{}, }) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to update workflow: %v", err) } return map[string]interface{}{ "workflow_id": workflowID, "update_name": updateName, }, "", "" } // ── Activity operations ────────────────────────────────────────────────────── func (h *Handler) heartbeatActivity(ctx context.Context, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } taskToken := getString(payload, "task_token") if taskToken == "" { return nil, "INVALID_REQUEST", "task_token is required" } _, err := h.grpcClient.GetWorkflowServiceStub().RecordActivityTaskHeartbeat(ctx, &workflowservice.RecordActivityTaskHeartbeatRequest{ TaskToken: []byte(taskToken), }) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to heartbeat: %v", err) } return map[string]interface{}{"status": "heartbeat_recorded"}, "", "" } func (h *Handler) completeActivity(ctx context.Context, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } taskToken := getString(payload, "task_token") if taskToken == "" { return nil, "INVALID_REQUEST", "task_token is required" } _, err := h.grpcClient.GetWorkflowServiceStub().RespondActivityTaskCompleted(ctx, &workflowservice.RespondActivityTaskCompletedRequest{ TaskToken: []byte(taskToken), }) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to complete activity: %v", err) } return map[string]interface{}{"status": "activity_completed"}, "", "" } func (h *Handler) failActivity(ctx context.Context, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } taskToken := getString(payload, "task_token") if taskToken == "" { return nil, "INVALID_REQUEST", "task_token is required" } _, err := h.grpcClient.GetWorkflowServiceStub().RespondActivityTaskFailed(ctx, &workflowservice.RespondActivityTaskFailedRequest{ TaskToken: []byte(taskToken), }) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to fail activity: %v", err) } return map[string]interface{}{"status": "activity_failed"}, "", "" } // ── Namespace operations ───────────────────────────────────────────────────── func (h *Handler) listNamespaces(ctx context.Context) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } resp, err := h.grpcClient.GetWorkflowServiceStub().ListNamespaces(ctx, &workflowservice.ListNamespacesRequest{PageSize: 100}) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to list namespaces: %v", err) } names := make([]string, 0, len(resp.Namespaces)) for _, ns := range resp.Namespaces { if ns.NamespaceInfo != nil { names = append(names, ns.NamespaceInfo.Name) } } return map[string]interface{}{"namespaces": names}, "", "" } func (h *Handler) describeNamespace(ctx context.Context, namespace string) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } resp, err := h.grpcClient.GetWorkflowServiceStub().DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{Namespace: namespace}) if err != nil { return nil, "NOT_FOUND", fmt.Sprintf("failed to describe namespace: %v", err) } state := "" if resp.NamespaceInfo != nil { state = resp.NamespaceInfo.State.String() } return map[string]interface{}{ "name": namespace, "state": state, }, "", "" } func (h *Handler) createNamespace(ctx context.Context, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } name := getString(payload, "namespace_name") if name == "" { return nil, "INVALID_REQUEST", "namespace_name is required" } _, err := h.grpcClient.GetWorkflowServiceStub().RegisterNamespace(ctx, &workflowservice.RegisterNamespaceRequest{Namespace: name}) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to create namespace: %v", err) } return map[string]interface{}{"namespace": name, "status": "created"}, "", "" } func (h *Handler) updateNamespace(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } _, err := h.grpcClient.GetWorkflowServiceStub().UpdateNamespace(ctx, &workflowservice.UpdateNamespaceRequest{Namespace: namespace}) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to update namespace: %v", err) } return map[string]interface{}{"namespace": namespace, "status": "updated"}, "", "" } func (h *Handler) deleteNamespace(ctx context.Context, namespace string) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } _, err := h.grpcClient.GetWorkflowServiceStub().UpdateNamespace(ctx, &workflowservice.UpdateNamespaceRequest{Namespace: namespace}) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to delete namespace: %v", err) } return map[string]interface{}{"namespace": namespace, "status": "deleted"}, "", "" } // ── Search attributes ──────────────────────────────────────────────────────── func (h *Handler) listSearchAttributes(ctx context.Context, namespace string) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } resp, err := h.grpcClient.GetWorkflowServiceStub().GetSearchAttributes(ctx, &workflowservice.GetSearchAttributesRequest{}) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to list search attributes: %v", err) } attrs := make(map[string]string) for k, v := range resp.GetKeys() { attrs[k] = v.String() } return map[string]interface{}{"attributes": attrs}, "", "" } // ── Task queue operations ──────────────────────────────────────────────────── func (h *Handler) listTaskQueues(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } taskQueue := getString(payload, "task_queue") if taskQueue == "" { return nil, "INVALID_REQUEST", "task_queue is required" } resp, err := h.grpcClient.GetWorkflowServiceStub().ListTaskQueuePartitions(ctx, &workflowservice.ListTaskQueuePartitionsRequest{ Namespace: namespace, TaskQueue: &taskqueue.TaskQueue{Name: taskQueue}, }) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to list task queues: %v", err) } queues := make([]string, 0) for _, p := range resp.GetActivityTaskQueuePartitions() { queues = append(queues, p.GetKey()) } return map[string]interface{}{"queues": queues}, "", "" } // ── Cluster operations ─────────────────────────────────────────────────────── func (h *Handler) getClusterInfo(ctx context.Context) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } resp, err := h.grpcClient.GetWorkflowServiceStub().GetClusterInfo(ctx, &workflowservice.GetClusterInfoRequest{}) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to get cluster info: %v", err) } return map[string]interface{}{ "cluster_name": resp.GetClusterName(), "version_info": resp.GetVersionInfo(), }, "", "" } func (h *Handler) listClusterMembers(ctx context.Context) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } // ListClusterMembers is on the operator service resp, err := h.grpcClient.GetOperatorServiceStub().ListClusters(ctx, nil) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to list cluster members: %v", err) } return map[string]interface{}{"clusters": resp.GetClusters()}, "", "" } func (h *Handler) getSystemInfo(ctx context.Context) (interface{}, string, string) { if h.grpcClient == nil { return h.unavailable() } resp, err := h.grpcClient.GetWorkflowServiceStub().GetSystemInfo(ctx, &workflowservice.GetSystemInfoRequest{}) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to get system info: %v", err) } return map[string]interface{}{ "server_version": resp.GetServerVersion(), }, "", "" } // ── Helpers ────────────────────────────────────────────────────────────────── func (h *Handler) writeSuccess(w http.ResponseWriter, action, namespace string, data interface{}) { json.NewEncoder(w).Encode(ResponsePayload{ Success: true, Action: action, Namespace: namespace, Data: data, Timestamp: time.Now(), }) } func (h *Handler) writeError(w http.ResponseWriter, action, errorCode, message string) { json.NewEncoder(w).Encode(ResponsePayload{ Success: false, Action: action, Error: errorCode, Message: message, Timestamp: time.Now(), }) } func (h *Handler) writeErrorWithCode(w http.ResponseWriter, action, namespace, errorCode, message string) { json.NewEncoder(w).Encode(ResponsePayload{ Success: false, Action: action, Namespace: namespace, Error: errorCode, Message: message, Timestamp: time.Now(), }) } func getString(payload map[string]interface{}, key string) string { if val, ok := payload[key]; ok { if str, ok := val.(string); ok { return str } } return "" } 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 }