package cloud_storage

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/url"
	"strings"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
)

const checkAccessRangeCount = 1

type AzBlob struct {
	Host            string
	Container       string
	Prefix          *string
	ContainerClient *container.Client
}

type AzBlobConfig struct {
	URI             string
	SasToken        *string
	containerClient *container.Client
}

type AzBlobOption func(*AzBlobConfig)

func WithAzBlobSasToken(sasToken string) AzBlobOption {
	return func(cfg *AzBlobConfig) {
		cfg.SasToken = &sasToken
	}
}

func WithAzBlobContainerClient(client *container.Client) AzBlobOption {
	return func(cfg *AzBlobConfig) {
		cfg.containerClient = client
	}
}

func NewAzBlob(ctx context.Context, uri string, opts ...AzBlobOption) (CloudStorage, bizerr.BizError) {
	cfg := &AzBlobConfig{URI: uri}
	for _, opt := range opts {
		opt(cfg)
	}

	host, containerName, prefix, bizErr := parseAzureBlobParams(cfg.URI)
	if bizErr != nil {
		return nil, bizErr
	}
	if cfg.containerClient != nil {
		// If a container client is provided, use it directly
		return &AzBlob{
			ContainerClient: cfg.containerClient,
			Container:       containerName,
			Host:            host,
			Prefix:          prefix,
		}, nil
	}

	if cfg.SasToken == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "AzBlobConfig: must provide sasToken")
	}
	client, err := azblob.NewClientWithNoCredential(fmt.Sprintf("https://%s?%s", host, *cfg.SasToken), nil)
	if err != nil {
		log.Error(ctx, "failed to create client", zap.Error(err))
		return nil, bizerr.New(ErrCodeAzureBlobAccessDeny, errMsgAzureBlobAccessDeny)
	}

	containerClient := client.ServiceClient().NewContainerClient(containerName)
	return &AzBlob{
		ContainerClient: containerClient,
		Container:       containerName,
		Host:            host,
		Prefix:          prefix,
	}, nil
}

