package regionalapi

import (
	"context"
	"net/http"
	"net/http/httptest"
	"net/url"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	clusterv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/timestamppb"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/utils/ptr"
)

func TestSetBRDefaultEnv(t *testing.T) {
	tests := map[string]struct {
		input    map[string]string
		expected map[string]string
	}{
		"nil input should set default GOMEMLIMIT": {
			input: nil,
			expected: map[string]string{
				GoMemLimitEnv: BRDefaultGoMemLimit,
			},
		},
		"empty map should set default GOMEMLIMIT": {
			input: map[string]string{},
			expected: map[string]string{
				GoMemLimitEnv: BRDefaultGoMemLimit,
			},
		},
		"existing GOMEMLIMIT should not be overwritten": {
			input: map[string]string{
				GoMemLimitEnv: "2GiB",
			},
			expected: map[string]string{
				GoMemLimitEnv: "2GiB",
			},
		},
		"other env vars should be preserved and GOMEMLIMIT added": {
			input: map[string]string{
				"FOO": "bar",
				"BAZ": "qux",
			},
			expected: map[string]string{
				"FOO":         "bar",
				"BAZ":         "qux",
				GoMemLimitEnv: BRDefaultGoMemLimit,
			},
		},
		"other env vars with existing GOMEMLIMIT should not change": {
			input: map[string]string{
				"FOO":         "bar",
				GoMemLimitEnv: "512MiB",
			},
			expected: map[string]string{
				"FOO":         "bar",
				GoMemLimitEnv: "512MiB",
			},
		},
	}

	for name, tt := range tests {
		t.Run(name, func(t *testing.T) {
			result := setBRDefaultEnv(tt.input)
			assert.Equal(t, tt.expected, result)
		})
	}
}

