homelab-frontend/internal/temporal/operations_grpc.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
}