package serf

import (
	"testing"
	"time"

	"github.com/hashicorp/memberlist"
	"basic.com/valib/serf.git/testutil"
	"basic.com/valib/serf.git/testutil/retry"
)

func TestSerf_joinLeave_ltime(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	ip2, returnFn2 := testutil.TakeIP()
	defer returnFn2()

	s1Config := testConfig(t, ip1)
	s2Config := testConfig(t, ip2)

	s1, err := Create(s1Config)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s1.Shutdown()

	s2, err := Create(s2Config)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s2.Shutdown()

	waitUntilNumNodes(t, 1, s1, s2)

	_, err = s1.Join([]string{s2Config.NodeName + "/" + s2Config.MemberlistConfig.BindAddr}, false)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	waitUntilNumNodes(t, 2, s1, s2)
	retry.Run(t, func(r *retry.R) {
		if s2.members[s1.config.NodeName].statusLTime != 1 {
			r.Fatalf("join time is not valid %d",
				s2.members[s1.config.NodeName].statusLTime)
		}

		if s2.clock.Time() <= s2.members[s1.config.NodeName].statusLTime {
			r.Fatalf("join should increment")
		}
	})

	oldClock := s2.clock.Time()

	if err := s1.Leave(); err != nil {
		t.Fatalf("err: %v", err)
	}

	retry.Run(t, func(r *retry.R) {
		// s1 clock should exceed s2 due to leave
		if s2.clock.Time() <= oldClock {
			r.Fatalf("leave should increment (%d / %d)",
				s2.clock.Time(), oldClock)
		}
	})
}

func TestSerf_join_pendingIntent(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	upsertIntent(s.recentIntents, "test", messageJoinType, 5, time.Now)
	n := memberlist.Node{Name: "test",
		Addr: nil,
		Meta: []byte("test"),
	}

	s.handleNodeJoin(&n)

	mem := s.members["test"]
	if mem.statusLTime != 5 {
		t.Fatalf("bad join time")
	}
	if mem.Status != StatusAlive {
		t.Fatalf("bad status")
	}
}

func TestSerf_join_pendingIntents(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	upsertIntent(s.recentIntents, "test", messageJoinType, 5, time.Now)
	upsertIntent(s.recentIntents, "test", messageLeaveType, 6, time.Now)
	n := memberlist.Node{Name: "test",
		Addr: nil,
		Meta: []byte("test"),
	}

	s.handleNodeJoin(&n)

	mem := s.members["test"]
	if mem.statusLTime != 6 {
		t.Fatalf("bad join time")
	}
	if mem.Status != StatusLeaving {
		t.Fatalf("bad status")
	}
}

func TestSerf_leaveIntent_bufferEarly(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	// Deliver a leave intent message early
	j := messageLeave{LTime: 10, Node: "test"}
	if !s.handleNodeLeaveIntent(&j) {
		t.Fatalf("should rebroadcast")
	}
	if s.handleNodeLeaveIntent(&j) {
		t.Fatalf("should not rebroadcast")
	}

	// Check that we buffered
	if leave, ok := recentIntent(s.recentIntents, "test", messageLeaveType); !ok || leave != 10 {
		t.Fatalf("bad buffer")
	}
}

func TestSerf_leaveIntent_oldMessage(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	s.members["test"] = &memberState{
		Member: Member{
			Status: StatusAlive,
		},
		statusLTime: 12,
	}

	j := messageLeave{LTime: 10, Node: "test"}
	if s.handleNodeLeaveIntent(&j) {
		t.Fatalf("should not rebroadcast")
	}

	if _, ok := recentIntent(s.recentIntents, "test", messageLeaveType); ok {
		t.Fatalf("should not have buffered intent")
	}
}

func TestSerf_leaveIntent_newer(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	s.members["test"] = &memberState{
		Member: Member{
			Status: StatusAlive,
		},
		statusLTime: 12,
	}

	j := messageLeave{LTime: 14, Node: "test"}
	if !s.handleNodeLeaveIntent(&j) {
		t.Fatalf("should rebroadcast")
	}

	if _, ok := recentIntent(s.recentIntents, "test", messageLeaveType); ok {
		t.Fatalf("should not have buffered intent")
	}

	if s.members["test"].Status != StatusLeaving {
		t.Fatalf("should update status")
	}

	if s.clock.Time() != 15 {
		t.Fatalf("should update clock")
	}
}

func TestSerf_joinIntent_bufferEarly(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	// Deliver a join intent message early
	j := messageJoin{LTime: 10, Node: "test"}
	if !s.handleNodeJoinIntent(&j) {
		t.Fatalf("should rebroadcast")
	}
	if s.handleNodeJoinIntent(&j) {
		t.Fatalf("should not rebroadcast")
	}

	// Check that we buffered
	if join, ok := recentIntent(s.recentIntents, "test", messageJoinType); !ok || join != 10 {
		t.Fatalf("bad buffer")
	}
}

