Add command result tracking: server waits for response

- Go server: pendingCommands map tracks in-flight commands
- Server waits up to 30s for router response
- Returns: {success, output, error} or timeout
- Router: includes command_id in result response

Full flow:
  1. Client POST /api/command
  2. Server publishes to Redpanda
  3. Router receives via WS
  4. Router executes command
  5. Router sends command_result back
  6. Server receives, sends to waiting client

package/src/client2server-unified.lua

@@ -72,6 +72,7 @@ local function json_decode(str)
     local result = {}
     for k, v in str:gmatch('"([^"]+)":%s*"([^"]*)"') do result[k] = v end
     for k, v in str:gmatch('"([^"]+)":%s*(%d+)') do result[k] = tonumber(v) end
+    for k, v in str:gmatch('"([^"]+)":%s*(%a+)') do result[k] = v end
     return result
 end
 
@@ -356,6 +357,7 @@ local function co_commands()
                         event_type = "command_result",
                         payload = {
                             command = cmd_obj.command,
+                            command_id = cmd_obj.id or "",
                             success = result.success,
                             output = result.output,
                             error = result.error

server/main.go

@@ -36,7 +36,7 @@ type Config struct {
 var cfg = Config{
 	RedpandaBrokers: []string{"localhost:9092"},
 	WebsocketPort:   3843,
-	APIPort:        3844, // HTTP API on next port
+	APIPort:        3844,
 	Token:         "secret-token",
 }
 
@@ -49,7 +49,7 @@ type RouterEvent struct {
 	Timestamp  time.Time `json:"timestamp"`
 	Payload    map[string]interface{} `json:"payload"`
 	ReceivedAt time.Time `json:"received_at"`
-	Connection string    `json:"connection"` // "websocket" or "http"
+	Connection string    `json:"connection"`
 }
 
 // Command to router
@@ -61,6 +61,13 @@ type RouterCommand struct {
 	SentAt   time.Time       `json:"sent_at"`
 }
 
+// Command result received from router
+type CommandResult struct {
+	Success bool   `json:"success"`
+	Output string `json:"output"`
+	Error  string `json:"error"`
+}
+
 // Router state
 type Router struct {
 	ID          string
@@ -71,6 +78,10 @@ type Router struct {
 
 var routers = make(map[string]*Router)
 
+// Pending commands awaiting results (command_id -> result channel)
+var pendingCommands = make(map[string]chan CommandResult)
+const commandTimeout = 30 * time.Second
+
 // Redpanda
 var rp *redpanda.Client
 
@@ -85,7 +96,7 @@ func initRedpanda() error {
 		return fmt.Errorf("redpanda: %v", err)
 	}
 	
-	// Create topics if needed
+	// Create topics
 	topics := []string{"router-events", "router-commands"}
 	for _, topic := range topics {
 		err := rp.CreateTopic(topic, 1, 3)
@@ -105,29 +116,36 @@ func getEnvComma(key, def string) []string {
 	return []string{val}
 }
 
+func getEnvStr(key, def string) string {
+	if val := os.Getenv(key); val != "" {
+		return val
+	}
+	return def
+}
+
+func getEnvInt(key string, def int) int {
+	if val := os.Getenv(key); val != "" {
+		var v int
+		fmt.Sscanf(val, "%d", &v)
+		return v
+	}
+	return def
+}
+
 // Publish event to Redpanda
 func publishEvent(event RouterEvent) error {
-	data, err := json.Marshal(event)
-	if err != nil {
-		return err
-	}
-	
+	data, _ := json.Marshal(event)
 	return rp.Produce("router-events", []byte(event.ID), data)
 }
 
 // Publish command to Redpanda
 func publishCommand(cmd RouterCommand) error {
-	data, err := json.Marshal(cmd)
-	if err != nil {
-		return err
-	}
-	
+	data, _ := json.Marshal(cmd)
 	return rp.Produce("router-commands", []byte(cmd.ID), data)
 }
 
 // WebSocket handler
 func handleWebSocket(w http.ResponseWriter, r *http.Request) {
-	// Auth
 	token := r.URL.Query().Get("token")
 	if token != cfg.Token {
 		http.Error(w, "Unauthorized", http.StatusUnauthorized)
@@ -158,19 +176,15 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 		routerID = r.RemoteAddr
 	}
 	
-	// Store router
 	routers[routerID] = &Router{
 		ID:          routerID,
 		LastSeen:    time.Now(),
 		Conn:       conn,
-		Connected: true,
+		Connected:  true,
 	}
 	
 	log.Printf("Router connected: %s", routerID)
 	
-	// Update router last seen
-	routers[routerID].LastSeen = time.Now()
-	
 	// Message loop
 	for {
 		var event RouterEvent
@@ -183,28 +197,38 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 		event.ReceivedAt = time.Now()
 		event.Connection = "websocket"
 		
+		// Check if this is a command result
+		if event.EventType == "command_result" {
+			// Find pending command and send result
+			if cmdID := event.Payload["command_id"]; cmdID != nil {
+				cmdIDstr, _ := cmdID.(string)
+				if ch, ok := pendingCommands[cmdIDstr]; ok {
+					result := CommandResult{
+						Success: event.Payload["success"] == true,
+						Output: func() string { s, _ := event.Payload["output"].(string); return s }(),
+						Error:  func() string { s, _ := event.Payload["error"].(string); return s }(),
+					}
+					ch <- result
+				}
+			}
+		}
+		
 		// Publish to Redpanda
 		if err := publishEvent(event); err != nil {
 			log.Printf("Publish error: %v", err)
 		}
 		
 		log.Printf("[%s] %s", routerID, event.EventType)
-		
-		// Send ack
-		conn.Write(ctx, []byte(`{"ack":true}`))
-		
-		// Update last seen
 		routers[routerID].LastSeen = time.Now()
 	}
 	
-	// Cleanup
 	if routers[routerID] != nil {
 		routers[routerID].Connected = false
 	}
 	log.Printf("Router disconnected: %s", routerID)
 }
 
-// HTTP Event webhook (fallback)
+// HTTP Event webhook
 func handleHTTPEvent(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "POST" {
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -227,73 +251,28 @@ func handleHTTPEvent(w http.ResponseWriter, r *http.Request) {
 	event.ReceivedAt = time.Now()
 	event.Connection = "http"
 	
-	// Publish to Redpanda
 	if err := publishEvent(event); err != nil {
-		log.Printf("Publish error: %v", err)
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
 	
-	log.Printf("[%s] %s (HTTP)", event.RouterID, event.EventType)
-	
 	json.NewEncoder(w).Encode(map[string]string{"event_id": event.ID})
 }
 
-// Get routers list
+// Get routers
 func handleRouters(w http.ResponseWriter, r *http.Request) {
-	json.NewEncoder(w).Encode(map[string]interface{}{
-		"routers": func() []map[string]interface{} {
-			list := []map[string]interface{}{}
-			for id, router := range routers {
-				list = append(list, map[string]interface{}{
-					"id":           id,
-					"last_seen":    router.LastSeen,
-					"online":       time.Since(router.LastSeen) < 60*time.Second,
-				})
-			}
-			return list
-		}(),
-	})
-}
-
-// Get events
-func handleEvents(w http.ResponseWriter, r *http.Request) {
-	// Simple consumer - in production use proper offset management
-	consumer, err := rp.NewConsumer("router-events", "client2server-group")
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
+	list := []map[string]interface{}{}
+	for id, router := range routers {
+		list = append(list, map[string]interface{}{
+			"id":        id,
+			"last_seen": router.LastSeen,
+			"online":    time.Since(router.LastSeen) < 60*time.Second,
+		})
 	}
-	defer consumer.Close()
-	
-	limit := 100
-	if l := r.URL.Query().Get("limit"); l != "" {
-		fmt.Sscanf(l, "%d", &limit)
-	}
-	
-	events := []RouterEvent{}
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	
-	for i := 0; i < limit; i++ {
-		msg, err := consumer.Consume(ctx, 5*time.Second)
-		if err != nil {
-			break
-		}
-		
-		var event RouterEvent
-		if json.Unmarshal(msg.Value, &event) == nil {
-			events = append(events, event)
-		}
-	}
-	
-	json.NewEncoder(w).Encode(map[string]interface{}{
-		"events": events,
-		"total": len(events),
-	})
+	json.NewEncoder(w).Encode(map[string]interface{}{"routers": list})
 }
 
-// Send command to router
+// Send command and wait for result
 func handleCommand(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "POST" {
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -316,94 +295,90 @@ func handleCommand(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	
-	// Publish command
+	router := routers[routerID]
+	if router == nil || !router.Connected {
+		http.Error(w, "Router not connected", http.StatusServiceUnavailable)
+		return
+	}
+	
+	// Create result channel
+	resultChan := make(chan CommandResult, 1)
 	cmd.ID = uuid.New().String()
 	cmd.RouterID = routerID
 	cmd.SentAt = time.Now()
 	
+	// Store pending command
+	pendingCommands[cmd.ID] = resultChan
+	
 	if err := publishCommand(cmd); err != nil {
+		delete(pendingCommands, cmd.ID)
 		log.Printf("Publish command: %v", err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 	
-	log.Printf("Command to %s: %s", routerID, cmd.Command)
-	
-	json.NewEncoder(w).Encode(map[string]string{
-		"command_id": cmd.ID,
-		"status":     "sent",
-	})
+	log.Printf("Command to %s: %s (waiting...)", routerID, cmd.Command)
+	
+	// Wait for result with timeout
+	select {
+	case result := <-resultChan:
+		delete(pendingCommands, cmd.ID)
+		json.NewEncoder(w).Encode(map[string]interface{}{
+			"command_id": cmd.ID,
+			"status":    "completed",
+			"success":   result.Success,
+			"output":    result.Output,
+			"error":    result.Error,
+		})
+	case <-time.After(commandTimeout):
+		delete(pendingCommands, cmd.ID)
+		json.NewEncoder(w).Encode(map[string]string{
+			"command_id": cmd.ID,
+			"status":    "timeout",
+			"error":    "Router did not respond",
+		})
+	}
 }
 
-// Health check
+// Health
 func handleHealth(w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(map[string]interface{}{
-		"status":     "ok",
-		"routers":   len(routers),
-		"timestamp": time.Now(),
+		"status":   "ok",
+		"routers": len(routers),
 	})
 }
 
-// Main
 func main() {
-	// Config from env
 	cfg.Token = getEnvStr("TOKEN", cfg.Token)
 	cfg.WebsocketPort = getEnvInt("PORT", cfg.WebsocketPort)
 	
 	log.SetFlags(log.LstdFlags | log.Lshortfile)
 	log.Printf("=== client2server Go Server ===")
 	log.Printf("WebSocket: ws://localhost:%d", cfg.WebsocketPort)
-	log.Printf("HTTP API:   http://localhost:%d", cfg.APIPort)
 	
-	// Init Redpanda
 	if err := initRedpanda(); err != nil {
-		log.Printf("Redpanda init failed (running without): %v", err)
-	} else {
-		log.Printf("Redpanda: %v", cfg.RedpandaBrokers)
+		log.Printf("Redpanda init failed: %v", err)
 	}
 	
-	// HTTP handlers
 	http.HandleFunc("/", handleHealth)
 	http.HandleFunc("/health", handleHealth)
 	http.HandleFunc("/api/events", handleHTTPEvent)
 	http.HandleFunc("/api/events/", handleHTTPEvent)
 	http.HandleFunc("/api/routers", handleRouters)
 	http.HandleFunc("/api/command", handleCommand)
-	
-	// WebSocket (HAProxy protocol)
 	http.HandleFunc("/ws", handleWebSocket)
 	
-	// Graceful shutdown
 	go func() {
 		sigCh := make(chan os.Signal, 1)
 		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
 		<-sigCh
 		log.Println("Shutting down...")
-		
 		for _, router := range routers {
 			router.Conn.Close(websocket.StatusNormalClosure, "server shutdown")
 		}
-		
 		os.Exit(0)
 	}()
 	
-	// Start (using HTTP for both WS and API - need haproxy or prefix)
-	log.Printf("Server ready on port %d", cfg.WebiscrollPoint)
+	log.Printf("Server ready on port %d", cfg.WebsocketPort)
 	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", cfg.WebsocketPort), nil))
-}
-
-func getEnvStr(key, def string) string {
-	if val := os.Getenv(key); val != "" {
-		return val
-	}
-	return def
-}
-
-func getEnvInt(key string, def int) int {
-	if val := os.Getenv(key); val != "" {
-		var v int
-		fmt.Sscanf(val, "%d", &v)
-		return v
-	}
-	return def
 }