332 lines
10 KiB
Go
332 lines
10 KiB
Go
|
// 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
|
||
|
|
}
|