package storage

import (
	"context"
	"fmt"

	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"github.com/tidbcloud/dataflow-service/pkg/azblob"
)

type AzblobClient struct {
	azblob.AzureBlobStorage
}

func NewAzblobClient(ctx context.Context, sourceURL, sasToken string) (StorageClient, error) {
	abs, err := azblob.NewAzureBlobStorage(ctx, sourceURL, sasToken, nil)
	if err != nil {
		return nil, err
	}
	return &AzblobClient{abs}, nil
}

func (c *AzblobClient) ListObjects(ctx context.Context, bucketName, prefix string) ([]string, error) {

	containerCli, err := c.AzureBlobStorage.Container(ctx, bucketName)
	if err != nil {
		return nil, err
	}
	pager := containerCli.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
		Prefix: &prefix,
	})
	var ret []string
	for pager.More() {
		page, err := pager.NextPage(ctx)
		if err != nil {
			return nil, errors.Annotatef(err, "Failed to list azure blobs, bucket(container)='%s'", bucketName)
		}

		for _, blob := range page.Segment.BlobItems {
			name := fmt.Sprintf("%s://%s.blob.core.windows.net/%s/%s", constants.AzblobScheme, c.AzureBlobStorage.GetAccountName(), bucketName, *blob.Name)
			ret = append(ret, name)
		}
	}
	return ret, nil
}