//nolint:testpackage,funlen
package impl

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	backup_mock "github.com/tidbcloud/dataflow-service/app/models/backup/mock"
	"github.com/tidbcloud/dataflow-service/app/models/br"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	blob_mock "github.com/tidbcloud/dataflow-service/pkg/blob/mock"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// ts is a helper that converts a Unix seconds value to a *timestamppb.Timestamp.
func ts(sec int64) *timestamppb.Timestamp {
	return timestamppb.New(time.Unix(sec, 0))
}

// tsRange builds a PiTRTimestampRange from start/end Unix seconds.
func tsRange(start, end int64) *domain.PiTRTimestampRange {
	return &domain.PiTRTimestampRange{StartTS: ts(start), EndTS: ts(end)}
}

// ─── mergeSingleRegionPiTRTimeRanges ──────────────────────────────────────────

func TestMergeSingleRegionPiTRTimeRanges(t *testing.T) {
	m := &pitBackupModel{}
	ctx := context.Background()

	tests := []struct {
		name     string
		input    []*domain.PiTRTimestampRange
		expected []*domain.PiTRTimestampRange
	}{
		{
			name:     "empty input returns empty slice",
			input:    []*domain.PiTRTimestampRange{},
			expected: []*domain.PiTRTimestampRange{},
		},
		{
			name:     "single range is returned unchanged",
			input:    []*domain.PiTRTimestampRange{tsRange(1000, 2000)},
			expected: []*domain.PiTRTimestampRange{tsRange(1000, 2000)},
		},
		{
			// Ranges touch exactly at 1500: [1500,2000] + [1000,1500] → [1000,2000]
			name:     "two touching ranges are merged",
			input:    []*domain.PiTRTimestampRange{tsRange(1500, 2000), tsRange(1000, 1500)},
			expected: []*domain.PiTRTimestampRange{tsRange(1000, 2000)},
		},
		{
			// Start of newer range falls inside older range: [1200,2000] + [1000,1500] → [1000,2000]
			name:     "two overlapping ranges are merged",
			input:    []*domain.PiTRTimestampRange{tsRange(1200, 2000), tsRange(1000, 1500)},
			expected: []*domain.PiTRTimestampRange{tsRange(1000, 2000)},
		},
		{
			// Gap between ranges (1400 < 1600): no merge
			name:     "two non-adjacent ranges are not merged",
			input:    []*domain.PiTRTimestampRange{tsRange(1600, 2000), tsRange(1000, 1400)},
			expected: []*domain.PiTRTimestampRange{tsRange(1600, 2000), tsRange(1000, 1400)},
		},
		{
			// [1500,2000] + [1000,1500] merge → [1000,2000]; [500,900] is separate
			name:  "three ranges: first two merge, third stays separate",
			input: []*domain.PiTRTimestampRange{tsRange(1500, 2000), tsRange(1000, 1500), tsRange(500, 900)},
			expected: []*domain.PiTRTimestampRange{
				tsRange(1000, 2000),
				tsRange(500, 900),
			},
		},
		{
			// All three ranges chain-touch: [1500,2000]+[1000,1500]+[500,1000] → [500,2000]
			name:     "three chain-touching ranges merge into one",
			input:    []*domain.PiTRTimestampRange{tsRange(1500, 2000), tsRange(1000, 1500), tsRange(500, 1000)},
			expected: []*domain.PiTRTimestampRange{tsRange(500, 2000)},
		},
		{
			// pitrTimeRange.StartTS == timeRanges[i].StartTS → strict ">" fails → no merge
			name:     "ranges with equal start timestamps are not merged",
			input:    []*domain.PiTRTimestampRange{tsRange(500, 2000), tsRange(500, 1000)},
			expected: []*domain.PiTRTimestampRange{tsRange(500, 2000), tsRange(500, 1000)},
		},
		{
			// pitrTimeRange.StartTS > timeRanges[i].EndTS → "<=" fails → no merge
			name:     "newer start strictly after older end is not merged",
			input:    []*domain.PiTRTimestampRange{tsRange(1800, 2000), tsRange(1000, 1500)},
			expected: []*domain.PiTRTimestampRange{tsRange(1800, 2000), tsRange(1000, 1500)},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := m.mergeSingleRegionPiTRTimeRanges(ctx, tt.input)
			assert.Len(t, got, len(tt.expected))
			for i, want := range tt.expected {
				assert.Equal(t, want.StartTS.GetSeconds(), got[i].StartTS.GetSeconds(), "range[%d] StartTS", i)
				assert.Equal(t, want.EndTS.GetSeconds(), got[i].EndTS.GetSeconds(), "range[%d] EndTS", i)
			}
		})
	}
}

