main.go

package main

import (
	"fmt"
	"io"
	"net/http"
	"os"
	"time"

	"github.com/pranav718/hotaru/raft"
)

func httpDo(method, url string) (string, error) {
	req, err := http.NewRequest(method, url, nil)
	if err != nil {
		return "", err
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(body))
	}
	return string(body), nil
}

func httpPortFor(id int) int {
	return 8010 + id
}

func httpSet(port int, key, value string) error {
	url := fmt.Sprintf("http://127.0.0.1:%d/set?key=%s&value=%s", port, key, value)
	res, err := httpDo("POST", url)
	if err != nil {
		return fmt.Errorf("SET %s=%s on port %d: %v", key, value, port, err)
	}
	fmt.Printf("[HTTP] SET %s=%s on :%d → %s\n", key, value, port, res)
	return nil
}

func httpGet(port int, key string) (string, error) {
	url := fmt.Sprintf("http://127.0.0.1:%d/get?key=%s", port, key)
	res, err := httpDo("GET", url)
	if err != nil {
		return "", fmt.Errorf("GET %s on port %d: %v", key, port, err)
	}
	return res, nil
}

func httpDel(port int, key string) error {
	url := fmt.Sprintf("http://127.0.0.1:%d/del?key=%s", port, key)
	res, err := httpDo("DELETE", url)
	if err != nil {
		return fmt.Errorf("DEL %s on port %d: %v", key, port, err)
	}
	fmt.Printf("[HTTP] DEL %s on :%d → %s\n", key, port, res)
	return nil
}

func httpJoin(port int, id int, addr string) error {
	url := fmt.Sprintf("http://127.0.0.1:%d/join?id=%d&addr=%s", port, id, addr)
	res, err := httpDo("POST", url)
	if err != nil {
		return fmt.Errorf("JOIN node %d on port %d: %v", id, port, err)
	}
	fmt.Printf("[HTTP] JOIN node %d (%s) on :%d → %s\n", id, addr, port, res)
	return nil
}

func httpLeave(port int, id int) error {
	url := fmt.Sprintf("http://127.0.0.1:%d/leave?id=%d", port, id)
	res, err := httpDo("POST", url)
	if err != nil {
		return fmt.Errorf("LEAVE node %d on port %d: %v", id, port, err)
	}
	fmt.Printf("[HTTP] LEAVE node %d on :%d → %s\n", id, port, res)
	return nil
}

