package blob

import (
	"context"
	"net/url"
	"strings"
	"sync/atomic"

	"github.com/gammazero/workerpool"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb/br/pkg/storage"
	pkglog "github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
)

var DefaultDeleteOption = &DeleteOption{
	Concurrency: 1,
}

var DefualtGetStorageFunc = func(ctx context.Context, urlStr string, opt *StorageOption) (Storage, error) {
	return NewStorage(ctx, urlStr, opt)
}

type (
	GetStorageFunc func(ctx context.Context, urlStr string, opt *StorageOption) (Storage, error)
)

type Storage interface {
	storage.ExternalStorage
	DeleteSubDir(ctx context.Context, subDir string, opt *DeleteOption) error
}

type storageImpl struct {
	storage.ExternalStorage
}

type StorageOption struct {
	ProjectID     uint64
	ClusterRegion string
	AWSRoleARN    string
}

func NewStorage(ctx context.Context, urlStr string, opt *StorageOption) (Storage, error) {
	if opt == nil {
		return nil, errors.New("opt can't be nil")
	}

	backendOpt := &storage.BackendOptions{}

	u, err := url.Parse(urlStr)
	if err != nil {
		return nil, errors.Annotate(err, "parse url")
	}

	switch u.Scheme {
	case "s3":
		if opt.AWSRoleARN != "" {
			backendOpt.S3.RoleARN = opt.AWSRoleARN
		}
	}

	backend, err := storage.ParseBackend(urlStr, backendOpt)
	if err != nil {
		return nil, errors.Annotate(err, "parse backend")
	}
	externalOpts := &storage.ExternalStorageOptions{}
	s, err := storage.New(ctx, backend, externalOpts)
	if err != nil {
		return nil, errors.Annotate(err, "new storage")
	}

	return &storageImpl{
		ExternalStorage: s,
	}, nil
}

type DeleteOption struct {
	Concurrency int
}

func (s *storageImpl) DeleteSubDir(ctx context.Context, subDir string, opt *DeleteOption) error {
	if opt == nil {
		opt = DefaultDeleteOption
	}
	wp := workerpool.New(opt.Concurrency)
	defer func() {
		wp.StopWait()
	}()
	var total, failedNum, successfulNum int32
	delFun := func(path string, size int64) error {
		total += 1

		if wp.WaitingQueueSize() > 1000 {
			pkglog.GetLogger(ctx).Info("waiting queue size is too large, wait for a while")
			wp.SubmitWait(func() {
				s.deleteSubFile(ctx, subDir, path, &failedNum, &successfulNum)
			})
			pkglog.GetLogger(ctx).Info("waiting queue size is too large, wait done")
			return nil
		}

		wp.Submit(func() {
			s.deleteSubFile(ctx, subDir, path, &failedNum, &successfulNum)
		})
		return nil
	}

	walkOpt := &storage.WalkOption{
		SubDir: subDir,
	}
	err := s.WalkDir(ctx, walkOpt, delFun)
	if err != nil {
		return errors.Annotate(err, "walk dir")
	}

	wp.StopWait()
	if failedNum > 0 || failedNum+successfulNum != total {
		return errors.Errorf("get %d objects, delete successfully %d, delete failed %d", total, successfulNum, failedNum)
	}
	pkglog.GetLogger(ctx).Info("delete sub dir over")
	return nil
}

func (s *storageImpl) deleteSubFile(ctx context.Context, subDir string, path string,
	failedNum *int32, successfulNum *int32,
) {
	if err := s.DeleteFile(ctx, path); err != nil {
		log.L().Warn("delete file error", zap.Error(err),
			zap.String("backup path", redactURI(s.ExternalStorage.URI())), zap.String("sub dir", subDir))
		atomic.AddInt32(failedNum, 1)
	} else {
		atomic.AddInt32(successfulNum, 1)
	}
}

func IsNotFoundError(err error) bool {
	return strings.Contains(err.Error(), "Not Found") ||
		strings.Contains(err.Error(), "bucket doesn't exist")
}

// redactURI strips query params from a URI before logging to prevent leaking
// secrets such as Azure SAS tokens embedded as query parameters.
func redactURI(rawURI string) string {
	u, err := url.Parse(rawURI)
	if err != nil {
		return "[invalid URI]"
	}
	u.RawQuery = ""
	return u.String()
}