// client2server - Go server with WebSocket + Redpanda
// Copyright (c) 2026 Luis Rosales - MIT License
//
// Build: go build -o client2server-server
// Run: ./client2server-server
// WebSocket: ws://localhost:3843
// HTTP API: http://localhost:3843

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/google/uuid"
	"github.com/redpanda-data/redpanda-sdk-go/redpanda"
	"github.com/redpanda-data/redpanda-sdk-go/schema"
	"nhooyr.io/websocket"
)

// Config
type Config struct {
	RedpandaBrokers []string
	WebsocketPort   int
	APIPort        int
	Token          string
}

var cfg = Config{
	RedpandaBrokers: []string{"localhost:9092"},
	WebsocketPort:   3843,
	APIPort:        3844,
	Token:         "secret-token",
}

// Event from router
type RouterEvent struct {
	ID          string    `json:"id"`
	RouterID    string    `json:"router_id"`
	Hostname   string    `json:"hostname"`
	EventType  string    `json:"event_type"`
	Timestamp  time.Time `json:"timestamp"`
	Payload    map[string]interface{} `json:"payload"`
	ReceivedAt time.Time `json:"received_at"`
	Connection string    `json:"connection"`
}

// Command to router
type RouterCommand struct {
	ID        string          `json:"id"`
	RouterID string          `json:"router_id"`
	Command  string          `json:"command"`
	Args     map[string]string `json:"args,omitempty"`
	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
	LastSeen   time.Time
	Conn      *websocket.Conn
	Connected bool
}

var routers = make(map[string]*Router)

// Per-router command queue (for offline routers - auto-flush on reconnect)
var routerQueues = make(map[string][]RouterCommand)

// Pending commands awaiting results (command_id -> result channel)
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,
	})
	if err != nil {
		return fmt.Errorf("redpanda: %v", err)
	}

	// Create topics
	topics := []string{"router-events", "router-commands"}
	for _, topic := range topics {
		err := rp.CreateTopic(topic, 1, 3)
		if err != nil && !schema.ErrTopicExists.Exists(err) {
			log.Printf("Topic %s: %v", topic, err)
		}
	}

	return nil
}

func getEnvComma(key, def string) []string {
	val := os.Getenv(key)
	if val == "" {
		return []string{def}
	}
	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, _ := json.Marshal(event)
	return rp.Produce("router-events", []byte(event.ID), data)
}

// Publish command to Redpanda
func publishCommand(cmd RouterCommand) error {
	data, _ := json.Marshal(cmd)
	return rp.Produce("router-commands", []byte(cmd.ID), data)
}

// WebSocket handler
func handleWebSocket(w http.ResponseWriter, r *http.Request) {
	token := r.URL.Query().Get("token")
	if token != cfg.Token {
		http.Error(w, "Unauthorized", http.StatusUnauthorized)
		return
	}

	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
		CompressionMode: websocket.CompressionContextTakeover,
	})
	if err != nil {
		log.Printf("WS accept: %v", err)
		return
	}
	defer conn.Close(websocket.StatusNormalClosure, "")

	ctx := context.Background()

	// Read router registration
	var regMsg RouterEvent
	err = conn.Read(ctx, &regMsg)
	if err != nil {
		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 (flushing queue)", routerID)

	// Flush queued commands for this router
	if queuedCmds, ok := routerQueues[routerID]; ok && len(queuedCmds) > 0 {
		log.Printf("Flushing %d queued commands to %s", len(queuedCmds), routerID)
		for _, cmd := range queuedCmds {
			resultChan := make(chan CommandResult, 1)
			cmd.SentAt = time.Now()
			pendingCommands[cmd.ID] = resultChan
			publishCommand(cmd)
			log.Printf("Queued cmd sent: %s", cmd.Command)
			// Fire and forget - waiter will handle result
		}
		delete(routerQueues, routerID)
	}

	// Message loop
	for {
		var event RouterEvent
		err := conn.Read(ctx, &event)
		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
			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
				}

				// 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
	}
	log.Printf("Router disconnected: %s", routerID)
}

// HTTP Event webhook
func handleHTTPEvent(w http.ResponseWriter, r *http.Request) {
	if r.Method != "POST" {
		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})
}

// Get routers
func handleRouters(w http.ResponseWriter, r *http.Request) {
	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,
		})
	}
	json.NewEncoder(w).Encode(map[string]interface{}{"routers": list})
}

// 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)
		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]

	// ──────────────────────────────────────────────────────────
	// IDEMPOTENCY CHECK - Dedup before publishing!
	// ──────────────────────────────────────────────────────────
	// If client provides an idempotency key, reuse it
	xecID := cmd.ID
	if xecID == "" {
		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
		}
		delete(executedCommands, xecID)
	}

	// ──────────────────────────────────────────────────────────
	// Queue vs Live delivery
	// ──────────────────────────────────────────────────────────
	cmd.ID = xecID
	cmd.RouterID = routerID
	cmd.SentAt = time.Now()

	if router == nil || !router.Connected {
		// Router offline - QUEUE it!
		routerQueues[routerID] = append(routerQueues[routerID], cmd)
		log.Printf("Queued command for %s: %s (queue depth: %d)", routerID, cmd.Command, len(routerQueues[routerID]))
		json.NewEncoder(w).Encode(map[string]interface{}{
			"command_id": cmd.ID,
			"status":     "queued",
			"queued_for": routerID,
			"queue_depth": len(routerQueues[routerID]),
			"message":   "Router offline, command queued",
		})
		return
	}

	// Router online - send directly
	resultChan := make(chan CommandResult, 1)
	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:
		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",
		})
	}

	// Also mark timeout for idempotency (allows retry after TTL)
	// executedCommands stays, will expire naturally
}

// Health
func handleHealth(w http.ResponseWriter, r *http.Request) {
	json.NewEncoder(w).Encode(map[string]interface{}{
		"status":   "ok",
		"routers": len(routers),
	})
}

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)
	http.HandleFunc("/api/events/", handleHTTPEvent)
	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)
		<-sigCh
		log.Println("Shutting down...")
		for _, router := range routers {
			router.Conn.Close(websocket.StatusNormalClosure, "server shutdown")
		}
		os.Exit(0)
	}()

	log.Printf("Server ready on port %d", cfg.WebsocketPort)
	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", cfg.WebsocketPort), nil))
}