package main

import (
	"context"
	"errors"
	"io"
	"os"
	"path/filepath"
	"reflect"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
)

type MockObjectStorage struct {
	mock.Mock
}

func (m *MockObjectStorage) Object(ctx context.Context, bucket, key string) (*storage.Object, error) {
	//TODO implement me
	panic("implement me")
}

func (m *MockObjectStorage) ObjectExists(ctx context.Context, bucket, key string) (bool, error) {
	//TODO implement me
	panic("implement me")
}

func (m *MockObjectStorage) ObjectSize(ctx context.Context, bucket, key string) (int64, error) {
	//TODO implement me
	panic("implement me")
}

func (m *MockObjectStorage) Objects(ctx context.Context, bucket, prefix string) ([]storage.ObjectMeta, error) {
	args := m.Called(ctx, bucket, prefix)
	return args.Get(0).([]storage.ObjectMeta), args.Error(1)
}

func (m *MockObjectStorage) ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error) {
	//TODO implement me
	panic("implement me")
}

func (m *MockObjectStorage) ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(storage.ObjectMeta) bool) error {
	args := m.Called(ctx, bucket, prefix, callback)
	return args.Error(0)
}

func (m *MockObjectStorage) ObjectsCallbackWithStartKey(ctx context.Context, bucket, prefix, startKey string, callback func(storage.ObjectMeta) bool) error {
	args := m.Called(ctx, bucket, prefix, startKey, callback)
	return args.Error(0)
}

func (m *MockObjectStorage) Mkdir(ctx context.Context, bucket, key string, body io.Reader) error {
	//TODO implement me
	panic("implement me")
}

func (m *MockObjectStorage) SaveObject(ctx context.Context, bucket, key string, body io.Reader) error {
	args := m.Called(ctx, key, bucket, body)
	return args.Error(0)
}

func (m *MockObjectStorage) ListObject(ctx context.Context, bucket string) ([]storage.ObjectMeta, error) {
	args := m.Called(ctx, bucket)
	return args.Get(0).([]storage.ObjectMeta), args.Error(1)
}

func (m *MockObjectStorage) ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) (objects []storage.ObjectMeta, nextPageToken string, err error) {
	args := m.Called(ctx, bucket, prefix, pageSize, pageToken)
	return args.Get(0).([]storage.ObjectMeta), args.String(1), args.Error(2)
}

func (m *MockObjectStorage) ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]storage.ObjectMeta, error) {
	args := m.Called(ctx, bucket, prefix)
	return args.Get(0).([]storage.ObjectMeta), args.Error(1)
}

func (m *MockObjectStorage) PutObject(ctx context.Context, key string, bucket string, filepath string) error {
	args := m.Called(ctx, key, bucket, filepath)
	return args.Error(0)
}

func (m *MockObjectStorage) DeleteObject(ctx context.Context, bucket string, key string) error {
	args := m.Called(ctx, bucket, key)
	return args.Error(0)
}

func (m *MockObjectStorage) DeleteObjects(ctx context.Context, bucket string, keys []string) error {
	args := m.Called(ctx, bucket, keys)
	return args.Error(0)
}

func (m *MockObjectStorage) GetObject(ctx context.Context, key string, bucket string) (io.ReadCloser, error) {
	args := m.Called(ctx, key, bucket)
	return args.Get(0).(io.ReadCloser), args.Error(1)
}

func (m *MockObjectStorage) Close() error {
	args := m.Called()
	return args.Error(0)
}

type MockStreamParser struct {
	mock.Mock
}

func (m *MockStreamParser) ReadNext() (*SlowLog, error) {
	args := m.Called()
	if args.Get(0) == nil {
		return nil, args.Error(1)
	}
	return args.Get(0).(*SlowLog), args.Error(1)
}

