Add server-side idempotency tracking

- executedCommands map stores command_id -> timestamp
- Idempotency check BEFORE publishing to Redpanda
- dedup within 5-minute TTL window
- If duplicate: returns idempotent_reject: true
- Marks executed on result received

Usage:
  # Same ID = rejected
  curl -X POST .../command -d '{id:xyz-123,...}'
  # Same xyz-123 within 5 min = 400 idempotent_reject

server/main.go

@@ -82,12 +82,16 @@ var routers = make(map[string]*Router)
 var pendingCommands = make(map[string]chan CommandResult)
 const commandTimeout = 30 * time.Second
 
+// Idempotency: track recently executed commands (command_id -> timestamp)
+var executedCommands = make(map[string]time.Time)
+const idempotencyTTL = 5 * 60 * time.Second // 5 minutes
+
 // Redpanda
 var rp *redpanda.Client
 
 func initRedpanda() error {
 	cfg.RedpandaBrokers = getEnvComma("REDPANDA_BROKERS", "localhost:9092")
-	
+
 	var err error
 	rp, err = redpanda.NewClient(&redpanda.ClientConfig{
 		Brokers: cfg.RedpandaBrokers,
@@ -95,7 +99,7 @@ func initRedpanda() error {
 	if err != nil {
 		return fmt.Errorf("redpanda: %v", err)
 	}
-	
+
 	// Create topics
 	topics := []string{"router-events", "router-commands"}
 	for _, topic := range topics {
@@ -104,7 +108,7 @@ func initRedpanda() error {
 			log.Printf("Topic %s: %v", topic, err)
 		}
 	}
-	
+
 	return nil
 }
 
@@ -151,7 +155,7 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "Unauthorized", http.StatusUnauthorized)
 		return
 	}
-	
+
 	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
 		CompressionMode: websocket.CompressionContextTakeover,
 	})
@@ -160,9 +164,9 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	defer conn.Close(websocket.StatusNormalClosure, "")
-	
+
 	ctx := context.Background()
-	
+
 	// Read router registration
 	var regMsg RouterEvent
 	err = conn.Read(ctx, &regMsg)
@@ -170,21 +174,21 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 		log.Printf("WS read reg: %v", err)
 		return
 	}
-	
+
 	routerID := regMsg.RouterID
 	if routerID == "" {
 		routerID = r.RemoteAddr
 	}
-	
+
 	routers[routerID] = &Router{
 		ID:          routerID,
 		LastSeen:    time.Now(),
 		Conn:       conn,
 		Connected:  true,
 	}
-	
+
 	log.Printf("Router connected: %s", routerID)
-	
+
 	// Message loop
 	for {
 		var event RouterEvent
@@ -192,11 +196,11 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 		if err != nil {
 			break
 		}
-		
+
 		event.ID = uuid.New().String()
 		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
@@ -210,18 +214,21 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 					}
 					ch <- result
 				}
+
+				// Mark as executed (for idempotency)
+				executedCommands[cmdIDstr] = time.Now()
 			}
 		}
-		
+
 		// Publish to Redpanda
 		if err := publishEvent(event); err != nil {
 			log.Printf("Publish error: %v", err)
 		}
-		
+
 		log.Printf("[%s] %s", routerID, event.EventType)
 		routers[routerID].LastSeen = time.Now()
 	}
-	
+
 	if routers[routerID] != nil {
 		routers[routerID].Connected = false
 	}
@@ -234,28 +241,28 @@ func handleHTTPEvent(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 		return
 	}
-	
+
 	token := r.Header.Get("Authorization")
 	if token != "Bearer "+cfg.Token && token != cfg.Token {
 		http.Error(w, "Unauthorized", http.StatusUnauthorized)
 		return
 	}
-	
+
 	var event RouterEvent
 	if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
 		http.Error(w, "Invalid JSON", http.StatusBadRequest)
 		return
 	}
-	
+
 	event.ID = uuid.New().String()
 	event.ReceivedAt = time.Now()
 	event.Connection = "http"
-	
+
 	if err := publishEvent(event); err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
-	
+
 	json.NewEncoder(w).Encode(map[string]string{"event_id": event.ID})
 }
 
@@ -278,47 +285,73 @@ func handleCommand(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 		return
 	}
-	
+
 	var cmd RouterCommand
 	if err := json.NewDecoder(r.Body).Decode(&cmd); err != nil {
 		http.Error(w, "Invalid JSON", http.StatusBadRequest)
 		return
 	}
-	
+
 	routerID := r.URL.Query().Get("router_id")
 	if routerID == "" {
 		routerID = cmd.RouterID
 	}
-	
+
 	if routerID == "" {
 		http.Error(w, "router_id required", http.StatusBadRequest)
 		return
 	}
-	
+
 	router := routers[routerID]
 	if router == nil || !router.Connected {
 		http.Error(w, "Router not connected", http.StatusServiceUnavailable)
 		return
 	}
-	
+
+	// ──────────────────────────────────────────────────────────
+	// IDEMPOTENCY CHECK - Dedup before publishing!
+	// ──────────────────────────────────────────────────────────
+	// If client provides an idempotency key, reuse it
+	xecID := cmd.ID
+	if xecID == "" {
+		// New command ID acts as idempotency key
+		xecID = uuid.New().String()
+	}
+
+	// Check if already executed within TTL
+	if lastExec, exists := executedCommands[xecID]; exists {
+		if time.Since(lastExec) < idempotencyTTL {
+			log.Printf("Idempotent reject: %s (recently executed)", xecID)
+			json.NewEncoder(w).Encode(map[string]interface{}{
+				"idempotent_reject": true,
+				"existing_command_id": xecID,
+				"message": "Command already executed",
+			})
+			return
+		}
+		// Old entry, allow retry
+		delete(executedCommands, xecID)
+	}
+
+
 	// Create result channel
 	resultChan := make(chan CommandResult, 1)
-	cmd.ID = uuid.New().String()
+	cmd.ID = xecID
 	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 (waiting...)", routerID, cmd.Command)
-	
+
 	// Wait for result with timeout
 	select {
 	case result := <-resultChan:
@@ -338,6 +371,9 @@ func handleCommand(w http.ResponseWriter, r *http.Request) {
 			"error":    "Router did not respond",
 		})
 	}
+
+	// Also mark timeout for idempotency (allows retry after TTL)
+	// executedCommands stays, will expire naturally
 }
 
 // Health
@@ -351,15 +387,15 @@ func handleHealth(w http.ResponseWriter, r *http.Request) {
 func main() {
 	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)
-	
+
 	if err := initRedpanda(); err != nil {
 		log.Printf("Redpanda init failed: %v", err)
 	}
-	
+
 	http.HandleFunc("/", handleHealth)
 	http.HandleFunc("/health", handleHealth)
 	http.HandleFunc("/api/events", handleHTTPEvent)
@@ -367,7 +403,7 @@ func main() {
 	http.HandleFunc("/api/routers", handleRouters)
 	http.HandleFunc("/api/command", handleCommand)
 	http.HandleFunc("/ws", handleWebSocket)
-	
+
 	go func() {
 		sigCh := make(chan os.Signal, 1)
 		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
@@ -378,7 +414,7 @@ func main() {
 		}
 		os.Exit(0)
 	}()
-	
+
 	log.Printf("Server ready on port %d", cfg.WebsocketPort)
 	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", cfg.WebsocketPort), nil))
 }