package resourcemanager

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/tidbcloud/observability/pkg/common/configs"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/alicloud"
	s3aws "github.com/tidbcloud/observability/pkg/resourceworker/providers/aws"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/azure"
	s3gcp "github.com/tidbcloud/observability/pkg/resourceworker/providers/gcp"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	openapicred "github.com/aliyun/credentials-go/credentials"
	"github.com/patrickmn/go-cache"
	"github.com/prometheus-operator/prometheus-operator/pkg/client/versioned"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/utils"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
)

var clusterRW ClusterReaderWriter
var rdsclusterRW ClusterReaderWriter
var objectStorageRW ObjectStorageReaderWriter
var iamRW IAMReaderWriter
var namespaceRW NamespaceReaderWriter
var dataAccountRW DataAccountReaderWriter
var dataAccountNamespacedRW DataAccountReaderWriterFactory
var prometheusRuleRW PrometheusRuleReaderWriter
var vmRuleRW PrometheusRuleReaderWriter
var clusterRoleBindingRW ClusterRoleBindingReaderWriter
var prometheusOperatorClient versioned.Interface
var subscriptionRW SubscriptionReaderWriter
var pvcRW PVCReaderWriter
var jobRW JobReaderWriter
var configMapRW ConfigMapReaderWriter
var prometheusAPIClient PrometheusAPIClient
var archivedClusterRW ArchivedClusterReaderWriter
var integrationTaskRW IntegrationTaskReaderWriter
var s3Client providers.S3Client
var helmReleaseRW HelmReleaseReaderWriter
var clusterUpgradeOperationRW UpgradeOperationReaderWriter

func InitReaderWriter(kubeConfig *rest.Config, mgr ctrl.Manager, mysqlCli client.Client, regional bool) {
	kubeClient, err := kubernetes.NewForConfig(kubeConfig)
	cli := mgr.GetClient()
	readCli := mgr.GetAPIReader()
	if err != nil {
		log.Error("init config failed", zap.Error(err))
		os.Exit(1)
	}

	promClient, err := versioned.NewForConfig(kubeConfig)
	if err != nil {
		log.Error("create prometheusRule client failed", zap.Error(err))
		os.Exit(1)
	}

	var s3cli providers.S3Client
	if regional {
		s3cli, err = getS3Client()
		if err != nil {
			log.Error("create s3 client failed", zap.Error(err))
			os.Exit(1)
		}
		s3Client = s3cli // Store globally for access
	}

	prometheusOperatorClient = promClient
	initRW(readCli, cli, kubeClient, s3cli, mysqlCli, regional)
}

func initRW(readCli client.Reader, cli client.Client, kubeClient kubernetes.Interface, s3cli providers.S3Client, mysqlCli client.Client, regional bool) {
	if regional {
		clusterRW = clusterReaderWriter{
			cli:     cli,
			readCli: readCli,
		}
		rdsclusterRW = rdsclusterReaderWriter{
			clusterReaderWriter: clusterReaderWriter{
				cli:     cli,
				readCli: readCli,
			},
			rdscli: mysqlCli,
		}
		objectStorageRW = objectStorageReaderWriter{
			cli: cli,
		}
		iamRW = iamReaderWriter{
			cli: cli,
		}
		prometheusRuleRW = prometheusRuleReaderWriter{
			cli: cli,
		}

		vmRuleRW = vmRuleReaderWriter{
			cli: cli,
		}

		namespaceRW = namespaceReaderWriter{
			kubeClient: kubeClient,
		}
		clusterRoleBindingRW = clusterRoleBindingReaderWriter{
			kubeClient: kubeClient,
		}

		integrationTaskRW = integrationTaskReaderWriter{
			cli: cli,
		}

		dataAccountRW = dataAccountReaderWriter{
			generator: utils.NewIdGenerator(utils.NewConfig(10), configMapSegmentMgr{
				name:       constants.DataAccountSegmentConfigMap,
				keyStartId: constants.DataAccountKeyStartID,
				retries:    10,
				kubeClient: kubeClient,
				namespace:  constants.ObservabilityNamespace,
			}),
			kubeClient: kubeClient,
		}
		dataAccountNamespacedRW = NewdataAccountReaderWriterFactory(kubeClient, utils.NewConfig(10))

		prometheusAPIClient = &cachedPrometheusAPIClient{
			promClientPool: cache.New(time.Hour, 6*time.Hour),
		}

		archivedClusterRW = &archivedClusterReaderWriter{
			s3Cli: s3cli,
		}
	}

	pvcRW = pvcReaderWriter{
		kubeClient: kubeClient,
	}

	jobRW = jobReaderWriter{
		kubeClient: kubeClient,
	}

	configMapRW = configMapReaderWriter{
		cli: cli,
	}

}