// ─── groupPiTRTimeRangesByRegionV2 ────────────────────────────────────────────

// fakeCluster returns a minimal *mgmt.Cluster for use in test cases.
func fakeCluster(projectID uint64, provider, region, regionDisplay string) *mgmt.Cluster {
	return &mgmt.Cluster{
		ProjectId:     projectID,
		CloudProvider: provider,
		Region:        region,
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				RegionDisplayName: regionDisplay,
			},
		},
	}
}

// fakeCloudRegions returns a cloud-regions map for use in mock returns.
func fakeCloudRegions(provider string, regions map[string]string) map[string]map[string]*core.CloudRegion {
	inner := make(map[string]*core.CloudRegion, len(regions))
	for name, displayName := range regions {
		inner[name] = &core.CloudRegion{DisplayName: displayName}
	}
	return map[string]map[string]*core.CloudRegion{provider: inner}
}

// logBackupWithBuckets builds a PiTRTimeRange whose LogBackup has the given buckets.
func logBackupWithBuckets(startTS, endTS int64, buckets []entity.BackupSettingBucket) *domain.PiTRTimeRange {
	return &domain.PiTRTimeRange{
		StartTS: startTS,
		EndTS:   endTS,
		LogBackup: entity.LogBackup{
			BucketSettingSnapshot: &entity.BucketSetting{Buckets: buckets},
		},
	}
}

