refactor: retire /workflows endpoint, use X-Service: workflow instead
CI / CI (pull_request) Successful in 3m12s
CI / CI (pull_request) Successful in 3m12s
Remove deprecated /workflows HTTP endpoint in favor of unified X-Service
header routing. All workflow operations now route through:
X-Service: workflow
X-Resource: {action} (start, describe, signal, query, etc)
This consolidates routing patterns and allows users to specify domain/
namespace via request payload instead of path prefixes.
Changes:
- Remove internal/proxy/workflows.go (484 lines of predefined workflows)
- Remove internal/proxy/workflows_test.go
- Remove /workflows handler from proxy.ServeHTTP()
- Update README.md to document X-Service routing pattern
- Add migration note: use X-Service: workflow instead of /workflows
WorkflowAdapter in serviceadapter/ handles X-Service: workflow requests
and forwards to Temporal gRPC API. Users can now specify domain/namespace
in request payload for multi-tenant workflow access.
Related: #26
This commit is contained in:
@@ -38,22 +38,22 @@ Production API gateway for the homelab cluster. Single entry point (`api.riotpia
|
||||
│ (routing, auth, limits) │
|
||||
└──────┬───────────────────────┘
|
||||
│
|
||||
┌──────┴──────────────────────────────────┐
|
||||
┌──────┴───────────────────────────────────┐
|
||||
│ │
|
||||
/v1/* /workflow /sqs /
|
||||
(LLM) (Temporal gRPC) (Queues) (X-Service)
|
||||
│ │ │ │
|
||||
▼ ▼ ▼ ▼
|
||||
llm-serving temporal:7233 kmsvc/Kafka IAM, S3
|
||||
(vLLM, Ollama) (WorkflowService) Memory
|
||||
(TEI) (gRPC bridge) (poimen)
|
||||
/v1/* X-Service header routing /
|
||||
(LLM) (workflow, sqs, s3, iam, memory) /
|
||||
│ │ │
|
||||
▼ ▼ ▼
|
||||
llm-serving temporal:7233 kmsvc/Kafka, MinIO,
|
||||
(vLLM, Ollama) (gRPC) Authentik, poimen-memory
|
||||
(TEI)
|
||||
```
|
||||
|
||||
**Design principles:**
|
||||
- ✅ Single hostname, multiple path prefixes
|
||||
- ✅ HTTP REST gateway → gRPC Temporal bridge
|
||||
- ✅ Single hostname, unified X-Service + X-Resource header routing
|
||||
- ✅ HTTP REST gateway → gRPC Temporal bridge (via X-Service: workflow)
|
||||
- ✅ Bearer token auth via Authentik (JWT + RBAC)
|
||||
- ✅ Streaming unbuffered (SSE, WebSocket)
|
||||
- ✅ Streaming unbuffered (SSE, WebSocket, HTTP/2 multiplexing)
|
||||
- ✅ Per-route timeouts & rate limits
|
||||
- ✅ No cluster credentials held by gateway
|
||||
|
||||
@@ -61,16 +61,16 @@ llm-serving temporal:7233 kmsvc/Kafka IAM, S3
|
||||
|
||||
## Services & Capabilities
|
||||
|
||||
| Service | Prefix | Upstream | Status |
|
||||
| Service | Method | Upstream | Status |
|
||||
|---------|--------|----------|--------|
|
||||
| **LLM Chat** | `/v1/chat/completions` | llm-serving (vLLM) | ✅ Live |
|
||||
| **Embeddings** | `/v1/embeddings` | llm-serving (TEI) | ✅ Live |
|
||||
| **Reranking** | `/v1/rerank` | llm-serving (TEI) | ✅ Live |
|
||||
| **Workflows** | `/workflow` | Temporal gRPC (7233) | ✅ Live (START, DESCRIBE, SIGNAL, QUERY, etc) |
|
||||
| **Queues** | `/` + `X-Service: sqs` | kmsvc/Kafka | ⏳ Ready (ServiceAdapter) |
|
||||
| **Memory** | `/` + `X-Service: memory` | poimen-memory | ✅ Live |
|
||||
| **IAM** | `/` + `X-Service: iam` | Authentik API | ✅ Live |
|
||||
| **S3** | `/` + `X-Service: s3` | MinIO | ✅ Live |
|
||||
| **LLM Chat** | `POST /v1/chat/completions` | llm-serving (vLLM) | ✅ Live |
|
||||
| **Embeddings** | `POST /v1/embeddings` | llm-serving (TEI) | ✅ Live |
|
||||
| **Reranking** | `POST /v1/rerank` | llm-serving (TEI) | ✅ Live |
|
||||
| **Workflows** | `X-Service: workflow` + `X-Resource: {action}` | Temporal gRPC (7233) | ✅ Live (START, DESCRIBE, SIGNAL, QUERY, etc) |
|
||||
| **Queues** | `X-Service: sqs` + `X-Resource: {action}` | kmsvc/Kafka | ✅ Live |
|
||||
| **Memory** | `X-Service: memory` + `X-Resource: {action}` | poimen-memory | ✅ Live |
|
||||
| **IAM** | `X-Service: iam` + `X-Resource: {action}` | Authentik API | ✅ Live |
|
||||
| **S3** | `X-Service: s3` + `X-Resource: {action}` | MinIO | ✅ Live |
|
||||
|
||||
---
|
||||
|
||||
@@ -102,18 +102,17 @@ curl -X POST https://api.riotpiao.com/v1/chat/completions \
|
||||
}'
|
||||
```
|
||||
|
||||
**Workflow:**
|
||||
**Workflow (via X-Service header):**
|
||||
```bash
|
||||
curl -X POST https://api.riotpiao.com/workflow \
|
||||
curl -X POST https://api.riotpiao.com/ \
|
||||
-H "Authorization: Bearer $TOKEN" \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: start" \
|
||||
-d '{
|
||||
"action": "START_WORKFLOW",
|
||||
"namespace": "default",
|
||||
"payload": {
|
||||
"workflow_id": "my-workflow",
|
||||
"workflow_type": "MyWorkflow",
|
||||
"task_queue": "default"
|
||||
}
|
||||
}'
|
||||
```
|
||||
|
||||
|
||||
@@ -271,13 +271,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Handle /workflows endpoint (workflow orchestration)
|
||||
if r.URL.Path == "/workflows" {
|
||||
h.handleWorkflow(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
// Try to find a matching route (including body-based dispatch for /v1/chat/completions)
|
||||
// Try to find a matching route (including body-based dispatch for /v1/chat/completions).
|
||||
// Note: /workflows endpoint is deprecated. Use X-Service: workflow + X-Resource headers instead.
|
||||
route, err := h.RouteRequest(r)
|
||||
|
||||
// Check if this is a model validation error (from body-based dispatch)
|
||||
|
||||
@@ -1,484 +0,0 @@
|
||||
// Package proxy provides request routing and forwarding.
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
// WorkflowRequest represents a workflow execution request
|
||||
type WorkflowRequest struct {
|
||||
// Workflow ID or name
|
||||
Workflow string `json:"workflow"`
|
||||
|
||||
// Input parameters for the workflow
|
||||
Input map[string]interface{} `json:"input"`
|
||||
|
||||
// Optional: timeout in seconds
|
||||
Timeout int `json:"timeout,omitempty"`
|
||||
|
||||
// Optional: wait for result (default: true)
|
||||
Wait *bool `json:"wait,omitempty"`
|
||||
}
|
||||
|
||||
// WorkflowResponse represents the response from workflow execution
|
||||
type WorkflowResponse struct {
|
||||
// Workflow execution ID
|
||||
ID string `json:"id"`
|
||||
|
||||
// Workflow name
|
||||
Workflow string `json:"workflow"`
|
||||
|
||||
// Execution status: pending, running, completed, failed
|
||||
Status string `json:"status"`
|
||||
|
||||
// Output of the workflow
|
||||
Output interface{} `json:"output,omitempty"`
|
||||
|
||||
// Error message if workflow failed
|
||||
Error string `json:"error,omitempty"`
|
||||
|
||||
// Timestamp when workflow was created
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
|
||||
// Timestamp when workflow completed
|
||||
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
||||
}
|
||||
|
||||
// PredefinedWorkflow defines a workflow template that combines multiple API calls
|
||||
type PredefinedWorkflow struct {
|
||||
Name string
|
||||
Description string
|
||||
Handler func(*http.Request, *Handler, map[string]interface{}) (interface{}, error)
|
||||
}
|
||||
|
||||
// handleWorkflow handles the /workflows endpoint
|
||||
// It accepts workflow definitions and orchestrates API calls
|
||||
func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) {
|
||||
// Only POST is supported
|
||||
if r.Method != "POST" {
|
||||
w.Header().Set("Content-Type", "application/problem+json")
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
fmt.Fprintf(w, `{"type":"https://api.example.com/problems/method-not-allowed","title":"Method Not Allowed","status":405,"detail":"Only POST is supported for /workflows"}`)
|
||||
return
|
||||
}
|
||||
|
||||
// Parse request body
|
||||
var workflowReq WorkflowRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&workflowReq); err != nil {
|
||||
writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/invalid-workflow-request", "Invalid Workflow Request", "Failed to parse workflow request: "+err.Error(), nil)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate workflow name
|
||||
if workflowReq.Workflow == "" {
|
||||
writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/missing-workflow", "Missing Workflow", "The 'workflow' field is required", nil)
|
||||
return
|
||||
}
|
||||
|
||||
// Get predefined workflow
|
||||
workflow, ok := h.getWorkflow(workflowReq.Workflow)
|
||||
if !ok {
|
||||
availableWorkflows := h.getAvailableWorkflows()
|
||||
writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/unknown-workflow", "Unknown Workflow", fmt.Sprintf("Workflow %q is not available", workflowReq.Workflow), availableWorkflows)
|
||||
return
|
||||
}
|
||||
|
||||
// Default wait to true
|
||||
wait := true
|
||||
if workflowReq.Wait != nil {
|
||||
wait = *workflowReq.Wait
|
||||
}
|
||||
|
||||
// Set default timeout if not provided
|
||||
timeout := time.Duration(30) * time.Second
|
||||
if workflowReq.Timeout > 0 {
|
||||
timeout = time.Duration(workflowReq.Timeout) * time.Second
|
||||
}
|
||||
|
||||
// Create a context with timeout for workflow execution
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
// Execute workflow
|
||||
output, err := workflow.Handler(r.WithContext(ctx), h, workflowReq.Input)
|
||||
|
||||
// Build response
|
||||
workflowResp := WorkflowResponse{
|
||||
ID: generateWorkflowID(),
|
||||
Workflow: workflowReq.Workflow,
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
workflowResp.Status = "failed"
|
||||
workflowResp.Error = err.Error()
|
||||
} else {
|
||||
if wait {
|
||||
workflowResp.Status = "completed"
|
||||
workflowResp.Output = output
|
||||
now := time.Now()
|
||||
workflowResp.CompletedAt = &now
|
||||
} else {
|
||||
workflowResp.Status = "pending"
|
||||
}
|
||||
}
|
||||
|
||||
// Write response
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
} else {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
json.NewEncoder(w).Encode(workflowResp)
|
||||
}
|
||||
|
||||
// getWorkflow returns a predefined workflow by name
|
||||
func (h *Handler) getWorkflow(name string) (*PredefinedWorkflow, bool) {
|
||||
workflows := h.getPredefinedWorkflows()
|
||||
for _, wf := range workflows {
|
||||
if wf.Name == name {
|
||||
return &wf, true
|
||||
}
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// getPredefinedWorkflows returns all available workflows
|
||||
func (h *Handler) getPredefinedWorkflows() []PredefinedWorkflow {
|
||||
return []PredefinedWorkflow{
|
||||
{
|
||||
Name: "chat-and-embed",
|
||||
Description: "Chat with a model and then embed the response",
|
||||
Handler: h.chatAndEmbedWorkflow,
|
||||
},
|
||||
{
|
||||
Name: "multi-model-chat",
|
||||
Description: "Chat with multiple models sequentially",
|
||||
Handler: h.multiModelChatWorkflow,
|
||||
},
|
||||
{
|
||||
Name: "rag-pipeline",
|
||||
Description: "RAG pipeline: embed query, rerank, then chat with context",
|
||||
Handler: h.ragPipelineWorkflow,
|
||||
},
|
||||
{
|
||||
Name: "batch-embeddings",
|
||||
Description: "Generate embeddings for multiple texts",
|
||||
Handler: h.batchEmbeddingsWorkflow,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// getAvailableWorkflows returns a list of available workflow names
|
||||
func (h *Handler) getAvailableWorkflows() []string {
|
||||
workflows := h.getPredefinedWorkflows()
|
||||
names := make([]string, len(workflows))
|
||||
for i, wf := range workflows {
|
||||
names[i] = wf.Name
|
||||
}
|
||||
return names
|
||||
}
|
||||
|
||||
// Workflow implementations
|
||||
|
||||
// chatAndEmbedWorkflow: Chat with a model, then embed the response
|
||||
func (h *Handler) chatAndEmbedWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
model, ok := input["model"].(string)
|
||||
if !ok || model == "" {
|
||||
return nil, fmt.Errorf("missing required parameter: model")
|
||||
}
|
||||
|
||||
embedModel, ok := input["embed_model"].(string)
|
||||
if !ok {
|
||||
embedModel = "nomic-ai/nomic-embed-text-v2-moe"
|
||||
}
|
||||
|
||||
messages, ok := input["messages"].([]interface{})
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("missing required parameter: messages")
|
||||
}
|
||||
|
||||
// Step 1: Chat
|
||||
chatReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
chatBody, _ := json.Marshal(chatReq)
|
||||
chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody)))
|
||||
chatHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
// Create a response writer to capture the chat response
|
||||
chatResp := &responseCapture{}
|
||||
handler.ServeHTTP(chatResp, chatHTTPReq)
|
||||
|
||||
var chatResult map[string]interface{}
|
||||
if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse chat response: %v", err)
|
||||
}
|
||||
|
||||
// Extract message content
|
||||
var messageContent string
|
||||
if choices, ok := chatResult["choices"].([]interface{}); ok && len(choices) > 0 {
|
||||
if choice, ok := choices[0].(map[string]interface{}); ok {
|
||||
if message, ok := choice["message"].(map[string]interface{}); ok {
|
||||
if content, ok := message["content"].(string); ok {
|
||||
messageContent = content
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Step 2: Embed the response
|
||||
embedReq := map[string]interface{}{
|
||||
"model": embedModel,
|
||||
"input": messageContent,
|
||||
}
|
||||
|
||||
embedBody, _ := json.Marshal(embedReq)
|
||||
embedHTTPReq, _ := http.NewRequest("POST", "/v1/embeddings", io.NopCloser(bytes.NewReader(embedBody)))
|
||||
embedHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
embedResp := &responseCapture{}
|
||||
handler.ServeHTTP(embedResp, embedHTTPReq)
|
||||
|
||||
var embedResult map[string]interface{}
|
||||
if err := json.Unmarshal(embedResp.body.Bytes(), &embedResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse embedding response: %v", err)
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"chat_response": chatResult,
|
||||
"embedding_response": embedResult,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// multiModelChatWorkflow: Chat with multiple models sequentially
|
||||
func (h *Handler) multiModelChatWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
models, ok := input["models"].([]interface{})
|
||||
if !ok || len(models) == 0 {
|
||||
return nil, fmt.Errorf("missing required parameter: models (array)")
|
||||
}
|
||||
|
||||
messages, ok := input["messages"].([]interface{})
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("missing required parameter: messages")
|
||||
}
|
||||
|
||||
results := make([]map[string]interface{}, 0)
|
||||
|
||||
for _, modelInterface := range models {
|
||||
model, ok := modelInterface.(string)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
chatReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
chatBody, _ := json.Marshal(chatReq)
|
||||
chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody)))
|
||||
chatHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
chatResp := &responseCapture{}
|
||||
handler.ServeHTTP(chatResp, chatHTTPReq)
|
||||
|
||||
var chatResult map[string]interface{}
|
||||
if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil {
|
||||
results = append(results, map[string]interface{}{
|
||||
"model": model,
|
||||
"error": err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
results = append(results, map[string]interface{}{
|
||||
"model": model,
|
||||
"result": chatResult,
|
||||
})
|
||||
}
|
||||
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// ragPipelineWorkflow: RAG pipeline - embed query, rerank, chat with context
|
||||
func (h *Handler) ragPipelineWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
query, ok := input["query"].(string)
|
||||
if !ok || query == "" {
|
||||
return nil, fmt.Errorf("missing required parameter: query")
|
||||
}
|
||||
|
||||
documents, ok := input["documents"].([]interface{})
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("missing required parameter: documents")
|
||||
}
|
||||
|
||||
model, ok := input["model"].(string)
|
||||
if !ok {
|
||||
model = "reasoning"
|
||||
}
|
||||
|
||||
rerankModel, ok := input["rerank_model"].(string)
|
||||
if !ok {
|
||||
rerankModel = "BAAI/bge-reranker-base"
|
||||
}
|
||||
|
||||
topK := 3
|
||||
if tk, ok := input["top_k"].(float64); ok {
|
||||
topK = int(tk)
|
||||
}
|
||||
|
||||
// Step 1: Rerank documents based on query
|
||||
rerankReq := map[string]interface{}{
|
||||
"model": rerankModel,
|
||||
"query": query,
|
||||
"texts": documents,
|
||||
"top_k": topK,
|
||||
}
|
||||
|
||||
rerankBody, _ := json.Marshal(rerankReq)
|
||||
rerankHTTPReq, _ := http.NewRequest("POST", "/v1/rerank", io.NopCloser(bytes.NewReader(rerankBody)))
|
||||
rerankHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
rerankResp := &responseCapture{}
|
||||
handler.ServeHTTP(rerankResp, rerankHTTPReq)
|
||||
|
||||
var rerankResult map[string]interface{}
|
||||
if err := json.Unmarshal(rerankResp.body.Bytes(), &rerankResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse rerank response: %v", err)
|
||||
}
|
||||
|
||||
// Extract top documents
|
||||
var topDocs []string
|
||||
if results, ok := rerankResult["results"].([]interface{}); ok {
|
||||
for i, resultInterface := range results {
|
||||
if i >= topK {
|
||||
break
|
||||
}
|
||||
if result, ok := resultInterface.(map[string]interface{}); ok {
|
||||
if text, ok := result["text"].(string); ok {
|
||||
topDocs = append(topDocs, text)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Step 2: Chat with context
|
||||
context := fmt.Sprintf("Context from documents:\n%v\n\nQuery: %s", topDocs, query)
|
||||
|
||||
chatReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{
|
||||
"role": "user",
|
||||
"content": context,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
chatBody, _ := json.Marshal(chatReq)
|
||||
chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody)))
|
||||
chatHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
chatResp := &responseCapture{}
|
||||
handler.ServeHTTP(chatResp, chatHTTPReq)
|
||||
|
||||
var chatResult map[string]interface{}
|
||||
if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse chat response: %v", err)
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"reranked_documents": topDocs,
|
||||
"chat_response": chatResult,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// batchEmbeddingsWorkflow: Generate embeddings for multiple texts
|
||||
func (h *Handler) batchEmbeddingsWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
texts, ok := input["texts"].([]interface{})
|
||||
if !ok || len(texts) == 0 {
|
||||
return nil, fmt.Errorf("missing required parameter: texts (array)")
|
||||
}
|
||||
|
||||
model, ok := input["model"].(string)
|
||||
if !ok {
|
||||
model = "nomic-ai/nomic-embed-text-v2-moe"
|
||||
}
|
||||
|
||||
// Convert interface{} to []string
|
||||
textStrings := make([]string, 0)
|
||||
for _, t := range texts {
|
||||
if str, ok := t.(string); ok {
|
||||
textStrings = append(textStrings, str)
|
||||
}
|
||||
}
|
||||
|
||||
if len(textStrings) == 0 {
|
||||
return nil, fmt.Errorf("no valid text strings in texts array")
|
||||
}
|
||||
|
||||
embedReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"input": textStrings,
|
||||
}
|
||||
|
||||
embedBody, _ := json.Marshal(embedReq)
|
||||
embedHTTPReq, _ := http.NewRequest("POST", "/v1/embeddings", io.NopCloser(bytes.NewReader(embedBody)))
|
||||
embedHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
embedResp := &responseCapture{}
|
||||
handler.ServeHTTP(embedResp, embedHTTPReq)
|
||||
|
||||
var embedResult map[string]interface{}
|
||||
if err := json.Unmarshal(embedResp.body.Bytes(), &embedResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse embedding response: %v", err)
|
||||
}
|
||||
|
||||
return embedResult, nil
|
||||
}
|
||||
|
||||
// Utility functions
|
||||
|
||||
// responseCapture captures HTTP response for reuse within workflows
|
||||
type responseCapture struct {
|
||||
status int
|
||||
header http.Header
|
||||
body bytes.Buffer
|
||||
}
|
||||
|
||||
func (w *responseCapture) Header() http.Header {
|
||||
if w.header == nil {
|
||||
w.header = make(http.Header)
|
||||
}
|
||||
return w.header
|
||||
}
|
||||
|
||||
func (w *responseCapture) Write(b []byte) (int, error) {
|
||||
if w.status == 0 {
|
||||
w.status = http.StatusOK
|
||||
}
|
||||
return w.body.Write(b)
|
||||
}
|
||||
|
||||
func (w *responseCapture) WriteHeader(statusCode int) {
|
||||
if w.status == 0 {
|
||||
w.status = statusCode
|
||||
}
|
||||
}
|
||||
|
||||
// generateWorkflowID generates a unique workflow execution ID
|
||||
func generateWorkflowID() string {
|
||||
return fmt.Sprintf("wf_%d", time.Now().UnixNano())
|
||||
}
|
||||
|
||||
|
||||
@@ -1,258 +0,0 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
|
||||
)
|
||||
|
||||
func TestWorkflowEndpointNotFound(t *testing.T) {
|
||||
// Create a minimal config
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: map[string]*config.ModelUpstream{
|
||||
"reasoning": {
|
||||
Address: "localhost:8001",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test POST /workflows with unknown workflow
|
||||
body := map[string]interface{}{
|
||||
"workflow": "unknown-workflow",
|
||||
"input": map[string]interface{}{},
|
||||
}
|
||||
|
||||
bodyBytes, _ := json.Marshal(body)
|
||||
req := httptest.NewRequest("POST", "/workflows", bytes.NewReader(bodyBytes))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("Expected 400, got %d", w.Code)
|
||||
}
|
||||
|
||||
var response map[string]interface{}
|
||||
json.Unmarshal(w.Body.Bytes(), &response)
|
||||
|
||||
if response["type"] != "https://api.example.com/problems/unknown-workflow" {
|
||||
t.Errorf("Expected unknown-workflow error, got %v", response["type"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowEndpointMissingWorkflow(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test POST /workflows with missing workflow field
|
||||
body := map[string]interface{}{
|
||||
"input": map[string]interface{}{},
|
||||
}
|
||||
|
||||
bodyBytes, _ := json.Marshal(body)
|
||||
req := httptest.NewRequest("POST", "/workflows", bytes.NewReader(bodyBytes))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("Expected 400, got %d", w.Code)
|
||||
}
|
||||
|
||||
var response map[string]interface{}
|
||||
json.Unmarshal(w.Body.Bytes(), &response)
|
||||
|
||||
if response["type"] != "https://api.example.com/problems/missing-workflow" {
|
||||
t.Errorf("Expected missing-workflow error, got %v", response["type"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowEndpointInvalidMethod(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test GET /workflows (should be 405)
|
||||
req := httptest.NewRequest("GET", "/workflows", nil)
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusMethodNotAllowed {
|
||||
t.Errorf("Expected 405, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowEndpointInvalidJSON(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test POST /workflows with invalid JSON
|
||||
req := httptest.NewRequest("POST", "/workflows", bytes.NewReader([]byte("not json")))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("Expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetAvailableWorkflows(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
workflows := handler.getAvailableWorkflows()
|
||||
|
||||
expectedWorkflows := []string{
|
||||
"chat-and-embed",
|
||||
"multi-model-chat",
|
||||
"rag-pipeline",
|
||||
"batch-embeddings",
|
||||
}
|
||||
|
||||
if len(workflows) != len(expectedWorkflows) {
|
||||
t.Errorf("Expected %d workflows, got %d", len(expectedWorkflows), len(workflows))
|
||||
}
|
||||
|
||||
// Check that all expected workflows are present
|
||||
for _, expected := range expectedWorkflows {
|
||||
found := false
|
||||
for _, actual := range workflows {
|
||||
if actual == expected {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("Expected workflow %q not found", expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetWorkflow(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test getting a valid workflow
|
||||
workflow, ok := handler.getWorkflow("chat-and-embed")
|
||||
if !ok {
|
||||
t.Error("Expected to find chat-and-embed workflow")
|
||||
}
|
||||
if workflow.Name != "chat-and-embed" {
|
||||
t.Errorf("Expected workflow name chat-and-embed, got %s", workflow.Name)
|
||||
}
|
||||
|
||||
// Test getting an invalid workflow
|
||||
workflow, ok = handler.getWorkflow("invalid-workflow")
|
||||
if ok {
|
||||
t.Error("Expected not to find invalid-workflow")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateWorkflowID(t *testing.T) {
|
||||
id1 := generateWorkflowID()
|
||||
id2 := generateWorkflowID()
|
||||
|
||||
if id1 == id2 {
|
||||
t.Error("Generated workflow IDs should be unique")
|
||||
}
|
||||
|
||||
if !bytes.HasPrefix([]byte(id1), []byte("wf_")) {
|
||||
t.Errorf("Workflow ID should start with 'wf_', got %s", id1)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseCapture(t *testing.T) {
|
||||
rc := &responseCapture{}
|
||||
|
||||
// Test Header
|
||||
rc.Header().Set("X-Test", "value")
|
||||
if rc.Header().Get("X-Test") != "value" {
|
||||
t.Error("Header not set correctly")
|
||||
}
|
||||
|
||||
// Test Write
|
||||
n, err := rc.Write([]byte("test content"))
|
||||
if err != nil {
|
||||
t.Errorf("Unexpected error: %v", err)
|
||||
}
|
||||
if n != 12 {
|
||||
t.Errorf("Expected 12 bytes written, got %d", n)
|
||||
}
|
||||
if rc.body.String() != "test content" {
|
||||
t.Errorf("Expected 'test content', got %s", rc.body.String())
|
||||
}
|
||||
|
||||
// Test WriteHeader
|
||||
rc.WriteHeader(http.StatusOK)
|
||||
if rc.status != http.StatusOK {
|
||||
t.Errorf("Expected status 200, got %d", rc.status)
|
||||
}
|
||||
|
||||
// Test WriteHeader doesn't override
|
||||
rc.WriteHeader(http.StatusInternalServerError)
|
||||
if rc.status != http.StatusOK {
|
||||
t.Error("WriteHeader should not override existing status")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowResponseSerialization(t *testing.T) {
|
||||
resp := WorkflowResponse{
|
||||
ID: "wf_123",
|
||||
Workflow: "test-workflow",
|
||||
Status: "completed",
|
||||
Output: map[string]interface{}{
|
||||
"key": "value",
|
||||
},
|
||||
Error: "",
|
||||
}
|
||||
|
||||
data, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
t.Errorf("Failed to marshal response: %v", err)
|
||||
}
|
||||
|
||||
var unmarshaled WorkflowResponse
|
||||
if err := json.Unmarshal(data, &unmarshaled); err != nil {
|
||||
t.Errorf("Failed to unmarshal response: %v", err)
|
||||
}
|
||||
|
||||
if unmarshaled.ID != resp.ID {
|
||||
t.Errorf("Expected ID %s, got %s", resp.ID, unmarshaled.ID)
|
||||
}
|
||||
if unmarshaled.Workflow != resp.Workflow {
|
||||
t.Errorf("Expected Workflow %s, got %s", resp.Workflow, unmarshaled.Workflow)
|
||||
}
|
||||
if unmarshaled.Status != resp.Status {
|
||||
t.Errorf("Expected Status %s, got %s", resp.Status, unmarshaled.Status)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user