func TestGetConfigMapData(t *testing.T) {
	tempDir := t.TempDir()

	tests := []struct {
		name        string
		setup       func()
		cleanup     func()
		wantErr     bool
		wantNil     bool
		description string
	}{
		{
			name: "ValidConfigFile",
			setup: func() {
				content := `{
					"tenants": {
						"tenant1": {
							"01": {
								"frequency": "10m",
								"taskType": "slow_query_etl"
							}
						}
					}
				}`
				confPath = filepath.Join(tempDir, "valid_config.json")
				if err := os.WriteFile(confPath, []byte(content), 0644); err != nil {
					t.Fatalf("Failed to create test config file: %v", err)
				}
			},
			cleanup: func() {
				os.Remove(confPath)
			},
			wantErr:     false,
			wantNil:     false,
			description: "test valida json config file",
		},
		{
			name: "EmptyConfigPath",
			setup: func() {
				confPath = ""
			},
			cleanup:     func() {},
			wantErr:     true,
			wantNil:     true,
			description: "test empty config file path",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.setup()
			defer tt.cleanup()

			got, err := GetConfigData()

			if (err != nil) != tt.wantErr {
				t.Errorf("%s: GetConfigMapData() error = %v, wantErr %v", tt.description, err, tt.wantErr)
				return
			}

			if (got == nil) != tt.wantNil {
				t.Errorf("%s: GetConfigMapData() = %v, wantNil %v", tt.description, got, tt.wantNil)
			}

			if !tt.wantErr && !tt.wantNil {
				if got == nil {
					t.Error("Expected non-nil config, got nil")
					return
				}

				if tenant, ok := got.Tenants["tenant1"]; !ok {
					t.Error("Expected tenant1 in config, not found")
				} else {
					if task, ok := tenant["01"]; !ok {
						t.Error("Expected task 01 in tenant1, not found")
					} else {
						if task.Frequency != "10m" || task.TaskType != "slow_query_etl" {
							t.Errorf("Unexpected task values: got %+v", task)
						}
					}
				}
			}
		})
	}
}

func TestExtractTimestamp(t *testing.T) {
	tests := []struct {
		name     string
		filename string
		want     int64
		wantErr  bool
	}{
		{
			name:     "stand name",
			filename: "1709025967-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz",
			want:     1709025967,
			wantErr:  false,
		},
		{
			name:     "no suffix",
			filename: "1709025967-2b64e974",
			want:     1709025967,
			wantErr:  false,
		},
		{
			name:     "invalid",
			filename: "invalid.log",
			want:     0,
			wantErr:  true,
		},
		{
			name:     "ts is no number format",
			filename: "abc-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz",
			want:     0,
			wantErr:  true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := extractTimestamp(tt.filename)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestFilterNewerFiles(t *testing.T) {
	tests := []struct {
		name          string
		mockObjects   []storage.ObjectMeta
		mockError     error
		markerFile    string
		expectedFiles []string
		expectedErr   bool
	}{
		{
			name: "new file",
			mockObjects: []storage.ObjectMeta{
				{Key: "1709025965-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"},
				{Key: "1709025966-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"},
				{Key: "1709025967-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"},
				{Key: "1709025968-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"},
				{Key: "1709025969-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"},
			},
			markerFile:    "1709025966-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz",
			expectedFiles: []string{"1709025967-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz", "1709025968-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz", "1709025969-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"},
			expectedErr:   false,
		},
		{
			name:          "no new file",
			mockObjects:   []storage.ObjectMeta{{Key: "1709025967-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"}},
			markerFile:    "1709025967-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz",
			expectedFiles: []string{},
			expectedErr:   false,
		},
		{
			name:        "error",
			mockError:   errors.New("list error"),
			expectedErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			mockStorage := new(MockObjectStorage)
			mockStorage.On("ListObjectsPaged", mock.Anything, "source-bucket", "/source-bucket/share_data/meta/clusterid/logs/slow_log/", 1000, "").
				Return(tt.mockObjects, "", tt.mockError)

			files, err := FilterNewerFiles(context.Background(), mockStorage, "source-bucket", "/source-bucket/share_data/meta/clusterid/logs/slow_log/", tt.markerFile)

			if tt.expectedErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.expectedFiles, files)
			}
		})
	}
}