func TestListLogBackupsByRegion_ParsesSubCommandStatuses(t *testing.T) {
	t.Helper()

	failedAt := time.Date(2026, 4, 29, 16, 5, 9, 0, time.UTC)
	payload := &clusterv1.ListAllLogBackupsResponse{
		LogBackups: []*clusterv1.LogBackup{
			{
				Id:            "log-ab-10143988032832670331-360709",
				LogStatus:     clusterv1.LogBackup_RUNNING,
				StartTs:       "458142299193344000",
				Bucket:        "gcs://bucket/path",
				CheckpointTs:  "466221498642989182",
				TruncateUntil: "465773218328215644",
				LogSubCommandStatuses: map[string]*clusterv1.LogBackup_SubCommandStatus{
					"log-truncate": {
						Command:            clusterv1.LogBackup_LOG_TRUNCATE,
						Phase:              clusterv1.LogBackup_FAILED,
						LogTruncatingUntil: "465773218328215644",
						TimeStarted:        timestamppb.New(failedAt.Add(-2 * time.Minute)),
						TimeCompleted:      timestamppb.New(failedAt),
						Conditions: []*clusterv1.LogBackup_SubCommandCondition{
							{
								Command:            clusterv1.LogBackup_LOG_TRUNCATE,
								Type:               clusterv1.LogBackup_FAILED,
								Status:             clusterv1.ClusterCondition_True,
								LastTransitionTime: timestamppb.New(failedAt),
								Reason:             "TruncateLogBackupFailed",
								Message:            "googleapi: Error 503",
							},
						},
					},
				},
			},
		},
	}

	body, err := protojson.Marshal(payload)
	require.NoError(t, err)

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, listLogBackupPath, r.URL.Path)
		_, _ = w.Write(body)
	}))
	defer server.Close()

	backups, err := (&brClient{}).listLogBackupsByRegion(context.Background(), server.URL)
	require.NoError(t, err)
	require.Len(t, backups, 1)

	truncate := backups[0].Status.LogSubCommandStatuses[cloudv1.LogTruncateCommand]
	require.Equal(t, cloudv1.BackupFailed, truncate.Phase)
	require.Equal(t, "465773218328215644", truncate.LogTruncatingUntil)
	require.Len(t, truncate.Conditions, 1)
	require.Equal(t, corev1.ConditionTrue, truncate.Conditions[0].Status)
	require.Equal(t, "TruncateLogBackupFailed", truncate.Conditions[0].Reason)
	require.Equal(t, "googleapi: Error 503", truncate.Conditions[0].Message)
	require.True(t, truncate.Conditions[0].LastTransitionTime.Time.Equal(failedAt))
}

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

	cases := map[string]struct {
		input clusterv1.LogBackup_Status
		want  cloudv1.BackupConditionType
	}{
		"failed":     {input: clusterv1.LogBackup_FAILED, want: cloudv1.BackupFailed},
		"running":    {input: clusterv1.LogBackup_RUNNING, want: cloudv1.BackupRunning},
		"stopped":    {input: clusterv1.LogBackup_STOPPED, want: cloudv1.BackupStopped},
		"prepare":    {input: clusterv1.LogBackup_PREPARE, want: cloudv1.BackupPrepare},
		"scheduled":  {input: clusterv1.LogBackup_SCHEDULED, want: cloudv1.BackupScheduled},
		"truncating": {input: clusterv1.LogBackup_TRUNCATING, want: cloudv1.BackupTruncating},
		"truncated":  {input: clusterv1.LogBackup_TRUNCATED, want: cloudv1.BackupTruncated},
		"stopping":   {input: clusterv1.LogBackup_STOPPING, want: cloudv1.BackupStopping},
		"default":    {input: clusterv1.LogBackup_STATUS_UNSPECIFIED, want: cloudv1.BackupScheduled},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			require.Equal(t, tc.want, logBackupStatusToPhase(tc.input))
		})
	}
}

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

	cases := map[string]struct {
		input clusterv1.LogBackup_SubCommandType
		want  cloudv1.LogSubCommandType
	}{
		"log-start":    {input: clusterv1.LogBackup_LOG_START, want: cloudv1.LogStartCommand},
		"log-truncate": {input: clusterv1.LogBackup_LOG_TRUNCATE, want: cloudv1.LogTruncateCommand},
		"log-stop":     {input: clusterv1.LogBackup_LOG_STOP, want: cloudv1.LogStopCommand},
		"log-pause":    {input: clusterv1.LogBackup_LOG_PAUSE, want: cloudv1.LogPauseCommand},
		"log-resume":   {input: clusterv1.LogBackup_LOG_RESUME, want: cloudv1.LogResumeCommand},
		"default":      {input: clusterv1.LogBackup_SUB_COMMAND_TYPE_UNSPECIFIED, want: cloudv1.LogUnknownCommand},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			require.Equal(t, tc.want, pbSubCommandTypeToCloud(tc.input))
		})
	}
}

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

	cases := map[string]struct {
		input clusterv1.ClusterCondition_ConditionStatus
		want  corev1.ConditionStatus
	}{
		"true":    {input: clusterv1.ClusterCondition_True, want: corev1.ConditionTrue},
		"false":   {input: clusterv1.ClusterCondition_False, want: corev1.ConditionFalse},
		"default": {input: clusterv1.ClusterCondition_CONDITION_STATUS_UNSPECIFIED, want: corev1.ConditionUnknown},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			require.Equal(t, tc.want, pbConditionStatusToCore(tc.input))
		})
	}
}

func TestPBSubCommandStatusesToCloud_Empty(t *testing.T) {
	t.Parallel()
	require.Nil(t, pbSubCommandStatusesToCloud(nil))
	require.Nil(t, pbSubCommandStatusesToCloud(map[string]*clusterv1.LogBackup_SubCommandStatus{}))
}

// stubRegionalClient is a minimal hand-written stub for RegionalClient used in tests
// that live inside the regionalapi package (where generated mocks would cause an import cycle).
var _ RegionalClient = &stubRegionalClient{}

type stubRegionalClient struct {
	capturedPostBody interface{}
	postErr          bizerr.BizError
	getErr           bizerr.BizError
	getRestore       *clusterv1.Restore
}

func (s *stubRegionalClient) Post(_ context.Context, _ *Region, _ string, _ map[string]string, _ url.Values, body, _ interface{}) bizerr.BizError {
	s.capturedPostBody = body
	return s.postErr
}

func (s *stubRegionalClient) Get(_ context.Context, _ *Region, _ string, _ map[string]string, _ url.Values, out interface{}) bizerr.BizError {
	if s.getErr != nil {
		return s.getErr
	}
	if s.getRestore == nil {
		return nil
	}

	restoreOut, ok := out.(*clusterv1.Restore)
	if ok {
		*restoreOut = *s.getRestore
	}
	return nil
}

