//go:generate ${GOBIN}/mockgen -destination premium_client_conn_mock_test.go -package tidbtools google.golang.org/grpc ClientConnInterface

package tidbtools

import (
	"context"
	"fmt"
	"math"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/devops/api"
	"github.com/tidbcloud/devops/pkg/infra/errs"
	tidbtoolsapi "github.com/tidbcloud/devops/pkg/service/tidbtools/api"
	nextgenglobal "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func (s *svc) ListPremiumBackups(ctx context.Context, orgID, projectID, clusterID uint64, page, pageSize int32) (*nextgenglobal.ListClusterBackupsResponse, error) {
	skip, err := premiumPageSkip(page, pageSize)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()

	return s.nextGenGlobal.ListClusterBackups(ctx, &nextgenglobal.ListClusterBackupsRequest{
		ClusterId: strconv.FormatUint(clusterID, 10),
		PageSize:  pageSize,
		Skip:      skip,
	})
}

func premiumPageSkip(page, pageSize int32) (int32, error) {
	if page < 1 || pageSize < 1 {
		return 0, fmt.Errorf("page and page size must be positive")
	}
	skip := int64(page-1) * int64(pageSize)
	if skip > math.MaxInt32 {
		return 0, fmt.Errorf("page offset exceeds the Premium API limit")
	}
	return int32(skip), nil
}

func (s *svc) GetPremiumBackupSetting(ctx context.Context, orgID, projectID, clusterID uint64) (*nextgenglobal.BackupSetting, *nextgenglobal.ListBackupPoliciesResponse, error) {
	settingCtx, settingCancel := context.WithTimeout(ctx, rpcTimeout)
	setting, err := s.nextGenGlobal.GetBackupSetting(settingCtx, &nextgenglobal.GetBackupSettingRequest{ClusterId: strconv.FormatUint(clusterID, 10)})
	settingCancel()
	if err != nil {
		// A Premium instance created by restore does not have a backup-setting
		// record until the restore has finished. Management exposes that state as
		// its application-specific NotFound code, rather than a setting payload.
		// It is an expected transient state, so let the caller render an
		// unavailable setting instead of failing the entire Tools page.
		if isPremiumBackupSettingUnavailable(err) {
			return nil, nil, nil
		}
		return nil, nil, err
	}
	policiesCtx, policiesCancel := context.WithTimeout(ctx, rpcTimeout)
	policies, err := s.nextGenBackup.ListBackupPolicies(policiesCtx, &nextgenglobal.ListBackupPoliciesRequest{TidbId: strconv.FormatUint(clusterID, 10)})
	policiesCancel()
	if err != nil {
		return nil, nil, err
	}
	return setting, policies, nil
}

func isPremiumBackupSettingUnavailable(err error) bool {
	code := status.Code(err)
	return code == codes.NotFound || code == codeImportClusterNotFound
}

func (s *svc) ListPremiumExports(ctx context.Context, orgID, projectID, clusterID uint64, page, pageSize int32) (*nextgenglobal.ListExportsResponse, error) {
	skip, err := premiumPageSkip(page, pageSize)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	if s.coreSvc == nil || s.coreSvc.Mgmt == nil || s.coreSvc.Mgmt.NextGenExport == nil {
		return nil, fmt.Errorf("Premium export management client is not configured")
	}
	return s.coreSvc.Mgmt.NextGenExport.ListExports(ctx, &nextgenglobal.ListExportsRequest{
		ClusterId: strconv.FormatUint(clusterID, 10),
		PageSize:  pageSize,
		Skip:      skip,
	})
}

func (s *svc) GetPremiumRestoreStatus(ctx context.Context, orgID, projectID, clusterID uint64) (*nextgenglobal.RestoreStatusResponse, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	resp, err := s.nextGenGlobal.RestoreStatus(ctx, &nextgenglobal.RestoreStatusRequest{
		ClusterId: strconv.FormatUint(clusterID, 10),
	})
	if err != nil {
		if isPremiumRestoreUnavailable(err) {
			return nil, nil
		}
		return nil, err
	}
	return resp, nil
}

