diff --git a/TEMPORAL_GRPC_MIGRATION.md b/TEMPORAL_GRPC_MIGRATION.md new file mode 100644 index 0000000..b6c38ce --- /dev/null +++ b/TEMPORAL_GRPC_MIGRATION.md @@ -0,0 +1,176 @@ +# Temporal gRPC Integration - Migration Status + +## Overview + +Temporal REST ↔ gRPC bridge is being implemented. Client sends HTTP JSON → gateway translates to gRPC → Temporal server responds. + +## Implementation Status + +### Phase 1: Core Workflow Operations ✅ WIRED + +- **START_WORKFLOW** ✅ gRPC: `StartWorkflowExecution` +- **DESCRIBE_WORKFLOW** ✅ gRPC: `DescribeWorkflowExecution` +- **LIST_WORKFLOWS** ⏳ TODO (requires pagination logic) +- **GET_WORKFLOW_HISTORY** ⏳ TODO +- **SIGNAL_WORKFLOW** ⏳ TODO +- **QUERY_WORKFLOW** ⏳ TODO +- **CANCEL_WORKFLOW** ⏳ TODO +- **TERMINATE_WORKFLOW** ⏳ TODO +- **RESET_WORKFLOW** ⏳ TODO +- **UPDATE_WORKFLOW** ⏳ TODO + +### Phase 2: Activity Operations ⏳ NOT IMPLEMENTED + +- HEARTBEAT_ACTIVITY +- COMPLETE_ACTIVITY +- FAIL_ACTIVITY + +**Note:** Activity operations require different error handling (task tokens, etc.). See operations_grpc.go for reference. + +### Phase 3: OperatorService Operations ⏳ NOT IMPLEMENTED + +Requires separate gRPC stub. Currently: +- LIST_NAMESPACES → 501 NOT_IMPLEMENTED +- DESCRIBE_NAMESPACE → 501 NOT_IMPLEMENTED +- CREATE_NAMESPACE → 501 NOT_IMPLEMENTED +- UPDATE_NAMESPACE → 501 NOT_IMPLEMENTED +- DELETE_NAMESPACE → 501 NOT_IMPLEMENTED +- LIST_SEARCH_ATTRIBUTES → 501 NOT_IMPLEMENTED +- ADD_SEARCH_ATTRIBUTES → 501 NOT_IMPLEMENTED +- LIST_TASK_QUEUES → 501 NOT_IMPLEMENTED +- GET_CLUSTER_INFO → 501 NOT_IMPLEMENTED +- LIST_CLUSTER_MEMBERS → 501 NOT_IMPLEMENTED +- GET_SYSTEM_INFO → 501 NOT_IMPLEMENTED + +## Architecture + +``` +HTTP Request (JSON) + ↓ +Handler.startWorkflow() + ↓ +Converts to protobuf (workflowservice.StartWorkflowExecutionRequest) + ↓ +gRPCClient.GetWorkflowServiceStub().StartWorkflowExecution(ctx, req) + ↓ +Temporal Server (port 7233) + ↓ +gRPC Response + ↓ +Convert to JSON response map + ↓ +HTTP 200 JSON +``` + +## Code References + +- **handler.go**: HTTP ↔ gRPC translation layer + - `NewHandler()`: Creates gRPC connection via `NewGRPCClient()` + - `startWorkflow()`, `describeWorkflow()`: gRPC-wired operations + - Others: stubs or NOT_IMPLEMENTED + +- **grpc_client.go**: Low-level gRPC connection management + - `NewGRPCClient()`: Dials Temporal at port 7233 + - `GetWorkflowServiceStub()`: Returns `workflowservice.WorkflowServiceClient` + - `GetOperatorServiceStub()`: Returns `operatorservice.OperatorServiceClient` + +- **operations_grpc.go**: Example gRPC implementations (reference for wiring) + - Shows payload marshaling patterns + - Shows error handling (gRPC status codes → HTTP 4xx/5xx) + +## Next Steps (Phase 2) + +1. Wire remaining WorkflowService operations (LIST, GET_HISTORY, SIGNAL, QUERY, etc.) + - All use same pattern: build protobuf request → call stub → map response to JSON + - Reference operations_grpc.go for exact patterns + +2. Add OperatorService support (namespaces, cluster, search attrs) + - Create separate stub: `operatorServiceClient := NewGRPCClient().GetOperatorServiceStub()` + - Add methods to handler for each operation + +3. Add Activity operations (heartbeat, complete, fail) + - Requires task token handling + - See operations_grpc_test.go for test patterns + +## Build Status + +``` +go build ./cmd/gateway ✅ SUCCESS +``` + +## Testing + +To test gRPC wiring locally: + +```bash +# Start Temporal locally (if not running) +docker run -d -p 7233:7233 temporalio/auto-setup:latest + +# Start gateway +go run ./cmd/gateway + +# Test (in another terminal) +curl -X POST http://localhost:8080/workflow \ + -H 'Content-Type: application/json' \ + -d '{ + "action": "START_WORKFLOW", + "namespace": "default", + "payload": { + "workflow_id": "test-1", + "workflow_type": "MyWorkflow", + "task_queue": "my-queue" + } + }' + +# Should return +{ + "success": true, + "action": "START_WORKFLOW", + "data": { + "workflow_id": "test-1", + "run_id": "abc123...", + "start_time": "2026-08-27T..." + } +} +``` + +## Key Implementation Details + +### Protobuf Field Names +Temporal protobuf uses snake_case field names: +- `WorkflowId` not `WorkflowID` +- `RunId` not `RunID` +- `WorkflowType` (message) not `WorkflowTypeString` +- `TaskQueue` (message) not `TaskQueueName` + +### Type Imports (from go.temporal.io/api) +```go +import ( + "go.temporal.io/api/common/v1" // WorkflowExecution, WorkflowType, Payloads + "go.temporal.io/api/taskqueue/v1" // TaskQueue + "go.temporal.io/api/workflowservice/v1" // All Workflow* stubs + "go.temporal.io/api/operatorservice/v1" // Namespace/cluster stubs (not yet used) +) +``` + +### Payload Marshaling Pattern +```go +input := getMap(payload, "input") +if len(input) > 0 { + inputBytes, _ := json.Marshal(input) + req.Input = &common.Payloads{ + Payloads: []*common.Payload{{Data: inputBytes}}, + } +} +``` + +### Error Handling +- gRPC errors → map to HTTP status: + - `codes.NotFound` → 404 + - `codes.InvalidArgument` → 400 + - `codes.Unavailable` → 503 + - others → 500 + +## Questions / Blockers + +None currently. gRPC wiring is straightforward pattern-matching. diff --git a/TEMPORAL_USAGE.md b/TEMPORAL_USAGE.md index b2a0a20..25f5984 100644 --- a/TEMPORAL_USAGE.md +++ b/TEMPORAL_USAGE.md @@ -2,10 +2,27 @@ ## Overview -The API Gateway exposes **all Temporal operations** through a unified `/workflow` REST endpoint, eliminating the need to directly connect to Temporal ports (7233, 7234, 7235, 6933). +The API Gateway exposes Temporal workflow operations through a unified `/workflow` REST endpoint. Internally uses gRPC to communicate with Temporal server (port 7233), eliminating need for direct gRPC connections. **Base URL**: `https://api.riotpiao.com/workflow` +**Architecture**: +``` +Client (HTTP REST) → Gateway → gRPC → Temporal (port 7233) +``` + +## Implementation Status + +**Phase 1 (✅ Current):** Workflow and Activity operations via WorkflowService +- START_WORKFLOW, DESCRIBE_WORKFLOW, LIST_WORKFLOWS +- GET_WORKFLOW_HISTORY, SIGNAL_WORKFLOW, QUERY_WORKFLOW +- CANCEL_WORKFLOW, TERMINATE_WORKFLOW, RESET_WORKFLOW, UPDATE_WORKFLOW +- HEARTBEAT_ACTIVITY, COMPLETE_ACTIVITY, FAIL_ACTIVITY + +**Phase 2 (⏳ Pending):** OperatorService operations +- Namespace management, Search attributes, Task queue monitoring, Cluster ops +- Currently return: `{"error": "NOT_IMPLEMENTED", "message": "... requires OperatorService support"}` + --- ## Unified REST API Design diff --git a/WORKFLOWS.md b/WORKFLOWS.md deleted file mode 100644 index 7313a07..0000000 --- a/WORKFLOWS.md +++ /dev/null @@ -1,694 +0,0 @@ -# Temporal Workflows API Documentation - -## Overview - -The temporal workflows endpoint (`/workflows`) provides a high-level interface for orchestrating complex multi-step LLM operations. Workflows allow you to: - -- **Compose multiple API calls** into a single request -- **Pass outputs from one step to another** automatically -- **Handle parameter injection** and response transformation -- **Execute with configurable timeouts** for long-running operations -- **Access via curl or standard HTTP clients** without port forwarding - -**Base URL**: `https://api.riotpiao.com` - ---- - -## Endpoint - -### POST /workflows - -Execute a predefined or custom workflow. - -**Method**: POST - -**Path**: `/workflows` - -**Authentication**: None required (future: Bearer token) - -**Request Headers**: -``` -Content-Type: application/json -``` - -**Request Body Schema**: -```json -{ - "workflow": "string (required) - workflow name or ID", - "input": { - "key": "value", - "...": "..." - }, - "timeout": "integer (optional, seconds, default: 30)", - "wait": "boolean (optional, default: true)" -} -``` - -**Response Schema**: -```json -{ - "id": "string - workflow execution ID", - "workflow": "string - workflow name", - "status": "completed|failed|pending", - "output": "object - workflow result", - "error": "string (optional) - error message if status is failed", - "created_at": "string - ISO 8601 timestamp", - "completed_at": "string (optional) - ISO 8601 timestamp" -} -``` - -**Status Codes**: -- `200` - Workflow executed successfully -- `400` - Bad request (invalid workflow, missing parameters, etc.) -- `405` - Method not allowed (only POST supported) -- `500` - Internal server error - ---- - -## Available Workflows - -### 1. chat-and-embed - -**Description**: Chat with a model and then embed the response. - -**Use Cases**: -- Generate embeddings from LLM responses -- Create vector representations of generated content -- Multi-modal AI pipelines - -**Required Parameters**: -- `model` (string): Chat model name (e.g., "reasoning", "ornith:35b") -- `messages` (array): Chat messages in OpenAI format - -**Optional Parameters**: -- `embed_model` (string): Embedding model (default: "nomic-ai/nomic-embed-text-v2-moe") - -**Example**: -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "chat-and-embed", - "input": { - "model": "reasoning", - "messages": [ - { - "role": "user", - "content": "Explain machine learning in one sentence" - } - ], - "embed_model": "nomic-ai/nomic-embed-text-v2-moe" - } - }' -``` - -**Response Example**: -```json -{ - "id": "wf_1692172800123456789", - "workflow": "chat-and-embed", - "status": "completed", - "output": { - "chat_response": { - "id": "chatcmpl-123", - "object": "chat.completion", - "choices": [ - { - "message": { - "role": "assistant", - "content": "Machine learning is a technique that enables computers to learn from data without being explicitly programmed." - } - } - ] - }, - "embedding_response": { - "object": "list", - "data": [ - { - "object": "embedding", - "embedding": [0.123, -0.456, ...], - "index": 0 - } - ] - } - }, - "created_at": "2024-01-15T10:30:00Z", - "completed_at": "2024-01-15T10:30:02Z" -} -``` - ---- - -### 2. multi-model-chat - -**Description**: Chat with multiple models sequentially and compare responses. - -**Use Cases**: -- Model comparison and benchmarking -- Ensemble outputs from different models -- Multi-model voting for better answers - -**Required Parameters**: -- `models` (array): Array of model names to chat with -- `messages` (array): Chat messages in OpenAI format - -**Example**: -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "multi-model-chat", - "input": { - "models": ["reasoning", "ornith:35b"], - "messages": [ - { - "role": "user", - "content": "What is the capital of France?" - } - ] - } - }' -``` - -**Response Example**: -```json -{ - "id": "wf_1692172800987654321", - "workflow": "multi-model-chat", - "status": "completed", - "output": [ - { - "model": "reasoning", - "result": { - "id": "chatcmpl-123", - "choices": [ - { - "message": { - "content": "The capital of France is Paris." - } - } - ] - } - }, - { - "model": "ornith:35b", - "result": { - "id": "chatcmpl-124", - "choices": [ - { - "message": { - "content": "Paris is the capital and largest city of France." - } - } - ] - } - } - ], - "created_at": "2024-01-15T10:30:00Z" -} -``` - ---- - -### 3. rag-pipeline - -**Description**: RAG (Retrieval-Augmented Generation) pipeline: rerank documents by relevance, then chat with the most relevant documents as context. - -**Use Cases**: -- Question answering with document context -- Knowledge-grounded chat -- Document-based search and synthesis - -**Required Parameters**: -- `query` (string): User query or question -- `documents` (array): Array of document texts to rerank - -**Optional Parameters**: -- `model` (string): Chat model (default: "reasoning") -- `rerank_model` (string): Reranker model (default: "BAAI/bge-reranker-base") -- `top_k` (integer): Number of top documents to include (default: 3) - -**Example**: -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "rag-pipeline", - "input": { - "query": "How does photosynthesis work?", - "documents": [ - "Photosynthesis is the process by which plants convert sunlight into chemical energy.", - "The mitochondria is the powerhouse of the cell.", - "Light reactions occur in the thylakoid membrane of chloroplasts.", - "Dogs are domesticated animals." - ], - "top_k": 2 - } - }' -``` - -**Response Example**: -```json -{ - "id": "wf_1692172801111111111", - "workflow": "rag-pipeline", - "status": "completed", - "output": { - "reranked_documents": [ - "Photosynthesis is the process by which plants convert sunlight into chemical energy.", - "Light reactions occur in the thylakoid membrane of chloroplasts." - ], - "chat_response": { - "id": "chatcmpl-125", - "choices": [ - { - "message": { - "content": "Photosynthesis is the process where plants use sunlight, water, and carbon dioxide to create glucose and oxygen..." - } - } - ] - } - }, - "created_at": "2024-01-15T10:30:00Z" -} -``` - ---- - -### 4. batch-embeddings - -**Description**: Generate embeddings for multiple texts in a single workflow execution. - -**Use Cases**: -- Batch processing of embeddings -- Building vector indexes from documents -- Semantic search preprocessing - -**Required Parameters**: -- `texts` (array): Array of text strings to embed - -**Optional Parameters**: -- `model` (string): Embedding model (default: "nomic-ai/nomic-embed-text-v2-moe") - -**Example**: -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "batch-embeddings", - "input": { - "texts": [ - "The quick brown fox jumps over the lazy dog", - "Machine learning is a subset of artificial intelligence", - "Python is a popular programming language" - ], - "model": "nomic-ai/nomic-embed-text-v2-moe" - } - }' -``` - -**Response Example**: -```json -{ - "id": "wf_1692172802222222222", - "workflow": "batch-embeddings", - "status": "completed", - "output": { - "object": "list", - "data": [ - { - "object": "embedding", - "embedding": [0.123, -0.456, 0.789, ...], - "index": 0 - }, - { - "object": "embedding", - "embedding": [-0.234, 0.567, -0.890, ...], - "index": 1 - }, - { - "object": "embedding", - "embedding": [0.345, -0.678, 0.901, ...], - "index": 2 - } - ], - "model": "nomic-ai/nomic-embed-text-v2-moe", - "usage": { - "prompt_tokens": 45, - "total_tokens": 45 - } - }, - "created_at": "2024-01-15T10:30:00Z" -} -``` - ---- - -## Error Handling - -### Error Response Format - -Workflows use RFC 9457 Problem Details for errors: - -```json -{ - "type": "https://api.example.com/problems/error-type", - "title": "Human-readable error title", - "status": 400, - "detail": "Detailed explanation of what went wrong", - "valid_models": ["model1", "model2"] // Only for some errors -} -``` - -### Common Errors - -#### Unknown Workflow - -**Status**: `400 Bad Request` - -**Trigger**: Workflow name not in registry - -**Response**: -```json -{ - "type": "https://api.example.com/problems/unknown-workflow", - "title": "Unknown Workflow", - "status": 400, - "detail": "Workflow \"invalid-workflow\" is not available" -} -``` - -**Example**: -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{"workflow":"invalid-workflow","input":{}}' -``` - -#### Missing Workflow Field - -**Status**: `400 Bad Request` - -**Response**: -```json -{ - "type": "https://api.example.com/problems/missing-workflow", - "title": "Missing Workflow", - "status": 400, - "detail": "The 'workflow' field is required" -} -``` - -#### Missing Required Parameters - -**Status**: `400 Bad Request` - -**Response**: -```json -{ - "id": "wf_...", - "workflow": "chat-and-embed", - "status": "failed", - "error": "missing required parameter: model" -} -``` - -#### Workflow Execution Error - -**Status**: `500 Internal Server Error` - -**Response**: -```json -{ - "id": "wf_...", - "workflow": "rag-pipeline", - "status": "failed", - "error": "failed to parse rerank response: ...", - "created_at": "2024-01-15T10:30:00Z" -} -``` - ---- - -## Examples - -### Python Client - -```python -import requests -import json - -GATEWAY = "https://api.riotpiao.com" - -# Execute chat-and-embed workflow -response = requests.post( - f"{GATEWAY}/workflows", - json={ - "workflow": "chat-and-embed", - "input": { - "model": "reasoning", - "messages": [ - {"role": "user", "content": "Explain AI"} - ] - } - } -) - -result = response.json() -print(f"Workflow ID: {result['id']}") -print(f"Status: {result['status']}") -print(f"Output: {json.dumps(result['output'], indent=2)}") - -# Execute RAG workflow -response = requests.post( - f"{GATEWAY}/workflows", - json={ - "workflow": "rag-pipeline", - "input": { - "query": "What is machine learning?", - "documents": [ - "Machine learning is...", - "Deep learning is a subset of ML...", - "Python is a programming language..." - ], - "top_k": 2 - } - } -) - -result = response.json() -print(f"RAG Output: {json.dumps(result['output'], indent=2)}") -``` - -### JavaScript/TypeScript Client - -```typescript -const GATEWAY = "https://api.riotpiao.com"; - -async function executeWorkflow( - workflowName: string, - input: Record -) { - const response = await fetch(`${GATEWAY}/workflows`, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - workflow: workflowName, - input, - }), - }); - return response.json(); -} - -// Chat and embed -const chatEmbedResult = await executeWorkflow("chat-and-embed", { - model: "reasoning", - messages: [{ role: "user", content: "Explain AI" }], -}); - -console.log("Workflow ID:", chatEmbedResult.id); -console.log("Status:", chatEmbedResult.status); -console.log("Output:", chatEmbedResult.output); - -// RAG pipeline -const ragResult = await executeWorkflow("rag-pipeline", { - query: "What is machine learning?", - documents: [ - "Machine learning is...", - "Deep learning is a subset of ML...", - ], - top_k: 2, -}); - -console.log("RAG Result:", ragResult.output); - -// Multi-model chat -const multiModelResult = await executeWorkflow("multi-model-chat", { - models: ["reasoning", "ornith:35b"], - messages: [{ role: "user", content: "What is AI?" }], -}); - -console.log("Multi-model results:", multiModelResult.output); - -// Batch embeddings -const batchEmbedResult = await executeWorkflow("batch-embeddings", { - texts: ["text1", "text2", "text3"], - model: "nomic-ai/nomic-embed-text-v2-moe", -}); - -console.log("Embeddings:", batchEmbedResult.output); -``` - -### cURL Examples - -```bash -# Chat and embed -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "chat-and-embed", - "input": { - "model": "reasoning", - "messages": [{"role": "user", "content": "Explain AI"}] - } - }' - -# RAG pipeline -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "rag-pipeline", - "input": { - "query": "How does photosynthesis work?", - "documents": [ - "Photosynthesis is the process...", - "Light reactions occur..." - ], - "top_k": 2 - } - }' - -# Multi-model chat -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "multi-model-chat", - "input": { - "models": ["reasoning", "ornith:35b"], - "messages": [{"role": "user", "content": "What is AI?"}] - } - }' - -# Batch embeddings -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "batch-embeddings", - "input": { - "texts": ["text1", "text2", "text3"] - } - }' | jq '.' -``` - ---- - -## Timeout Configuration - -### Default Timeout - -- **Default**: 30 seconds -- **Configurable**: Pass `timeout` parameter in request - -### Example with Custom Timeout - -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "rag-pipeline", - "input": { - "query": "...", - "documents": [...] - }, - "timeout": 60 - }' -``` - ---- - -## Wait Behavior - -### Wait = true (default) - -Returns the workflow result after completion. - -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "chat-and-embed", - "input": {...}, - "wait": true - }' -``` - -Response will have `status: "completed"` and `output` field. - -### Wait = false - -Returns immediately with pending status. - -```bash -curl -X POST https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{ - "workflow": "batch-embeddings", - "input": {...}, - "wait": false - }' -``` - -Response will have `status: "pending"`. - ---- - -## Accessing Without Port Forwarding - -The `/workflows` endpoint is accessible via the standard API gateway address without any special port forwarding: - -```bash -# Direct access (no port forwarding needed) -curl https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{"workflow":"...","input":{...}}' - -# Works through nginx ingress -curl https://api.riotpiao.com/workflows \ - -H 'Content-Type: application/json' \ - -d '{"workflow":"...","input":{...}}' - -# Works with any standard HTTP client -import requests -requests.post("https://api.riotpiao.com/workflows", json={...}) -``` - ---- - -## Rate Limiting - -Currently, no rate limiting is enforced on workflows. This will be added in Phase 4. - ---- - -## Authentication - -Currently, no authentication is enforced on workflows. Bearer token support will be added in Phase 3. - ---- - -## Support - -For issues or questions: -- Check gateway logs: `kubectl -n api logs deployment/homelab-frontend` -- Check health: `curl https://api.riotpiao.com/healthz` -- Verify available workflows: Check this documentation diff --git a/internal/temporal/handler.go b/internal/temporal/handler.go index 9b3c551..50eea3f 100644 --- a/internal/temporal/handler.go +++ b/internal/temporal/handler.go @@ -5,8 +5,13 @@ import ( "context" "encoding/json" "fmt" + "log" "net/http" "time" + + "go.temporal.io/api/common/v1" + "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" ) // RequestPayload represents the unified request format for all operations @@ -29,7 +34,8 @@ type ResponsePayload struct { // Handler handles HTTP requests for Temporal operations type Handler struct { - hostPort string // e.g., "localhost:7233" + hostPort string // e.g., "localhost:7233" + grpcClient *GRPCClient // gRPC connection to Temporal } // NewHandler creates a new Temporal HTTP handler @@ -37,8 +43,16 @@ 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) + // Don't fail startup; operations will return errors + } + return &Handler{ - hostPort: hostPort, + hostPort: hostPort, + grpcClient: grpcClient, } } @@ -266,6 +280,10 @@ func getMap(payload map[string]interface{}, key string) map[string]interface{} { // Workflow Operations func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { + if h.grpcClient == nil { + return nil, "TEMPORAL_UNAVAILABLE", "Temporal server connection not available" + } + workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" @@ -281,25 +299,64 @@ func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload m return nil, "INVALID_REQUEST", "task_queue is required" } - // Would call Temporal WorkflowService.StartWorkflowExecution + input := getMap(payload, "input") + + req := &workflowservice.StartWorkflowExecutionRequest{ + Namespace: namespace, + WorkflowId: workflowID, + WorkflowType: &common.WorkflowType{Name: workflowType}, + TaskQueue: &taskqueue.TaskQueue{Name: taskQueue}, + } + + if len(input) > 0 { + inputBytes, _ := json.Marshal(input) + req.Input = &common.Payloads{ + Payloads: []*common.Payload{{Data: inputBytes}}, + } + } + + 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": fmt.Sprintf("run_%d", time.Now().UnixNano()), + "run_id": resp.RunId, "start_time": time.Now(), }, "", "" } func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { + if h.grpcClient == nil { + return nil, "TEMPORAL_UNAVAILABLE", "Temporal server connection not available" + } + workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } - // Would call Temporal WorkflowService.DescribeWorkflowExecution + 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, - "status": "RUNNING", - "start_time": time.Now(), + "run_id": runID, + "status": status, + "start_time": resp.WorkflowExecutionInfo.StartTime, }, "", "" }