|
|
|
@@ -10,9 +10,7 @@ import (
|
|
|
|
|
"time"
|
|
|
|
|
|
|
|
|
|
"go.temporal.io/api/common/v1"
|
|
|
|
|
query "go.temporal.io/api/query/v1"
|
|
|
|
|
"go.temporal.io/api/taskqueue/v1"
|
|
|
|
|
update "go.temporal.io/api/update/v1"
|
|
|
|
|
"go.temporal.io/api/workflowservice/v1"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
@@ -36,8 +34,8 @@ type ResponsePayload struct {
|
|
|
|
|
|
|
|
|
|
// Handler handles HTTP requests for Temporal operations
|
|
|
|
|
type Handler struct {
|
|
|
|
|
hostPort string
|
|
|
|
|
grpcClient *GRPCClient
|
|
|
|
|
hostPort string // e.g., "localhost:7233"
|
|
|
|
|
grpcClient *GRPCClient // gRPC connection to Temporal
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// NewHandler creates a new Temporal HTTP handler
|
|
|
|
@@ -49,6 +47,7 @@ func NewHandler(hostPort string) *Handler {
|
|
|
|
|
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{
|
|
|
|
@@ -57,14 +56,10 @@ func NewHandler(hostPort string) *Handler {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 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)
|
|
|
|
@@ -78,6 +73,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 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)
|
|
|
|
@@ -85,13 +81,15 @@ func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Parse request
|
|
|
|
|
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")
|
|
|
|
|
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")
|
|
|
|
@@ -105,10 +103,14 @@ func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
|
|
// Route to appropriate handler
|
|
|
|
|
var result interface{}
|
|
|
|
|
var errCode, errMsg string
|
|
|
|
|
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":
|
|
|
|
@@ -129,12 +131,16 @@ func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
|
|
|
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.Payload)
|
|
|
|
|
result, errCode, errMsg = h.heartbeatActivity(ctx, req.Namespace, req.Payload)
|
|
|
|
|
case "COMPLETE_ACTIVITY":
|
|
|
|
|
result, errCode, errMsg = h.completeActivity(ctx, req.Payload)
|
|
|
|
|
result, errCode, errMsg = h.completeActivity(ctx, req.Namespace, req.Payload)
|
|
|
|
|
case "FAIL_ACTIVITY":
|
|
|
|
|
result, errCode, errMsg = h.failActivity(ctx, req.Payload)
|
|
|
|
|
result, errCode, errMsg = h.failActivity(ctx, req.Namespace, req.Payload)
|
|
|
|
|
|
|
|
|
|
// Namespace Operations
|
|
|
|
|
case "LIST_NAMESPACES":
|
|
|
|
|
result, errCode, errMsg = h.listNamespaces(ctx)
|
|
|
|
|
case "DESCRIBE_NAMESPACE":
|
|
|
|
@@ -144,36 +150,48 @@ func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
|
|
|
case "UPDATE_NAMESPACE":
|
|
|
|
|
result, errCode, errMsg = h.updateNamespace(ctx, req.Namespace, req.Payload)
|
|
|
|
|
case "DELETE_NAMESPACE":
|
|
|
|
|
result, errCode, errMsg = h.deleteNamespace(ctx, req.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
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
statusCode := http.StatusOK
|
|
|
|
|
// Determine HTTP status code
|
|
|
|
|
statusCode = http.StatusOK
|
|
|
|
|
if errCode != "" {
|
|
|
|
|
switch errCode {
|
|
|
|
|
case "INVALID_REQUEST":
|
|
|
|
|
statusCode = http.StatusBadRequest
|
|
|
|
|
case "NOT_FOUND", "WORKFLOW_NOT_FOUND":
|
|
|
|
|
case "NOT_FOUND":
|
|
|
|
|
statusCode = http.StatusNotFound
|
|
|
|
|
case "ALREADY_EXISTS":
|
|
|
|
|
statusCode = http.StatusConflict
|
|
|
|
|
case "TEMPORAL_UNAVAILABLE":
|
|
|
|
|
statusCode = http.StatusServiceUnavailable
|
|
|
|
|
default:
|
|
|
|
|
case "INTERNAL_ERROR":
|
|
|
|
|
statusCode = http.StatusInternalServerError
|
|
|
|
|
default:
|
|
|
|
|
statusCode = http.StatusBadRequest
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
@@ -185,77 +203,143 @@ func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// handleHealth checks Temporal server health
|
|
|
|
|
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,
|
|
|
|
|
})
|
|
|
|
|
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 gateway metrics\n"))
|
|
|
|
|
w.Write([]byte("# Temporal Metrics\n# Prometheus endpoint\n"))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// ── Workflow operations ──────────────────────────────────────────────────────
|
|
|
|
|
// 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 h.unavailable()
|
|
|
|
|
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 input := getMap(payload, "input"); len(input) > 0 {
|
|
|
|
|
b, _ := json.Marshal(input)
|
|
|
|
|
req.Input = &common.Payloads{Payloads: []*common.Payload{{Data: b}}}
|
|
|
|
|
|
|
|
|
|
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,
|
|
|
|
|
"started_at": time.Now(),
|
|
|
|
|
"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 h.unavailable()
|
|
|
|
|
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{
|
|
|
|
|
resp, err := h.grpcClient.GetWorkflowServiceStub().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
|
|
|
|
|
Namespace: namespace,
|
|
|
|
|
Execution: &common.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
|
|
|
|
|
})
|
|
|
|
@@ -267,106 +351,43 @@ func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payloa
|
|
|
|
|
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) {
|
|
|
|
|
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)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Would call Temporal WorkflowService.ListWorkflowExecutions
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"executions": executions,
|
|
|
|
|
"next_page_token": string(resp.NextPageToken),
|
|
|
|
|
"executions": []interface{}{},
|
|
|
|
|
"next_page_token": "",
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (h *Handler) getWorkflowHistory(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")
|
|
|
|
|
|
|
|
|
|
resp, err := h.grpcClient.GetWorkflowServiceStub().GetWorkflowExecutionHistory(ctx,
|
|
|
|
|
&workflowservice.GetWorkflowExecutionHistoryRequest{
|
|
|
|
|
Namespace: namespace,
|
|
|
|
|
Execution: &common.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
|
|
|
|
|
})
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to get history: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
events := make([]interface{}, 0, len(resp.History.GetEvents()))
|
|
|
|
|
for _, e := range resp.History.GetEvents() {
|
|
|
|
|
events = append(events, map[string]interface{}{
|
|
|
|
|
"event_id": e.EventId,
|
|
|
|
|
"event_type": e.EventType.String(),
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal WorkflowService.GetWorkflowExecutionHistory
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"events": events,
|
|
|
|
|
"next_page_token": string(resp.NextPageToken),
|
|
|
|
|
"events": []interface{}{},
|
|
|
|
|
"next_page_token": "",
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (h *Handler) terminateWorkflow(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"
|
|
|
|
|
}
|
|
|
|
|
reason := getString(payload, "reason")
|
|
|
|
|
|
|
|
|
|
_, err := h.grpcClient.GetWorkflowServiceStub().TerminateWorkflowExecution(ctx,
|
|
|
|
|
&workflowservice.TerminateWorkflowExecutionRequest{
|
|
|
|
|
Namespace: namespace,
|
|
|
|
|
WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID},
|
|
|
|
|
Reason: reason,
|
|
|
|
|
})
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to terminate workflow: %v", err)
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal WorkflowService.TerminateWorkflowExecution
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"workflow_id": workflowID,
|
|
|
|
|
"terminated_at": time.Now(),
|
|
|
|
@@ -374,22 +395,12 @@ func (h *Handler) terminateWorkflow(ctx context.Context, namespace string, paylo
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (h *Handler) cancelWorkflow(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"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
_, err := h.grpcClient.GetWorkflowServiceStub().RequestCancelWorkflowExecution(ctx,
|
|
|
|
|
&workflowservice.RequestCancelWorkflowExecutionRequest{
|
|
|
|
|
Namespace: namespace,
|
|
|
|
|
WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID},
|
|
|
|
|
})
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to cancel workflow: %v", err)
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal WorkflowService.RequestCancelWorkflowExecution
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"workflow_id": workflowID,
|
|
|
|
|
"status": "canceling",
|
|
|
|
@@ -397,32 +408,17 @@ func (h *Handler) cancelWorkflow(ctx context.Context, namespace string, payload
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (h *Handler) signalWorkflow(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"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
signalName := getString(payload, "signal_name")
|
|
|
|
|
if signalName == "" {
|
|
|
|
|
return nil, "INVALID_REQUEST", "signal_name is required"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
req := &workflowservice.SignalWorkflowExecutionRequest{
|
|
|
|
|
Namespace: namespace,
|
|
|
|
|
WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID},
|
|
|
|
|
SignalName: signalName,
|
|
|
|
|
}
|
|
|
|
|
if data := getMap(payload, "signal_data"); len(data) > 0 {
|
|
|
|
|
b, _ := json.Marshal(data)
|
|
|
|
|
req.Input = &common.Payloads{Payloads: []*common.Payload{{Data: b}}}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
_, err := h.grpcClient.GetWorkflowServiceStub().SignalWorkflowExecution(ctx, req)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to signal workflow: %v", err)
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal WorkflowService.SignalWorkflowExecution
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"workflow_id": workflowID,
|
|
|
|
|
"signal_name": signalName,
|
|
|
|
@@ -431,347 +427,182 @@ func (h *Handler) signalWorkflow(ctx context.Context, namespace string, payload
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (h *Handler) queryWorkflow(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"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
queryType := getString(payload, "query_type")
|
|
|
|
|
if queryType == "" {
|
|
|
|
|
return nil, "INVALID_REQUEST", "query_type is required"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
resp, err := h.grpcClient.GetWorkflowServiceStub().QueryWorkflow(ctx,
|
|
|
|
|
&workflowservice.QueryWorkflowRequest{
|
|
|
|
|
Namespace: namespace,
|
|
|
|
|
Execution: &common.WorkflowExecution{WorkflowId: workflowID},
|
|
|
|
|
Query: &query.WorkflowQuery{QueryType: queryType},
|
|
|
|
|
})
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to query workflow: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
var result interface{}
|
|
|
|
|
if resp.QueryResult != nil && len(resp.QueryResult.Payloads) > 0 {
|
|
|
|
|
json.Unmarshal(resp.QueryResult.Payloads[0].Data, &result)
|
|
|
|
|
}
|
|
|
|
|
return map[string]interface{}{"query_result": result}, "", ""
|
|
|
|
|
// 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) {
|
|
|
|
|
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)
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal WorkflowService.ResetWorkflowExecution
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"workflow_id": workflowID,
|
|
|
|
|
"new_run_id": resp.RunId,
|
|
|
|
|
"reset_at": time.Now(),
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
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)
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal WorkflowService.UpdateWorkflowExecution
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"workflow_id": workflowID,
|
|
|
|
|
"update_name": updateName,
|
|
|
|
|
"status": "pending",
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// ── Activity operations ──────────────────────────────────────────────────────
|
|
|
|
|
// Activity Operations
|
|
|
|
|
|
|
|
|
|
func (h *Handler) heartbeatActivity(ctx context.Context, payload map[string]interface{}) (interface{}, string, string) {
|
|
|
|
|
if h.grpcClient == nil {
|
|
|
|
|
return h.unavailable()
|
|
|
|
|
}
|
|
|
|
|
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"
|
|
|
|
|
}
|
|
|
|
|
_, 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"}, "", ""
|
|
|
|
|
|
|
|
|
|
// Would call Temporal WorkflowService.RecordActivityTaskHeartbeat
|
|
|
|
|
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()
|
|
|
|
|
}
|
|
|
|
|
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"
|
|
|
|
|
}
|
|
|
|
|
_, 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"}, "", ""
|
|
|
|
|
|
|
|
|
|
// Would call Temporal WorkflowService.RespondActivityTaskCompleted
|
|
|
|
|
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()
|
|
|
|
|
}
|
|
|
|
|
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"
|
|
|
|
|
}
|
|
|
|
|
_, 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"}, "", ""
|
|
|
|
|
|
|
|
|
|
// Would call Temporal WorkflowService.RespondActivityTaskFailed
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"status": "activity_failed",
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// ── Namespace operations ─────────────────────────────────────────────────────
|
|
|
|
|
// 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}, "", ""
|
|
|
|
|
// Would call Temporal OperatorService.ListNamespaces
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"namespaces": []interface{}{},
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
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()
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal OperatorService.DescribeNamespace
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"name": namespace,
|
|
|
|
|
"state": state,
|
|
|
|
|
"state": "ACTIVE",
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
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 == "" {
|
|
|
|
|
namespaceName := getString(payload, "namespace_name")
|
|
|
|
|
if namespaceName == "" {
|
|
|
|
|
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"}, "", ""
|
|
|
|
|
|
|
|
|
|
// 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) {
|
|
|
|
|
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"}, "", ""
|
|
|
|
|
// Would call Temporal OperatorService.UpdateNamespace
|
|
|
|
|
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"}, "", ""
|
|
|
|
|
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 ────────────────────────────────────────────────────────
|
|
|
|
|
// Search Attributes Operations
|
|
|
|
|
|
|
|
|
|
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}, "", ""
|
|
|
|
|
// Would call Temporal OperatorService.ListSearchAttributes
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"attributes": map[string]string{},
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// ── Task queue operations ────────────────────────────────────────────────────
|
|
|
|
|
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) {
|
|
|
|
|
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}, "", ""
|
|
|
|
|
// Would call Temporal OperatorService.ListTaskQueuePartitions
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"queues": []interface{}{},
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// ── Cluster operations ───────────────────────────────────────────────────────
|
|
|
|
|
// 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)
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal OperatorService.GetClusterInfo
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"cluster_name": resp.GetClusterName(),
|
|
|
|
|
"version_info": resp.GetVersionInfo(),
|
|
|
|
|
"cluster_name": "temporal-cluster",
|
|
|
|
|
"version": "1.24.0",
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
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)
|
|
|
|
|
}
|
|
|
|
|
// Would call Temporal OperatorService.ListClusterMembers
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"server_version": resp.GetServerVersion(),
|
|
|
|
|
"members": []interface{}{},
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// ── 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
|
|
|
|
|
func (h *Handler) getSystemInfo(ctx context.Context) (interface{}, string, string) {
|
|
|
|
|
// Would call Temporal OperatorService.GetSystemInfo
|
|
|
|
|
return map[string]interface{}{
|
|
|
|
|
"server_version": "1.24.0",
|
|
|
|
|
}, "", ""
|
|
|
|
|
}
|
|
|
|
|