package tidbtools

import (
	"testing"
	"time"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	dm "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	cloudconfig "github.com/tidbcloud/devops/api/cloudconfig"
	tidbtoolsapi "github.com/tidbcloud/devops/pkg/service/tidbtools/api"
	commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// TestToGetRestoreProgressResp_BugDRegression locks in the fix for a
// historical bug where the converter scaled `pb.Progress * 100`,
// producing values like 10000% on success because the upstream
// dataflow service already returns Progress in the [0, 100] range
// (see dataflow-service/app/controllers/console/backup.go where
// resp.Progress is set to 100 on RESTORE_STATUS_SUCCESS).
func TestToGetRestoreProgressResp_BugDRegression(t *testing.T) {
	pb := &dataflow.RestoreProgressResp{Progress: 100, RestoreStatus: dataflow.RestoreStatus_RESTORE_STATUS_SUCCESS}
	out := ToGetRestoreProgressResp(42, pb)
	if out == nil {
		t.Fatal("expected non-nil output")
	}
	if out.ProgressPercent != 100 {
		t.Errorf("ProgressPercent: got %d, want 100 (no scaling)", out.ProgressPercent)
	}
	if out.ID != "42" {
		t.Errorf("ID: got %q, want \"42\"", out.ID)
	}
}

func TestToGetRestoreProgressResp_NilProto(t *testing.T) {
	out := ToGetRestoreProgressResp(7, nil)
	if out == nil {
		t.Fatal("expected non-nil output even for nil proto")
	}
	if out.ID != "7" {
		t.Errorf("ID: got %q, want \"7\"", out.ID)
	}
	if out.ProgressPercent != 0 {
		t.Errorf("ProgressPercent: got %d, want 0", out.ProgressPercent)
	}
}

func TestToGetRestoreProgressResp_StageFromStep(t *testing.T) {
	pb := &dataflow.RestoreProgressResp{
		Progress: 50,
		Steps: []*dataflow.RestoreStepProgress{
			{StepName: "import_kvs"},
		},
		RestoreStatus: dataflow.RestoreStatus_RESTORE_STATUS_RUNNING,
	}
	out := ToGetRestoreProgressResp(1, pb)
	if out.Stage != "import_kvs" {
		t.Errorf("Stage: got %q, want \"import_kvs\"", out.Stage)
	}
}

func TestToGetRestoreProgressResp_StageFallbackToStatus(t *testing.T) {
	pb := &dataflow.RestoreProgressResp{
		Progress:      0,
		RestoreStatus: dataflow.RestoreStatus_RESTORE_STATUS_PENDING,
	}
	out := ToGetRestoreProgressResp(1, pb)
	if out.Stage != string(RestoreStatusPending) {
		t.Errorf("Stage: got %q, want %q", out.Stage, RestoreStatusPending)
	}
}

func TestToListRestoresResp_NoProgressScaling(t *testing.T) {
	pb := &dataflow.ListRestoresResp{
		Total: 1,
		Items: []*dataflow.ListRestoresItem{
			{Id: 7, Name: "r1", Progress: 100, Status: dataflow.RestoreStatus_RESTORE_STATUS_SUCCESS},
		},
	}
	out := ToListRestoresResp(pb)
	if len(out.Items) != 1 {
		t.Fatalf("Items: got %d, want 1", len(out.Items))
	}
	if out.Items[0].ProgressPercent != 100 {
		t.Errorf("ProgressPercent: got %d, want 100 (no scaling)", out.Items[0].ProgressPercent)
	}
	if out.Total != 1 {
		t.Errorf("Total: got %d, want 1", out.Total)
	}
}

func TestToListRestoresResp_NilProto(t *testing.T) {
	out := ToListRestoresResp(nil)
	if out == nil || out.Items == nil {
		t.Fatal("expected non-nil result with non-nil Items slice")
	}
	if len(out.Items) != 0 {
		t.Errorf("Items: got %d, want 0", len(out.Items))
	}
}

func TestRestoreStatusMapping(t *testing.T) {
	cases := []struct {
		in   dataflow.RestoreStatus
		want RestoreStatus
	}{
		{dataflow.RestoreStatus_RESTORE_STATUS_PENDING, RestoreStatusPending},
		{dataflow.RestoreStatus_RESTORE_STATUS_RUNNING, RestoreStatusRunning},
		{dataflow.RestoreStatus_RESTORE_STATUS_SUCCESS, RestoreStatusSucceeded},
		{dataflow.RestoreStatus_RESTORE_STATUS_FAILED, RestoreStatusFailed},
		{dataflow.RestoreStatus_RESTORE_STATUS_TERMINATED, RestoreStatusTerminated},
	}
	for _, tc := range cases {
		if got := restoreStatus(tc.in); got != tc.want {
			t.Errorf("restoreStatus(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
	// Unknown / unspecified maps to "unknown".
	if got := restoreStatus(dataflow.RestoreStatus_RESTORE_STATUS_UNSPECIFIED); got != RestoreStatusUnknown {
		t.Errorf("UNSPECIFIED: got %q, want %q", got, RestoreStatusUnknown)
	}
}

func TestRestoreModeMapping(t *testing.T) {
	cases := []struct {
		in   dataflow.RestoreMode
		want RestoreMode
	}{
		{dataflow.RestoreMode_RESTORE_MODE_PITR, RestoreModePITR},
		{dataflow.RestoreMode_RESTORE_MODE_SNAPSHOT, RestoreModeSnapshot},
		{dataflow.RestoreMode_RESTORE_MODE_VOLUME_SNAPSHOT, RestoreModeSnapshot},
	}
	for _, tc := range cases {
		if got := restoreMode(tc.in); got != tc.want {
			t.Errorf("restoreMode(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
	// Unspecified / out-of-range falls through to "".
	if got := restoreMode(dataflow.RestoreMode_RESTORE_MODE_UNSPECIFIED); got != "" {
		t.Errorf("UNSPECIFIED: got %q, want empty", got)
	}
}

func TestDmSubTaskStageToStatus(t *testing.T) {
	cases := []struct {
		in   dm.SubTask_Stage
		want DMJobStatus
	}{
		{dm.SubTask_Running, DMJobStatusRunning},
		{dm.SubTask_Creating, DMJobStatusRunning},
		{dm.SubTask_Deleting, DMJobStatusRunning},
		{dm.SubTask_Scaling, DMJobStatusRunning},
		{dm.SubTask_Pausing, DMJobStatusPaused},
		{dm.SubTask_Paused, DMJobStatusPaused},
		{dm.SubTask_Failed, DMJobStatusFailed},
		{dm.SubTask_Finished, DMJobStatusFinished},
	}
	for _, tc := range cases {
		if got := dmSubTaskStageToStatus(tc.in); got != tc.want {
			t.Errorf("dmSubTaskStageToStatus(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

// TestToDMJob_StatusPriority verifies the aggregation rule: when a job
// has multiple subtasks in different stages, the highest-priority status
// wins (Failed > Running > Paused > Finished).
func TestToDMJob_StatusPriority(t *testing.T) {
	cases := []struct {
		name   string
		stages []dm.SubTask_Stage
		want   DMJobStatus
	}{
		{"failed wins over running", []dm.SubTask_Stage{dm.SubTask_Running, dm.SubTask_Failed}, DMJobStatusFailed},
		{"running wins over paused", []dm.SubTask_Stage{dm.SubTask_Paused, dm.SubTask_Running}, DMJobStatusRunning},
		{"paused wins over finished", []dm.SubTask_Stage{dm.SubTask_Finished, dm.SubTask_Paused}, DMJobStatusPaused},
		{"only finished", []dm.SubTask_Stage{dm.SubTask_Finished, dm.SubTask_Finished}, DMJobStatusFinished},
		{"failed wins everywhere", []dm.SubTask_Stage{dm.SubTask_Failed, dm.SubTask_Running, dm.SubTask_Paused, dm.SubTask_Finished}, DMJobStatusFailed},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			job := &dm.Job{Id: 1, Name: "j", SubTasks: map[uint64]*dm.SubTask{}}
			for i, s := range tc.stages {
				job.SubTasks[uint64(i+1)] = &dm.SubTask{Stage: s}
			}
			got := toDMJob(job)
			if got == nil {
				t.Fatal("toDMJob returned nil")
			}
			if got.Status != tc.want {
				t.Errorf("Status: got %q, want %q", got.Status, tc.want)
			}
		})
	}
}

// TestToDMJob_NoSubtasks verifies the empty-subtasks fallback to Unknown.
func TestToDMJob_NoSubtasks(t *testing.T) {
	got := toDMJob(&dm.Job{Id: 1})
	if got == nil {
		t.Fatal("toDMJob returned nil")
	}
	if got.Status != DMJobStatusUnknown {
		t.Errorf("Status: got %q, want %q", got.Status, DMJobStatusUnknown)
	}
}

func TestToDMJob_NilJob(t *testing.T) {
	if got := toDMJob(nil); got != nil {
		t.Errorf("toDMJob(nil): got %+v, want nil", got)
	}
}

// TestToListDMJobsResp_DropsExcessArgs verifies the converter no longer
// references caller-supplied page/pageSize arguments (they were dead
// after the refactor; Total now reflects len(items) of the upstream
// non-paginated response).
func TestToListDMJobsResp(t *testing.T) {
	pb := &dm.ListJobsResp{
		Total: 2,
		Jobs: []*dm.Job{
			{Id: 1, Name: "j1"},
			{Id: 2, Name: "j2"},
		},
	}
	out := ToListDMJobsResp(pb)
	if len(out.Items) != 2 {
		t.Errorf("Items: got %d, want 2", len(out.Items))
	}
	if out.Total != 2 {
		t.Errorf("Total: got %d, want 2", out.Total)
	}
}

func TestToListDMJobsResp_Nil(t *testing.T) {
	out := ToListDMJobsResp(nil)
	if out == nil || out.Items == nil {
		t.Fatal("expected non-nil result with non-nil Items slice")
	}
	if len(out.Items) != 0 || out.Total != 0 {
		t.Errorf("got items=%d total=%d, want 0/0", len(out.Items), out.Total)
	}
}

func TestToListDMPrechecksResp_NormalizesStatus(t *testing.T) {
	pb := &dm.ListPrechecksResp{
		Total: 4,
		Prechecks: []*dm.PrecheckSummary{
			{PrecheckId: 1, Status: "success", SuccessCnt: 3, Total: 3},
			{PrecheckId: 2, Status: "success", SuccessCnt: 2, WarnCnt: 1, Total: 3},
			{PrecheckId: 3, Status: "success", SuccessCnt: 2, FailedCnt: 1, Total: 3},
			{PrecheckId: 4, Status: "warn", WarnCnt: 1, Total: 1},
		},
	}

	out := ToListDMPrechecksResp(pb)
	if out.Total != 4 {
		t.Fatalf("Total: got %d, want 4", out.Total)
	}
	if len(out.Items) != 4 {
		t.Fatalf("Items: got %d, want 4", len(out.Items))
	}
	wants := []DMPrecheckStatus{
		DMPrecheckStatusPassed,
		DMPrecheckStatusWarning,
		DMPrecheckStatusFailed,
		DMPrecheckStatusWarning,
	}
	for i, want := range wants {
		if out.Items[i].Status != want {
			t.Errorf("Items[%d].Status: got %q, want %q", i, out.Items[i].Status, want)
		}
	}
	if out.Items[2].Finished != 3 {
		t.Errorf("Finished: got %d, want 3", out.Items[2].Finished)
	}
}

func TestToListDMPrechecksResp_Nil(t *testing.T) {
	out := ToListDMPrechecksResp(nil)
	if out == nil || out.Items == nil {
		t.Fatal("expected non-nil result with non-nil Items slice")
	}
	if len(out.Items) != 0 || out.Total != 0 {
		t.Errorf("got items=%d total=%d, want 0/0", len(out.Items), out.Total)
	}
}

func TestImportStatus(t *testing.T) {
	cases := []struct {
		in   dataflow.ImportGetResp_Status
		want ImportStatus
	}{
		{dataflow.ImportGetResp_PREPARING, ImportStatusPreparing},
		{dataflow.ImportGetResp_IMPORTING, ImportStatusRunning},
		{dataflow.ImportGetResp_COMPLETED, ImportStatusSucceeded},
		{dataflow.ImportGetResp_FAILED, ImportStatusFailed},
		{dataflow.ImportGetResp_CANCELING, ImportStatusCanceled},
		{dataflow.ImportGetResp_CANCELED, ImportStatusCanceled},
	}
	for _, tc := range cases {
		if got := importStatus(tc.in); got != tc.want {
			t.Errorf("importStatus(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

func TestImportSourceType(t *testing.T) {
	cases := []struct {
		in   dataflow.ImportDataSource
		want ImportSourceType
	}{
		{dataflow.ImportDataSource_S3_SOURCE, ImportSourceTypeS3},
		{dataflow.ImportDataSource_GCS_SOURCE, ImportSourceTypeGCS},
		{dataflow.ImportDataSource_LOCAL_SOURCE, ImportSourceTypeLocalFile},
	}
	for _, tc := range cases {
		if got := importSourceType(tc.in); got != tc.want {
			t.Errorf("importSourceType(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

func TestImportDataFormat(t *testing.T) {
	cases := []struct {
		in   dataflow.ImportDataFormat
		want ImportFileType
	}{
		{dataflow.ImportDataFormat_SQL, ImportFileTypeSQL},
		{dataflow.ImportDataFormat_CSV, ImportFileTypeCSV},
		{dataflow.ImportDataFormat_PARQUET, ImportFileTypeParquet},
	}
	for _, tc := range cases {
		if got := importDataFormat(tc.in); got != tc.want {
			t.Errorf("importDataFormat(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

func TestSnapshotBackupTypeMapping(t *testing.T) {
	cases := []struct {
		in   dataflow.BackupType
		want SnapshotBackupType
	}{
		{dataflow.BackupType_MANUAL, SnapshotBackupTypeManual},
		{dataflow.BackupType_AUTO, SnapshotBackupTypeAuto},
	}
	for _, tc := range cases {
		if got := snapshotBackupType(tc.in); got != tc.want {
			t.Errorf("snapshotBackupType(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

func TestSnapshotBackupStatusMapping(t *testing.T) {
	cases := []struct {
		in   dataflow.Backup_ExternalStatus
		want BackupStatus
	}{
		{dataflow.Backup_RUNNING, BackupStatusRunning},
		{dataflow.Backup_SUCCESS, BackupStatusSucceeded},
		{dataflow.Backup_FAILED, BackupStatusFailed},
		{dataflow.Backup_PENDING, BackupStatusPending},
	}
	for _, tc := range cases {
		if got := snapshotBackupStatus(tc.in); got != tc.want {
			t.Errorf("snapshotBackupStatus(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

func TestLogBackupStatusMapping(t *testing.T) {
	cases := []struct {
		in   dataflow.LogBackup_Status
		want LogBackupStatus
	}{
		{dataflow.LogBackup_PENDING, LogBackupStatusPending},
		{dataflow.LogBackup_RUNNING, LogBackupStatusRunning},
		{dataflow.LogBackup_STOPPING, LogBackupStatusStopping},
		{dataflow.LogBackup_STOPPED, LogBackupStatusStopped},
		{dataflow.LogBackup_PAUSING, LogBackupStatusPausing},
		{dataflow.LogBackup_PAUSED, LogBackupStatusPaused},
		{dataflow.LogBackup_RESUMING, LogBackupStatusResuming},
		{dataflow.LogBackup_FAILED, LogBackupStatusFailed},
		{dataflow.LogBackup_SCHEDULED, LogBackupStatusScheduled},
	}
	for _, tc := range cases {
		if got := logBackupStatus(tc.in); got != tc.want {
			t.Errorf("logBackupStatus(%v): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

func TestToListBackupsResp(t *testing.T) {
	pb := &dataflow.Backups{
		Total: 2,
		Items: []*dataflow.Backup{
			{Id: 1, Name: "b1", Type: dataflow.BackupType_MANUAL, Status: dataflow.Backup_SUCCESS},
			{Id: 2, Name: "b2", Type: dataflow.BackupType_AUTO, Status: dataflow.Backup_RUNNING},
		},
	}
	out := ToListBackupsResp(pb)
	if out == nil || len(out.Items) != 2 {
		t.Fatalf("Items: got %d, want 2", len(out.Items))
	}
	if out.Total != 2 {
		t.Errorf("Total: got %d, want 2", out.Total)
	}
	if out.Items[0].Status != BackupStatusSucceeded {
		t.Errorf("Items[0].Status: got %q, want %q", out.Items[0].Status, BackupStatusSucceeded)
	}
	if out.Items[0].Type != SnapshotBackupTypeManual {
		t.Errorf("Items[0].Type: got %q, want %q", out.Items[0].Type, SnapshotBackupTypeManual)
	}
}

func TestToListBackupsResp_Nil(t *testing.T) {
	out := ToListBackupsResp(nil)
	if out == nil || out.Items == nil {
		t.Fatal("expected non-nil result with non-nil Items slice")
	}
	if len(out.Items) != 0 {
		t.Errorf("Items: got %d, want 0", len(out.Items))
	}
}

func TestToListLogBackupsResp(t *testing.T) {
	pb := &dataflow.LogBackupListResp{
		Total: 1,
		Items: []*dataflow.LogBackup{
			{Id: 9, Name: "lb1", Status: dataflow.LogBackup_RUNNING},
		},
	}
	out := ToListLogBackupsResp(pb)
	if out == nil || len(out.Items) != 1 {
		t.Fatalf("Items: got %d, want 1", len(out.Items))
	}
	if out.Items[0].Status != LogBackupStatusRunning {
		t.Errorf("Status: got %q, want %q", out.Items[0].Status, LogBackupStatusRunning)
	}
	if out.Items[0].ID != "9" {
		t.Errorf("ID: got %q, want \"9\"", out.Items[0].ID)
	}
}

func TestToListLogBackupsResp_Nil(t *testing.T) {
	out := ToListLogBackupsResp(nil)
	if out == nil || out.Items == nil {
		t.Fatal("expected non-nil result with non-nil Items slice")
	}
}

func TestToGetImportIntoTaskResp_Nil(t *testing.T) {
	out := ToGetImportIntoTaskResp(nil)
	if out == nil {
		t.Fatal("expected non-nil result")
	}
	if out.ID != "" || out.ProgressPercent != 0 {
		t.Errorf("got %+v, want zero-value task", out)
	}
}

func TestDailyTimeToCron(t *testing.T) {
	cases := []struct {
		in, want string
	}{
		{"03:30", "30 03 * * *"},
		{"00:00", "00 00 * * *"},
		{"23:59", "59 23 * * *"},
		{"", ""},                 // empty input → empty (single-element split)
		{"badinput", "badinput"}, // unparseable → returned unchanged per contract
	}
	for _, tc := range cases {
		if got := dailyTimeToCron(tc.in); got != tc.want {
			t.Errorf("dailyTimeToCron(%q): got %q, want %q", tc.in, got, tc.want)
		}
	}
}

// TestToChangefeedSummaries verifies the shared projection used by both
// list and overview converters.
func TestToChangefeedSummaries(t *testing.T) {
	resp := &cloudconfig.ListInternalChangefeedsResp{
		Changefeeds: []*cloudconfig.InternalChangefeed{
			{Id: "1", DisplayName: "cf1", Status: "normal"},
			nil, // exercises nil-skip
			{Id: "2", DisplayName: "cf2", Status: "stopped"},
			{DisplayId: "3", DisplayName: "cf3", Status: "creating"},
		},
	}
	got := toChangefeedSummaries(resp)
	if len(got) != 3 {
		t.Fatalf("got %d items, want 3 (nil entry must be skipped)", len(got))
	}
	if got[0].ID != "1" || got[0].DisplayName != "cf1" || got[0].Status != ChangefeedStatusNormal {
		t.Errorf("got[0]: %+v", got[0])
	}
	if got[1].Status != ChangefeedStatusStopped {
		t.Errorf("got[1].Status: got %q, want stopped", got[1].Status)
	}
	if got[2].Status != ChangefeedStatusCreating {
		t.Errorf("got[2].Status: got %q, want creating", got[2].Status)
	}
	if got[2].ID != "" {
		t.Errorf("creating changefeed ID: got %q, want empty kernel ID", got[2].ID)
	}
}

func TestToChangefeedSummaries_Nil(t *testing.T) {
	got := toChangefeedSummaries(nil)
	if got == nil {
		t.Fatal("expected non-nil empty slice")
	}
	if len(got) != 0 {
		t.Errorf("len: got %d, want 0", len(got))
	}
}

func TestToListChangefeedsResp_TotalEqualsLen(t *testing.T) {
	resp := &cloudconfig.ListInternalChangefeedsResp{
		Changefeeds: []*cloudconfig.InternalChangefeed{
			{Id: "1", DisplayName: "cf1", Status: "normal"},
			{Id: "2", DisplayName: "cf2", Status: "stopped"},
		},
	}
	out := ToListChangefeedsResp(resp)
	if out.Total != int64(len(out.Items)) {
		t.Errorf("Total %d != len(Items) %d", out.Total, len(out.Items))
	}
	if out.Total != 2 {
		t.Errorf("Total: got %d, want 2", out.Total)
	}
}

func TestToListChangefeedsOverviewResp(t *testing.T) {
	out := ToListChangefeedsOverviewResp(nil)
	if out == nil || out.Items == nil {
		t.Fatal("expected non-nil result")
	}
	if out.Total != 0 {
		t.Errorf("Total: got %d, want 0", out.Total)
	}
}

func TestToListBackupExportsResp(t *testing.T) {
	pb := &dataflow.ListBackupExportsResp{
		Total: 2,
		Items: []*dataflow.BackupExport{
			{
				Id:         11,
				Name:       "export-to-s3",
				BackupId:   22,
				BackupName: "snapshot-22",
				BackupTs:   timestamppb.New(time.Unix(100, 0)),
				Provider:   "AWS",
				Region:     "us-west-2",
				Storage: &dataflow.BackupExport_S3_{S3: &dataflow.BackupExport_S3{
					TargetUri: "s3://customer/backup/",
				}},
				Status:     dataflow.BackupExport_SUCCESS,
				CreatedAt:  timestamppb.New(time.Unix(200, 0)),
				FinishedAt: timestamppb.New(time.Unix(300, 0)),
			},
			nil,
		},
	}

	out := ToListBackupExportsResp(pb)
	if out.Total != 2 || len(out.Items) != 1 {
		t.Fatalf("unexpected list response: %#v", out)
	}
	got := out.Items[0]
	if got.ID != "11" || got.BackupID != "22" {
		t.Errorf("IDs: got export=%q backup=%q", got.ID, got.BackupID)
	}
	if got.StorageType != "s3" || got.StorageURI != "s3://customer/backup/" {
		t.Errorf("storage: got type=%q uri=%q", got.StorageType, got.StorageURI)
	}
	if got.Status != BackupExportStatusSucceeded {
		t.Errorf("status: got %q", got.Status)
	}
	if got.BackupAt != 100 || got.CreatedAt != 200 || got.FinishedAt != 300 {
		t.Errorf("timestamps: got backup=%d created=%d finished=%d", got.BackupAt, got.CreatedAt, got.FinishedAt)
	}
}

func TestBackupExportStorage(t *testing.T) {
	tests := []struct {
		name        string
		export      *dataflow.BackupExport
		storageType string
		uri         string
	}{
		{
			name: "gcs",
			export: &dataflow.BackupExport{Storage: &dataflow.BackupExport_Gcs{
				Gcs: &dataflow.BackupExport_GCS{TargetUri: "gs://customer/backup/"},
			}},
			storageType: "gcs",
			uri:         "gs://customer/backup/",
		},
		{
			name: "azure blob",
			export: &dataflow.BackupExport{Storage: &dataflow.BackupExport_AzBlob{
				AzBlob: &dataflow.BackupExport_AZBlob{TargetUri: "https://account.blob.core.windows.net/container/"},
			}},
			storageType: "azblob",
			uri:         "https://account.blob.core.windows.net/container/",
		},
		{name: "missing", export: &dataflow.BackupExport{}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			storageType, uri := backupExportStorage(tt.export)
			if storageType != tt.storageType || uri != tt.uri {
				t.Fatalf("got type=%q uri=%q, want type=%q uri=%q", storageType, uri, tt.storageType, tt.uri)
			}
		})
	}
}

func TestBackupExportStatusMapping(t *testing.T) {
	tests := []struct {
		in   dataflow.BackupExport_Status
		want BackupExportStatus
	}{
		{dataflow.BackupExport_PENDING, BackupExportStatusPending},
		{dataflow.BackupExport_RUNNING, BackupExportStatusRunning},
		{dataflow.BackupExport_SUCCESS, BackupExportStatusSucceeded},
		{dataflow.BackupExport_FAILED, BackupExportStatusFailed},
		{dataflow.BackupExport_CANCELED, BackupExportStatusCanceled},
		{dataflow.BackupExport_STATUS_UNSPECIFIED, BackupExportStatusUnknown},
	}
	for _, tt := range tests {
		if got := backupExportStatus(tt.in); got != tt.want {
			t.Errorf("backupExportStatus(%v): got %q, want %q", tt.in, got, tt.want)
		}
	}
}

func TestPremiumBackupRestoreAndExportConverters(t *testing.T) {
	backups := ToListPremiumBackupsResp(&globalngv1.ListClusterBackupsResponse{
		TotalSize: 1,
		Backups: []*globalngv1.Backup{{
			Id:             "7",
			DisplayName:    "backup-7",
			BackupTs:       "468279672299061252",
			Size:           99,
			State:          globalngv1.Backup_STATE_SUCCESS,
			TriggerType:    globalngv1.Backup_TRIGGER_TYPE_AUTO,
			RegionId:       "aws-us-west-2",
			CreateTime:     timestamppb.New(time.Unix(100, 0)),
			ExpirationTime: timestamppb.New(time.Unix(200, 0)),
		}},
	})
	if backups.Total != 1 || len(backups.Items) != 1 {
		t.Fatalf("unexpected premium backups: %#v", backups)
	}
	if backups.Items[0].ID != "7" || backups.Items[0].SizeBytes != 99 || backups.Items[0].Status != BackupStatusSucceeded {
		t.Errorf("unexpected premium backup: %#v", backups.Items[0])
	}
	if backups.Items[0].FinishedAt != 0 {
		t.Errorf("premium backup completion must remain unset without an upstream completion timestamp: %#v", backups.Items[0])
	}
	manualBackups := ToListPremiumBackupsResp(&globalngv1.ListClusterBackupsResponse{Backups: []*globalngv1.Backup{
		{Id: "manual-expiring", TriggerType: globalngv1.Backup_TRIGGER_TYPE_MANUAL, ExpirationTime: timestamppb.New(time.Unix(200, 0))},
		{Id: "manual-no-expiry", TriggerType: globalngv1.Backup_TRIGGER_TYPE_MANUAL},
	}})
	for _, backup := range manualBackups.Items {
		if backup.Protected {
			t.Errorf("manual backup %q must not infer protection from trigger type: %#v", backup.ID, backup)
		}
	}

	setting := ToGetNextGenBackupSettingResp(&globalngv1.BackupSetting{
		Mode:               globalngv1.BackupSetting_STANDARD_BUNDLE,
		RetentionDays:      func() *int32 { v := int32(7); return &v }(),
		DailyBackupTimeUtc: func() *string { v := "00:00"; return &v }(),
	}, &globalngv1.ListBackupPoliciesResponse{BackupPolicies: []*globalngv1.BackupSetting_BackupPolicy{{
		Name:         globalngv1.BackupSetting_BackupPolicy_TIER1,
		Pitr:         &globalngv1.BackupSetting_BackupPolicy_PiTRPolicy{RetentionDays: 7},
		HourlyBackup: &globalngv1.BackupSetting_BackupPolicy_HourlyPolicy{RetentionDays: 7},
		DailyBackup:  &globalngv1.BackupSetting_BackupPolicy_DailyPolicy{RetentionDays: 33, StartTimeUtc: "00:00"},
	}}})
	if setting.RetentionDays == nil || *setting.RetentionDays != 7 || setting.Mode != "standard_bundle" || len(setting.BackupPolicies) != 1 || setting.BackupPolicies[0].DailyRetentionDays == nil || *setting.BackupPolicies[0].DailyRetentionDays != 33 {
		t.Errorf("unexpected premium backup setting: %#v", setting)
	}
	partialPolicy := ToGetNextGenBackupSettingResp(&globalngv1.BackupSetting{Mode: globalngv1.BackupSetting_STANDARD_BUNDLE}, &globalngv1.ListBackupPoliciesResponse{BackupPolicies: []*globalngv1.BackupSetting_BackupPolicy{{Name: globalngv1.BackupSetting_BackupPolicy_TIER1}}})
	if len(partialPolicy.BackupPolicies) != 1 || partialPolicy.BackupPolicies[0].PITRRetentionDays != nil || partialPolicy.BackupPolicies[0].HourlyRetentionDays != nil || partialPolicy.BackupPolicies[0].DailyRetentionDays != nil {
		t.Errorf("missing policy branches must remain unavailable: %#v", partialPolicy)
	}
	if unavailable := ToGetNextGenBackupSettingResp(nil, nil); !unavailable.Unavailable {
		t.Errorf("expected neutral unavailable setting while a restore is in progress: %#v", unavailable)
	}

	exports := ToListPremiumExportsResp(&globalngv1.ListExportsResponse{
		TotalSize: 1,
		Exports: []*globalngv1.Export{{
			ExportId:    "exp-1",
			TidbId:      "42",
			State:       globalngv1.Export_SUCCEEDED,
			SnapshotTso: 12345,
			DisplayName: "export-job",
			CloudStorage: &globalngv1.CloudStorage{
				Storage: &globalngv1.CloudStorage_S3{S3: &globalngv1.S3CloudStorage{Uri: "s3://bucket/path"}},
			},
			CreateTime:   timestamppb.New(time.Unix(100, 0)),
			CompleteTime: timestamppb.New(time.Unix(200, 0)),
		}},
	})
	if exports.Total != 1 || len(exports.Items) != 1 {
		t.Fatalf("unexpected premium exports: %#v", exports)
	}
	if exports.Items[0].ID != "exp-1" || exports.Items[0].StorageURI != "s3://bucket/path" || exports.Items[0].Status != ExportStatusSucceeded {
		t.Errorf("unexpected premium export: %#v", exports.Items[0])
	}
	pendingExports := ToListPremiumExportsResp(&globalngv1.ListExportsResponse{Exports: []*globalngv1.Export{{
		ExportId: "exp-pending",
		State:    globalngv1.Export_STATE_UNSPECIFIED,
	}}})
	if len(pendingExports.Items) != 1 || pendingExports.Items[0].Status != ExportStatusPending {
		t.Errorf("expected pending Premium export, got %#v", pendingExports)
	}

	restorePB := &globalngv1.RestoreStatusResponse{
		State:    globalngv1.RestoreState_RESTORE_STATE_RUNNING,
		Message:  "restoring",
		Progress: 35,
	}
	restores := ToListPremiumRestoresResp("42", restorePB)
	if restores.Total != 1 || len(restores.Items) != 1 || restores.Items[0].Status != RestoreStatusRunning {
		t.Errorf("unexpected premium restores: %#v", restores)
	}
	if restores.Items[0].ErrorMsg != "" {
		t.Errorf("running Premium restore must not expose a healthy status message as an error: %#v", restores.Items[0])
	}
	failedRestores := ToListPremiumRestoresResp("42", &globalngv1.RestoreStatusResponse{
		State:   globalngv1.RestoreState_RESTORE_STATE_FAILED,
		Message: "restore failed",
	})
	if len(failedRestores.Items) != 1 || failedRestores.Items[0].ErrorMsg != "restore failed" {
		t.Errorf("failed Premium restore must expose its error message: %#v", failedRestores)
	}
	progress := ToGetPremiumRestoreProgressResp("42", restorePB)
	if progress.ProgressPercent != 35 || progress.Stage != "running" {
		t.Errorf("unexpected premium restore progress: %#v", progress)
	}
}

func TestSanitizeExportStorageURI_DropsMalformedCredentialURI(t *testing.T) {
	if got := sanitizeExportStorageURI("s3://key:secret@bucket/%zz"); got != "" {
		t.Fatalf("malformed credential URI must be omitted, got %q", got)
	}
}

func TestToGetPremiumRestoreDiagnosticsResp(t *testing.T) {
	requestAt := time.Unix(95, 0)
	updatedAt := time.Unix(205, 0)
	diagnostics := &tidbtoolsapi.PremiumRestore{
		Status: &globalngv1.RestoreStatusResponse{
			State:    globalngv1.RestoreState_RESTORE_STATE_SUCCESS,
			Message:  "restore completed",
			Progress: 100,
		},
		Target: &mgmtv1.Cluster{
			ClusterId:      42,
			ProjectId:      2,
			OrganizationId: 1,
			ClusterName:    "restored-target",
			CloudProvider:  "aws",
			CreatedBy:      "123",
			Region:         "us-west-2",
			CreateTime:     timestamppb.New(time.Unix(100, 0)),
			UpdateTime:     timestamppb.New(time.Unix(200, 0)),
			ClusterInfo: &mgmtv1.Cluster_NextGenCluster{NextGenCluster: &globalngv1.Cluster{
				State:              globalngv1.Cluster_AVAILABLE,
				ServicePlan:        commonv1.ServicePlan_Premium,
				CreateFinishedTime: timestamppb.New(time.Unix(190, 0)),
				RegionDisplayName:  "Oregon",
				Version:            "v9.0.0",
				SharedPoolId:       "shared-pool",
				DataplaneId:        "dataplane",
			}},
		},
		Source: &tidbtoolsapi.PremiumRestoreSource{
			ClusterID:                7,
			RestoreMode:              "SNAPSHOT",
			BackupID:                 "72169",
			EventAt:                  &requestAt,
			UpdatedAt:                &updatedAt,
			ExpectedRestoreDataBytes: 103414155,
			Region:                   "us-west-2",
			CloudProvider:            "aws",
			ServicePlan:              "Essential_V2",
		},
		Backup: &globalngv1.Backup{
			Id:          "72169",
			DisplayName: "auto-backup-202608100700",
			BackupTs:    "468279672299061252",
			State:       globalngv1.Backup_STATE_SUCCESS,
			Type:        globalngv1.Backup_TYPE_SNAPSHOT,
			Size:        103414155,
			RegionId:    "aws-us-west-2",
			CreateTime:  timestamppb.New(time.Unix(90, 0)),
		},
	}

	out := ToGetPremiumRestoreDiagnosticsResp("42", diagnostics)
	if out.NotFound || out.ProgressPercent != 100 || out.Stage != "succeeded" {
		t.Fatalf("unexpected premium restore status: %#v", out)
	}
	if out.Target == nil || out.Target.Name != "restored-target" || out.Target.State != "AVAILABLE" || out.Target.ServicePlan != "Premium" {
		t.Fatalf("unexpected target diagnostics: %#v", out.Target)
	}
	if out.RestoreStartedAt != 95 || out.RestoreFinishedAt != 205 || out.DurationSeconds != 110 {
		t.Errorf("unexpected restore timeline: %#v", out)
	}
	if !out.SourceMetadataAvailable || out.SourceClusterID != "7" {
		t.Errorf("expected source association: %#v", out)
	}
	if out.Source == nil || out.Source.ServicePlan != "Essential_V2" || out.Source.ExpectedRestoreDataBytes != 103414155 {
		t.Errorf("unexpected source diagnostics: %#v", out.Source)
	}
	if out.Backup == nil || out.Backup.ID != "72169" || out.Backup.TSO != "468279672299061252" {
		t.Errorf("unexpected source backup: %#v", out.Backup)
	}
	if out.Backup.FinishedAt != 0 {
		t.Errorf("source backup completion must remain unset without an upstream completion timestamp: %#v", out.Backup)
	}
	if out.Audit == nil || out.Audit.RestoreMode != "SNAPSHOT" || out.Audit.EventAt != 95 {
		t.Errorf("unexpected restore request metadata: %#v", out.Audit)
	}
}

func TestToGetPremiumRestoreDiagnosticsResp_PartialSourceDoesNotFabricateClusterID(t *testing.T) {
	out := ToGetPremiumRestoreDiagnosticsResp("42", &tidbtoolsapi.PremiumRestore{
		Status: &globalngv1.RestoreStatusResponse{State: globalngv1.RestoreState_RESTORE_STATE_RUNNING},
		Source: &tidbtoolsapi.PremiumRestoreSource{RestoreMode: "SNAPSHOT"},
	})
	if out.SourceClusterID != "" || out.Source != nil || out.SourceMetadataAvailable {
		t.Fatalf("partial source must remain unavailable, got %#v", out)
	}
}

func TestToGetPremiumRestoreDiagnosticsResp_NoRestore(t *testing.T) {
	out := ToGetPremiumRestoreDiagnosticsResp("42", &tidbtoolsapi.PremiumRestore{})
	if !out.NotFound || out.Stage != "" || out.ProgressPercent != 0 {
		t.Fatalf("ordinary Premium instance must be an empty state, not a pending restore: %#v", out)
	}
}

func TestToGetPremiumRestoreResp_NoRestore(t *testing.T) {
	if detail := ToGetPremiumRestoreResp("42", "42", nil); detail != nil {
		t.Fatalf("expected no restore detail, got %#v", detail)
	}
}