// Copyright 2023 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//	http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package cloudstorage

import (
	"context"
	"fmt"
	"net/url"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/pingcap/ticdc/pkg/clock"
	commonType "github.com/pingcap/ticdc/pkg/common"
	appcontext "github.com/pingcap/ticdc/pkg/common/context"
	"github.com/pingcap/ticdc/pkg/config"
	"github.com/pingcap/ticdc/pkg/pdutil"
	"github.com/pingcap/ticdc/pkg/util"
	timodel "github.com/pingcap/tidb/pkg/meta/model"
	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/parser/mysql"
	"github.com/pingcap/tidb/pkg/parser/types"
	"github.com/stretchr/testify/require"
	"github.com/tikv/client-go/v2/oracle"
)

func testFilePathGenerator(ctx context.Context, t *testing.T, dir string) *FilePathGenerator {
	uri := fmt.Sprintf("file:///%s?flush-interval=2s", dir)
	storage, err := util.GetExternalStorageWithDefaultTimeout(ctx, uri)
	require.NoError(t, err)

	sinkURI, err := url.Parse(uri)
	require.NoError(t, err)
	replicaConfig := config.GetDefaultReplicaConfig()
	replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String())
	replicaConfig.Sink.Protocol = util.AddressOf(config.ProtocolOpen.String())
	replicaConfig.Sink.FileIndexWidth = util.AddressOf(6)
	cfg := NewConfig()
	err = cfg.Apply(ctx, sinkURI, replicaConfig.Sink, true)
	require.NoError(t, err)

	mockPDClock := pdutil.NewClock4Test()
	appcontext.SetService(appcontext.DefaultPDClock, mockPDClock)
	f := NewFilePathGenerator(commonType.ChangeFeedID{}, cfg, storage, ".json")
	return f
}

func TestGenerateDataFilePath(t *testing.T) {
	t.Parallel()
	ctx, cancel := context.WithCancel(context.TODO())
	defer cancel()

	table := VersionedTableName{
		TableNameWithPhysicTableID: commonType.TableName{
			Schema: "test",
			Table:  "table1",
		},
		TableInfoVersion: 5,
		DispatcherID:     commonType.NewDispatcherID(),
	}

	dir := t.TempDir()
	f := testFilePathGenerator(ctx, t, dir)
	f.versionMap[table] = table.TableInfoVersion
	date := f.GenerateDateStr()
	// date-separator: none
	path, err := f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/CDC_%s_000001.json", table.DispatcherID.String()), path)

	// date-separator: year
	mockClock := clock.NewMock()
	f = testFilePathGenerator(ctx, t, dir)
	f.versionMap[table] = table.TableInfoVersion
	f.config.DateSeparator = config.DateSeparatorYear.String()
	f.SetClock(pdutil.NewMonotonicClock(mockClock))
	mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC))
	date = f.GenerateDateStr()
	path, err = f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/2022/CDC_%s_000001.json", table.DispatcherID.String()), path)
	// year changed
	mockClock.Set(time.Date(2023, 1, 1, 0, 0, 20, 0, time.UTC))
	date = f.GenerateDateStr()
	path, err = f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/2023/CDC_%s_000001.json", table.DispatcherID.String()), path)

	// date-separator: month
	mockClock = clock.NewMock()
	f = testFilePathGenerator(ctx, t, dir)
	f.versionMap[table] = table.TableInfoVersion
	f.config.DateSeparator = config.DateSeparatorMonth.String()
	f.SetClock(pdutil.NewMonotonicClock(mockClock))

	mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC))
	date = f.GenerateDateStr()
	path, err = f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/2022-12/CDC_%s_000001.json", table.DispatcherID.String()), path)
	// month changed
	mockClock.Set(time.Date(2023, 1, 1, 0, 0, 20, 0, time.UTC))
	date = f.GenerateDateStr()
	path, err = f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/2023-01/CDC_%s_000001.json", table.DispatcherID.String()), path)

	// date-separator: day
	mockClock = clock.NewMock()
	f = testFilePathGenerator(ctx, t, dir)
	f.versionMap[table] = table.TableInfoVersion
	f.config.DateSeparator = config.DateSeparatorDay.String()
	f.SetClock(pdutil.NewMonotonicClock(mockClock))

	mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC))
	date = f.GenerateDateStr()
	path, err = f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/2022-12-31/CDC_%s_000001.json", table.DispatcherID.String()), path)
	// day changed
	mockClock.Set(time.Date(2023, 1, 1, 0, 0, 20, 0, time.UTC))
	date = f.GenerateDateStr()
	path, err = f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/2023-01-01/CDC_%s_000001.json", table.DispatcherID.String()), path)
}

