package processor

import (
	_ "context"
	"encoding/csv"
	"fmt"
	"os"
	"testing"
	"time"
	_ "time"

	_ "github.com/aws/aws-sdk-go-v2/aws"
	_ "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/golang/mock/gomock"
	"github.com/stretchr/testify/assert"
	_ "github.com/stretchr/testify/assert"
	mock_data_processor "github.com/tidbcloud/observability/data-collector/test/mockdataprocessor"
)

func newMockDataProcessor(mock S3Client) DataProcessor {
	return DataProcessor{
		cli: mock,
	}
}

func TestUploadProcessedData(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	mockS3 := newMockS3Client(s3)
	processor := newMockDataProcessor(mockS3)
	prepareData("test", "ossinsight", "v1", "default.csv", [][]string{
		{"id", "title", "status", "user_id", "query_sql", "result", "created_at"},
		{"f5b3171a4f547b2b0aee9c327b92b3f", "bạn là ai", "error", "1170032", "null", "", "2023-03-07 12:21:42"},
	})
	defer deleteData("./test")
	t.Run("error", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("error")).Times(1)
		err := processor.UploadProcessedData("test", "test", "ossinsight", "v1")
		fmt.Println(err)
		assert.Error(t, err)
	})
	t.Run("success", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).Times(1)
		err := processor.UploadProcessedData("test", "test", "ossinsight", "v1")
		assert.NoError(t, err)
	})
}

func TestUploadMergedData(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	mockS3 := newMockS3Client(s3)
	processor := newMockDataProcessor(mockS3)
	prepareData("test", "ossinsight", "v1", "20230306.csv", [][]string{
		{"id", "title", "status", "user_id", "query_sql", "result", "created_at"},
		{"f5b3171a4f547b2b0aee9c327b92b3f", "bạn là ai", "error", "1170032", "null", "", "2023-03-07 12:21:42"},
	})
	defer deleteData("./test")

	t.Run("error", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("error")).Times(1)
		err := processor.UploadMergedData("test", "test", "ossinsight", "v1")
		assert.Error(t, err)
	})

	t.Run("success", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).Times(1)
		err := processor.UploadMergedData("test", "test", "ossinsight", "v1")
		assert.NoError(t, err)
	})
}

func TestUploadJoinedData(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	mockS3 := newMockS3Client(s3)
	processor := newMockDataProcessor(mockS3)
	prepareData("test", "ossinsight", "v1", "20230307.csv", [][]string{
		{"id", "title", "status", "user_id", "query_sql", "result", "created_at"},
		{"f5b3171a4f547b2b0aee9c327b92b3f", "bạn là ai", "error", "1170032", "null", "", "2023-03-07 12:21:42"},
	})
	defer deleteData("./test")
	t.Run("error", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("error")).Times(1)
		err := processor.UploadJoinedData("test", "test", "ossinsight", "v1")
		assert.Error(t, err)
	})
	t.Run("success", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).Times(1)
		err := processor.UploadJoinedData("test", "test", "ossinsight", "v1")
		assert.NoError(t, err)
	})
}

func TestUploadConstructedDB(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	mockS3 := newMockS3Client(s3)
	processor := newMockDataProcessor(mockS3)
	prepareData("test", "ossinsight", "v1/1234567", "20230307.csv", [][]string{
		{"id", "title", "status", "user_id", "query_sql", "result", "created_at"},
		{"f5b3171a4f547b2b0aee9c327b92b3f", "bạn là ai", "error", "1170032", "null", "", "2023-03-07 12:21:42"},
	})
	SaveCheckPoint("./test/ossinsight/v1/checkpoint.json", time.Now().Unix())
	defer deleteData("./test")
	t.Run("error", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("error")).Times(1)
		err := processor.UploadConstructedDB("test", "test", "ossinsight", "v1")
		assert.Error(t, err)
	})
	t.Run("success", func(t *testing.T) {
		s3.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).Times(1)
		err := processor.UploadConstructedDB("test", "test", "ossinsight", "v1")
		assert.NoError(t, err)
	})
}

func deleteData(dir string) {
	os.RemoveAll(dir)
}

func prepareData(dir string, ds string, v string, nm string, rows [][]string) {
	fn := fmt.Sprintf("%s/%s/%s/%s", dir, ds, v, nm)
	if _, err := os.Stat(fn); os.IsNotExist(err) {
		fmt.Println(err)
		os.MkdirAll(fmt.Sprintf("%s/%s/%s/", dir, ds, v), 0700)
	}
	f, _ := os.OpenFile(fn, os.O_WRONLY|os.O_CREATE, os.ModePerm)
	defer f.Close()

	w := csv.NewWriter(f)

	for _, r := range rows {
		WriteCsvFile(w, r)
	}
}