From 6b089ff2231b5960331ce21c49d34972b7588f8f Mon Sep 17 00:00:00 2001 From: Rock Date: Sat, 26 Sep 2026 10:50:16 +0000 Subject: [PATCH] feat: add SSE streaming endpoint for workflow events (#36) New SSE endpoint: GET /workflow/stream?workflow_id=X&namespace=default - Polls Temporal GetWorkflowExecutionHistory every 2s - Maps event types to semantic SSE events (activity_started/completed/failed, workflow_completed, etc.) - Heartbeat every 15s - Clean close on workflow completion or client disconnect - Added stream resource to workflow ServiceAdapter spec Closes #33 --------- Co-authored-by: Poimen Reviewed-on: https://forgejo.riotpiao.com/riotpiao-poimen/homelab-frontend/pulls/36 --- internal/serviceadapter/workflow_adapter.go | 13 ++ internal/temporal/handler.go | 2 + internal/temporal/sse.go | 224 ++++++++++++++++++++ 3 files changed, 239 insertions(+) create mode 100644 internal/temporal/sse.go diff --git a/internal/serviceadapter/workflow_adapter.go b/internal/serviceadapter/workflow_adapter.go index 773404e..c843e79 100644 --- a/internal/serviceadapter/workflow_adapter.go +++ b/internal/serviceadapter/workflow_adapter.go @@ -359,6 +359,19 @@ func GetWorkflowSpec() *Spec { }, }, }, + { + Name: "stream", + Methods: []Method{ + { + Verb: "GET", + UpstreamPath: "/workflow/stream", + Auth: &Auth{ + Required: true, + Capability: "workflow:read", + }, + }, + }, + }, }, } } diff --git a/internal/temporal/handler.go b/internal/temporal/handler.go index 7dde90a..7e9d076 100644 --- a/internal/temporal/handler.go +++ b/internal/temporal/handler.go @@ -72,6 +72,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.handleHealth(w, r) case "/workflow/metrics": h.handleMetrics(w, r) + case "/workflow/stream": + h.StreamWorkflowEvents(w, r) default: w.WriteHeader(http.StatusNotFound) h.writeError(w, "", "NOT_FOUND", "Endpoint not found") diff --git a/internal/temporal/sse.go b/internal/temporal/sse.go new file mode 100644 index 0000000..8986769 --- /dev/null +++ b/internal/temporal/sse.go @@ -0,0 +1,224 @@ +package temporal + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +// SSEEvent represents a Server-Sent Event +type SSEEvent struct { + Event string `json:"event"` + Data interface{} `json:"data"` +} + +// StreamWorkflowEvents streams workflow execution history as SSE events. +// Polls Temporal GetWorkflowExecutionHistory every 2s and sends new events. +// Sends heartbeat every 15s. Closes on workflow completion or client disconnect. +func (h *Handler) StreamWorkflowEvents(w http.ResponseWriter, r *http.Request) { + // Only GET allowed + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + // Extract workflow_id from query params + workflowID := r.URL.Query().Get("workflow_id") + if workflowID == "" { + http.Error(w, `{"error":"workflow_id query parameter required"}`, http.StatusBadRequest) + return + } + runID := r.URL.Query().Get("run_id") + namespace := r.URL.Query().Get("namespace") + if namespace == "" { + namespace = "default" + } + + // Check gRPC client + if h.grpcClient == nil { + http.Error(w, `{"error":"Temporal server not available"}`, http.StatusServiceUnavailable) + return + } + + // Set SSE headers + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "Streaming not supported", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") // disable nginx buffering + + // Use request context for cancellation (client disconnect) + ctx := r.Context() + + var lastEventID int64 + pollTicker := time.NewTicker(2 * time.Second) + heartbeatTicker := time.NewTicker(15 * time.Second) + defer pollTicker.Stop() + defer heartbeatTicker.Stop() + + // Send initial connected event + writeSSE(w, flusher, "connected", map[string]interface{}{ + "workflow_id": workflowID, + "run_id": runID, + "namespace": namespace, + "timestamp": time.Now().Format(time.RFC3339), + }) + + for { + select { + case <-ctx.Done(): + // Client disconnected + return + + case <-heartbeatTicker.C: + writeSSE(w, flusher, "heartbeat", map[string]string{ + "timestamp": time.Now().Format(time.RFC3339), + }) + + case <-pollTicker.C: + events, workflowDone, err := h.pollNewEvents(ctx, namespace, workflowID, runID, lastEventID) + if err != nil { + writeSSE(w, flusher, "error", map[string]string{ + "error": err.Error(), + "timestamp": time.Now().Format(time.RFC3339), + }) + continue + } + + for _, evt := range events { + writeSSE(w, flusher, evt.Event, evt.Data) + if evtData, ok := evt.Data.(map[string]interface{}); ok { + if id, ok := evtData["event_id"]; ok { + if idInt, ok := id.(int64); ok { + if idInt > lastEventID { + lastEventID = idInt + } + } + } + } + } + + if workflowDone { + writeSSE(w, flusher, "workflow_completed", map[string]string{ + "workflow_id": workflowID, + "timestamp": time.Now().Format(time.RFC3339), + }) + return + } + } + } +} + +// pollNewEvents fetches workflow history from Temporal and returns new events since lastEventID. +// Returns (events, workflowDone, error). +func (h *Handler) pollNewEvents(ctx context.Context, namespace, workflowID, runID string, lastEventID int64) ([]SSEEvent, bool, error) { + pollCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + grpcImpl := NewWorkflowGRPCImpl(h.grpcClient) + result, err := grpcImpl.GetWorkflowExecutionHistory(pollCtx, namespace, workflowID, runID) + if err != nil { + return nil, false, fmt.Errorf("failed to get history: %w", err) + } + + eventsRaw, ok := result["events"] + if !ok { + return nil, false, nil + } + + events, ok := eventsRaw.([]map[string]interface{}) + if !ok { + return nil, false, nil + } + + var sseEvents []SSEEvent + workflowDone := false + + for _, evt := range events { + eventID, _ := evt["event_id"].(int64) + if eventID <= lastEventID { + continue + } + + // operations_grpc.go stores event type under "type" key + eventType, _ := evt["type"].(string) + sseEvent := mapTemporalEvent(eventType, evt) + if sseEvent != nil { + sseEvents = append(sseEvents, *sseEvent) + } + + if isTerminalEvent(eventType) { + workflowDone = true + } + } + + return sseEvents, workflowDone, nil +} + +// mapTemporalEvent converts a Temporal history event to an SSE event. +func mapTemporalEvent(eventType string, evt map[string]interface{}) *SSEEvent { + data := map[string]interface{}{ + "event_id": evt["event_id"], + "event_type": eventType, + "timestamp": evt["timestamp"], + } + + switch { + case strings.Contains(eventType, "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED"): + return &SSEEvent{Event: "activity_scheduled", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_ACTIVITY_TASK_STARTED"): + return &SSEEvent{Event: "activity_started", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_ACTIVITY_TASK_COMPLETED"): + return &SSEEvent{Event: "activity_completed", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_ACTIVITY_TASK_FAILED"): + return &SSEEvent{Event: "activity_failed", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT"): + return &SSEEvent{Event: "activity_timed_out", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED"): + return &SSEEvent{Event: "workflow_started", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED"): + return &SSEEvent{Event: "workflow_completed", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_WORKFLOW_EXECUTION_FAILED"): + return &SSEEvent{Event: "workflow_failed", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT"): + return &SSEEvent{Event: "workflow_timed_out", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED"): + return &SSEEvent{Event: "workflow_canceled", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED"): + return &SSEEvent{Event: "workflow_terminated", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_CHILD_WORKFLOW"): + return &SSEEvent{Event: "child_workflow", Data: data} + case strings.Contains(eventType, "EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED"): + return &SSEEvent{Event: "workflow_signaled", Data: data} + default: + // Skip internal/uninteresting events (timer, decision task, etc.) + return nil + } +} + +// isTerminalEvent returns true if the event type indicates workflow completion. +func isTerminalEvent(eventType string) bool { + return strings.Contains(eventType, "COMPLETED") || + strings.Contains(eventType, "FAILED") || + strings.Contains(eventType, "TIMED_OUT") || + strings.Contains(eventType, "CANCELED") || + strings.Contains(eventType, "TERMINATED") +} + +// writeSSE writes a single SSE event to the response writer. +func writeSSE(w http.ResponseWriter, flusher http.Flusher, event string, data interface{}) { + jsonData, err := json.Marshal(data) + if err != nil { + return + } + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, string(jsonData)) + flusher.Flush() +}