func TestFetchIndexFromFileName(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithCancel(context.TODO())
	defer cancel()

	dir := t.TempDir()
	f := testFilePathGenerator(ctx, t, dir)
	testCases := []struct {
		fileName string
		wantErr  string
	}{
		{
			fileName: "CDC000011.json",
			wantErr:  "",
		},
		{
			fileName: "CDC1000000.json",
			wantErr:  "",
		},
		{
			fileName: "CDC1.json",
			wantErr:  "filename in storage sink is invalid",
		},
		{
			fileName: "cdc000001.json",
			wantErr:  "filename in storage sink is invalid",
		},
		{
			fileName: "CDC000005.xxx",
			wantErr:  "filename in storage sink is invalid",
		},
		{
			fileName: "CDChello.json",
			wantErr:  "filename in storage sink is invalid",
		},
	}

	for _, tc := range testCases {
		_, err := FetchIndexFromFileName(tc.fileName, f.extension)
		if len(tc.wantErr) != 0 {
			require.Contains(t, err.Error(), tc.wantErr)
		} else {
			require.NoError(t, err)
		}
	}
}

func TestGenerateDataFilePathWithIndexFile(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithCancel(context.TODO())
	defer cancel()

	dir := t.TempDir()
	f := testFilePathGenerator(ctx, t, dir)
	mockClock := clock.NewMock()
	f.config.DateSeparator = config.DateSeparatorDay.String()
	f.SetClock(pdutil.NewMonotonicClock(mockClock))

	mockClock.Set(time.Date(2023, 3, 9, 23, 59, 59, 0, time.UTC))
	dispatcherID := commonType.NewDispatcherID()
	table := VersionedTableName{
		TableNameWithPhysicTableID: commonType.TableName{
			Schema: "test",
			Table:  "table1",
		},
		TableInfoVersion: 5,
		DispatcherID:     dispatcherID,
	}
	f.versionMap[table] = table.TableInfoVersion
	date := f.GenerateDateStr()
	indexFilePath := f.GenerateIndexFilePath(table, date)
	err := f.storage.WriteFile(ctx, indexFilePath, []byte(fmt.Sprintf("CDC_%s_000005.json\n", dispatcherID.String())))
	require.NoError(t, err)

	dataFilePath, err := f.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/2023-03-09/CDC_%s_000006.json", dispatcherID.String()), dataFilePath)
}