func TestGroupPiTRTimeRangesByRegionV2(t *testing.T) {
	type tsRange struct{ start, end int64 }
	type wantRegion struct {
		displayName string
		timeRanges  []tsRange
	}

	awsCluster := fakeCluster(1, "aws", "us-east-1", "US East")

	tests := map[string]struct {
		cluster        *mgmt.Cluster
		cloudRegions   map[string]map[string]*core.CloudRegion
		logBackups     []*domain.PiTRTimeRange
		checkpointFunc func(context.Context, entity.LogBackup, string, blob.GetStorageFunc) (int64, bool, error)
		wantErr        bool
		wantRegions    map[string]wantRegion
	}{
		"empty log backups returns empty result": {
			cluster:      awsCluster,
			cloudRegions: fakeCloudRegions("aws", nil),
			wantRegions:  map[string]wantRegion{},
		},
		"master bucket: region found in cloud map uses cloud display name": {
			cluster: awsCluster,
			cloudRegions: fakeCloudRegions("aws", map[string]string{
				"us-east-1": "US East (N. Virginia)",
			}),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{"us-east-1"}, Master: true},
				}),
			},
			wantRegions: map[string]wantRegion{
				"us-east-1": {displayName: "US East (N. Virginia)", timeRanges: []tsRange{{1000, 2000}}},
			},
		},
		// Primary bucket absent from cloud map → DisplayName falls back to region name.
		"primary bucket region absent from cloud map falls back to name as display name": {
			cluster: fakeCluster(1, "aws", "us-west-2", "US West"),
			cloudRegions: fakeCloudRegions("aws", map[string]string{
				"us-east-1": "US East (N. Virginia)",
			}),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{"us-west-2"}, Master: true},
				}),
			},
			wantRegions: map[string]wantRegion{
				"us-west-2": {displayName: "us-west-2", timeRanges: []tsRange{{1000, 2000}}},
			},
		},
		"secondary region: checkpoint found sets EndTS to checkpoint value": {
			cluster: awsCluster,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc) (int64, bool, error) {
				return 1800, true, nil
			},
			cloudRegions: fakeCloudRegions("aws", nil),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{"us-west-2"}},
				}),
			},
			wantRegions: map[string]wantRegion{
				// EndTS is the checkpoint (1800), not the original EndTS (2000)
				"us-west-2": {displayName: "us-west-2", timeRanges: []tsRange{{1000, 1800}}},
			},
		},
		"secondary region: checkpoint not found falls back to primary end timestamp": {
			cluster: awsCluster,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc) (int64, bool, error) {
				return 0, false, nil
			},
			cloudRegions: fakeCloudRegions("aws", nil),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{"us-west-2"}},
				}),
			},
			wantRegions: map[string]wantRegion{
				"us-west-2": {displayName: "us-west-2", timeRanges: []tsRange{{1000, 2000}}},
			},
		},
		"secondary region: checkpoint error falls back to primary end timestamp without propagating error": {
			cluster: awsCluster,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc) (int64, bool, error) {
				return 0, false, errors.New("storage unavailable")
			},
			cloudRegions: fakeCloudRegions("aws", nil),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{"us-west-2"}},
				}),
			},
			wantRegions: map[string]wantRegion{
				"us-west-2": {displayName: "us-west-2", timeRanges: []tsRange{{1000, 2000}}},
			},
		},
		// Checkpoint at 500 < startTS 1000 → inverted interval, region is skipped.
		"secondary region: checkpoint before range start is skipped": {
			cluster: awsCluster,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc) (int64, bool, error) {
				return 500, true, nil
			},
			cloudRegions: fakeCloudRegions("aws", nil),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{"us-west-2"}},
				}),
			},
			wantRegions: map[string]wantRegion{}, // skipped — no recoverable data in this interval
		},
		"multiple log backups for same region accumulate time ranges": {
			cluster:      awsCluster,
			cloudRegions: fakeCloudRegions("aws", nil),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(2000, 3000, []entity.BackupSettingBucket{
					{Regions: []string{"us-east-1"}, Master: true},
				}),
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{"us-east-1"}, Master: true},
				}),
			},
			wantRegions: map[string]wantRegion{
				"us-east-1": {displayName: "us-east-1", timeRanges: []tsRange{{2000, 3000}, {1000, 2000}}},
			},
		},
		"master and non-master buckets produce two separate region entries": {
			cluster: awsCluster,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, region string, _ blob.GetStorageFunc) (int64, bool, error) {
				if region == "us-west-2" {
					return 2500, true, nil
				}
				return 0, false, nil
			},
			cloudRegions: fakeCloudRegions("aws", map[string]string{
				"us-east-1": "US East (N. Virginia)",
				"us-west-2": "US West (Oregon)",
			}),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 3000, []entity.BackupSettingBucket{
					{Regions: []string{"us-east-1"}, Master: true},
					{Regions: []string{"us-west-2"}, Master: false},
				}),
			},
			wantRegions: map[string]wantRegion{
				"us-east-1": {displayName: "US East (N. Virginia)", timeRanges: []tsRange{{1000, 3000}}},
				"us-west-2": {displayName: "US West (Oregon)", timeRanges: []tsRange{{1000, 2500}}},
			},
		},
		"bucket with no regions is skipped": {
			cluster:      awsCluster,
			cloudRegions: fakeCloudRegions("aws", nil),
			logBackups: []*domain.PiTRTimeRange{
				logBackupWithBuckets(1000, 2000, []entity.BackupSettingBucket{
					{Regions: []string{}, Master: true},
				}),
			},
			wantRegions: map[string]wantRegion{},
		},
	}

	for name, tt := range tests {
		t.Run(name, func(t *testing.T) {
			m := &pitBackupModel{
				getCheckpointForCRPiTR: tt.checkpointFunc,
			}

			got, err := m.groupPiTRTimeRangesByRegionV2(context.Background(), tt.logBackups, tt.cluster, tt.cloudRegions)

			if tt.wantErr {
				assert.Error(t, err)
				return
			}

			assert.NoError(t, err)
			assert.Len(t, got, len(tt.wantRegions))

			gotMap := make(map[string]*domain.PiTRRegionTimeRange, len(got))
			for _, r := range got {
				gotMap[r.Region.Name] = r
			}

			for regionName, want := range tt.wantRegions {
				r, ok := gotMap[regionName]
				if !assert.True(t, ok, "expected region %q not in result", regionName) {
					continue
				}
				assert.Equal(t, want.displayName, r.Region.DisplayName, "region %q DisplayName", regionName)
				assert.Len(t, r.TimeRanges, len(want.timeRanges), "region %q TimeRanges count", regionName)
				for i, wtr := range want.timeRanges {
					if i >= len(r.TimeRanges) {
						break
					}
					assert.Equal(t, wtr.start, r.TimeRanges[i].StartTS.GetSeconds(), "region %q range[%d] StartTS", regionName, i)
					assert.Equal(t, wtr.end, r.TimeRanges[i].EndTS.GetSeconds(), "region %q range[%d] EndTS", regionName, i)
				}
			}
		})
	}
}