func main() {
	ports := map[int]string{
		0: "127.0.0.1:8000",
		1: "127.0.0.1:8001",
		2: "127.0.0.1:8002",
	}

	for i := 0; i < 4; i++ {
		os.Remove(fmt.Sprintf("raft_state_%d.json", i))
		os.Remove(fmt.Sprintf("raft_snapshot_%d.bin", i))
	}

	node0 := raft.NewRaftNode(0, []int{1, 2}, ports)
	node1 := raft.NewRaftNode(1, []int{0, 2}, ports)
	node2 := raft.NewRaftNode(2, []int{0, 1}, ports)

	nodes := []*raft.RaftNode{node0, node1, node2}

	for _, n := range nodes {
		if err := n.StartServer(); err != nil {
			fmt.Printf("error starting server %d: %v\n", n.GetId(), err)
			return
		}
		defer n.StopServer()
	}

	if len(os.Args) > 1 && os.Args[1] == "--dashboard" {
		fmt.Println("=====================================================")
		fmt.Println(" Hotaru Raft Cluster running in Dashboard Mode")
		fmt.Println(" Nodes: 0 (:8010), 1 (:8011), 2 (:8012)")
		fmt.Println(" SSE Event Streams: http://127.0.0.1:801x/events")
		fmt.Println(" Status Endpoints:   http://127.0.0.1:801x/status")
		fmt.Println(" Press Ctrl+C to stop.")
		fmt.Println("=====================================================")
		select {}
	}

	time.Sleep(600 * time.Millisecond)

	var leader *raft.RaftNode
	for _, node := range nodes {
		state, _ := node.GetState()
		if state == raft.Leader {
			leader = node
			break
		}
	}

	if leader == nil {
		fmt.Println("[Test] No leader found.")
		return
	}

	fmt.Println("=== Test 1: Write via leader HTTP API ===")
	if err := httpSet(httpPortFor(leader.GetId()), "x", "100"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	if err := httpSet(httpPortFor(leader.GetId()), "y", "200"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	time.Sleep(400 * time.Millisecond)

	fmt.Println("\n=== Test 2: Read via HTTP on all nodes (linearizable) ===")
	for i := 0; i < 3; i++ {
		val, err := httpGet(httpPortFor(i), "x")
		if err != nil {
			fmt.Printf("FAIL: node %d: %v\n", i, err)
		} else {
			fmt.Printf("[Test] Node %d: key 'x' = %q\n", i, val)
		}
		val, err = httpGet(httpPortFor(i), "y")
		if err != nil {
			fmt.Printf("FAIL: node %d: %v\n", i, err)
		} else {
			fmt.Printf("[Test] Node %d: key 'y' = %q\n", i, val)
		}
	}

	fmt.Println("\n=== Test 3: Write to follower via HTTP (expect proxy) ===")
	var followerPort int
	for i := 0; i < 3; i++ {
		if nodes[i] != leader {
			followerPort = httpPortFor(i)
			break
		}
	}
	if err := httpSet(followerPort, "z", "300"); err != nil {
		fmt.Printf("FAIL: write to follower proxied: %v\n", err)
		return
	}
	time.Sleep(200 * time.Millisecond)

	val, err := httpGet(httpPortFor(leader.GetId()), "z")
	if err != nil {
		fmt.Printf("FAIL: read z from leader: %v\n", err)
	} else {
		fmt.Printf("[Test] Key 'z' after follower proxy write = %q\n", val)
	}

	fmt.Println("\n=== Test 4: Delete via HTTP ===")
	if err := httpDel(httpPortFor(leader.GetId()), "y"); err != nil {
		fmt.Printf("FAIL: delete via leader: %v\n", err)
		return
	}
	time.Sleep(200 * time.Millisecond)

	for i := 0; i < 3; i++ {
		val, err := httpGet(httpPortFor(i), "y")
		if err != nil {
			fmt.Printf("[Test] Node %d: key 'y' deleted — %v\n", i, err)
		} else {
			fmt.Printf("[Test] Node %d: key 'y' = %q\n", i, val)
		}
	}

	fmt.Println("\n=== Test 5: Crash-recovery via HTTP ===")
	fmt.Println("[Test] Killing Node 2...")
	node2.StopServer()
	time.Sleep(300 * time.Millisecond)

	var newLeader *raft.RaftNode
	for _, node := range nodes {
		if node == node2 {
			continue
		}
		state, _ := node.GetState()
		if state == raft.Leader {
			newLeader = node
			break
		}
	}
	if newLeader == nil {
		fmt.Println("[Test] No new leader found after killing Node 2.")
		return
	}
	fmt.Printf("[Test] New leader is Node %d\n", newLeader.GetId())

	fmt.Println("[Test] Writing SET x 500 while Node 2 is offline...")
	if err := httpSet(httpPortFor(newLeader.GetId()), "x", "500"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	time.Sleep(500 * time.Millisecond)

	fmt.Println("[Test] Restarting Node 2...")
	newNode2 := raft.NewRaftNode(2, []int{0, 1}, ports)
	if err := newNode2.StartServer(); err != nil {
		fmt.Printf("error restarting server 2: %v\n", err)
		return
	}
	defer newNode2.StopServer()
	time.Sleep(600 * time.Millisecond)

	fmt.Println("[Test] Verifying Node 2 caught up via HTTP...")
	val, err = httpGet(httpPortFor(2), "x")
	if err != nil {
		fmt.Printf("FAIL: %v\n", err)
	} else {
		fmt.Printf("[Test] Restarted Node 2: key 'x' = %q (expected \"500\")\n", val)
	}
	val, err = httpGet(httpPortFor(2), "y")
	if err != nil {
		fmt.Printf("[Test] Restarted Node 2: key 'y' deleted (expected) — %v\n", err)
	} else {
		fmt.Printf("[Test] Restarted Node 2: key 'y' = %q\n", val)
	}

	fmt.Println("\n=== Test 6: Log Compaction & InstallSnapshot ===")
	// Step 1: Write keys to grow log
	fmt.Println("[Test] Writing a, b, c to leader...")
	if err := httpSet(httpPortFor(newLeader.GetId()), "a", "1"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	if err := httpSet(httpPortFor(newLeader.GetId()), "b", "2"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	if err := httpSet(httpPortFor(newLeader.GetId()), "c", "3"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	time.Sleep(300 * time.Millisecond)

	// Step 2: Trigger snapshot on the leader up to its lastApplied index
	leaderApplied := newLeader.GetLastApplied()
	fmt.Printf("[Test] Triggering snapshot on leader (Node %d) up to index %d...\n", newLeader.GetId(), leaderApplied)
	if err := newLeader.TakeSnapshot(leaderApplied); err != nil {
		fmt.Printf("FAIL: taking snapshot: %v\n", err)
		return
	}

	// Step 3: Stop one follower (newNode2, which is index 2)
	fmt.Println("[Test] Stopping Node 2...")
	newNode2.StopServer()
	time.Sleep(300 * time.Millisecond)

	// Step 4: Write more keys to the leader while Node 2 is offline
	fmt.Println("[Test] Writing d, e to leader while Node 2 is offline...")
	if err := httpSet(httpPortFor(newLeader.GetId()), "d", "4"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	if err := httpSet(httpPortFor(newLeader.GetId()), "e", "5"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	time.Sleep(300 * time.Millisecond)

	// Step 5: Trigger another snapshot on the leader to compact the new entries
	leaderApplied2 := newLeader.GetLastApplied()
	fmt.Printf("[Test] Triggering second snapshot on leader up to index %d...\n", leaderApplied2)
	if err := newLeader.TakeSnapshot(leaderApplied2); err != nil {
		fmt.Printf("FAIL: taking second snapshot: %v\n", err)
		return
	}

	// Step 6: Restart Node 2
	fmt.Println("[Test] Restarting Node 2...")
	node2Recovered := raft.NewRaftNode(2, []int{0, 1}, ports)
	if err := node2Recovered.StartServer(); err != nil {
		fmt.Printf("error restarting server 2: %v\n", err)
		return
	}
	defer node2Recovered.StopServer()
	time.Sleep(800 * time.Millisecond)

	// Step 7: Verify Node 2 successfully caught up and holds the state from the snapshot + new entries!
	fmt.Println("[Test] Verifying Node 2 caught up after InstallSnapshot...")
	for _, key := range []string{"a", "b", "c", "d", "e"} {
		val, err := httpGet(httpPortFor(2), key)
		if err != nil {
			fmt.Printf("FAIL: node 2 get key %s: %v\n", key, err)
		} else {
			fmt.Printf("[Test] Node 2: key %q = %q\n", key, val)
		}
	}

	fmt.Println("\n=== Test 7: Dynamic Cluster Membership Changes (/join & /leave) ===")
	// Step 1: Start Node 3 server
	ports[3] = "127.0.0.1:8003"
	node3 := raft.NewRaftNode(3, []int{0, 1, 2}, ports)
	if err := node3.StartServer(); err != nil {
		fmt.Printf("error starting server 3: %v\n", err)
		return
	}
	defer node3.StopServer()
	time.Sleep(300 * time.Millisecond)

	// Step 2: Dynamically join Node 3 to cluster via HTTP
	fmt.Println("[Test] Proposing JOIN for Node 3 via HTTP...")
	if err := httpJoin(httpPortFor(0), 3, "127.0.0.1:8003"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	time.Sleep(1 * time.Second)

	// Step 3: Write key to cluster and read from Node 3
	fmt.Println("[Test] Writing 'membership_key=active' to leader...")
	if err := httpSet(httpPortFor(newLeader.GetId()), "membership_key", "active"); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	time.Sleep(400 * time.Millisecond)

	fmt.Println("[Test] Reading 'membership_key' directly from newly joined Node 3...")
	val, err = httpGet(httpPortFor(3), "membership_key")
	if err != nil {
		fmt.Printf("FAIL: reading from Node 3: %v\n", err)
	} else {
		fmt.Printf("[Test] Node 3 read result: membership_key = %q\n", val)
	}

	// Step 4: Dynamically remove Node 3 via HTTP
	fmt.Println("[Test] Proposing LEAVE for Node 3 via HTTP...")
	if err := httpLeave(httpPortFor(0), 3); err != nil {
		fmt.Printf("FAIL: %v\n", err)
		return
	}
	time.Sleep(500 * time.Millisecond)

	fmt.Println("\n=== Test 8: High-Throughput & Eager Replication Benchmark ===")
	startBench := time.Now()
	numReqs := 20
	fmt.Printf("[Test] Firing %d rapid HTTP SET proposals to leader...\n", numReqs)
	for i := 0; i < numReqs; i++ {
		key := fmt.Sprintf("bench_%d", i)
		val := fmt.Sprintf("val_%d", i)
		if err := httpSet(httpPortFor(newLeader.GetId()), key, val); err != nil {
			fmt.Printf("FAIL: benchmark write %d: %v\n", i, err)
			return
		}
	}
	elapsed := time.Since(startBench)
	fmt.Printf("[Benchmark Result] Completed %d proposals in %v (avg %.2f ms per proposal)\n", numReqs, elapsed, float64(elapsed.Milliseconds())/float64(numReqs))

	fmt.Println("\n=== All HTTP integration tests complete ===")
}