package store

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	osscred "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/credentials-go/credentials"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/pingcap/log"
	"github.com/tidbcloud/serverless-metering/pkg/config"
	"github.com/tidbcloud/serverless-metering/pkg/util"
	"go.uber.org/zap"
)

type OssRemoteStorage struct {
	ossClient *oss.Client
	bucket    string
	hostname  string
}

func NewOssRemoteStorage(hostname string) (RemoteStorage, error) {
	cfg := config.Global()

	arnCredential, gerr := credentials.NewCredential(nil)
	provider := osscred.CredentialsProviderFunc(func(ctx context.Context) (osscred.Credentials, error) {
		if gerr != nil {
			return osscred.Credentials{}, gerr
		}
		cred, err := arnCredential.GetCredential()
		if err != nil {
			return osscred.Credentials{}, err
		}

		return osscred.Credentials{
			AccessKeyID:     *cred.AccessKeyId,
			AccessKeySecret: *cred.AccessKeySecret,
			SecurityToken:   *cred.SecurityToken,
		}, nil
	})
	ossCfg := oss.LoadDefaultConfig().WithRegion(cfg.Region).WithCredentialsProvider(provider)
	ossClient := oss.NewClient(ossCfg)

	return &OssRemoteStorage{
		ossClient: ossClient,
		bucket:    cfg.Store.S3Bucket,
		hostname:  hostname,
	}, nil
}

func (s *OssRemoteStorage) Upload(ctx context.Context, file *LocalFile) error {
	f, err := os.Open(file.Path)
	if err != nil {
		return fmt.Errorf("failed to read local metric file: %s", err)
	}
	defer f.Close()

	dir := util.LocateLatestOnTheHour(file.End)
	key := fmt.Sprintf("data/%d/%s/%s", dir, s.hostname, filepath.Base(file.Path))
	_, err = s.ossClient.PutObject(ctx, &oss.PutObjectRequest{
		Bucket: aws.String(s.bucket),
		Key:    aws.String(key),
		Body:   f,
	})
	if err != nil {
		return fmt.Errorf("failed to upload metric file to oss: %s", err)
	}
	return nil
}

func (s *OssRemoteStorage) BuildScanner(ctx context.Context, ts int64) (RemoteScanner, error) {
	dir := util.LocateLatestOnTheHour(ts)
	// TODO: page
	output, err := s.ossClient.ListObjectsV2(ctx, &oss.ListObjectsV2Request{
		Bucket: aws.String(s.bucket),
		Prefix: aws.String(fmt.Sprintf("data/%d", dir)),
	})
	if err != nil {
		return nil, fmt.Errorf("failed to build oss scanner: %s", err)
	}

	return &OssParquetScanner{
		ossClient: s.ossClient,
		bucket:    s.bucket,
		objects:   output.Contents,
		cursor:    0,
	}, nil
}

func (s *OssRemoteStorage) PutData(ctx context.Context, key string, data []byte) error {
	_, err := s.ossClient.PutObject(ctx, &oss.PutObjectRequest{
		Bucket: aws.String(s.bucket),
		Key:    aws.String(key),
		Body:   bytes.NewReader(data),
	})
	if err != nil {
		return fmt.Errorf("failed to upload data to oss: %s", err)
	}
	return nil
}

func (s *OssRemoteStorage) GetData(ctx context.Context, key string) ([]byte, error) {
	output, err := s.ossClient.GetObject(ctx, &oss.GetObjectRequest{
		Bucket: aws.String(s.bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		return nil, fmt.Errorf("failed to get data from oss: %s", err)
	}
	defer output.Body.Close()

	data, err := io.ReadAll(output.Body)
	if err != nil {
		return nil, fmt.Errorf("failed to read s3 data: %s", err)
	}

	return data, nil
}

type OssParquetScanner struct {
	ossClient *oss.Client
	bucket    string
	objects   []oss.ObjectProperties
	cursor    int
}

func (s *OssParquetScanner) ScanMetrics(ctx context.Context) ([]*Metric, bool, error) {
	return nil, false, errors.New("not implemented")
}

func (s *OssParquetScanner) ScanRemoteFiles(ctx context.Context) (*RemoteFile, bool, error) {
	if !s.hasMore() {
		return nil, false, nil
	}
	obj := s.objects[s.cursor]

	log.Debug("get s3 object", zap.String("key", *obj.Key))
	output, err := s.ossClient.GetObject(ctx, &oss.GetObjectRequest{
		Bucket: aws.String(s.bucket),
		Key:    obj.Key,
	})
	if err != nil {
		return nil, false, fmt.Errorf("failed to get metrics file from oss: %s", err)
	}
	log.Debug("get s3 object finished", zap.String("key", *obj.Key))

	name := strings.ReplaceAll(*obj.Key, "/", "-")
	s.cursor++
	return &RemoteFile{Body: output.Body, Name: name}, true, nil
}

func (s *OssParquetScanner) hasMore() bool {
	return s.cursor < len(s.objects)
}