func TestCheckForNewFiles(t *testing.T) {
	const (
		sourceBucket        = "source-bucket"
		sourceMarkPointPath = "/source-bucket/share_data/meta/clusterid/markpoint/"
		sourceLogPath       = "/source-bucket/share_data/meta/clusterid/logs/slow_log/"
	)

	originalTimeNow := timeNow
	timeNow = func() time.Time { return time.Unix(1709025967, 0) }
	defer func() { timeNow = originalTimeNow }()

	markerKey := sourceMarkPointPath + "1709025966-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"
	startKey := sourceLogPath + "1709025966-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"
	initialStartKey := sourceLogPath + "170902"

	tests := []struct {
		name       string
		markerObjs []storage.ObjectMeta
		setupMock  func(*MockObjectStorage)
		want       bool
		wantErr    bool
	}{
		{
			name:       "has newer file after marker",
			markerObjs: []storage.ObjectMeta{{Key: markerKey}},
			setupMock: func(mockStorage *MockObjectStorage) {
				mockStorage.On("ObjectsCallbackWithStartKey", mock.Anything, sourceBucket, sourceLogPath, startKey, mock.Anything).
					Run(func(args mock.Arguments) {
						callback := args.Get(4).(func(storage.ObjectMeta) bool)
						callback(storage.ObjectMeta{Key: startKey})
						callback(storage.ObjectMeta{Key: sourceLogPath + "1709025967-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"})
					}).
					Return(nil)
			},
			want: true,
		},
		{
			name:       "no newer file after marker",
			markerObjs: []storage.ObjectMeta{{Key: markerKey}},
			setupMock: func(mockStorage *MockObjectStorage) {
				mockStorage.On("ObjectsCallbackWithStartKey", mock.Anything, sourceBucket, sourceLogPath, startKey, mock.Anything).
					Run(func(args mock.Arguments) {
						callback := args.Get(4).(func(storage.ObjectMeta) bool)
						callback(storage.ObjectMeta{Key: startKey})
					}).
					Return(nil)
			},
			want: false,
		},
		{
			name:       "no marker and has files",
			markerObjs: nil,
			setupMock: func(mockStorage *MockObjectStorage) {
				mockStorage.On("ObjectsCallbackWithStartKey", mock.Anything, sourceBucket, sourceLogPath, initialStartKey, mock.Anything).
					Run(func(args mock.Arguments) {
						callback := args.Get(4).(func(storage.ObjectMeta) bool)
						callback(storage.ObjectMeta{Key: sourceLogPath + "1709025965-2b64e974-3b82-45cc-aaca-b0b98b1d4796.log.gz"})
					}).
					Return(nil)
			},
			want: true,
		},
		{
			name:       "no marker and no files",
			markerObjs: nil,
			setupMock: func(mockStorage *MockObjectStorage) {
				mockStorage.On("ObjectsCallbackWithStartKey", mock.Anything, sourceBucket, sourceLogPath, initialStartKey, mock.Anything).
					Return(nil)
			},
			want: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			mockStorage := new(MockObjectStorage)
			mockStorage.On("Objects", mock.Anything, sourceBucket, sourceMarkPointPath).
				Return(tt.markerObjs, nil)
			tt.setupMock(mockStorage)

			got, err := checkForNewFiles(context.Background(), mockStorage, sourceBucket, sourceMarkPointPath, sourceLogPath)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want, got)
			}
		})
	}
}

func TestInitialListingPrefix(t *testing.T) {
	originalTimeNow := timeNow
	originalInitialMarkerPrefix := initialMarkerPrefix
	timeNow = func() time.Time { return time.Unix(1709025967, 0) }
	defer func() {
		timeNow = originalTimeNow
		initialMarkerPrefix = originalInitialMarkerPrefix
	}()

	assert.Equal(t, int64(170902), initialListingPrefix())

	initialMarkerPrefix = 170901
	assert.Equal(t, int64(170901), initialListingPrefix())
}

func TestSplitFiles(t *testing.T) {
	tests := []struct {
		name   string
		files  []string
		groups int
		want   [][]string
	}{
		{
			name:   "4 files 3 groups",
			files:  []string{"file1", "file2", "file3", "file4"},
			groups: 3,
			want: [][]string{
				{"file1", "file2"},
				{"file3"},
				{"file4"},
			},
		},
		{
			name:   "5 files 3 groups",
			files:  []string{"file1", "file2", "file3", "file4", "file5"},
			groups: 3,
			want: [][]string{
				{"file1", "file2"},
				{"file3", "file4"},
				{"file5"},
			},
		},
		{
			name:   "7 files 3 groups",
			files:  []string{"file1", "file2", "file3", "file4", "file5", "file6", "file7"},
			groups: 3,
			want: [][]string{
				{"file1", "file2", "file3"},
				{"file4", "file5"},
				{"file6", "file7"},
			},
		},
		{
			name:   "file number less than group number",
			files:  []string{"file1", "file2"},
			groups: 3,
			want: [][]string{
				{"file1"},
				{"file2"},
			},
		},
		{
			name:   "file number less than group number",
			files:  []string{"file1"},
			groups: 5,
			want: [][]string{
				{"file1"},
			},
		},
		{
			name:   "file number less than group number",
			files:  []string{"file1", "file2", "file3"},
			groups: 3,
			want: [][]string{
				{"file1"},
				{"file2"},
				{"file3"},
			},
		},
		{
			name:   "file number equal to group number",
			files:  []string{"file1", "file2"},
			groups: 2,
			want: [][]string{
				{"file1"},
				{"file2"},
			},
		},
		{
			name:   "empty",
			files:  []string{},
			groups: 3,
			want:   nil,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := splitFiles(tt.files, tt.groups)
			if !reflect.DeepEqual(got, tt.want) {
				t.Errorf("splitFiles() = %v, want %v", got, tt.want)
			}
		})
	}
}