func TestGenerateDataFilePathResyncIndexFile(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithCancel(context.TODO())
	defer cancel()

	dir := t.TempDir()
	f1 := testFilePathGenerator(ctx, t, dir)
	f2 := testFilePathGenerator(ctx, t, dir)

	dispatcherID := commonType.NewDispatcherID()
	table := VersionedTableName{
		TableNameWithPhysicTableID: commonType.TableName{
			Schema: "test",
			Table:  "table1",
		},
		TableInfoVersion: 5,
		DispatcherID:     dispatcherID,
	}
	f1.versionMap[table] = table.TableInfoVersion
	f2.versionMap[table] = table.TableInfoVersion

	date := ""
	indexFilePath := f1.GenerateIndexFilePath(table, date)

	// Simulate dispatcher moved between captures:
	// 1) f1 generates CDC_..._000001 and writes index file.
	dataFilePath, err := f1.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/CDC_%s_000001.json", dispatcherID.String()), dataFilePath)
	err = f1.storage.WriteFile(ctx, dataFilePath, []byte("test1"))
	require.NoError(t, err)
	err = f1.storage.WriteFile(ctx, indexFilePath, []byte(fmt.Sprintf("CDC_%s_000001.json\n", dispatcherID.String())))
	require.NoError(t, err)

	// 2) f2 continues from index file and generates CDC_..._000002, then updates index file.
	dataFilePath, err = f2.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	err = f2.storage.WriteFile(ctx, dataFilePath, []byte("test2"))
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/CDC_%s_000002.json", dispatcherID.String()), dataFilePath)
	err = f2.storage.WriteFile(ctx, indexFilePath, []byte(fmt.Sprintf("CDC_%s_000002.json\n", dispatcherID.String())))
	require.NoError(t, err)

	// 3) f1 generates again after being scheduled back. It must reconcile with index file and
	//    generate CDC_..._000003 instead of overwriting CDC_..._000002.
	dataFilePath, err = f1.GenerateDataFilePath(ctx, table, date)
	require.NoError(t, err)
	require.Equal(t, fmt.Sprintf("test/table1/5/CDC_%s_000003.json", dispatcherID.String()), dataFilePath)
}

func TestIsSchemaFile(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name   string
		path   string
		expect bool
	}{
		{
			"valid database schema <schema>/meta/",
			"schema2/meta/schema_123_0123456789.json", true,
		},
		{
			"valid table schema <schema>/<table>/meta/",
			"schema1/table1/meta/schema_123_0123456789.json", true,
		},
		{"valid special prefix", "meta/meta/schema_123_0123456789.json", true},
		{"valid schema1", "meta/schema_123_0123456789.json", true},
		{"missing field1", "meta/schema_012345678_.json", false},
		{"missing field2", "meta/schema_012345678.json", false},
		{"invalid checksum1", "meta/schema_123_012345678.json", false},
		{"invalid checksum2", "meta/schema_123_012a4567c9.json", false},
		{"invalid table version", "meta/schema_abc_0123456789.json", false},
		{"invalid extension1", "meta/schema_123_0123456789.txt", false},
		{"invalid extension2", "meta/schema_123_0123456789.json ", false},
		{"invalid path", "meta/schema1/schema_123_0123456789.json", false},
	}

	for _, tt := range tests {
		require.Equal(t, tt.expect, IsSchemaFile(tt.path),
			"testCase: %s, path: %v", tt.name, tt.path)
	}
}

func TestCheckOrWriteSchema(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	dir := t.TempDir()
	f := testFilePathGenerator(ctx, t, dir)

	var columns []*timodel.ColumnInfo
	ft := types.NewFieldType(mysql.TypeLong)
	ft.SetFlag(mysql.PriKeyFlag | mysql.NotNullFlag)
	col := &timodel.ColumnInfo{
		Name:         ast.NewCIStr("Id"),
		FieldType:    *ft,
		DefaultValue: 10,
	}
	columns = append(columns, col)
	tidbInfo := &timodel.TableInfo{
		ID:      20,
		Name:    ast.NewCIStr("table1"),
		Columns: columns,
		Version: 100,
	}
	tableInfo := commonType.WrapTableInfo("test", tidbInfo)

	table := VersionedTableName{
		TableNameWithPhysicTableID: tableInfo.TableName,
		TableInfoVersion:           100,
	}

	hasNewerSchemaVersion, err := f.CheckOrWriteSchema(ctx, table, tableInfo)
	require.NoError(t, err)
	require.False(t, hasNewerSchemaVersion)
	require.Equal(t, table.TableInfoVersion, f.versionMap[table])

	// test old dml file can be ignored
	table.TableInfoVersion = 99
	hasNewerSchemaVersion, err = f.CheckOrWriteSchema(ctx, table, tableInfo)
	require.NoError(t, err)
	require.True(t, hasNewerSchemaVersion)
	require.Equal(t, 1, len(f.versionMap))

	// test only table version changed, schema file should be reused
	table.TableInfoVersion = 101
	hasNewerSchemaVersion, err = f.CheckOrWriteSchema(ctx, table, tableInfo)
	require.NoError(t, err)
	require.False(t, hasNewerSchemaVersion)
	require.Equal(t, uint64(tidbInfo.Version), f.versionMap[table])

	dir = filepath.Join(dir, "test/table1/meta")
	files, err := os.ReadDir(dir)
	require.NoError(t, err)
	require.Equal(t, 1, len(files))

	// test schema file is invalid
	err = os.WriteFile(filepath.Join(dir,
		fmt.Sprintf("%s.tmp.%s", files[0].Name(), uuid.NewString())),
		[]byte("invalid"), 0o644)
	require.NoError(t, err)
	err = os.Remove(filepath.Join(dir, files[0].Name()))
	require.NoError(t, err)
	delete(f.versionMap, table)
	hasNewerSchemaVersion, err = f.CheckOrWriteSchema(ctx, table, tableInfo)
	require.NoError(t, err)
	require.False(t, hasNewerSchemaVersion)
	require.Equal(t, table.TableInfoVersion, f.versionMap[table])

	files, err = os.ReadDir(dir)
	require.NoError(t, err)
	require.Equal(t, 2, len(files))
}

