// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package raft

import (
	"fmt"
	"io/ioutil"
	"os"
	"path"
	"strings"
	"testing"
	"time"

	"v.io/x/lib/vlog"
	"v.io/x/ref/test"
)

func tempDir(t *testing.T) string {
	// Get a temporary file name that is unlikely to clash with anyone else's.
	dir, err := ioutil.TempDir("", "raftstate")
	if err != nil {
		t.Fatalf("TempDir: %s", err)
	}
	os.Remove(dir)
	vlog.Infof("log is %s", dir)
	return dir
}

func compareLogs(t *testing.T, p persistent, expected []LogEntry, tag string) {
	found := 0
outer:
	for i := Index(0); i < 1000; i++ {
		le := p.Lookup(i)
		if le == nil {
			continue
		}
		for _, v := range expected {
			if v.Index != le.Index {
				continue
			}
			if v.Term != le.Term || string(v.Cmd) != string(le.Cmd) {
				t.Fatalf("%s: expected %v, got %v", tag, v, *le)
			}
			found++
			continue outer
		}
		t.Fatalf("unexpected: %v", *le)
	}
	if found != len(expected) {
		t.Fatalf("%s: not all entries found, got %d out of %v", tag, found, expected)
	}
}

// TestCreation verifies that a log is created when we start a fresh instance and is still there after we stop it.
func TestCreation(t *testing.T) {
	vlog.Infof("TestCreation")
	ctx, shutdown := test.V23Init()
	defer shutdown()
	td := tempDir(t)
	defer os.RemoveAll(td)

	// Launch a raft.  This should create the log file and append at least one control entry to it.
	config := RaftConfig{HostPort: "127.0.0.1:0", LogDir: td}
	r, err := newRaft(ctx, &config, new(client))
	if err != nil {
		t.Fatalf("newRaft: %s", err)
	}
	vlog.Infof("id is %s", r.Id())

	// Make sure the file is there.
	info, err := os.Stat(path.Join(td, "00000000000000000000.00000000000000000000.log"))
	if err != nil {
		t.Fatalf("File didn't last: %s", err)
	}
	if info.Size() == 0 {
		t.Fatalf("File too short: %d", info.Size())
	}
	r.Stop()

	// Make sure the file is still there after shutdown.
	info, err = os.Stat(path.Join(td, "00000000000000000000.00000000000000000000.log"))
	if err != nil {
		t.Fatalf("File didn't last: %s", err)
	}
	if info.Size() == 0 {
		t.Fatalf("File too short: %d", info.Size())
	}
	vlog.Infof("TestCreation passed")
}

