// Package temporal provides gRPC implementations for Temporal operations package temporal import ( "context" "encoding/json" "fmt" "time" "google.golang.org/protobuf/types/known/durationpb" "go.temporal.io/api/common/v1" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/api/operatorservice/v1" "go.temporal.io/api/taskqueue/v1" "go.temporal.io/api/query/v1" enumsv1 "go.temporal.io/api/enums/v1" ) // WorkflowGRPCImpl provides gRPC implementations for workflow operations type WorkflowGRPCImpl struct { grpc *GRPCClient } // NewWorkflowGRPCImpl creates a new workflow gRPC implementation func NewWorkflowGRPCImpl(grpcClient *GRPCClient) *WorkflowGRPCImpl { return &WorkflowGRPCImpl{grpc: grpcClient} } // StartWorkflowExecution starts a workflow via gRPC func (w *WorkflowGRPCImpl) StartWorkflowExecution(ctx context.Context, namespace, workflowID, workflowType, taskQueueName string, input map[string]interface{}) (map[string]interface{}, error) { inputBytes, _ := json.Marshal(input) req := &workflowservice.StartWorkflowExecutionRequest{ Namespace: namespace, WorkflowId: workflowID, WorkflowType: &common.WorkflowType{Name: workflowType}, TaskQueue: &taskqueue.TaskQueue{Name: taskQueueName}, WorkflowExecutionTimeout: durationpb.New(24 * time.Hour), WorkflowRunTimeout: durationpb.New(24 * time.Hour), WorkflowTaskTimeout: durationpb.New(10 * time.Minute), Input: &common.Payloads{ Payloads: []*common.Payload{ { Data: inputBytes, Metadata: map[string][]byte{ "encoding": []byte("json/plain"), }, }, }, }, } resp, err := w.grpc.GetWorkflowServiceStub().StartWorkflowExecution(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC StartWorkflowExecution failed: %w", err) } return map[string]interface{}{ "workflow_id": workflowID, "run_id": resp.RunId, "started_at": time.Now(), }, nil } // DescribeWorkflowExecution gets workflow details via gRPC func (w *WorkflowGRPCImpl) DescribeWorkflowExecution(ctx context.Context, namespace, workflowID, runID string) (map[string]interface{}, error) { req := &workflowservice.DescribeWorkflowExecutionRequest{ Namespace: namespace, Execution: &common.WorkflowExecution{ WorkflowId: workflowID, RunId: runID, }, } resp, err := w.grpc.GetWorkflowServiceStub().DescribeWorkflowExecution(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC DescribeWorkflowExecution failed: %w", err) } info := resp.WorkflowExecutionInfo if info == nil { return nil, fmt.Errorf("workflow execution info not found") } return map[string]interface{}{ "workflow_id": workflowID, "run_id": runID, "workflow_type": info.Type.Name, "status": info.Status.String(), "start_time": info.StartTime.AsTime(), "close_time": info.CloseTime.AsTime(), "history_length": info.HistoryLength, "task_queue": info.TaskQueue, }, nil } // TerminateWorkflowExecution terminates a workflow via gRPC func (w *WorkflowGRPCImpl) TerminateWorkflowExecution(ctx context.Context, namespace, workflowID, runID, reason string) (map[string]interface{}, error) { req := &workflowservice.TerminateWorkflowExecutionRequest{ Namespace: namespace, WorkflowExecution: &common.WorkflowExecution{ WorkflowId: workflowID, RunId: runID, }, Reason: reason, } _, err := w.grpc.GetWorkflowServiceStub().TerminateWorkflowExecution(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC TerminateWorkflowExecution failed: %w", err) } return map[string]interface{}{ "workflow_id": workflowID, "run_id": runID, "status": "TERMINATED", "terminated_at": time.Now(), "reason": reason, }, nil } // CancelWorkflowExecution cancels a workflow via gRPC func (w *WorkflowGRPCImpl) CancelWorkflowExecution(ctx context.Context, namespace, workflowID, runID string) (map[string]interface{}, error) { req := &workflowservice.RequestCancelWorkflowExecutionRequest{ Namespace: namespace, WorkflowExecution: &common.WorkflowExecution{ WorkflowId: workflowID, RunId: runID, }, } _, err := w.grpc.GetWorkflowServiceStub().RequestCancelWorkflowExecution(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC RequestCancelWorkflowExecution failed: %w", err) } return map[string]interface{}{ "workflow_id": workflowID, "run_id": runID, "status": "CANCEL_REQUESTED", "cancelled_at": time.Now(), }, nil } // SignalWorkflowExecution sends a signal to a workflow via gRPC func (w *WorkflowGRPCImpl) SignalWorkflowExecution(ctx context.Context, namespace, workflowID, runID, signalName string, input map[string]interface{}) (map[string]interface{}, error) { inputBytes, _ := json.Marshal(input) req := &workflowservice.SignalWorkflowExecutionRequest{ Namespace: namespace, WorkflowExecution: &common.WorkflowExecution{ WorkflowId: workflowID, RunId: runID, }, SignalName: signalName, Input: &common.Payloads{ Payloads: []*common.Payload{ { Data: inputBytes, Metadata: map[string][]byte{ "encoding": []byte("json/plain"), }, }, }, }, } _, err := w.grpc.GetWorkflowServiceStub().SignalWorkflowExecution(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC SignalWorkflowExecution failed: %w", err) } return map[string]interface{}{ "workflow_id": workflowID, "run_id": runID, "signal_name": signalName, "signaled_at": time.Now(), }, nil } // QueryWorkflowExecution queries a workflow via gRPC func (w *WorkflowGRPCImpl) QueryWorkflowExecution(ctx context.Context, namespace, workflowID, runID, queryType string) (map[string]interface{}, error) { req := &workflowservice.QueryWorkflowRequest{ Namespace: namespace, Execution: &common.WorkflowExecution{ WorkflowId: workflowID, RunId: runID, }, Query: &query.WorkflowQuery{ QueryType: queryType, }, } resp, err := w.grpc.GetWorkflowServiceStub().QueryWorkflow(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC QueryWorkflow failed: %w", err) } var queryResult interface{} = nil if resp.QueryResult != nil && len(resp.QueryResult.Payloads) > 0 { json.Unmarshal(resp.QueryResult.Payloads[0].Data, &queryResult) } return map[string]interface{}{ "workflow_id": workflowID, "run_id": runID, "query_type": queryType, "query_result": queryResult, "queried_at": time.Now(), }, nil } // ListWorkflowExecutions lists workflows via gRPC func (w *WorkflowGRPCImpl) ListWorkflowExecutions(ctx context.Context, namespace string, pageSize int32) (map[string]interface{}, error) { if pageSize <= 0 { pageSize = 10 } req := &workflowservice.ListWorkflowExecutionsRequest{ Namespace: namespace, PageSize: pageSize, Query: "ExecutionStatus != 'CLOSED'", } resp, err := w.grpc.GetWorkflowServiceStub().ListWorkflowExecutions(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC ListWorkflowExecutions failed: %w", err) } executions := make([]map[string]interface{}, len(resp.Executions)) for i, exec := range resp.Executions { executions[i] = map[string]interface{}{ "workflow_id": exec.Execution.WorkflowId, "run_id": exec.Execution.RunId, "type": exec.Type.Name, "status": exec.Status.String(), "start_time": exec.StartTime.AsTime(), } } return map[string]interface{}{ "executions": executions, "count": len(executions), "next_page_token": string(resp.NextPageToken), }, nil } // GetWorkflowExecutionHistory gets workflow history via gRPC func (w *WorkflowGRPCImpl) GetWorkflowExecutionHistory(ctx context.Context, namespace, workflowID, runID string) (map[string]interface{}, error) { req := &workflowservice.GetWorkflowExecutionHistoryRequest{ Namespace: namespace, Execution: &common.WorkflowExecution{ WorkflowId: workflowID, RunId: runID, }, } resp, err := w.grpc.GetWorkflowServiceStub().GetWorkflowExecutionHistory(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC GetWorkflowExecutionHistory failed: %w", err) } events := make([]map[string]interface{}, len(resp.History.Events)) for i, event := range resp.History.Events { events[i] = map[string]interface{}{ "event_id": event.EventId, "type": event.EventType.String(), "timestamp": event.EventTime.AsTime(), } } return map[string]interface{}{ "workflow_id": workflowID, "run_id": runID, "events": events, "event_count": len(events), }, nil } // SearchAttributesGRPCImpl provides gRPC implementations for search attributes type SearchAttributesGRPCImpl struct { grpc *GRPCClient } // NewSearchAttributesGRPCImpl creates a new search attributes gRPC implementation func NewSearchAttributesGRPCImpl(grpcClient *GRPCClient) *SearchAttributesGRPCImpl { return &SearchAttributesGRPCImpl{grpc: grpcClient} } // ListSearchAttributes lists search attributes via gRPC func (s *SearchAttributesGRPCImpl) ListSearchAttributes(ctx context.Context) (map[string]interface{}, error) { req := &operatorservice.ListSearchAttributesRequest{} resp, err := s.grpc.GetOperatorServiceStub().ListSearchAttributes(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC ListSearchAttributes failed: %w", err) } attributes := make(map[string]interface{}) for name, attrType := range resp.CustomAttributes { attributes[name] = attrType.String() } return map[string]interface{}{ "custom_attributes": attributes, "count": len(attributes), }, nil } // AddSearchAttributes adds search attributes via gRPC func (s *SearchAttributesGRPCImpl) AddSearchAttributes(ctx context.Context, attributes map[string]interface{}) (map[string]interface{}, error) { customAttrs := make(map[string]enumsv1.IndexedValueType) for name := range attributes { customAttrs[name] = enumsv1.INDEXED_VALUE_TYPE_TEXT } req := &operatorservice.AddSearchAttributesRequest{ SearchAttributes: customAttrs, } _, err := s.grpc.GetOperatorServiceStub().AddSearchAttributes(ctx, req) if err != nil { return nil, fmt.Errorf("gRPC AddSearchAttributes failed: %w", err) } return map[string]interface{}{ "attributes_added": len(customAttrs), "attributes": attributes, "added_at": time.Now(), }, nil }