package s3

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"mime/multipart"
	"sync"

	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	s3api "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/metamanager"
	"go.uber.org/zap"
	"gopkg.in/yaml.v3"
)

type evaluationMeta struct {
	awsClient *s3api.Client
}

var metaSvc metamanager.EvaluationMeta

var once sync.Once

func GetEvaluationMetaService() metamanager.EvaluationMeta {
	once.Do(func() {
		if metaSvc == nil {
			cfg, err := awsconfig.LoadDefaultConfig(context.TODO())
			if err != nil {
				panic(err)
			}
			cfg.Region = "us-west-2"
			metaSvc = &evaluationMeta{
				awsClient: s3api.NewFromConfig(cfg),
			}
		}
	})

	return metaSvc
}

func (e evaluationMeta) SaveMeta(ctx context.Context, meta structs.Evaluation) (err error) {
	bucket := constants.EvaluationBucketName()
	fileKey := fmt.Sprintf(constants.MetaFileKey, meta.User, meta.Name)

	metaYaml, err := yaml.Marshal(meta)
	if err != nil {
		return err
	}

	reader := bytes.NewReader(metaYaml)
	_, err = e.awsClient.PutObject(ctx, &s3api.PutObjectInput{
		Bucket: &bucket,
		Key:    &fileKey,
		Body:   reader,
	})
	if err != nil {
		err = errors.Wrap(errors.EvaluationWriteError, "put evaluation meta to s3 failed", err)
		log.Error("put evaluation meta to s3 failed", zap.Error(err))
		return
	}

	return
}

func (e evaluationMeta) GetMeta(ctx context.Context, user string, evaluationName string) (evaluation structs.Evaluation, err error) {
	bucket := constants.EvaluationBucketName()
	fileKey := fmt.Sprintf(constants.MetaFileKey, user, evaluationName)

	output, err := e.awsClient.GetObject(ctx, &s3api.GetObjectInput{
		Bucket: &bucket,
		Key:    &fileKey,
	})
	if err != nil {
		err = errors.Wrap(errors.EvaluationReadError, "get evaluation meta from s3 failed", err)
		log.Error("get evaluation meta from s3 failed", zap.Error(err))
		return
	}

	defer output.Body.Close()

	content, err := io.ReadAll(output.Body)
	if err != nil {
		err = errors.Wrap(errors.EvaluationReadError, "read evaluation meta from s3 failed", err)
		log.Error("read evaluation meta from s3 failed", zap.Error(err))
		return
	}

	err = yaml.Unmarshal(content, &evaluation)
	if err != nil {
		err = errors.Wrap(errors.EvaluationReadError, "parse evaluation meta yaml failed", err)
		log.Error("parse evaluation meta yaml failed", zap.Error(err))
		return
	}
	return
}

func (e evaluationMeta) SavePred(ctx context.Context, meta structs.Evaluation, file multipart.File) (err error) {

	return
}

func (e evaluationMeta) SaveResult(ctx context.Context, meta structs.Evaluation) (err error) {
	bucket := constants.EvaluationBucketName()
	fileKey := fmt.Sprintf(constants.ResultFileKey, meta.User, meta.Name)

	reader := bytes.NewReader([]byte(meta.Result))
	_, err = e.awsClient.PutObject(ctx, &s3api.PutObjectInput{
		Bucket: &bucket,
		Key:    &fileKey,
		Body:   reader,
	})
	if err != nil {
		err = errors.Wrap(errors.EvaluationWriteError, "put evaluation result to s3", err)
		log.Error("put evaluation result to s3 failed", zap.Error(err))
		return
	}

	return
}

func (e evaluationMeta) GetResult(ctx context.Context, user string, evaluationName string) (result string, resultSummary string, err error) {
	bucket := constants.EvaluationBucketName()
	fileKey := fmt.Sprintf(constants.ResultFileKey, user, evaluationName)

	result, err = e.getResultFromS3(ctx, bucket, fileKey)
	if err != nil {
		log.Error("get result from s3 failed", zap.String("File_Key", fileKey))
		return
	}

	summary, summaryErr := e.getResultFromS3(ctx, bucket, fileKey+".json")
	if summaryErr != nil {
		// ignore this error
		log.Warn("get result.txt.json failed")
		return
	} else {
		resultSummary = summary
		log.Info("get evaluation result summary", zap.String("Summary", resultSummary))
	}

	return
}

func (e evaluationMeta) getResultFromS3(ctx context.Context, bucket, fileKey string) (string, error) {
	output, err := e.awsClient.GetObject(ctx, &s3api.GetObjectInput{
		Bucket: &bucket,
		Key:    &fileKey,
	})
	if err != nil {
		err = errors.Wrap(errors.EvaluationReadError, "get evaluation result from s3 failed", err)
		log.Warn("get evaluation result from s3 failed", zap.Error(err))
		return "", err
	}

	defer output.Body.Close()

	content, err := io.ReadAll(output.Body)

	if err != nil {
		err = errors.Wrap(errors.EvaluationReadError, "read evaluation result from s3 failed", err)
		log.Error("read evaluation result from s3 failed", zap.Error(err))
		return "", err
	}
	return string(content), err
}

func (e evaluationMeta) GetDatasetRecords(ctx context.Context, dataset string, date string) (objects []*s3api.GetObjectOutput, err error) {
	bucket := constants.EvaluationBucketName()
	prefix := fmt.Sprintf(constants.DatasetRecordsKey, dataset, date)

	obj, err := e.awsClient.ListObjects(ctx, &s3api.ListObjectsInput{
		Bucket: &bucket,
		Prefix: &prefix,
	})
	if err != nil {
		err = errors.Wrap(errors.EvaluationReadError, "get dataset records from s3 failed", err)
		log.Error("get dataset records from s3 failed", zap.Error(err))
		return
	}
	log.Info("get dataset", zap.Int("Content_Length", len(obj.Contents)))
	for _, o := range obj.Contents {
		output, err := e.awsClient.GetObject(ctx, &s3api.GetObjectInput{
			Bucket: &bucket,
			Key:    o.Key,
		})
		if err != nil {
			err = errors.Wrap(errors.EvaluationReadError, "get dataset records from s3 failed", err)
			log.Error("get dataset records from s3 failed", zap.Error(err))
			return objects, err
		}
		objects = append(objects, output)
	}
	return objects, nil
}

func Mock(mock metamanager.EvaluationMeta) {
	metaSvc = mock
}