func (a *AzBlob) ListAll(ctx context.Context) ([]string, bizerr.BizError) {
	var objectPaths []string
	o := &container.ListBlobsFlatOptions{}
	if a.Prefix != nil {
		o.Prefix = a.Prefix
	}

	pager := a.ContainerClient.NewListBlobsFlatPager(o)
	for pager.More() {
		page, err := pager.NextPage(ctx)
		if err != nil {
			log.Error(ctx, "failed to list objects", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return nil, bizerr.New(ErrCodeAzureBlobAccessDeny, errMsgAzureBlobAccessDeny)
		}

		for _, item := range page.Segment.BlobItems {
			objectPaths = append(objectPaths, fmt.Sprintf("azure://%s/%s/%s", a.Host, a.Container, *item.Name))
		}
	}
	return objectPaths, nil
}

func (a *AzBlob) CheckImportPermission(ctx context.Context) bizerr.BizError {
	o := &container.ListBlobsFlatOptions{
		MaxResults: aws.Int32(1),
	}
	if a.Prefix != nil {
		o.Prefix = a.Prefix
	}
	// check list access
	pager := a.ContainerClient.NewListBlobsFlatPager(o)
	if pager.More() {
		p, err := pager.NextPage(ctx)
		if err != nil {
			log.Error(ctx, "failed to list objects", zap.Error(err))
			return bizerr.New(ErrCodeAzureBlobAccessDeny, errMsgAzureBlobAccessDeny)
		}
		if len(p.Segment.BlobItems) == 0 {
			if isOneFile(a.Prefix) {
				return bizerr.New(ErrCodeImportDataFileNotFound, "Can't find data file: %s", *a.Prefix)
			}
			return bizerr.New(ErrCodeImportEmptyFolder, errMsgImportEmptyFolder)
		}

		var blobName string
		if isOneFile(a.Prefix) {
			blobName = *a.Prefix
		} else {
			blobName = *p.Segment.BlobItems[0].Name
		}

		// check get access

		client := a.ContainerClient.NewBlockBlobClient(blobName)
		rangeOption := &blob.DownloadStreamOptions{
			Range: blob.HTTPRange{
				Count: checkAccessRangeCount,
			},
		}
		stream, err := client.DownloadStream(ctx, rangeOption)
		if err != nil {
			var e *azcore.ResponseError
			if errors.As(err, &e) {
				if e.ErrorCode == string(bloberror.BlobNotFound) {
					return bizerr.New(ErrCodeImportDataFileNotFound, "Can't find data file: %s", *a.Prefix)
				}
			}
			log.Error(ctx, "failed to get download stream", zap.Error(err))
			return bizerr.New(ErrCodeAzureBlobAccessDeny, errMsgAzureBlobAccessDeny)
		}
		body := stream.NewRetryReader(ctx, &blob.RetryReaderOptions{})
		defer func() {
			_ = body.Close()
		}()
		_, err = io.ReadAll(body)
		if err != nil {
			log.Error(ctx, "failed to download stream", zap.Error(err))
			return bizerr.New(ErrCodeAzureBlobAccessDeny, errMsgAzureBlobAccessDeny)
		}
	}

	return nil
}

func (s *AzBlob) RetrieveSchemaFile(ctx context.Context, schema string, table string, subDir string) (*string, bizerr.BizError) {
	schemaFilePath := getSchemaFilePath(s.Prefix, subDir, schema, table)
	o := &container.ListBlobsFlatOptions{
		Prefix: &schemaFilePath,
	}

	pager := s.ContainerClient.NewListBlobsFlatPager(o)
	for pager.More() {
		page, err := pager.NextPage(ctx)
		if err != nil {
			log.Error(ctx, "failed to list objects from Azure Blob", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return nil, bizerr.New(bizerr.ErrCodeUnknown, err.Error())
		}

		for _, item := range page.Segment.BlobItems {
			if item.Name != nil && matchSchemaFileExtension(*item.Name, schemaFilePath) {
				result := getRelativeFilePath(s.Prefix, *item.Name)
				return &result, nil
			}
		}
	}

	return nil, nil
}

func (s *AzBlob) ListLimited(ctx context.Context, opts *ListFilesOptions) ([]string, bool, bizerr.BizError) {
	maxResults := GetMaxFilesFromOptions(opts)

	o := &container.ListBlobsFlatOptions{
		MaxResults: aws.Int32(int32(maxResults)),
	}

	if s.Prefix != nil {
		o.Prefix = s.Prefix
	}

	pager := s.ContainerClient.NewListBlobsFlatPager(o)
	var files []string

	if pager.More() {
		page, err := pager.NextPage(ctx)
		if err != nil {
			log.Error(ctx, "failed to list objects from Azure Blob", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil, false, bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return nil, false, bizerr.New(bizerr.ErrCodeUnknown, err.Error())
		}

		for _, item := range page.Segment.BlobItems {
			if item.Name != nil && !strings.HasSuffix(*item.Name, "/") {
				files = append(files, getRelativeFilePath(s.Prefix, *item.Name))
			}
		}

		// Check if there are more pages
		return files, pager.More(), nil
	}

	return files, false, nil
}

func parseAzureBlobParams(uri string) (storageAccount, container string, prefix *string, _ bizerr.BizError) {
	p, err := url.Parse(EscapePath(uri))
	if err != nil {
		return "", "", nil, bizerr.New(bizerr.ErrCodeParams, "Invalid URL")
	}
	if p.Host == "" {
		return "", "", nil, bizerr.New(bizerr.ErrCodeParams, "Invalid URL: Host is missing or improperly specified.")
	}
	// path: /containerName/prefix
	parts := strings.SplitN(strings.TrimPrefix(p.Path, "/"), "/", 2)
	if len(parts) == 0 || parts[0] == "" {
		return "", "", nil, bizerr.New(bizerr.ErrCodeParams, "Invalid URL: Container name is missing or improperly specified.")
	}
	container = parts[0]
	if len(parts) == 2 && parts[1] != "" {
		return p.Host, container, aws.String(parts[1]), nil
	}
	return p.Host, container, nil, nil
}