func (s *stubRegionalClient) Put(_ context.Context, _ *Region, _ string, _ map[string]string, _, _ interface{}) bizerr.BizError {
	return nil
}

func (s *stubRegionalClient) Delete(_ context.Context, _ *Region, _ string, _ map[string]string, _ url.Values) bizerr.BizError {
	return nil
}

func (s *stubRegionalClient) ListRegions(_ context.Context) ([]RegionInfo, bizerr.BizError) {
	return nil, nil
}

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

	tests := map[string]struct {
		cloudProvider    string
		bucketRegion     string
		wantBucketRegion *string
	}{
		"aws: sets BucketRegion pointer": {
			cloudProvider:    "aws",
			bucketRegion:     "us-east-1",
			wantBucketRegion: ptr.To("us-east-1"),
		},
		"gcp: sets BucketRegion pointer": {
			cloudProvider:    "gcp",
			bucketRegion:     "us-central1",
			wantBucketRegion: ptr.To("us-central1"),
		},
		"azure: BucketRegion is nil for empty region": {
			cloudProvider:    "azure",
			bucketRegion:     "",
			wantBucketRegion: nil,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			stub := &stubRegionalClient{}
			client := &brClient{regionalCli: stub}
			param := &infraapi.RestoreParam{
				ClusterID:   12345,
				TenantID:    67890,
				RestoreName: "test-restore",
				BucketInfo: &infraapi.RestoreBucketInfo{
					CloudProvider: tc.cloudProvider,
					Region:        tc.bucketRegion,
					Name:          "my-bucket",
					BackupPrefix:  "backups/",
				},
				Configuration: &cloudv1.DPConfiguration{
					Config:    `{}`,
					BaseImage: "pingcap/br",
					Tag:       "latest",
				},
			}

			err := client.CreateRestoreIgnoreExists(context.Background(), tc.cloudProvider, "us-east-1", param)
			require.NoError(t, err)

			req, ok := stub.capturedPostBody.(*clusterv1.CreateRestoreRequest)
			require.True(t, ok, "expected Post body to be *clusterv1.CreateRestoreRequest")

			if tc.wantBucketRegion == nil {
				assert.Nil(t, req.Restore.BucketRegion)
			} else {
				require.NotNil(t, req.Restore.BucketRegion)
				assert.Equal(t, *tc.wantBucketRegion, *req.Restore.BucketRegion)
			}
		})
	}
}

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

	lastTransition := time.Date(2026, 6, 11, 10, 30, 0, 0, time.UTC)
	stub := &stubRegionalClient{
		getRestore: &clusterv1.Restore{
			Status: clusterv1.Restore_RUNNING,
			Progresses: []*clusterv1.Restore_Progress{
				{
					Step:               "restore_full",
					Progress:           67,
					LastTransitionTime: timestamppb.New(lastTransition),
				},
			},
		},
	}

	client := &brClient{regionalCli: stub}
	resp, err := client.GetRestore(context.Background(), "aws", "us-west-2", 1, 2, 3, "restore-1")
	require.NoError(t, err)
	require.NotNil(t, resp)
	require.Equal(t, cloudv1.RestoreRunning, resp.Status.Phase)

	require.Len(t, resp.Status.Progresses, 1)
	assert.Equal(t, "restore_full", resp.Status.Progresses[0].Step)
	assert.Equal(t, int32(67), resp.Status.Progresses[0].Progress)
	assert.True(t, resp.Status.Progresses[0].LastTransitionTime.Time.Equal(lastTransition))
}

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

	t.Run("empty input", func(t *testing.T) {
		require.Nil(t, convertRestoreProgressesToCloud(nil))
		require.Nil(t, convertRestoreProgressesToCloud([]*clusterv1.Restore_Progress{}))
	})

	t.Run("skip nil and convert fields", func(t *testing.T) {
		lastTransition := time.Date(2026, 6, 11, 11, 0, 0, 0, time.UTC)
		got := convertRestoreProgressesToCloud([]*clusterv1.Restore_Progress{
			nil,
			{
				Step:               "restore_log",
				Progress:           42,
				LastTransitionTime: timestamppb.New(lastTransition),
			},
		})

		require.Len(t, got, 1)
		assert.Equal(t, "restore_log", got[0].Step)
		assert.Equal(t, int32(42), got[0].Progress)
		assert.True(t, got[0].LastTransitionTime.Time.Equal(lastTransition))
	})
}