package lightning

import (
	"context"
	"fmt"
	"sort"
	"testing"

	lnconfig "github.com/pingcap/tidb/pkg/lightning/config"
	lnlog "github.com/pingcap/tidb/pkg/lightning/log"
	"github.com/pingcap/tidb/pkg/lightning/mydump"
	filter "github.com/pingcap/tidb/pkg/util/table-filter"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/pkgv2/log"
)

type fileScanSuite struct {
	suite.Suite
}

func TestFileScanSuite(t *testing.T) {
	suite.Run(t, new(fileScanSuite))
}

func (s *fileScanSuite) TestBasic() {
	var err error
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fileRouter, err := mydump.NewDefaultFileRouter(lnlog.Wrap(log.WithContextL(ctx)))
	s.Require().NoError(err)
	theFilter, err := filter.Parse(lnconfig.GetDefaultFilter())
	s.Require().NoError(err)
	testCases := []struct {
		DataFiles           int
		ExpectedDataFiles   int
		Suffix              string
		ExpectedCompression mydump.Compression
	}{
		{
			DataFiles:           3,
			ExpectedDataFiles:   3,
			Suffix:              "",
			ExpectedCompression: mydump.CompressionNone,
		},
		{
			DataFiles:           200,
			ExpectedDataFiles:   100,
			Suffix:              "",
			ExpectedCompression: mydump.CompressionNone,
		},
		{
			DataFiles:           3,
			ExpectedDataFiles:   3,
			Suffix:              ".gz",
			ExpectedCompression: mydump.CompressionGZ,
		},
		{
			DataFiles:           200,
			ExpectedDataFiles:   100,
			Suffix:              ".gz",
			ExpectedCompression: mydump.CompressionGZ,
		},
	}
	for _, tc := range testCases {
		mockLister := &mockObjectLister{
			prefix: "some_prefix",
			setting: &mockDataSourceSetting{
				TableSettings: []*mockTableSetting{
					{
						DBName:        "db01",
						TableName:     "tbl01",
						Suffix:        fmt.Sprintf("csv%s", tc.Suffix),
						DataFileCount: tc.DataFiles,
					},
					{
						DBName:        "db01",
						TableName:     "tbl02",
						Suffix:        fmt.Sprintf("csv%s", tc.Suffix),
						DataFileCount: tc.DataFiles,
					},
					{
						DBName:        "db02",
						TableName:     "tbl01",
						Suffix:        fmt.Sprintf("sql%s", tc.Suffix),
						DataFileCount: tc.DataFiles,
					},
					{
						DBName:        "db02",
						TableName:     "tbl02",
						Suffix:        fmt.Sprintf("sql%s", tc.Suffix),
						DataFileCount: tc.DataFiles,
					},
				},
			},
		}
		fileIter := &partialFileIterator{
			fileRouter:           fileRouter,
			filter:               theFilter,
			lister:               mockLister,
			maxDataFilesPerTable: 100,
		}
		fileScan := NewFileScan(fileRouter, fileIter, theFilter, "auto")
		dbMetas, err := fileScan.GetMDDatabaseMetas(ctx)
		s.Require().NoErrorf(err, "test case: %+v", tc)
		s.Require().Equalf(2, len(dbMetas), "test case: %+v", tc)
		sort.Slice(dbMetas, func(i, j int) bool {
			return dbMetas[i].Name < dbMetas[j].Name
		})
		for i, dbMeta := range dbMetas {
			s.Require().Equalf(fmt.Sprintf("db%02d", i+1), dbMeta.Name, "test case: %+v", tc)
			s.Require().Equalf(2, len(dbMeta.Tables), "test case: %+v", tc)
			tables := dbMeta.Tables
			sort.Slice(tables, func(i, j int) bool {
				return tables[i].Name < tables[j].Name
			})
			for j, tblMeta := range tables {
				s.T().Logf("dbName: %s; tblMeta.Name: %s", dbMeta.Name, tblMeta.Name)
				s.Require().Equalf(fmt.Sprintf("tbl%02d", j+1), tblMeta.Name, "test case: %+v", tc)
				s.Require().Equalf(tc.ExpectedDataFiles, len(tblMeta.DataFiles), "test case: %+v", tc)
				for _, dataFile := range tblMeta.DataFiles {
					switch i {
					case 0:
						s.Require().Equalf(dataFile.FileMeta.Type, mydump.SourceTypeCSV, "test case: %+v", tc)
					case 1:
						s.Require().Equalf(dataFile.FileMeta.Type, mydump.SourceTypeSQL, "test case: %+v", tc)
					default:
						s.FailNowf("unexpected index", "test case: %+v", tc)
					}
					s.Require().Equalf(dataFile.FileMeta.Compression, tc.ExpectedCompression, "test case: %+v", tc)
				}
			}
		}
	}
}

func (s *fileScanSuite) TestScanSubDir() {
	var err error
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fileRouter, err := mydump.NewDefaultFileRouter(lnlog.Wrap(log.WithContextL(ctx)))
	s.Require().NoError(err)
	theFilter, err := filter.Parse(lnconfig.GetDefaultFilter())
	s.Require().NoError(err)
	mockLister := &mockObjectLister{
		prefix:  "some_prefix",
		setting: nil,
		fileList: []string{
			"some_prefix/part01/shardingdb.tbl01.csv",
			"some_prefix/part02/shardingdb.tbl01.csv",
			"some_prefix/part03/shardingdb.tbl01.csv",
			"some_prefix/part04/shardingdb.tbl01.csv",
			"some_prefix/shardingdb-schema-create.sql",
			"some_prefix/shardingdb.tbl01-schema.sql",
		},
	}
	fileIter := &partialFileIterator{
		fileRouter:           fileRouter,
		filter:               theFilter,
		lister:               mockLister,
		maxDataFilesPerTable: 100,
	}
	fileScan := NewFileScan(fileRouter, fileIter, theFilter, "auto")
	dbMetas, err := fileScan.GetMDDatabaseMetas(ctx)
	s.Require().NoError(err)
	s.Require().Equal(1, len(dbMetas))
	dbMeta := dbMetas[0]
	s.Require().Equal("shardingdb", dbMeta.Name)
	s.Require().Equal(1, len(dbMeta.Tables))
	s.Require().NotNil(dbMeta.SchemaFile)
	s.Require().Equal("shardingdb-schema-create.sql", dbMeta.SchemaFile.FileMeta.Path)
	s.Require().Equal(mydump.SourceTypeSchemaSchema, dbMeta.SchemaFile.FileMeta.Type)
	tblMeta := dbMeta.Tables[0]
	s.Require().Equal("tbl01", tblMeta.Name)
	s.Require().NotNil(tblMeta.SchemaFile)
	s.Require().Equal("shardingdb.tbl01-schema.sql", tblMeta.SchemaFile.FileMeta.Path)
	s.Require().Equal(mydump.SourceTypeTableSchema, tblMeta.SchemaFile.FileMeta.Type)
	s.Require().Equal(4, len(tblMeta.DataFiles))
	dataFiles := tblMeta.DataFiles
	sort.Slice(dataFiles, func(i, j int) bool {
		return dataFiles[i].FileMeta.Path < dataFiles[j].FileMeta.Path
	})
	for i, dataFile := range dataFiles {
		s.Require().Equal(mydump.SourceTypeCSV, dataFile.FileMeta.Type)
		s.Require().Equal(fmt.Sprintf("part%02d/shardingdb.tbl01.csv", i+1), dataFile.FileMeta.Path)
	}
}