package aws

import (
	"fmt"
	"io"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/storage/iface"
)

const (
	// region = "us-west-2"
	DefaultBucketName = "clinic"
	TmpPrefix         = "tmp"
)

var keyFunc = func(uuid, location string) *string {
	return aws.String(fmt.Sprintf("%s/%s", location, uuid))
}

var tmpKey = func(uuid string) *string {
	return keyFunc(uuid, TmpPrefix)
}

type s3Client interface {
	CreateMultipartUpload(input *s3.CreateMultipartUploadInput) (*s3.CreateMultipartUploadOutput, error)

	UploadPart(input *s3.UploadPartInput) (*s3.UploadPartOutput, error)

	CompleteMultipartUpload(input *s3.CompleteMultipartUploadInput) (*s3.CompleteMultipartUploadOutput, error)

	AbortMultipartUpload(input *s3.AbortMultipartUploadInput) (*s3.AbortMultipartUploadOutput, error)

	GetObject(input *s3.GetObjectInput) (*s3.GetObjectOutput, error)

	HeadObject(input *s3.HeadObjectInput) (*s3.HeadObjectOutput, error)

	DeleteObject(input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error)
}

type AWSStorage struct {
	client     s3Client
	uploader   *s3manager.Uploader
	downloader *s3manager.Downloader
	bucketName string
}

func NewAWSStorage(cfg config.FileStorage) (*AWSStorage, error) {
	awsConfig := &aws.Config{
		Endpoint:         aws.String(cfg.AWSStorage.Endpoint),
		Region:           aws.String(cfg.AWSStorage.Region),
		S3ForcePathStyle: aws.Bool(cfg.AWSStorage.ForcePathStyle),
		// debug purpose
		// LogLevel: aws.LogLevel(aws.LogDebugWithRequestErrors),
	}

	// if cfg.AWSStorage.
	// awsConfig := aws.NewConfig().WithRegion(cfg.AWSStorage.Region).WithMaxRetries(maxRetry).WithEndpoint(cfg.AWSStorage.Endpoint)
	// if cfg.AWSStorage.AccessKey != "" && cfg.AWSStorage.SecretKey != "" {
	// 	awsConfig = awsConfig.WithCredentials(credentials.NewStaticCredentials(cfg.AWSStorage.AccessKey, cfg.AWSStorage.SecretKey, ""))
	// }
	awsSessionOpts := session.Options{
		Config: *awsConfig,
	}
	ses, err := session.NewSessionWithOptions(awsSessionOpts)
	if err != nil {
		return nil, err
	}
	//cli := s3.New(ses)
	//// since we set LogLevel to LogDebugWithRequestErrors, we need to remove following logging handler to avoid noisy log.
	//cli.Handlers.Send.SwapNamed(awsclient.LogHTTPRequestHandler)
	//cli.Handlers.Send.SwapNamed(awsclient.LogHTTPResponseHandler)
	//cli.Handlers.Send.SwapNamed(awsclient.LogHTTPRequestHeaderHandler)
	//cli.Handlers.Send.SwapNamed(awsclient.LogHTTPResponseHeaderHandler)

	uploader := s3manager.NewUploader(ses)
	// set partsize to 50M, so max size of whole file could be 500G
	uploader.PartSize = 1024 * 1024 * 50

	return &AWSStorage{
		client:     s3.New(ses),
		uploader:   uploader,
		downloader: s3manager.NewDownloader(ses),
		bucketName: cfg.AWSStorage.Bucket,
	}, nil
}

func NewAWSStorageWithCli(client s3Client, bucketName string) *AWSStorage {
	return &AWSStorage{
		client:     client,
		bucketName: bucketName,
	}
}

func (s *AWSStorage) PreUpload(fileUUID string) (*string, error) {
	output, err := s.client.CreateMultipartUpload(&s3.CreateMultipartUploadInput{
		Bucket: aws.String(s.bucketName),
		Key:    tmpKey(fileUUID),
	})
	if err != nil {
		return nil, err
	}

	return output.UploadId, nil
}

func (s *AWSStorage) Upload(fileUUID string, content io.ReadSeeker, config iface.UploadConfigRequest) (*string, error) {
	uploadOutput, err := s.client.UploadPart(&s3.UploadPartInput{
		Bucket:     aws.String(s.bucketName),
		Key:        tmpKey(fileUUID),
		PartNumber: aws.Int64(int64(config.SerialNumber)),
		UploadId:   aws.String(config.UploadId),
		Body:       content,
		// ContentLength: aws.Int64(int64(len(content))),
	})
	if err != nil {
		log.Error("Upload failed, err: ", zap.Error(err))
		return nil, err
	}

	return uploadOutput.ETag, nil
}