func TestCheckOrWriteSchemaWriteSchemaIndex(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	dir := t.TempDir()
	f := testFilePathGenerator(ctx, t, dir)
	f.config.EnableSchemaIndexByGetObject = true

	var columns []*timodel.ColumnInfo
	ft := types.NewFieldType(mysql.TypeLong)
	ft.SetFlag(mysql.PriKeyFlag | mysql.NotNullFlag)
	col := &timodel.ColumnInfo{
		Name:         ast.NewCIStr("Id"),
		FieldType:    *ft,
		DefaultValue: 10,
	}
	columns = append(columns, col)
	tidbInfo := &timodel.TableInfo{
		ID:      20,
		Name:    ast.NewCIStr("table1"),
		Columns: columns,
		Version: 100,
	}
	tableInfo := commonType.WrapTableInfo("test", tidbInfo)
	table := VersionedTableName{
		TableNameWithPhysicTableID: tableInfo.TableName,
		TableInfoVersion:           100,
	}

	hasNewerSchemaVersion, err := f.CheckOrWriteSchema(ctx, table, tableInfo)
	require.NoError(t, err)
	require.False(t, hasNewerSchemaVersion)

	latestPath := GenerateSchemaMetaIndexFilePath("test", "table1")
	content, err := f.storage.ReadFile(ctx, latestPath)
	require.NoError(t, err)
	require.Equal(t, "SCHEMA00000000000000000001\n", string(content))
	var def TableDefinition
	def.FromTableInfo(tableInfo.GetSchemaName(), tableInfo.GetTableName(), tableInfo, table.TableInfoVersion, f.config.OutputColumnID)
	expectedSchemaPath, err := def.GenerateSchemaFilePath()
	require.NoError(t, err)
	versionIndexPath := GenerateSchemaMetaIndexDataFilePath("test", "table1", 1)
	content, err = f.storage.ReadFile(ctx, versionIndexPath)
	require.NoError(t, err)
	require.Equal(t, expectedSchemaPath+"\n", string(content))
}

func TestFetchSchemaMetaIndexFromFileName(t *testing.T) {
	t.Parallel()

	seq, err := FetchSchemaMetaIndexFromFileName("SCHEMA00000000000000000123")
	require.NoError(t, err)
	require.Equal(t, uint64(123), seq)

	_, err = FetchSchemaMetaIndexFromFileName("SCHEMA123")
	require.Error(t, err)

	_, err = FetchSchemaMetaIndexFromFileName("CDC00000000000000000123")
	require.Error(t, err)
}

