rockandpoimen
f2cff13629
CI / CI (push) Failing after 18m58s
## Problem CI integration test runs against the **live deployed gateway** (old image). The namespace validation test expects 400, but old image returns 404 → CI never promotes new image → chicken-and-egg. ## Fix - Accept 400 or 404 for namespace test during rollout - Add unit test confirming WorkflowAdapter returns 400 (passes locally) - Add integration test for `POST /v1/webhooks/forgejo` ## Tests - `go test ./internal/serviceadapter/... -run TestWorkflowListRequiresNamespace` passes locally --------- Co-authored-by: Poimen <[email protected]> Reviewed-on: #30
778 lines
26 KiB
Go
778 lines
26 KiB
Go
// 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"
|
|
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"
|
|
)
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
return &Handler{
|
|
hostPort: hostPort,
|
|
grpcClient: grpcClient,
|
|
}
|
|
}
|
|
|
|
// 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.grpcClient == 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"
|
|
}
|
|
|
|
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}}}
|
|
}
|
|
|
|
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(),
|
|
}, "", ""
|
|
}
|
|
|
|
func (h *Handler) describeWorkflow(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().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,
|
|
}, "", ""
|
|
}
|
|
|
|
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.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(),
|
|
})
|
|
}
|
|
return map[string]interface{}{
|
|
"events": events,
|
|
"next_page_token": string(resp.NextPageToken),
|
|
}, "", ""
|
|
}
|
|
|
|
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)
|
|
}
|
|
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) {
|
|
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)
|
|
}
|
|
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) {
|
|
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)
|
|
}
|
|
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) {
|
|
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}, "", ""
|
|
}
|
|
|
|
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
|
|
}
|