func isPremiumRestoreUnavailable(err error) bool {
	code := status.Code(err)
	return code == codes.NotFound || code == codeImportClusterNotFound
}

func premiumRestoreMode(mode nextgenglobal.RestoreMode) string {
	return strings.TrimPrefix(mode.String(), "RESTORE_MODE_")
}

func protobufTime(ts *timestamppb.Timestamp) *time.Time {
	if ts == nil {
		return nil
	}
	t := ts.AsTime()
	return &t
}

func premiumRestoreSource(details *nextgenglobal.RestoreDetails) *tidbtoolsapi.PremiumRestoreSource {
	if details == nil {
		return nil
	}
	backupID := ""
	if details.GetBackupId() != 0 {
		backupID = strconv.FormatUint(details.GetBackupId(), 10)
	}
	return &tidbtoolsapi.PremiumRestoreSource{
		ClusterID:                details.GetSourceClusterId(),
		RestoreMode:              premiumRestoreMode(details.GetRestoreMode()),
		BackupID:                 backupID,
		PointInTime:              protobufTime(details.GetPointInTime()),
		EventAt:                  protobufTime(details.GetRestoreRequestedTime()),
		UpdatedAt:                protobufTime(details.GetRestoreUpdatedTime()),
		ExpectedRestoreDataBytes: int64(details.GetExpectedRestoreDataSizeBytes()),
		Region:                   details.GetSourceRegionId(),
		CloudProvider:            details.GetSourceCloudProvider().String(),
		ServicePlan:              details.GetSourceServicePlan().String(),
	}
}

func (s *svc) getPremiumBackupById(ctx context.Context, backupId string) (*nextgenglobal.Backup, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	resp, err := s.nextGenGlobal.ListClusterBackups(ctx, &nextgenglobal.ListClusterBackupsRequest{
		BackupIds: []string{backupId},
	})
	if err != nil {
		return nil, err
	}
	if cnt := len(resp.GetBackups()); cnt != 1 {
		return nil, fmt.Errorf("unexpected backup count, cnt=%d", cnt)
	}

	return resp.GetBackups()[0], nil
}

func (s *svc) GetPremiumRestore(
	ctx context.Context,
	orgID, projectID, clusterID uint64,
) (*tidbtoolsapi.PremiumRestore, error) {
	statusCtx, cancel := context.WithTimeout(ctx, rpcTimeout)

	globalStatus, globalErr := s.nextGenGlobal.RestoreStatus(statusCtx, &nextgenglobal.RestoreStatusRequest{
		ClusterId: strconv.FormatUint(clusterID, 10),
	})
	cancel()

	result := &tidbtoolsapi.PremiumRestore{}
	if globalErr == nil {
		result.Status = globalStatus
		result.Source = premiumRestoreSource(globalStatus.GetDetails())
		if result.Source == nil {
			result.SourceError = "the ng-global restore record does not contain durable source metadata"
		} else if result.Source.BackupID != "" {
			var backupErr error
			result.Backup, backupErr = s.getPremiumBackupById(ctx, result.Source.BackupID)
			if backupErr != nil {
				result.SourceError = fmt.Sprintf("failed to load source backup metadata from ng-global: %v", backupErr)
			}
		}
	} else if isPremiumRestoreUnavailable(globalErr) {
		return result, nil
	}
	if globalErr != nil {
		return nil, globalErr
	}
	if result.Status == nil {
		return result, nil
	}

	if s.coreSvc == nil || s.coreSvc.Mgmt == nil {
		return nil, fmt.Errorf("Premium TiDB management client is not configured")
	}
	targetCtx, targetCancel := context.WithTimeout(ctx, rpcTimeout)
	target, err := s.coreSvc.Mgmt.GetCluster(targetCtx, clusterID)
	targetCancel()
	if err != nil {
		return nil, err
	}
	result.Target = target

	// cluster-service-ng diagnoses a restore by its region-unique restore ID.
	// ng-global's current target status does not expose that ID, and the target
	// cluster ID is not a valid substitute. Keep the durable ng-global status
	// authoritative rather than querying a potentially unrelated runtime record.
	result.RuntimeError = "runtime diagnostics are unavailable because ng-global does not expose the cluster-service-ng restore ID"
	return result, nil
}

