package impl

import (
	"context"
	"errors"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	restore_mock "github.com/tidbcloud/dataflow-service/app/repo/restore/mock"
	"github.com/tidbcloud/dataflow-service/internal/config"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

type pitrRestoreTestSuite struct {
	suite.Suite
}

func TestPiTRestoreTestSuite(t *testing.T) {
	suite.Run(t, new(pitrRestoreTestSuite))
}

// newTestPiTRestore constructs a minimal PiTRestore with a mock repo for unit testing.
func newTestPiTRestore(restoreRepo *restore_mock.RestoreRepo) *PiTRestore {
	return &PiTRestore{
		baseRestoreModel: baseRestoreModel{
			restoreRepo:           restoreRepo,
			dedicatedGlobalClient: nil,
		},
	}
}

// minimalCreateParam builds a createRestoreParam sufficient for makeCreateRestoreEntity
// to succeed when the serialization check passes (existing cluster path).
func minimalCreateParam(logBackupID, backupID entity.ID) *createRestoreParam {
	return &createRestoreParam{
		logBackup: &entity.LogBackup{
			Updatable: entity.Updatable{Base: entity.Base{ID: logBackupID}},
		},
		backup: &entity.Backup{
			Updatable: entity.Updatable{Base: entity.Base{ID: backupID}},
		},
		createRestoreReq: &domain.CreateRestoreReqV2{
			ClusterConfig: domain.ClusterConfig{
				ClusterID: 99,
				Region:    "us-west-2",
			},
		},
	}
}

func (s *pitrRestoreTestSuite) TestMakeCreateRestoreEntity_ConcurrencyCheck() {
	tests := map[string]struct {
		allowConcurrent bool // if true, installs a global AllowConcurrentPiTR=true override
		countReturn     int64
		countErr        bizerr.BizError
		expectCountCall bool
		wantErrMsg      string
		wantNilErr      bool
	}{
		// Any active PiTR restore (PENDING or RUNNING) against the same log backup is
		// counted atomically by the repo filter; count>0 triggers the rejection.
		// Note: mock.Anything is used for the filter func since functional-option
		// predicates cannot be inspected via mock assertions.
		"concurrent restore (count>0) → reject with HTTP 409": {
			countReturn:     1,
			expectCountCall: true,
			wantErrMsg:      appconfig.ErrMsgPiTRConcurrentRestoreNotAllowed,
		},
		"count query fails → internal error": {
			countErr:        bizerr_wrapper.NewInternalErr("db error"),
			expectCountCall: true,
			wantErrMsg:      appconfig.ErrInternalError,
		},
		"no concurrent restore → entity created": {
			countReturn:     0,
			expectCountCall: true,
			wantNilErr:      true,
		},
		// Verifies AllowConcurrentPiTR bypass: Count must NOT be called.
		// testify/mock enforces this by failing on any unexpected method call.
		"AllowConcurrentPiTR=true → bypass, Count not called": {
			allowConcurrent: true,
			expectCountCall: false,
			wantNilErr:      true,
		},
	}

	for name, tc := range tests {
		s.Run(name, func() {
			if tc.allowConcurrent {
				orig := config.Global().BackupConfig.RestoreOverrides
				defer func() { config.Global().BackupConfig.RestoreOverrides = orig }()
				config.Global().BackupConfig.RestoreOverrides = config.RestoreOverrides{
					{Gate: config.FeatureGate{All: true}, AllowConcurrentPiTR: true},
				}
			}

			mockRepo := restore_mock.NewRestoreRepo(s.T())
			if tc.expectCountCall {
				mockRepo.On("Count", mock.Anything, mock.Anything).
					Return(tc.countReturn, tc.countErr)
			}

			p := newTestPiTRestore(mockRepo)
			param := minimalCreateParam(entity.ID(10), entity.ID(20))

			got, err := p.makeCreateRestoreEntity(context.Background(), param)

			if tc.wantNilErr {
				s.Require().Nil(err)
				s.Require().NotNil(got)
				s.Equal(entity.RestoreModePiTR, got.Mode)
			} else {
				s.Require().NotNil(err)
				s.Contains(err.Error(), tc.wantErrMsg)
				s.Nil(got)
			}
		})
	}
}

func (s *pitrRestoreTestSuite) TestCheckIsMeetTimeRanges() {
	mkRange := func(master bool, regions []string, startTS, endTS int64) *entity.PiTRTimeRange {
		return &entity.PiTRTimeRange{
			StartTS: startTS,
			EndTS:   endTS,
			Backup: entity.Backup{
				BucketSettingSnapshot: &entity.BucketSetting{
					Buckets: []entity.BackupSettingBucket{
						{
							BucketId: 1,
							Master:   master,
							Regions:  regions,
						},
					},
				},
			},
			LogBackup: entity.LogBackup{},
		}
	}
	mkParam := func(restoredTs int64, backupRegion, clusterRegion string) *createRestoreParam {
		return &createRestoreParam{
			createRestoreReq: &domain.CreateRestoreReqV2{
				RestoredTs:   restoredTs,
				BackupRegion: backupRegion,
				ClusterConfig: domain.ClusterConfig{
					Region: clusterRegion,
				},
			},
		}
	}

	tests := map[string]struct {
		timeRange            *entity.PiTRTimeRange
		param                *createRestoreParam
		enableCRRFOverride   bool
		checkpointFunc       func(context.Context, entity.LogBackup, string, blob.GetStorageFunc) (int64, bool, error)
		expectCheckpointCall bool
		want                 bool
	}{
		"nil snapshot returns false": {
			timeRange: &entity.PiTRTimeRange{
				StartTS: 100,
				EndTS:   300,
				Backup:  entity.Backup{},
			},
			param: mkParam(200, "", ""),
			want:  false,
		},
		"requested backup region not found returns false": {
			timeRange: mkRange(true, []string{"us-east-1"}, 100, 300),
			param:     mkParam(200, "us-west-2", ""),
			want:      false,
		},
		"master region in range returns true": {
			timeRange: mkRange(true, []string{"us-east-1"}, 100, 300),
			param:     mkParam(200, "us-east-1", ""),
			want:      true,
		},
		"secondary region uses checkpoint when override enabled": {
			timeRange:          mkRange(false, []string{"us-west-2"}, 100, 300),
			param:              mkParam(250, "us-west-2", ""),
			enableCRRFOverride: true,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc) (int64, bool, error) {
				return 200, true, nil
			},
			expectCheckpointCall: true,
			want:                 false,
		},
		"no explicit backup region but cluster region is secondary: checkpoint caps endTs": {
			timeRange:          mkRange(false, []string{"us-west-2"}, 100, 300),
			param:              mkParam(250, "", "us-west-2"),
			enableCRRFOverride: true,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc) (int64, bool, error) {
				return 200, true, nil
			},
			expectCheckpointCall: true,
			want:                 false,
		},
		"secondary region falls back to primary endTs on checkpoint error": {
			timeRange:          mkRange(false, []string{"us-west-2"}, 100, 300),
			param:              mkParam(250, "us-west-2", ""),
			enableCRRFOverride: true,
			checkpointFunc: func(_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc) (int64, bool, error) {
				return 0, false, errors.New("storage unavailable")
			},
			expectCheckpointCall: true,
			want:                 true,
		},
	}

	origOverrides := config.Global().BackupConfig.CRRFOverrides
	defer func() {
		config.Global().BackupConfig.CRRFOverrides = origOverrides
	}()

	cluster := &mgmt.Cluster{
		OrganizationId: 1,
		ProjectId:      2,
		ClusterId:      3,
		CloudProvider:  "aws",
		Region:         "us-east-1",
	}

	for name, tc := range tests {
		s.Run(name, func() {
			if tc.enableCRRFOverride {
				config.Global().BackupConfig.CRRFOverrides = config.CRRFOverrides{
					{
						Gate:                     config.FeatureGate{All: true},
						CalculatePiTRTimeRangeV2: true,
					},
				}
			} else {
				config.Global().BackupConfig.CRRFOverrides = nil
			}

			called := false
			checkpointFunc := tc.checkpointFunc
			if checkpointFunc == nil {
				checkpointFunc = func(
					_ context.Context, _ entity.LogBackup, _ string, _ blob.GetStorageFunc,
				) (int64, bool, error) {
					return 0, false, nil
				}
			}
			p := &PiTRestore{
				getCheckpointForCRPiTR: func(
					ctx context.Context, logBackup entity.LogBackup, region string, getStorageFunc blob.GetStorageFunc,
				) (int64, bool, error) {
					called = true
					return checkpointFunc(ctx, logBackup, region, getStorageFunc)
				},
			}

			got := p.checkIsMeetTimeRanges(context.Background(), tc.timeRange, tc.param, cluster)
			s.Equal(tc.want, got)
			s.Equal(tc.expectCheckpointCall, called)
		})
	}
}