func TestSerf_joinIntent_oldMessage(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	s.members["test"] = &memberState{
		statusLTime: 12,
	}

	j := messageJoin{LTime: 10, Node: "test"}
	if s.handleNodeJoinIntent(&j) {
		t.Fatalf("should not rebroadcast")
	}

	// Check that we didn't buffer anything
	if _, ok := recentIntent(s.recentIntents, "test", messageJoinType); ok {
		t.Fatalf("should not have buffered intent")
	}
}

func TestSerf_joinIntent_newer(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	s.members["test"] = &memberState{
		statusLTime: 12,
	}

	// Deliver a join intent message early
	j := messageJoin{LTime: 14, Node: "test"}
	if !s.handleNodeJoinIntent(&j) {
		t.Fatalf("should rebroadcast")
	}

	if _, ok := recentIntent(s.recentIntents, "test", messageJoinType); ok {
		t.Fatalf("should not have buffered intent")
	}

	if s.members["test"].statusLTime != 14 {
		t.Fatalf("should update join time")
	}

	if s.clock.Time() != 15 {
		t.Fatalf("should update clock")
	}
}

func TestSerf_joinIntent_resetLeaving(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	s.members["test"] = &memberState{
		Member: Member{
			Status: StatusLeaving,
		},
		statusLTime: 12,
	}

	j := messageJoin{LTime: 14, Node: "test"}
	if !s.handleNodeJoinIntent(&j) {
		t.Fatalf("should rebroadcast")
	}

	if _, ok := recentIntent(s.recentIntents, "test", messageJoinType); ok {
		t.Fatalf("should not have buffered intent")
	}

	if s.members["test"].statusLTime != 14 {
		t.Fatalf("should update join time")
	}
	if s.members["test"].Status != StatusAlive {
		t.Fatalf("should update status")
	}

	if s.clock.Time() != 15 {
		t.Fatalf("should update clock")
	}
}

func TestSerf_userEvent_oldMessage(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	// increase the ltime artificially
	s.eventClock.Witness(LamportTime(c.EventBuffer + 1000))

	msg := messageUserEvent{
		LTime:   1,
		Name:    "old",
		Payload: nil,
	}
	if s.handleUserEvent(&msg) {
		t.Fatalf("should not rebroadcast")
	}
}

func TestSerf_userEvent_sameClock(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	eventCh := make(chan Event, 4)
	c := testConfig(t, ip1)
	c.EventCh = eventCh
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	msg := messageUserEvent{
		LTime:   1,
		Name:    "first",
		Payload: []byte("test"),
	}
	if !s.handleUserEvent(&msg) {
		t.Fatalf("should rebroadcast")
	}
	msg = messageUserEvent{
		LTime:   1,
		Name:    "first",
		Payload: []byte("newpayload"),
	}
	if !s.handleUserEvent(&msg) {
		t.Fatalf("should rebroadcast")
	}
	msg = messageUserEvent{
		LTime:   1,
		Name:    "second",
		Payload: []byte("other"),
	}
	if !s.handleUserEvent(&msg) {
		t.Fatalf("should rebroadcast")
	}

	testUserEvents(t, eventCh,
		[]string{"first", "first", "second"},
		[][]byte{[]byte("test"), []byte("newpayload"), []byte("other")})
}

func TestSerf_query_oldMessage(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	c := testConfig(t, ip1)
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	// increase the ltime artificially
	s.queryClock.Witness(LamportTime(c.QueryBuffer + 1000))

	msg := messageQuery{
		LTime:   1,
		Name:    "old",
		Payload: nil,
	}
	if s.handleQuery(&msg) {
		t.Fatalf("should not rebroadcast")
	}
}

func TestSerf_query_sameClock(t *testing.T) {
	ip1, returnFn1 := testutil.TakeIP()
	defer returnFn1()

	eventCh := make(chan Event, 4)
	c := testConfig(t, ip1)
	c.EventCh = eventCh
	s, err := Create(c)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer s.Shutdown()

	msg := messageQuery{
		LTime:   1,
		ID:      1,
		Name:    "foo",
		Payload: []byte("test"),
	}
	if !s.handleQuery(&msg) {
		t.Fatalf("should rebroadcast")
	}
	if s.handleQuery(&msg) {
		t.Fatalf("should not rebroadcast")
	}
	msg = messageQuery{
		LTime:   1,
		ID:      2,
		Name:    "bar",
		Payload: []byte("newpayload"),
	}
	if !s.handleQuery(&msg) {
		t.Fatalf("should rebroadcast")
	}
	if s.handleQuery(&msg) {
		t.Fatalf("should not rebroadcast")
	}
	msg = messageQuery{
		LTime:   1,
		ID:      3,
		Name:    "baz",
		Payload: []byte("other"),
	}
	if !s.handleQuery(&msg) {
		t.Fatalf("should rebroadcast")
	}
	if s.handleQuery(&msg) {
		t.Fatalf("should not rebroadcast")
	}

	testQueryEvents(t, eventCh,
		[]string{"foo", "bar", "baz"},
		[][]byte{[]byte("test"), []byte("newpayload"), []byte("other")})
}