func getS3Client() (providers.S3Client, error) {
	var s3 providers.S3Client = nil
	switch configs.GetServiceInfo().Vendor {
	case string(constants.AWS):
		//we should  init aws config and client before start Observe
		cfg, err := awsconfig.LoadDefaultConfig(context.TODO())
		if err != nil {
			return nil, errors.New(errors.K8SClientError, "unable to get aws config")
		}
		cfg.Region = configs.GetServiceInfo().Region
		s3 = s3aws.NewS3(cfg)
	case string(constants.GCP):
		s3 = s3gcp.NewCloudStorage(configs.GetServiceInfo().AccountInfo.Project, configs.GetServiceInfo().Region)
	case string(constants.Azure):
		cred, err := azidentity.NewDefaultAzureCredential(nil)
		if err != nil {
			return nil, err
		}
		azureCli, err := azure.NewBlobStorageClient(configs.GetSusbcriptionID(), configs.GetStorageResourceGroup(), configs.GetAzureStorageAccount(), cred)
		if err != nil {
			return nil, err
		}
		s3 = azureCli
	case string(constants.AliCloud):
		cred, err := openapicred.NewCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create default AliCloud credentials: %w", err)
		}
		provider := credentials.CredentialsProviderFunc(func(ctx context.Context) (credentials.Credentials, error) {
			cred, err := cred.GetCredential()
			if err != nil {
				return credentials.Credentials{}, err
			}
			return credentials.Credentials{
				AccessKeyID:     *cred.AccessKeyId,
				AccessKeySecret: *cred.AccessKeySecret,
				SecurityToken:   *cred.SecurityToken,
			}, nil
		})
		cfg := oss.LoadDefaultConfig().WithRegion(configs.GetServiceInfo().Region).WithCredentialsProvider(provider)
		s3 = alicloud.NewS3(cfg, configs.GetServiceInfo().AccountInfo.OSSRedundancyType)
	}
	return s3, nil
}

func GetClusterRoleBindingReaderWriter(ctx context.Context) ClusterRoleBindingReaderWriter {
	return clusterRoleBindingRW
}

func GetPrometheusRulesReaderWriter(ctx context.Context, isVMRule bool) PrometheusRuleReaderWriter {
	if isVMRule {
		return vmRuleRW
	}
	return prometheusRuleRW
}

func GetClusterReaderWriter(ctx context.Context) ClusterReaderWriter {
	if b, ok := ctx.Value(constants.UseRDSKey).(bool); ok && b {
		return rdsclusterRW
	}
	return clusterRW
}

func GetObjectStorageReaderWriter(ctx context.Context) ObjectStorageReaderWriter {
	return objectStorageRW
}

func GetIAMReaderWriter(ctx context.Context) IAMReaderWriter {
	return iamRW
}

func GetPrometheusOperatorClient() versioned.Interface {
	return prometheusOperatorClient
}

func GetNamespaceReaderWriter() NamespaceReaderWriter {
	return namespaceRW
}

func GetDataAccountReaderWriter() DataAccountReaderWriter {
	return dataAccountRW
}

func GetDataAccountReaderWriterFactory() DataAccountReaderWriterFactory {
	return dataAccountNamespacedRW
}

func GetSubscriptionReaderWriter() SubscriptionReaderWriter {
	return subscriptionRW
}

