diff --git a/internal/temporal/handler.go b/internal/temporal/handler.go index 7e9d076..1a452ea 100644 --- a/internal/temporal/handler.go +++ b/internal/temporal/handler.go @@ -10,7 +10,6 @@ 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" @@ -38,6 +37,8 @@ type ResponsePayload struct { type Handler struct { hostPort string grpcClient *GRPCClient + ops *WorkflowGRPCImpl + searchOps *SearchAttributesGRPCImpl } // NewHandler creates a new Temporal HTTP handler @@ -51,10 +52,15 @@ func NewHandler(hostPort string) *Handler { log.Printf("WARNING: Failed to connect to Temporal at %s: %v", hostPort, err) } - return &Handler{ + 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 @@ -208,7 +214,7 @@ func (h *Handler) handleMetrics(w http.ResponseWriter, r *http.Request) { // ── Workflow operations ────────────────────────────────────────────────────── func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -223,31 +229,16 @@ func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload m 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) + 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 map[string]interface{}{ - "workflow_id": workflowID, - "run_id": resp.RunId, - "started_at": time.Now(), - }, "", "" + return result, "", "" } func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -255,25 +246,11 @@ func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payloa 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}, - }) + 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) } - - status := "UNKNOWN" - if resp.WorkflowExecutionInfo != nil { - status = resp.WorkflowExecutionInfo.Status.String() - } - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": runID, - "status": status, - }, "", "" + return result, "", "" } func (h *Handler) listWorkflows(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { @@ -319,7 +296,7 @@ func (h *Handler) listWorkflows(ctx context.Context, namespace string, payload m } func (h *Handler) getWorkflowHistory(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -327,79 +304,48 @@ func (h *Handler) getWorkflowHistory(ctx context.Context, namespace string, payl 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}, - }) + result, err := h.ops.GetWorkflowExecutionHistory(ctx, namespace, workflowID, 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), - }, "", "" + return result, "", "" } func (h *Handler) terminateWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + 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") - - _, err := h.grpcClient.GetWorkflowServiceStub().TerminateWorkflowExecution(ctx, - &workflowservice.TerminateWorkflowExecutionRequest{ - Namespace: namespace, - WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID}, - Reason: 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 map[string]interface{}{ - "workflow_id": workflowID, - "terminated_at": time.Now(), - }, "", "" + return result, "", "" } func (h *Handler) cancelWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == 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}, - }) + 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 map[string]interface{}{ - "workflow_id": workflowID, - "status": "canceling", - }, "", "" + return result, "", "" } func (h *Handler) signalWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -410,30 +356,17 @@ func (h *Handler) signalWorkflow(ctx context.Context, namespace string, payload 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) + 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 map[string]interface{}{ - "workflow_id": workflowID, - "signal_name": signalName, - "signaled_at": time.Now(), - }, "", "" + return result, "", "" } func (h *Handler) queryWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -444,22 +377,12 @@ func (h *Handler) queryWorkflow(ctx context.Context, namespace string, payload m 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}, - }) + 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) } - - 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}, "", "" + return result, "", "" } func (h *Handler) resetWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { diff --git a/internal/temporal/operations.go b/internal/temporal/operations.go deleted file mode 100644 index 43e82d8..0000000 --- a/internal/temporal/operations.go +++ /dev/null @@ -1,95 +0,0 @@ -// Package temporal provides operation wrappers for Temporal operations -package temporal - -import ( - "context" - "fmt" - "time" -) - -// OperationHandler handles Temporal operations -type OperationHandler struct { - grpc *GRPCClient -} - -// NewOperationHandler creates a new operation handler -func NewOperationHandler(grpcClient *GRPCClient) *OperationHandler { - return &OperationHandler{ - grpc: grpcClient, - } -} - -// StartWorkflowExecution starts a new workflow execution -func (oh *OperationHandler) StartWorkflowExecution(ctx context.Context, namespace, workflowID, workflowType, taskQueue string, input map[string]interface{}) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - // This is a placeholder for actual implementation - if oh.grpc == nil { - return nil, fmt.Errorf("gRPC client not initialized") - } - - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": fmt.Sprintf("run_%d", time.Now().UnixNano()), - "start_time": time.Now(), - }, nil -} - -// DescribeWorkflowExecution gets workflow details -func (oh *OperationHandler) DescribeWorkflowExecution(ctx context.Context, namespace, workflowID, runID string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": runID, - "status": "RUNNING", - "start_time": time.Now(), - }, nil -} - -// TerminateWorkflowExecution terminates a workflow -func (oh *OperationHandler) TerminateWorkflowExecution(ctx context.Context, namespace, workflowID, runID, reason string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": runID, - "terminated_at": time.Now(), - "reason": reason, - }, nil -} - -// CancelWorkflowExecution cancels a workflow -func (oh *OperationHandler) CancelWorkflowExecution(ctx context.Context, namespace, workflowID, runID string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "status": "canceling", - }, nil -} - -// SignalWorkflowExecution sends a signal to a workflow -func (oh *OperationHandler) SignalWorkflowExecution(ctx context.Context, namespace, workflowID, runID, signalName string, input map[string]interface{}) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "signal_name": signalName, - "signaled_at": time.Now(), - }, nil -} - -// QueryWorkflowExecution queries a workflow -func (oh *OperationHandler) QueryWorkflowExecution(ctx context.Context, namespace, workflowID, runID, queryType string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "query_type": queryType, - "query_result": map[string]interface{}{}, - }, nil -} - -// ListWorkflowExecutions lists workflows -func (oh *OperationHandler) ListWorkflowExecutions(ctx context.Context, namespace string, pageSize int32) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "executions": []interface{}{}, - "next_page_token": "", - }, nil -}