// ─── GetCheckpointForCRPiTR ───────────────────────────────────────────────────

func TestGetCheckpointForCRPiTR(t *testing.T) {
	const testRegion = "wus"

	logBackupWithSnapshot := func(storage string) entity.LogBackup {
		return entity.LogBackup{
			Storage: storage,
			BucketSettingSnapshot: &entity.BucketSetting{
				Buckets: []entity.BackupSettingBucket{
					{Regions: []string{testRegion}, Master: false, BucketId: 1},
				},
			},
		}
	}

	testBucket := &entity.Bucket{
		InfraProviderType: entity.InfraProviderTypeAWS,
		Provider:          string(entity.CloudProviderTypeAWS),
		Name:              "test-replica-bucket",
		ProjectID:         "1000",
	}

	tests := map[string]struct {
		logBackup      entity.LogBackup
		readFileData   []byte
		readFileErr    error
		wantCheckpoint int64
		wantFound      bool
		wantErr        bool
	}{
		"nil snapshot returns not-found": {
			logBackup: entity.LogBackup{
				Storage:               "s3://source-bucket/log",
				BucketSettingSnapshot: nil,
			},
			wantFound: false,
		},
		"normal checkpoint returns unix seconds": {
			logBackup:      logBackupWithSnapshot("s3://source-bucket/log"),
			readFileData:   []byte(`{"last_checkpoint": 443370000000000000}`),
			wantCheckpoint: util.TransToTS(443370000000000000),
			wantFound:      true,
		},
		"zero last_checkpoint treated as not-found": {
			logBackup:    logBackupWithSnapshot("s3://source-bucket/log"),
			readFileData: []byte(`{"last_checkpoint": 0}`),
			wantFound:    false,
		},
		"file not found returns not-found without error": {
			logBackup:   logBackupWithSnapshot("s3://source-bucket/log"),
			readFileErr: errors.New("Not Found"),
			wantFound:   false,
		},
		"region absent from snapshot returns not-found": {
			logBackup: entity.LogBackup{
				Storage: "s3://source-bucket/log",
				BucketSettingSnapshot: &entity.BucketSetting{
					Buckets: []entity.BackupSettingBucket{
						{Regions: []string{"other-region"}, Master: true},
					},
				},
			},
			wantFound: false,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			mockBucketRepo := repo_mock.NewBucketRepo(t)
			regionFound := false
			if tc.logBackup.BucketSettingSnapshot != nil {
				_, regionFound = tc.logBackup.BucketSettingSnapshot.GetBucketByRegion(testRegion)
			}
			if regionFound {
				mockBucketRepo.On("Load", mock.Anything, uint64(1)).Return(testBucket, nil)
			}

			mockStorage := blob_mock.NewStorage(t)
			if tc.readFileData != nil || tc.readFileErr != nil {
				mockStorage.On("ReadFile", mock.Anything, br.CRPiTRResumeStateFile).Return(tc.readFileData, tc.readFileErr)
			}

			storageFunc := func(_ context.Context, _ string, _ *blob.StorageOption) (blob.Storage, error) {
				return mockStorage, nil
			}

			m := &pitBackupModel{
				bucketRepo: mockBucketRepo,
				brCommon:   br.NewBRCommon(mockBucketRepo),
			}
			m.getCheckpointForCRPiTR = m.brCommon.GetCheckpointForCRPiTR
			checkpoint, found, err := m.getCheckpointForCRPiTR(context.Background(), tc.logBackup, testRegion, storageFunc)

			if tc.wantErr {
				assert.Error(t, err)
				return
			}
			assert.NoError(t, err)
			assert.Equal(t, tc.wantFound, found)
			if tc.wantFound {
				assert.Equal(t, tc.wantCheckpoint, checkpoint)
			}
		})
	}
}