// TestPersistence verifies that we store enough state to return to the previous log state
// after a orderly stop and restart.
//
// We also test that a truncated log remains truncated after an orderly stop and restart.
func TestPersistence(t *testing.T) {
	vlog.Infof("TestPersistence")
	ctx, shutdown := test.V23Init()
	defer shutdown()
	td := tempDir(t)
	defer os.RemoveAll(td)

	// Launch a raft.  This should create the file and populate the header with the
	// nil values.  The file should persist after the raft is closed.
	config := RaftConfig{HostPort: "127.0.0.1:0", LogDir: td}
	r, err := newRaft(ctx, &config, new(client))
	if err != nil {
		t.Fatalf("newRaft: %s", err)
	}
	vlog.Infof("id is %s", r.Id())

	// Set the persistent state.
	if err := r.p.SetCurrentTerm(2); err != nil {
		t.Fatalf("SetCurrentTerm: %s", err)
	}
	if err := r.p.SetVotedFor("whocares"); err != nil {
		t.Fatalf("SetVotedFor: %s", err)
	}
	cmds1 := []LogEntry{
		LogEntry{Term: 2, Index: 1, Cmd: []byte("cmd1")},
		LogEntry{Term: 2, Index: 2, Cmd: []byte("cmd2")},
		LogEntry{Term: 2, Index: 3, Cmd: []byte("cmd3")},
	}
	if err := r.p.AppendToLog(ctx, 0, 0, cmds1); err != nil {
		t.Fatalf("AppendToLog: %s", err)
	}
	cmds2 := []LogEntry{
		LogEntry{Term: 3, Index: 4, Cmd: []byte("cmd4")},
		LogEntry{Term: 3, Index: 5, Cmd: []byte("cmd5")},
		LogEntry{Term: 3, Index: 6, Cmd: []byte("cmd6")},
	}
	if err := r.p.AppendToLog(ctx, 2, 3, cmds2); err != nil {
		t.Fatalf("AppendToLog: %s", err)
	}
	info, err := os.Stat(path.Join(td, "00000000000000000000.00000000000000000000.log"))
	if err != nil {
		t.Fatalf("File didn't last: %s", err)
	}
	vlog.Infof("log size %d", info.Size())
	vlog.Infof("stopping %s", r.Id())
	r.Stop()

	// Reopen the persistent store and make sure it matches.
	r, err = newRaft(ctx, &config, new(client))
	if err != nil {
		t.Fatalf("newRaft: %s", err)
	}
	vlog.Infof("id is %s", r.Id())
	if r.p.CurrentTerm() != 2 {
		t.Fatalf("CurrentTerm: expected %d got %d", 2, r.p.CurrentTerm())
	}
	if r.p.VotedFor() != "whocares" {
		t.Fatalf("CurrentTerm: expected %s got %s", "whocares", r.p.VotedFor())
	}
	test := []LogEntry{
		LogEntry{Term: 2, Index: 1, Cmd: []byte("cmd1")},
		LogEntry{Term: 2, Index: 2, Cmd: []byte("cmd2")},
		LogEntry{Term: 2, Index: 3, Cmd: []byte("cmd3")},
		LogEntry{Term: 3, Index: 4, Cmd: []byte("cmd4")},
		LogEntry{Term: 3, Index: 5, Cmd: []byte("cmd5")},
		LogEntry{Term: 3, Index: 6, Cmd: []byte("cmd6")},
	}
	compareLogs(t, r.p, test, "after reopen")

	// Truncate the log by rewriting an index.
	if err := r.p.AppendToLog(ctx, 2, 3, []LogEntry{LogEntry{Term: 4, Index: 4, Cmd: []byte("cmd7")}}); err != nil {
		t.Fatalf("AppendToLog: %s", err)
	}
	test2 := []LogEntry{
		LogEntry{Term: 2, Index: 1, Cmd: []byte("cmd1")},
		LogEntry{Term: 2, Index: 2, Cmd: []byte("cmd2")},
		LogEntry{Term: 2, Index: 3, Cmd: []byte("cmd3")},
		LogEntry{Term: 4, Index: 4, Cmd: []byte("cmd7")},
	}
	compareLogs(t, r.p, test2, "after truncate")
	vlog.Infof("stopping %s", r.Id())
	r.Stop()

	// Make sure the log is still truncated when we reread it.
	r, err = newRaft(ctx, &config, new(client))
	if err != nil {
		t.Fatalf("newRaft: %s", err)
	}
	vlog.Infof("id is %s", r.Id())
	compareLogs(t, r.p, test2, "after truncate and close")
	vlog.Infof("stopping %s", r.Id())
	r.Stop()
	vlog.Infof("TestPersistence passed")
}

func getFileNames(t *testing.T, dir string) []string {
	d, err := os.Open(dir)
	if err != nil {
		t.Fatalf("opening %s: %s", dir, err)
	}
	defer d.Close()

	// Find all snapshot and log files.
	files, err := d.Readdirnames(0)
	if err != nil {
		t.Fatalf("reading %s: %s", dir, err)
	}

	return files
}