func (s *svc) GetBackupSetting(ctx context.Context, orgID, projectID, clusterID uint64) (*dataflow.QueryBackupSettingResp, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	return s.dataflow.Backup.QueryBackupSetting(ctx, &dataflow.QueryBackupSettingReq{
		OrgId:     orgID,
		ProjectId: projectID,
		ClusterId: clusterID,
	})
}

func (s *svc) ListBackupExports(ctx context.Context, orgID, projectID, clusterID uint64, page, pageSize uint32) (*dataflow.ListBackupExportsResp, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	return s.dataflow.Backup.ListBackupExports(ctx, &dataflow.ListBackupExportsReq{
		OrgId:     orgID,
		ProjectId: projectID,
		ClusterId: clusterID,
		Page:      page,
		PerPage:   pageSize,
	})
}

func (s *svc) ListRestores(ctx context.Context, orgID, projectID, clusterID uint64, page, pageSize uint32) (*dataflow.ListRestoresResp, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	return s.dataflow.Backup.ListRestores(ctx, &dataflow.ListRestoresReq{
		OrgId:     orgID,
		ProjectId: projectID,
		ClusterId: clusterID,
		Page:      page,
		PerPage:   pageSize,
	})
}

func (s *svc) ListBackups(ctx context.Context, orgID, projectID, clusterID uint64, page, pageSize uint32) (*dataflow.Backups, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	return s.dataflow.BackupSvc.List(ctx, &dataflow.BackupListReq{
		OrgId:     orgID,
		ProjectId: projectID,
		ClusterId: clusterID,
		Page:      page,
		PerPage:   pageSize,
	})
}

func (s *svc) ListLogBackups(ctx context.Context, orgID, projectID, clusterID uint64, page, pageSize uint32) (*dataflow.LogBackupListResp, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	var err error
	if ctx, err = withDataflowOpsUser(ctx, orgID, projectID); err != nil {
		return nil, err
	}
	return s.dataflow.BackupSvc.ListLogBackups(ctx, &dataflow.LogBackupListReq{
		OrgId:     orgID,
		ProjectId: projectID,
		ClusterId: clusterID,
		Page:      page,
		PerPage:   pageSize,
	})
}

func (s *svc) GetRestoreProgress(ctx context.Context, orgID, projectID, clusterID uint64) (*dataflow.RestoreProgressResp, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	var err error
	if ctx, err = withDataflowOpsUser(ctx, orgID, projectID); err != nil {
		return nil, err
	}
	resp, err := s.dataflow.BackupSvc.GetRestoreProgress(ctx, &dataflow.RestoreProgressReq{
		OrgId:     orgID,
		ProjectId: projectID,
		ClusterId: clusterID,
	})
	if err != nil {
		if status.Code(err) == codes.NotFound {
			// Dataflow uses gRPC NotFound to indicate that the cluster has no
			// active restore. Preserve that empty state for the UI.
			return nil, nil
		}
		return nil, err
	}
	return resp, nil
}

func (s *svc) GetRestore(ctx context.Context, orgID, projectID, clusterID, restoreID uint64) (*dataflow.GetRestoreResp, error) {
	ctx, cancel := context.WithTimeout(ctx, rpcTimeout)
	defer cancel()
	resp, err := s.dataflow.BackupInternal.GetRestore(ctx, &dataflow.GetRestoreReq{
		RestoreId: restoreID,
	})
	if err != nil {
		if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
			return nil, errs.APIErrorf(api.ErrCodeNotFound, "restore %d not found", restoreID)
		}
		return nil, err
	}
	return resp, nil
}