// ─── createLogBackupEntity: RTCEnabled propagation ───────────────────────────

// TestCreateLogBackupEntity_RTCEnabled verifies that createLogBackupEntity copies
// BucketSettingSnapshot.RTCEnabled from backup_setting.Bucket so that new log
// backups immediately reflect the actual RTC state.
func TestCreateLogBackupEntity_RTCEnabled(t *testing.T) {
	rtcTrue := true
	rtcFalse := false

	tests := map[string]struct {
		rtcEnabled     *bool
		wantRTCEnabled *bool
	}{
		"rtc enabled: propagated to new log backup":  {rtcEnabled: &rtcTrue, wantRTCEnabled: pointer.To(true)},
		"rtc disabled: propagated to new log backup": {rtcEnabled: &rtcFalse, wantRTCEnabled: pointer.To(false)},
		"rtc nil: preserved as nil":                  {rtcEnabled: nil, wantRTCEnabled: nil},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			backupSettingMock := backup_mock.NewBackupSettingModel(t)
			logBackupRepoMock := repo_mock.NewLogBackupRepo(t)

			prefix := "test-prefix"
			backupSettingMock.On("QueryBackupSetting", mock.Anything, mock.Anything).Return(
				&domain.QueryBackupSettingRespV2{
					BackupSetting: &entity.BackupSetting{
						Bucket: &entity.BucketSetting{
							Prefix:     &prefix,
							RTCEnabled: tc.rtcEnabled,
						},
					},
				}, nil,
			).Once()

			var saved *entity.LogBackup
			logBackupRepoMock.On("Save", mock.Anything, mock.AnythingOfType("*entity.LogBackup")).
				Run(func(args mock.Arguments) { saved = args.Get(1).(*entity.LogBackup) }).
				Return(nil).Once()

			m := &pitBackupModel{
				backupSettingModel: backupSettingMock,
				logBackupRepo:      logBackupRepoMock,
			}
			cluster := &mgmt.Cluster{ClusterId: 1001, ProjectId: 200, OrganizationId: 100}

			lb, err := m.createLogBackupEntity(context.Background(), cluster)
			require.NoError(t, err)
			require.NotNil(t, lb.BucketSettingSnapshot)
			assert.Equal(t, tc.wantRTCEnabled, lb.BucketSettingSnapshot.RTCEnabled)
			require.NotNil(t, saved)
			assert.Equal(t, tc.wantRTCEnabled, saved.BucketSettingSnapshot.RTCEnabled)
		})
	}
}