func GetPVCReaderWriter() PVCReaderWriter {
	return pvcRW
}

func GetJobReaderWriter() JobReaderWriter {
	return jobRW
}

func GetConfigMapReaderWriter() ConfigMapReaderWriter {
	return configMapRW
}

func GetPrometheusAPIClient() PrometheusAPIClient {
	return prometheusAPIClient
}

func GetArchivedClusterReaderWriter() ArchivedClusterReaderWriter {
	return archivedClusterRW
}

func GetIntegrationTaskReaderWriter() IntegrationTaskReaderWriter {
	return integrationTaskRW
}

func GetS3Client() providers.S3Client {
	return s3Client
}

func GetHelmReleaseReaderWriter() HelmReleaseReaderWriter {
	return helmReleaseRW
}

func GetClusterUpgradeOperationReaderWriter() UpgradeOperationReaderWriter {
	return clusterUpgradeOperationRW
}

func MockClusterReaderWriter(mock ClusterReaderWriter) {
	clusterRW = mock
}

func MockRDSClusterReaderWriter(mock ClusterReaderWriter) {
	rdsclusterRW = mock
}

func MockObjectStorageReaderWriter(mock ObjectStorageReaderWriter) {
	objectStorageRW = mock
}

func MockIAMReaderWriter(mock IAMReaderWriter) {
	iamRW = mock
}

func MockNamespaceReaderWriter(mock NamespaceReaderWriter) {
	namespaceRW = mock
}

func MockClusterRoleBindingReaderWriter(mock ClusterRoleBindingReaderWriter) {
	clusterRoleBindingRW = mock
}

func MockDataAccountReaderWriter(mock DataAccountReaderWriter) {
	dataAccountRW = mock
}

func MockPrometheusRuleReaderWriter(mock PrometheusRuleReaderWriter) {
	prometheusRuleRW = mock
}

func MockVMRuleReaderWriter(mock PrometheusRuleReaderWriter) {
	vmRuleRW = mock
}

func MockSubscriptionReaderWriter(mock SubscriptionReaderWriter) {
	subscriptionRW = mock
}

func MockPVCReaderWriter(mock PVCReaderWriter) {
	pvcRW = mock
}

func MockJobReaderWriter(mock JobReaderWriter) {
	jobRW = mock
}

func MockConfigMapReaderWriter(mock ConfigMapReaderWriter) {
	configMapRW = mock
}

func MockPrometheusAPIClient(mock PrometheusAPIClient) {
	prometheusAPIClient = mock
}

func MockArchivedClusterReaderWriter(mock ArchivedClusterReaderWriter) {
	archivedClusterRW = mock
}

func MockIntegrationTaskReaderWriter(mock IntegrationTaskReaderWriter) {
	integrationTaskRW = mock
}

func MockDataAccountReaderWriterFactory(mock DataAccountReaderWriterFactory) {
	dataAccountNamespacedRW = mock
}

func InitOpsReaderWriter(kubeConfig *rest.Config, mgr ctrl.Manager) {
	helmReleaseRW = helmReleaseReaderWriter{
		cli:     mgr.GetClient(),
		readCli: mgr.GetAPIReader(),
	}
	clusterUpgradeOperationRW = clusterUpgradeOperationReaderWriter{
		cli:     mgr.GetClient(),
		readCli: mgr.GetAPIReader(),
	}
}

func MockHelmReleaseReaderWriter(mock HelmReleaseReaderWriter) {
	helmReleaseRW = mock
}

func MockClusterUpgradeOperationReaderWriter(mock UpgradeOperationReaderWriter) {
	clusterUpgradeOperationRW = mock
}

func WrapClientError(err error) error {
	if err == nil {
		return nil
	}

	switch err.(type) {
	case errors.Error:
		return err
	default:
		if apierrors.IsConflict(err) {
			return errors.Wrap(errors.ClusterStatusConflict, "", err)
		} else if apierrors.IsAlreadyExists(err) {
			return errors.Wrap(errors.DuplicatedCluster, "", err)
		} else {
			return errors.New(errors.K8SClientError, err.Error())
		}
	}
}