func TestRemoveExpiredFilesWithoutPartition(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	dir := t.TempDir()
	uri := fmt.Sprintf("file:///%s?flush-interval=2s", dir)
	storage, err := util.GetExternalStorageWithDefaultTimeout(ctx, uri)
	require.NoError(t, err)
	sinkURI, err := url.Parse(uri)
	require.NoError(t, err)
	replicaConfig := config.GetDefaultReplicaConfig()
	replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorDay.String())
	replicaConfig.Sink.Protocol = util.AddressOf(config.ProtocolCsv.String())
	replicaConfig.Sink.FileIndexWidth = util.AddressOf(6)
	replicaConfig.Sink.CloudStorageConfig = &config.CloudStorageConfig{
		FileExpirationDays:  util.AddressOf(1),
		FileCleanupCronSpec: util.AddressOf("* * * * * *"),
	}
	cfg := NewConfig()
	err = cfg.Apply(ctx, sinkURI, replicaConfig.Sink, true)
	require.NoError(t, err)

	// generate some expired files
	filesWithoutPartition := []string{
		// schma1-table1
		"schema1/table1/5/2021-01-01/CDC000001.csv",
		"schema1/table1/5/2021-01-01/CDC000002.csv",
		"schema1/table1/5/2021-01-01/CDC000003.csv",
		"schema1/table1/5/2021-01-01/" + defaultIndexFileName, // index
		"schema1/table1/meta/schema_5_20210101.json",          // schema should never be cleaned
		// schma1-table2
		"schema1/table2/5/2021-01-01/CDC000001.csv",
		"schema1/table2/5/2021-01-01/CDC000002.csv",
		"schema1/table2/5/2021-01-01/CDC000003.csv",
		"schema1/table2/5/2021-01-01/" + defaultIndexFileName, // index
		"schema1/table2/meta/schema_5_20210101.json",          // schema should never be cleaned
	}
	for _, file := range filesWithoutPartition {
		err := storage.WriteFile(ctx, file, []byte("test"))
		require.NoError(t, err)
	}

	filesWithPartition := []string{
		// schma1-table1
		"schema1/table1/400200133/12/2021-01-01/20210101/CDC000001.csv",
		"schema1/table1/400200133/12/2021-01-01/20210101/CDC000002.csv",
		"schema1/table1/400200133/12/2021-01-01/20210101/CDC000003.csv",
		"schema1/table1/400200133/12/2021-01-01/20210101/" + defaultIndexFileName, // index
		"schema1/table1/meta/schema_5_20210101.json",                              // schema should never be cleaned
		// schma2-table1
		"schema2/table1/400200150/12/2021-01-01/20210101/CDC000001.csv",
		"schema2/table1/400200150/12/2021-01-01/20210101/CDC000002.csv",
		"schema2/table1/400200150/12/2021-01-01/20210101/CDC000003.csv",
		"schema2/table1/400200150/12/2021-01-01/20210101/" + defaultIndexFileName, // index
		"schema2/table1/meta/schema_5_20210101.json",                              // schema should never be cleaned
	}
	for _, file := range filesWithPartition {
		err := storage.WriteFile(ctx, file, []byte("test"))
		require.NoError(t, err)
	}

	filesNotExpired := []string{
		// schma1-table1
		"schema1/table1/5/2021-01-02/CDC000001.csv",
		"schema1/table1/5/2021-01-02/CDC000002.csv",
		"schema1/table1/5/2021-01-02/CDC000003.csv",
		"schema1/table1/5/2021-01-02/" + defaultIndexFileName, // index
		// schma1-table2
		"schema1/table2/5/2021-01-02/CDC000001.csv",
		"schema1/table2/5/2021-01-02/CDC000002.csv",
		"schema1/table2/5/2021-01-02/CDC000003.csv",
		"schema1/table2/5/2021-01-02/" + defaultIndexFileName, // index
	}
	for _, file := range filesNotExpired {
		err := storage.WriteFile(ctx, file, []byte("test"))
		require.NoError(t, err)
	}

	currTime := time.Date(2021, 1, 3, 0, 0, 0, 0, time.Local)
	checkpointTs := oracle.GoTimeToTS(currTime)
	cnt, err := RemoveExpiredFiles(ctx, commonType.ChangeFeedID{}, storage, cfg, checkpointTs)
	require.NoError(t, err)
	require.Equal(t, uint64(16), cnt)
}