// waitForLeadership waits for r to be elected leader.
func waitForLeadership(r *raft, timeout time.Duration) bool {
	start := time.Now()
	for {
		_, role, _ := r.Status()
		if role == RoleLeader {
			return true
		}
		if time.Now().Sub(start) > timeout {
			return false
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// TestSnapshot sets the limit for taking a snapshot down to 1 record and makes sure that snapshots are taken.
func TestSnapshot(t *testing.T) {
	vlog.Infof("TestSnapshot")
	ctx, shutdown := test.V23Init()
	defer shutdown()
	td := tempDir(t)
	defer os.RemoveAll(td)

	config := RaftConfig{HostPort: "127.0.0.1:0", LogDir: td, SnapshotThreshold: 1}
	rs, _ := buildRafts(t, ctx, 1, &config)
	defer cleanUp(rs)
	r := rs[0]
	c1 := r.client
	vlog.Infof("id is %s", r.Id())

	// This should cause a snapshot.
	if apperr, err := r.Append(ctx, []byte("string1")); err != nil || apperr != nil {
		t.Fatalf("AppendToLog: %s/%s", apperr, err)
	}
	if apperr, err := r.Append(ctx, []byte("string2")); err != nil || apperr != nil {
		t.Fatalf("AppendToLog: %s/%s", apperr, err)
	}
	if apperr, err := r.Append(ctx, []byte("string3")); err != nil || apperr != nil {
		t.Fatalf("AppendToLog: %s/%s", apperr, err)
	}

	// Wait for the snapshot to appear.  It is a background task so just loop.
	deadline := time.Now().Add(time.Minute)
outer:
	for {
		time.Sleep(100 * time.Millisecond)
		files := getFileNames(t, td)
		for _, s := range files {
			if strings.HasSuffix(s, ".snap") {
				log := strings.TrimSuffix(s, ".snap") + ".log"
				for _, l := range files {
					if l == log {
						break outer
					}
				}
			}
		}
		if time.Now().After(deadline) {
			t.Fatalf("timeout waiting for snap")
		}
	}
	r.Stop()

	// Restart.  We should read and restart from the snapshot.
	rs, _ = buildRafts(t, ctx, 1, &config)
	defer cleanUp(rs)
	r = rs[0]
	c2 := r.client
	vlog.Infof("new id is %s", r.Id())

	// Wait for leadership.
	if !waitForLeadership(r, time.Minute) {
		t.Fatalf("didn't become leader")
	}

	// Wait to commit.
	time.Sleep(time.Second)
	c1.(*client).Compare(t, c2.(*client))

	vlog.Infof("TestSnapshot passed")
}

// TestRemoteSnapshot makes sure a member can be recovered from a snapshot taken elsewhere.
func TestRemoteSnapshot(t *testing.T) {
	vlog.Infof("TestRemoteSnapshot")
	ctx, shutdown := test.V23Init()
	defer shutdown()
	config := RaftConfig{HostPort: "127.0.0.1:0", SnapshotThreshold: 30}
	rs, cs := buildRafts(t, ctx, 3, &config)
	defer cleanUp(rs)

	// This should cause a few snapshots.
	for i := 0; i < 100; i++ {
		if apperr, err := rs[1].Append(ctx, []byte(fmt.Sprintf("string%d", i))); err != nil || apperr != nil {
			t.Fatalf("Append: %s/%s", apperr, err)
		}
	}
	vlog.Infof("Appends done")

	// Wait for them all to get to the same point.
	if !waitForLogAgreement(rs, time.Minute) {
		t.Fatalf("no log agreement")
	}
	t.Log("Logs agree")
	if !waitForAppliedAgreement(rs, cs, time.Minute) {
		t.Fatalf("no applied agreement")
	}
	vlog.Infof("Applies agree")

	// Stop a server and remove all its log and snapshot info, i.e., act like its starting
	// from scratch.
	rs[0].Stop()
	os.RemoveAll(rs[0].logDir)
	restart(t, ctx, rs, cs, rs[0])

	// Wait for them all to get to the same point.
	if !waitForLogAgreement(rs, time.Minute) {
		t.Fatalf("no log agreement")
	}
	if !waitForAppliedAgreement(rs, cs, time.Minute) {
		t.Fatalf("no applied agreement")
	}

	vlog.Infof("TestRemoteSnapshot passed")
}