/*
Copyright 2022.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package s3

import (
	"context"
	"errors"
	"fmt"

	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"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"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/go-logr/logr"
	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/common/metrics"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler"
	"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"
)

// S3Reconciler reconciles a S3 object
type S3Reconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

const (
	s3finalizerName = "observability.pingcap.com/s3"
	errNotS3        = "not S3 resource"

	// s3ExportExpirationRuleName is the S3 lifecycle rule id used for tenant
	// export buckets that carry a retention window.
	s3ExportExpirationRuleName = "external-export-expiration"
)

func NewS3Reconciler(mgr ctrl.Manager, log logr.Logger, serviceInfo structs.ServiceInfo) *S3Reconciler {
	return &S3Reconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: mgr.GetClient(),
			ExternalObj: s3Connector{
				cli:         mgr.GetClient(),
				serviceInfo: serviceInfo,
			},
			NewResource: func() client.Object {
				return &observabilitypingcapcomv1.S3{}
			},
		},
	}
}

type s3Connector struct {
	serviceInfo structs.ServiceInfo
	cli         client.Client
}

func (s3c s3Connector) getClient() (providers.S3Client, error) {
	var s3 providers.S3Client
	switch constants.ProvidedVendor(s3c.serviceInfo.Vendor) {
	case 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("unable to get aws config")
		}
		cfg.Region = s3c.serviceInfo.Region
		s3 = s3aws.NewS3(cfg)
	case constants.GCP:
		s3 = s3gcp.NewCloudStorage(s3c.serviceInfo.AccountInfo.Project, s3c.serviceInfo.Region)
	case constants.Azure:
		cred, err := azidentity.NewDefaultAzureCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create default Azure credentials: %w", err)
		}
		if cli, err := azure.NewBlobStorageClient(s3c.serviceInfo.AccountInfo.SubscriptionID, s3c.serviceInfo.AccountInfo.BlobStorageResourceGroup, s3c.serviceInfo.AccountInfo.StorageAccount, cred); err != nil {
			return nil, err
		} else {
			s3 = cli
		}
	case 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(s3c.serviceInfo.Region).WithCredentialsProvider(provider)

		s3 = alicloud.NewS3(cfg, s3c.serviceInfo.AccountInfo.OSSRedundancyType)
	default:
		return nil, fmt.Errorf("not supported provider")
	}
	return s3, nil
}
func (s3c s3Connector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	s3cli, err := s3c.getClient()
	if err != nil {
		return nil, err
	}
	//get s3 client
	return &S3{
		s3cli:       s3cli,
		cli:         s3c.cli,
		serviceInfo: s3c.serviceInfo,
	}, nil
}

type S3 struct {
	s3cli       providers.S3Client
	cli         client.Client
	serviceInfo structs.ServiceInfo
}

func (s3 *S3) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	log := log.FromContext(ctx).WithName("Observe")
	s3v1, ok := obj.(*observabilitypingcapcomv1.S3)
	if !ok {
		s3.reconcilerRepoter(obj.GetName(), "", errNotS3)
		return reconciler.ExternalObservation{Droped: true}, errors.New(errNotS3)
	}
	log.V(1).Info("s3 crd info", "crd", fmt.Sprintf("%v", s3v1))
	if s3v1.Status.Phase == constants.ACTIVE || s3v1.Status.Phase == constants.CREATION_FAILED {
		// final state, nothing to do
		s3.reconcilerRepoter(s3v1.Name, "", "")
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}
	exists, err := s3.s3cli.BucketExists(ctx, s3v1.Spec.Name)
	if err != nil {
		s3.reconcilerRepoter(s3v1.Name, "", err.Error())
		return reconciler.ExternalObservation{}, err
	}
	if !exists {
		//if not exists then we should create
		return reconciler.ExternalObservation{ResourceExists: false}, nil
	}
	// exists but Phase not get final state so we should update state
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func (s3 *S3) Create(ctx context.Context, obj client.Object) (errResult error) {
	s3v1, ok := obj.(*observabilitypingcapcomv1.S3)
	if !ok {
		s3.reconcilerRepoter(obj.GetName(), "", errNotS3)
		return errors.New(errNotS3)
	}
	// update state and then create s3
	s3v1.Status.Phase = constants.CREATING
	if err := s3.cli.Status().Update(ctx, s3v1); err != nil {
		s3.reconcilerRepoter(s3v1.Name, "", err.Error())
		return err
	}
	if err := s3.s3cli.CreateBucket(ctx, s3v1.Spec.Name); err != nil {
		s3.reconcilerRepoter(s3v1.Name, "", err.Error())
		return err
	}
	s3.reconcilerRepoter(s3v1.Name, "", "")
	return nil
}

func (s3 *S3) Update(ctx context.Context, obj client.Object) (errResult error) {
	s3v1, ok := obj.(*observabilitypingcapcomv1.S3)
	if !ok {
		s3.reconcilerRepoter(obj.GetName(), "", errNotS3)
		return errors.New(errNotS3)
	}
	// Converge bucket-level desired state (retention) before marking the bucket
	// active, so a lifecycle failure keeps the resource reconciling instead of
	// silently going ACTIVE without its retention policy.
	if err := s3.ensureLifecycle(ctx, s3v1); err != nil {
		s3.reconcilerRepoter(s3v1.Name, "", err.Error())
		return err
	}
	s3v1.Status.Phase = constants.ACTIVE
	if err := s3.cli.Status().Update(ctx, s3v1); err != nil {
		s3.reconcilerRepoter(s3v1.Name, "", err.Error())
		return err
	}
	s3.reconcilerRepoter(s3v1.Name, "", "")
	return nil
}

// ensureLifecycle applies the object-expiration lifecycle rule when the S3
// resource requests a retention window. It is a no-op for buckets that do not
// set ExpirationDays, so existing shared buckets are left untouched.
func (s3 *S3) ensureLifecycle(ctx context.Context, s3v1 *observabilitypingcapcomv1.S3) error {
	if s3v1.Spec.ExpirationDays <= 0 {
		return nil
	}
	rules := []message.LifecycleRuleParam{
		{
			RuleName:       s3ExportExpirationRuleName,
			ExpirationDays: s3v1.Spec.ExpirationDays,
		},
	}
	return s3.s3cli.AddLifecycle(ctx, s3v1.Spec.Name, rules, nil)
}
func (s3 *S3) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	//TODO s3 delete path
	s3.reconcilerRepoter(obj.GetName(), "", "")
	return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the S3 object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.2/pkg/reconcile
func (r *S3Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := r.Log.WithValues("S3", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)
	return r.reconciler.Reconcile(ctx, req, s3finalizerName)
}

// SetupWithManager sets up the controller with the Manager.
func (r *S3Reconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&observabilitypingcapcomv1.S3{}).
		Complete(r)
}

func (s3 *S3) reconcilerRepoter(name string, interval string, err string) {
	metrics.HandleRegionalControllerRunsMetrics(s3.serviceInfo.Vendor, s3.serviceInfo.Region, "S3Controller", name, interval, err)
}