func (s *AWSStorage) Flush(fileUUID string, config iface.FlushConfigRequest) error {
	parts := make([]*s3.CompletedPart, 0)

	for _, p := range config.Etags {
		parts = append(parts, &s3.CompletedPart{
			ETag:       aws.String(p.ETag),
			PartNumber: aws.Int64(int64(p.SerialNumber)),
		})
	}
	//stream.OfMap(config.Etags).Map(func(entry stream.MapEntry) *s3.CompletedPart {
	//	return &s3.CompletedPart{
	//		ETag: aws.String(entry.Value.Interface().(string)),
	//		PartNumber: aws.Int64(int64(entry.Key.Interface().(int))),
	//	}
	//}).Collect(&parts)

	_, err := s.client.CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{
		Bucket:   aws.String(s.bucketName),
		Key:      tmpKey(fileUUID),
		UploadId: aws.String(config.UploadId),
		MultipartUpload: &s3.CompletedMultipartUpload{
			Parts: parts,
		},
	})

	return err
}

func (s *AWSStorage) GetObject(fileUUID, location string, cfg iface.GetObjectRequest) (*iface.ObjectResponse, error) {
	output, err := s.client.GetObject(&s3.GetObjectInput{
		Bucket: aws.String(s.bucketName),
		Key:    keyFunc(fileUUID, location),
		Range:  aws.String(cfg.Range),
	})
	if err != nil {
		return nil, err
	}

	return &iface.ObjectResponse{
		ETag:         aws.StringValue(output.ETag),
		Body:         output.Body,
		ContentLen:   aws.Int64Value(output.ContentLength),
		ContentRange: aws.StringValue(output.ContentRange),
		ContentType:  aws.StringValue(output.ContentType),
		AcceptRanges: aws.StringValue(output.AcceptRanges),
	}, nil
}

func (s *AWSStorage) DescribeObject(fileUUID, location string) (*iface.ObjectResponse, error) {
	output, err := s.client.HeadObject(&s3.HeadObjectInput{
		Bucket: aws.String(s.bucketName),
		Key:    keyFunc(fileUUID, location),
	})
	if err != nil {
		return nil, err
	}

	return &iface.ObjectResponse{
		ETag:         aws.StringValue(output.ETag),
		ContentLen:   aws.Int64Value(output.ContentLength),
		ContentType:  aws.StringValue(output.ContentType),
		AcceptRanges: aws.StringValue(output.AcceptRanges),
	}, nil
}

func (s *AWSStorage) Abort(fileUUID string, config iface.UploadConfigRequest) error {
	_, err := s.client.AbortMultipartUpload(&s3.AbortMultipartUploadInput{
		Bucket:   aws.String(s.bucketName),
		Key:      aws.String(fileUUID),
		UploadId: aws.String(config.UploadId),
	})

	return err
}

func (s *AWSStorage) Type() config.StorageType {
	return config.AWSStorageType
}

func (s *AWSStorage) DownloadCache(fileUUID string, w io.WriterAt) (int64, error) {
	return s.downloader.Download(w, &s3.GetObjectInput{
		Bucket: aws.String(s.bucketName),
		Key:    tmpKey(fileUUID),
	})
}

func (s *AWSStorage) Persist(fileUUID, clusterID string, reader io.Reader) error {
	_, err := s.uploader.Upload(&s3manager.UploadInput{
		Bucket: aws.String(s.bucketName),
		Key:    keyFunc(fileUUID, clusterID),
		Body:   reader,
	})
	return err
}

func (s *AWSStorage) DeleteCache(fileUUID string) error {
	_, err := s.client.DeleteObject(&s3.DeleteObjectInput{
		Bucket: aws.String(s.bucketName),
		Key:    tmpKey(fileUUID),
	})
	return err
}

// location can be clusterID
func (s *AWSStorage) DeleteObject(fileUUID, location string) error {
	_, err := s.client.DeleteObject(&s3.DeleteObjectInput{
		Bucket: aws.String(s.bucketName),
		Key:    keyFunc(fileUUID, location),
	})
	return err
}