package sink

import (
	"fmt"
	"testing"
	"time"

	"code.justin.tv/ear/kinesis/combiner"
	"code.justin.tv/release/trace/internal/kinesalite"
	"code.justin.tv/release/trace/internal/pipeline"
)

func TestKinesisSink(t *testing.T) {
	var (
		stream = "test-sink"
		n      = 100
	)

	kinesalite.Setup(t, stream, n)
	defer kinesalite.Teardown(t, stream)

	s, errors := fakeKinesisSink(stream)
	c := make(chan *combiner.Blob, n)
	want := make([][]byte, n)

	for i := 0; i < n; i++ {
		want[i] = []byte(fmt.Sprintf("test%d", i))
		c <- &combiner.Blob{
			Key:  fmt.Sprintf("%d", i),
			Data: want[i],
		}
	}

	close(c)
	s(c)
	kinesalite.AssertStreamContains(t, kinesalite.TestClient(), stream, want)
	for _, err := range *errors {
		t.Errorf("sink err=%q", err)
	}
}

// KinesisSink should flush on a timer
func TestKinesisSinkTimedFlush(t *testing.T) {
	var (
		stream = "test-sink"
		n      = 10
	)

	kinesalite.Setup(t, stream, 1)
	defer kinesalite.Teardown(t, stream)

	s, errors := fakeKinesisSink(stream)
	c := make(chan *combiner.Blob, n)
	want := make([][]byte, n)

	for i := 0; i < n; i++ {
		want[i] = []byte(fmt.Sprintf("test%d", i))
		c <- &combiner.Blob{
			Key:  fmt.Sprintf("%d", i),
			Data: want[i],
		}
	}
	done := make(chan struct{})
	go func() {
		s(c)
		close(done)
	}()
	time.Sleep(defaultFlushInterval * 2)
	kinesalite.AssertStreamContains(t, kinesalite.TestClient(), stream, want)
	close(c)
	<-done
	for _, err := range *errors {
		t.Errorf("sink err=%q", err)
	}
}

// KinesisSink should be able to flush 1MB records just fine
func TestKinesisSinkFlushBigRecords(t *testing.T) {
	var (
		stream      = "test-sink"
		n           = 20
		oneMegabyte = 1000000
	)

	kinesalite.Setup(t, stream, 1)
	defer kinesalite.Teardown(t, stream)

	s, errors := fakeKinesisSink(stream)
	c := make(chan *combiner.Blob, n)
	want := make([][]byte, n)

	for i := 0; i < n; i++ {
		want[i] = make([]byte, oneMegabyte)
		c <- &combiner.Blob{
			Key:  fmt.Sprintf("%d", i),
			Data: want[i],
		}
	}
	close(c)
	s(c)
	kinesalite.AssertStreamContains(t, kinesalite.TestClient(), stream, want)
	for _, err := range *errors {
		t.Errorf("sink err=%q", err)
	}
}

// KinesisSink should not error if it doesnt received any data for several flush cycles
func TestKinesisSinkNoData(t *testing.T) {
	var (
		stream = "test-sink"
		n      = 10
	)

	kinesalite.Setup(t, stream, n)
	defer kinesalite.Teardown(t, stream)

	s, errors := fakeKinesisSink(stream)
	c := make(chan *combiner.Blob, n)
	time.Sleep(defaultFlushInterval * 5)
	close(c)
	s(c)
	for _, err := range *errors {
		t.Errorf("sink err=%q", err)
	}
}

// KinesisSink should not error if it doesnt received any data for several flush cycles
func TestKinesisSinkPartialFailure(t *testing.T) {
	t.Skipf("Kinesalite does not properly implement partial failures")

	stream := "test-sink"
	kinesalite.Setup(t, stream, 1)
	defer kinesalite.Teardown(t, stream)

	s, _ := fakeKinesisSink(stream)
	c := make(chan *combiner.Blob, 5)
	// add 4 valid blobs and one invalid one
	c <- &combiner.Blob{Key: "1", Data: make([]byte, 100)}
	c <- &combiner.Blob{Key: "1", Data: make([]byte, 100)}
	c <- &combiner.Blob{Key: string(make([]byte, 1000)), Data: make([]byte, 100)}
	c <- &combiner.Blob{Key: "1", Data: make([]byte, 100)}
	c <- &combiner.Blob{Key: "1", Data: make([]byte, 100)}
	close(c)
	s(c)
	want := make([][]byte, 4)
	for i := 0; i < 4; i++ {
		want[i] = make([]byte, 100)
	}

	kinesalite.AssertStreamContains(t, kinesalite.TestClient(), stream, want)
}

func fakeKinesisSink(streamName string) (Sink, *[]error) {
	client := kinesalite.TestClient()
	cfg := NewKinesisConfig(streamName)
	var errors *[]error
	cfg.OnError, errors = pipeline.StoreErrors()

	sink := &kinesisSink{
		cfg:    cfg,
		client: client,
		buffer: []*combiner.Blob{},
	}
	return sink.Consume, errors
}
