package storage

import (
	"compress/gzip"
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"

	gcs "cloud.google.com/go/storage"
	"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/container"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"google.golang.org/api/iterator"
)

var (
	ErrNotFound = errors.New("object not found")
)

const deleteObjectsBatchSize = 1000
const azureDeleteObjectsBatchSize = 256 // Azure Blob Batch API limit per HTTP request
const gcpDeleteParallelism = 32         // GCS Go SDK has no multi-delete; parallel single deletes

func BucketName(env, region string, sharedTier, nextGen bool) string {
	bucketName := os.Getenv("BUCKET_NAME")
	if bucketName != "" {
		if env != "prod" {
			if os.Getenv("STAGING_BUCKET_NAME") != "" {
				return os.Getenv("STAGING_BUCKET_NAME")
			}
		}
		return bucketName
	}
	if sharedTier || nextGen {
		return fmt.Sprintf("o11y-prod-shared-%s", region)
	}
	switch env {
	case "prod":
		return fmt.Sprintf("o11y-prod-shared-%s", region)
	case "dev":
		return fmt.Sprintf("o11y-dev-shared-%s", region)
	default:
		return fmt.Sprintf("o11y-test-shared-%s", region)
	}
}

type ObjectMeta struct {
	Key  string
	Size int64
}

type Object struct {
	ETag          *string
	ContentType   *string
	ContentLength *int64
	ContentRange  *string
	AcceptRanges  *string
	Reader        io.ReadCloser
}

type Storage interface {
	Object(ctx context.Context, bucket, key string) (*Object, error)
	ObjectExists(ctx context.Context, bucket, key string) (bool, error)
	ObjectSize(ctx context.Context, bucket, key string) (int64, error)
	Objects(ctx context.Context, bucket, prefix string) ([]ObjectMeta, error)
	ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error)
	ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(ObjectMeta) bool) error
	ObjectsCallbackWithStartKey(ctx context.Context, bucket, prefix, startKey string, callback func(ObjectMeta) bool) error
	SaveObject(ctx context.Context, bucket, key string, body io.Reader) error
	DeleteObject(ctx context.Context, bucket, key string) error
	DeleteObjects(ctx context.Context, bucket string, keys []string) error
}

// ListStorage extends Storage with listing helpers used by data-process.
type ListStorage interface {
	Storage
	Mkdir(ctx context.Context, bucket, key string, body io.Reader) error
	ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]ObjectMeta, error)
	ListObject(ctx context.Context, bucket string) ([]ObjectMeta, error)
	ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) (
		objects []ObjectMeta, nextPageToken string, err error,
	)
}

type awsStorage struct {
	client *s3.Client
}

func NewAwsStorage(client *s3.Client) ListStorage {
	return &awsStorage{client: client}
}

func (s *awsStorage) Object(ctx context.Context, bucket, key string) (*Object, error) {
	output, err := s.client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: &bucket,
		Key:    &key,
	})
	if err != nil {
		var t *types.NoSuchKey
		if errors.As(err, &t) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	// Manually decompress. Ref to https://github.com/aws/aws-sdk-go-v2/blob/main/CHANGELOG.md#service-client-highlights-14
	reader := output.Body
	encoding := output.ContentEncoding
	if encoding != nil && *encoding == "gzip" {
		reader, err = gzip.NewReader(reader)
		if err != nil {
			return nil, err
		}
	}

	return &Object{
		ETag:          output.ETag,
		ContentType:   output.ContentType,
		ContentLength: output.ContentLength,
		ContentRange:  output.ContentRange,
		AcceptRanges:  output.AcceptRanges,
		Reader:        reader,
	}, nil
}

func (s *awsStorage) ObjectExists(ctx context.Context, bucket, key string) (bool, error) {
	_, err := s.client.GetObjectAttributes(ctx, &s3.GetObjectAttributesInput{
		Bucket: &bucket,
		Key:    &key,
		ObjectAttributes: []types.ObjectAttributes{
			types.ObjectAttributesObjectSize,
		},
	})
	if err != nil {
		var t *types.NoSuchKey
		if errors.As(err, &t) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (s *awsStorage) ObjectSize(ctx context.Context, bucket, key string) (int64, error) {
	output, err := s.client.GetObjectAttributes(ctx, &s3.GetObjectAttributesInput{
		Bucket: &bucket,
		Key:    &key,
		ObjectAttributes: []types.ObjectAttributes{
			types.ObjectAttributesObjectSize,
		},
	})
	if err != nil {
		var t *types.NoSuchKey
		if errors.As(err, &t) {
			return 0, nil
		}
		return 0, err
	}
	return *output.ObjectSize, nil
}

func (s *awsStorage) Objects(ctx context.Context, bucket, prefix string) ([]ObjectMeta, error) {
	input := &s3.ListObjectsV2Input{
		Bucket: &bucket,
		Prefix: &prefix,
	}
	p := s3.NewListObjectsV2Paginator(s.client, input)

	var keys []ObjectMeta
	for p.HasMorePages() {
		// next page takes a context
		page, err := p.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, c := range page.Contents {
			keys = append(keys, ObjectMeta{
				Key:  *c.Key,
				Size: *c.Size,
			})
		}
	}
	return keys, nil
}

func (s *awsStorage) ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error) {
	input := &s3.ListObjectsV2Input{
		Bucket:    &bucket,
		Prefix:    &prefix,
		Delimiter: aws.String("/"),
	}
	output, err := s.client.ListObjectsV2(ctx, input)
	if err != nil {
		return nil, err
	}
	if len(output.CommonPrefixes) == 0 {
		return nil, nil
	}
	prefixes := make([]string, len(output.CommonPrefixes))
	for n, p := range output.CommonPrefixes {
		prefixes[n] = *p.Prefix
	}
	return prefixes, nil
}

func (s *awsStorage) ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(ObjectMeta) bool) error {
	input := &s3.ListObjectsV2Input{
		Bucket: &bucket,
		Prefix: &prefix,
	}
	p := s3.NewListObjectsV2Paginator(s.client, input)
	for p.HasMorePages() {
		// next page takes a context
		page, err := p.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, c := range page.Contents {
			if !callback(ObjectMeta{
				Key:  *c.Key,
				Size: *c.Size,
			}) {
				return nil
			}
		}
	}
	return nil
}

func (s *awsStorage) ObjectsCallbackWithStartKey(
	ctx context.Context, bucket, prefix, startKey string, callback func(ObjectMeta) bool,
) error {
	if !callback(ObjectMeta{Key: startKey}) { // Putting it in the first place only for AWS
		return nil
	}
	input := &s3.ListObjectsV2Input{
		Bucket:     &bucket,
		Prefix:     &prefix,
		StartAfter: &startKey,
	}
	p := s3.NewListObjectsV2Paginator(s.client, input)
	for p.HasMorePages() {
		// next page takes a context
		page, err := p.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, c := range page.Contents {
			if !callback(ObjectMeta{
				Key:  *c.Key,
				Size: *c.Size,
			}) {
				return nil
			}
		}
	}
	return nil
}

func (s *awsStorage) SaveObject(ctx context.Context, bucket, key string, body io.Reader) error {
	_, err := s.client.PutObject(ctx, &s3.PutObjectInput{
		Bucket: &bucket,
		Key:    &key,
		Body:   body,
	})
	return err
}

func (s *awsStorage) Mkdir(ctx context.Context, bucket, key string, body io.Reader) error {
	if !strings.HasSuffix(key, "/") {
		key += "/"
	}
	_, err := s.client.PutObject(ctx, &s3.PutObjectInput{
		Bucket: &bucket,
		Key:    &key,
		Body:   strings.NewReader(""),
	})
	return err
}

func (s *awsStorage) ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]ObjectMeta, error) {
	resp, err := s.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
		Bucket: &bucket,
		Prefix: &prefix,
	})
	if err != nil {
		return nil, err
	}

	objects := make([]ObjectMeta, 0, len(resp.Contents))
	for _, item := range resp.Contents {
		objects = append(objects, ObjectMeta{
			Key:  *item.Key,
			Size: *item.Size,
		})
	}
	return objects, nil
}

func (s *awsStorage) ListObject(ctx context.Context, bucket string) ([]ObjectMeta, error) {
	return s.ListPrefixObject(ctx, bucket, "")
}

func (s *awsStorage) ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) ([]ObjectMeta, string, error) {
	maxKeys := int32(pageSize)
	input := &s3.ListObjectsV2Input{
		Bucket:  &bucket,
		Prefix:  &prefix,
		MaxKeys: &maxKeys,
	}
	if pageToken != "" {
		input.ContinuationToken = &pageToken
	}

	resp, err := s.client.ListObjectsV2(ctx, input)
	if err != nil {
		return nil, "", err
	}

	objects := make([]ObjectMeta, 0, len(resp.Contents))
	for _, item := range resp.Contents {
		objects = append(objects, ObjectMeta{
			Key:  *item.Key,
			Size: *item.Size,
		})
	}

	nextToken := ""
	if resp.NextContinuationToken != nil {
		nextToken = *resp.NextContinuationToken
	}
	return objects, nextToken, nil
}

func (s *awsStorage) DeleteObject(ctx context.Context, bucket, key string) error {
	_, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{
		Bucket: &bucket,
		Key:    &key,
	})
	return err
}

func (s *awsStorage) DeleteObjects(ctx context.Context, bucket string, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	for i := 0; i < len(keys); i += deleteObjectsBatchSize {
		end := i + deleteObjectsBatchSize
		if end > len(keys) {
			end = len(keys)
		}
		batch := keys[i:end]
		objects := make([]types.ObjectIdentifier, len(batch))
		for j, key := range batch {
			objects[j] = types.ObjectIdentifier{Key: aws.String(key)}
		}
		output, err := s.client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
			Bucket: aws.String(bucket),
			Delete: &types.Delete{
				Objects: objects,
				Quiet:   aws.Bool(true),
			},
		})
		if err != nil {
			return err
		}
		if len(output.Errors) > 0 {
			first := output.Errors[0]
			msg := ""
			if first.Message != nil {
				msg = *first.Message
			}
			key := ""
			if first.Key != nil {
				key = *first.Key
			}
			return fmt.Errorf("delete objects failed for key=%s: %s", key, msg)
		}
	}
	return nil
}

type gcpStorage struct {
	client *gcs.Client
}

func NewGcpStorage(client *gcs.Client) Storage {
	return &gcpStorage{client: client}
}

func (s *gcpStorage) Object(ctx context.Context, bucket, key string) (*Object, error) {
	obj := s.client.Bucket(bucket).Object(key)

	attrs, err := obj.Attrs(ctx)
	if err != nil {
		if errors.Is(err, gcs.ErrObjectNotExist) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	reader, err := obj.NewReader(ctx)
	if err != nil {
		return nil, err
	}

	return &Object{
		ETag:          &attrs.Etag,
		ContentLength: &attrs.Size,
		ContentType:   &attrs.ContentType,
		Reader:        reader,
	}, nil
}

func (s *gcpStorage) ObjectExists(ctx context.Context, bucket, key string) (bool, error) {
	obj := s.client.Bucket(bucket).Object(key)
	_, err := obj.Attrs(ctx)
	if err != nil {
		if errors.Is(err, gcs.ErrObjectNotExist) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (s *gcpStorage) ObjectSize(ctx context.Context, bucket, key string) (int64, error) {
	obj := s.client.Bucket(bucket).Object(key)
	attrs, err := obj.Attrs(ctx)
	if err != nil {
		if errors.Is(err, gcs.ErrObjectNotExist) {
			return 0, nil
		}
		return 0, err
	}
	return attrs.Size, nil
}

func (s *gcpStorage) Objects(ctx context.Context, bucket, prefix string) ([]ObjectMeta, error) {
	iter := s.client.Bucket(bucket).Objects(ctx, &gcs.Query{
		Prefix: prefix,
	})
	var keys []ObjectMeta
	for {
		attr, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return nil, err
		}
		if attr == nil {
			break
		}
		keys = append(keys, ObjectMeta{
			Key:  attr.Name,
			Size: attr.Size,
		})
	}
	return keys, nil
}

func (s *gcpStorage) ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error) {
	iter := s.client.Bucket(bucket).Objects(ctx, &gcs.Query{
		Delimiter: "/",
		Prefix:    prefix,
	})
	var prefixes []string
	for {
		attr, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return nil, err
		}
		if attr == nil {
			break
		}
		if attr.Prefix != "" {
			prefixes = append(prefixes, attr.Prefix)
		}
	}
	return prefixes, nil
}

func (s *gcpStorage) ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(ObjectMeta) bool) error {
	iter := s.client.Bucket(bucket).Objects(ctx, &gcs.Query{
		Prefix: prefix,
	})
	for {
		attr, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		if attr == nil {
			break
		}
		if !callback(ObjectMeta{
			Key:  attr.Name,
			Size: attr.Size,
		}) {
			return nil
		}
	}
	return nil
}

func (s *gcpStorage) ObjectsCallbackWithStartKey(
	ctx context.Context, bucket, prefix, startKey string, callback func(ObjectMeta) bool,
) error {
	iter := s.client.Bucket(bucket).Objects(ctx, &gcs.Query{
		Prefix:      prefix,
		StartOffset: startKey,
	})
	for {
		attr, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		if attr == nil {
			break
		}
		if !callback(ObjectMeta{
			Key:  attr.Name,
			Size: attr.Size,
		}) {
			return nil
		}
	}
	return nil
}

func (s *gcpStorage) SaveObject(ctx context.Context, bucket, key string, body io.Reader) error {
	w := s.client.Bucket(bucket).Object(key).NewWriter(ctx)
	if _, err := io.Copy(w, body); err != nil {
		return err
	}
	return w.Close()
}

func (s *gcpStorage) Mkdir(ctx context.Context, bucket, key string, body io.Reader) error {
	w := s.client.Bucket(bucket).Object(key).NewWriter(ctx)
	if _, err := io.Copy(w, body); err != nil {
		return err
	}
	return w.Close()
}

func (s *gcpStorage) ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]ObjectMeta, error) {
	return nil, fmt.Errorf("ListPrefixObject not implemented for gcp storage")
}

func (s *gcpStorage) ListObject(ctx context.Context, bucket string) ([]ObjectMeta, error) {
	return s.ListPrefixObject(ctx, bucket, "")
}

func (s *gcpStorage) ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) ([]ObjectMeta, string, error) {
	return nil, "", fmt.Errorf("ListObjectsPaged not implemented for gcp storage")
}

func (s *gcpStorage) DeleteObject(ctx context.Context, bucket, key string) error {
	return s.client.Bucket(bucket).Object(key).Delete(ctx)
}

func (s *gcpStorage) DeleteObjects(ctx context.Context, bucket string, keys []string) error {
	return deleteObjectsInParallel(ctx, keys, func(ctx context.Context, key string) error {
		return s.client.Bucket(bucket).Object(key).Delete(ctx)
	}, gcpDeleteParallelism)
}

type azureStorage struct {
	client *azblob.Client
}

func NewAzureStorage(client *azblob.Client) Storage {
	return &azureStorage{client: client}
}

func (s *azureStorage) Object(ctx context.Context, bucket, key string) (*Object, error) {
	client := s.client.ServiceClient().NewContainerClient(bucket).NewBlobClient(key)
	resp, err := client.DownloadStream(ctx, nil)
	if err != nil {
		var respErr *azcore.ResponseError
		if errors.As(err, &respErr) {
			if respErr.StatusCode == http.StatusNotFound {
				return nil, ErrNotFound
			}
		}
		return nil, err
	}
	return &Object{
		ETag:          (*string)(resp.ETag),
		ContentType:   resp.ContentType,
		ContentLength: resp.ContentLength,
		ContentRange:  resp.ContentRange,
		AcceptRanges:  resp.AcceptRanges,
		Reader:        resp.Body,
	}, nil
}

func (s *azureStorage) ObjectExists(ctx context.Context, bucket, key string) (bool, error) {
	client := s.client.ServiceClient().NewContainerClient(bucket).NewBlobClient(key)
	_, err := client.GetProperties(ctx, nil)
	if err != nil {
		var respErr *azcore.ResponseError
		if errors.As(err, &respErr) {
			if respErr.StatusCode == http.StatusNotFound {
				return false, nil
			}
		}
		return false, err
	}
	return true, nil
}

func (s *azureStorage) ObjectSize(ctx context.Context, bucket, key string) (int64, error) {
	client := s.client.ServiceClient().NewContainerClient(bucket).NewBlobClient(key)
	props, err := client.GetProperties(ctx, nil)
	if err != nil {
		var respErr *azcore.ResponseError
		if errors.As(err, &respErr) {
			if respErr.StatusCode == http.StatusNotFound {
				return 0, nil
			}
		}
		return 0, err
	}
	if props.ContentLength == nil {
		return 0, errors.New("content length is nil") // Should not happen for a valid blob
	}
	return *props.ContentLength, nil
}

func (s *azureStorage) Objects(ctx context.Context, bucket, prefix string) ([]ObjectMeta, error) {
	var objects []ObjectMeta
	client := s.client.ServiceClient().NewContainerClient(bucket)
	pager := client.NewListBlobsFlatPager(&azblob.ListBlobsFlatOptions{Prefix: &prefix})
	for pager.More() {
		resp, err := pager.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, blob := range resp.Segment.BlobItems {
			objects = append(objects, ObjectMeta{
				Key:  *blob.Name,
				Size: *blob.Properties.ContentLength,
			})
		}
	}
	return objects, nil
}

func (s *azureStorage) ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error) {
	var prefixes []string
	client := s.client.ServiceClient().NewContainerClient(bucket)
	delimiter := "/"
	pager := client.NewListBlobsHierarchyPager(delimiter, &container.ListBlobsHierarchyOptions{Prefix: &prefix})
	for pager.More() {
		resp, err := pager.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, prefixItem := range resp.Segment.BlobPrefixes {
			prefixes = append(prefixes, *prefixItem.Name)
		}
	}
	return prefixes, nil
}

func (s *azureStorage) ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(ObjectMeta) bool) error {
	client := s.client.ServiceClient().NewContainerClient(bucket)
	pager := client.NewListBlobsFlatPager(&azblob.ListBlobsFlatOptions{Prefix: &prefix})
	for pager.More() {
		resp, err := pager.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, blob := range resp.Segment.BlobItems {
			if !callback(ObjectMeta{
				Key:  *blob.Name,
				Size: *blob.Properties.ContentLength,
			}) {
				return nil
			}
		}
	}
	return nil
}

func (s *azureStorage) ObjectsCallbackWithStartKey(ctx context.Context, bucket, prefix, startKey string, callback func(ObjectMeta) bool) error {
	client := s.client.ServiceClient().NewContainerClient(bucket)
	pager := client.NewListBlobsFlatPager(&azblob.ListBlobsFlatOptions{
		Prefix: &prefix,
		Marker: &startKey,
	})
	for pager.More() {
		resp, err := pager.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, blob := range resp.Segment.BlobItems {
			if !callback(ObjectMeta{
				Key:  *blob.Name,
				Size: *blob.Properties.ContentLength,
			}) {
				return nil
			}
		}
	}
	return nil
}

func (s *azureStorage) SaveObject(ctx context.Context, bucket, key string, body io.Reader) error {
	client := s.client.ServiceClient().NewContainerClient(bucket).NewBlockBlobClient(key)
	_, err := client.UploadStream(ctx, body, nil)
	return err
}

func (s *azureStorage) Mkdir(ctx context.Context, bucket, key string, body io.Reader) error {
	client := s.client.ServiceClient().NewContainerClient(bucket).NewBlockBlobClient(key)
	_, err := client.UploadStream(ctx, body, nil)
	return err
}

func (s *azureStorage) ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]ObjectMeta, error) {
	return nil, fmt.Errorf("ListPrefixObject not implemented for azure storage")
}

func (s *azureStorage) ListObject(ctx context.Context, bucket string) ([]ObjectMeta, error) {
	return s.ListPrefixObject(ctx, bucket, "")
}

func (s *azureStorage) ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) ([]ObjectMeta, string, error) {
	return nil, "", fmt.Errorf("ListObjectsPaged not implemented for azure storage")
}

func (s *azureStorage) DeleteObject(ctx context.Context, bucket, key string) error {
	client := s.client.ServiceClient().NewContainerClient(bucket).NewBlobClient(key)
	_, err := client.Delete(ctx, nil)
	return err
}

func (s *azureStorage) DeleteObjects(ctx context.Context, bucket string, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	containerClient := s.client.ServiceClient().NewContainerClient(bucket)
	for i := 0; i < len(keys); i += azureDeleteObjectsBatchSize {
		end := i + azureDeleteObjectsBatchSize
		if end > len(keys) {
			end = len(keys)
		}
		batch := keys[i:end]
		bb, err := containerClient.NewBatchBuilder()
		if err != nil {
			return err
		}
		for _, key := range batch {
			if err := bb.Delete(key, nil); err != nil {
				return err
			}
		}
		resp, err := containerClient.SubmitBatch(ctx, bb, nil)
		if err != nil {
			return err
		}
		for _, item := range resp.Responses {
			if item.Error != nil {
				blobName := ""
				if item.BlobName != nil {
					blobName = *item.BlobName
				}
				return fmt.Errorf("delete blob %s failed: %w", blobName, item.Error)
			}
		}
	}
	return nil
}

type localStorage struct {
	rootDir string
}

func (s *localStorage) getObjectPath(bucket, key string) string {
	return filepath.Join(s.rootDir, bucket, key)
}

func (s *localStorage) Object(ctx context.Context, bucket, key string) (*Object, error) {
	filePath := s.getObjectPath(bucket, key)
	file, err := os.Open(filePath)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	fileInfo, err := file.Stat()
	if err != nil {
		file.Close()
		return nil, err
	}

	contentLength := fileInfo.Size()
	contentType := "application/octet-stream"

	return &Object{
		ContentLength: &contentLength,
		ContentType:   &contentType,
		Reader:        file,
	}, nil
}

func (s *localStorage) ObjectExists(ctx context.Context, bucket, key string) (bool, error) {
	filePath := s.getObjectPath(bucket, key)
	_, err := os.Stat(filePath)
	if err != nil {
		if os.IsNotExist(err) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (s *localStorage) ObjectSize(ctx context.Context, bucket, key string) (int64, error) {
	filePath := s.getObjectPath(bucket, key)
	fileInfo, err := os.Stat(filePath)
	if err != nil {
		if os.IsNotExist(err) {
			return 0, ErrNotFound
		}
		return 0, err
	}
	return fileInfo.Size(), nil
}

func (s *localStorage) Objects(ctx context.Context, bucket, prefix string) ([]ObjectMeta, error) {
	var result []ObjectMeta
	err := s.ObjectsCallback(ctx, bucket, prefix, func(meta ObjectMeta) bool {
		result = append(result, meta)
		return true
	})

	if err != nil {
		return nil, err
	}

	return result, nil
}

func (s *localStorage) ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error) {
	var result []string
	rootPath := filepath.Join(s.rootDir, bucket, prefix)

	err := filepath.WalkDir(rootPath, func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}

		if d.IsDir() {
			relPath, err := filepath.Rel(filepath.Join(s.rootDir, bucket), path)
			if err != nil {
				return err
			}
			if relPath != "." && !strings.HasSuffix(relPath, "/") {
				relPath += "/"
			}
			if relPath != "./" && relPath != "/" {
				result = append(result, relPath)
			}
		}

		return nil
	})

	if err != nil {
		if os.IsNotExist(err) {
			return []string{}, nil
		}
		return nil, err
	}

	return result, nil
}

func (s *localStorage) ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(ObjectMeta) bool) error {
	return s.ObjectsCallbackWithStartKey(ctx, bucket, prefix, "", callback)
}

func (s *localStorage) ObjectsCallbackWithStartKey(ctx context.Context, bucket, prefix, startKey string, callback func(ObjectMeta) bool) error {
	rootPath := filepath.Join(s.rootDir, bucket, prefix)

	return filepath.Walk(rootPath, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			if os.IsNotExist(err) {
				return nil
			}
			return err
		}

		if info.IsDir() {
			return nil
		}

		relPath, err := filepath.Rel(filepath.Join(s.rootDir, bucket), path)
		if err != nil {
			return err
		}

		if startKey != "" && relPath < startKey {
			return nil
		}

		meta := ObjectMeta{
			Key:  relPath,
			Size: info.Size(),
		}

		if !callback(meta) {
			return filepath.SkipAll
		}

		return nil
	})
}

func (s *localStorage) SaveObject(ctx context.Context, bucket, key string, body io.Reader) error {
	filePath := s.getObjectPath(bucket, key)

	// Create directory if it doesn't exist
	if err := os.MkdirAll(filepath.Dir(filePath), 0755); err != nil {
		return err
	}

	file, err := os.Create(filePath)
	if err != nil {
		return err
	}
	defer file.Close()

	_, err = io.Copy(file, body)
	return err
}

func (s *localStorage) Mkdir(ctx context.Context, bucket, key string, body io.Reader) error {
	dirPath := s.getObjectPath(bucket, key)
	if !strings.HasSuffix(dirPath, string(filepath.Separator)) {
		dirPath += string(filepath.Separator)
	}
	return os.MkdirAll(dirPath, 0755)
}

func (s *localStorage) ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]ObjectMeta, error) {
	return s.Objects(ctx, bucket, prefix)
}

func (s *localStorage) ListObject(ctx context.Context, bucket string) ([]ObjectMeta, error) {
	return s.ListPrefixObject(ctx, bucket, "")
}

func (s *localStorage) ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) ([]ObjectMeta, string, error) {
	allObjects, err := s.Objects(ctx, bucket, prefix)
	if err != nil {
		return nil, "", err
	}
	sort.Slice(allObjects, func(i, j int) bool {
		return allObjects[i].Key < allObjects[j].Key
	})

	startIdx := 0
	if pageToken != "" {
		for i, obj := range allObjects {
			if obj.Key >= pageToken {
				startIdx = i
				break
			}
		}
	}
	endIdx := startIdx + pageSize
	if endIdx > len(allObjects) {
		endIdx = len(allObjects)
	}
	page := allObjects[startIdx:endIdx]
	nextToken := ""
	if endIdx < len(allObjects) {
		nextToken = allObjects[endIdx].Key
	}
	return page, nextToken, nil
}

func (s *localStorage) DeleteObject(ctx context.Context, bucket, key string) error {
	filePath := s.getObjectPath(bucket, key)
	err := os.Remove(filePath)
	if err != nil && os.IsNotExist(err) {
		return nil
	}
	return err
}

func (s *localStorage) DeleteObjects(ctx context.Context, bucket string, keys []string) error {
	for _, key := range keys {
		if err := s.DeleteObject(ctx, bucket, key); err != nil {
			return err
		}
	}
	return nil
}

func NewLocalStorage() Storage {
	// Use current working directory as the root
	cwd, _ := os.Getwd()
	return &localStorage{
		rootDir: filepath.Join(cwd, "local-storage"),
	}
}

type alicloudStorage struct {
	client *oss.Client
}

func NewAlicloudStorage(client *oss.Client) ListStorage {
	return &alicloudStorage{client: client}
}

func (s *alicloudStorage) Object(ctx context.Context, bucket, key string) (*Object, error) {
	output, err := s.client.GetObject(ctx, &oss.GetObjectRequest{
		Bucket: oss.Ptr(bucket),
		Key:    oss.Ptr(key),
	})
	if err != nil {
		var serviceError *oss.ServiceError
		if errors.As(err, &serviceError) {
			if serviceError.Code == "NoSuchKey" || serviceError.StatusCode == http.StatusNotFound {
				return nil, ErrNotFound
			}
		}
		return nil, err
	}

	// Get metadata using HeadObject for additional information
	headOutput, err := s.client.HeadObject(ctx, &oss.HeadObjectRequest{
		Bucket: oss.Ptr(bucket),
		Key:    oss.Ptr(key),
	})
	if err != nil {
		// If HeadObject fails, we still return the object with basic info
		return &Object{
			ETag:          output.ETag,
			ContentType:   output.ContentType,
			ContentLength: nil,
			ContentRange:  output.ContentRange,
			AcceptRanges:  nil,
			Reader:        output.Body,
		}, nil
	}

	var contentLength *int64
	if headOutput.ContentLength > 0 {
		contentLength = &headOutput.ContentLength
	}

	return &Object{
		ETag:          output.ETag,
		ContentType:   output.ContentType,
		ContentLength: contentLength,
		ContentRange:  output.ContentRange,
		AcceptRanges:  nil,
		Reader:        output.Body,
	}, nil
}

func (s *alicloudStorage) ObjectExists(ctx context.Context, bucket, key string) (bool, error) {
	_, err := s.client.HeadObject(ctx, &oss.HeadObjectRequest{
		Bucket: oss.Ptr(bucket),
		Key:    oss.Ptr(key),
	})
	if err != nil {
		var serviceError *oss.ServiceError
		if errors.As(err, &serviceError) {
			if serviceError.Code == "NoSuchKey" || serviceError.StatusCode == http.StatusNotFound {
				return false, nil
			}
		}
		return false, err
	}
	return true, nil
}

func (s *alicloudStorage) ObjectSize(ctx context.Context, bucket, key string) (int64, error) {
	output, err := s.client.HeadObject(ctx, &oss.HeadObjectRequest{
		Bucket: oss.Ptr(bucket),
		Key:    oss.Ptr(key),
	})
	if err != nil {
		var serviceError *oss.ServiceError
		if errors.As(err, &serviceError) {
			if serviceError.Code == "NoSuchKey" || serviceError.StatusCode == http.StatusNotFound {
				return 0, nil
			}
		}
		return 0, err
	}
	return output.ContentLength, nil
}

func (s *alicloudStorage) Objects(ctx context.Context, bucket, prefix string) ([]ObjectMeta, error) {
	input := &oss.ListObjectsV2Request{
		Bucket: oss.Ptr(bucket),
		Prefix: oss.Ptr(prefix),
	}
	paginator := s.client.NewListObjectsV2Paginator(input)

	var keys []ObjectMeta
	for paginator.HasNext() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, obj := range page.Contents {
			keys = append(keys, ObjectMeta{
				Key:  *obj.Key,
				Size: obj.Size,
			})
		}
	}
	return keys, nil
}

func (s *alicloudStorage) ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error) {
	input := &oss.ListObjectsV2Request{
		Bucket:    oss.Ptr(bucket),
		Prefix:    oss.Ptr(prefix),
		Delimiter: oss.Ptr("/"),
	}
	output, err := s.client.ListObjectsV2(ctx, input)
	if err != nil {
		return nil, err
	}
	if len(output.CommonPrefixes) == 0 {
		return nil, nil
	}
	prefixes := make([]string, len(output.CommonPrefixes))
	for n, p := range output.CommonPrefixes {
		prefixes[n] = *p.Prefix
	}
	return prefixes, nil
}

func (s *alicloudStorage) ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(ObjectMeta) bool) error {
	input := &oss.ListObjectsV2Request{
		Bucket: oss.Ptr(bucket),
		Prefix: oss.Ptr(prefix),
	}
	paginator := s.client.NewListObjectsV2Paginator(input)
	for paginator.HasNext() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, obj := range page.Contents {
			if !callback(ObjectMeta{
				Key:  *obj.Key,
				Size: obj.Size,
			}) {
				return nil
			}
		}
	}
	return nil
}

func (s *alicloudStorage) ObjectsCallbackWithStartKey(
	ctx context.Context, bucket, prefix, startKey string, callback func(ObjectMeta) bool,
) error {
	// OSS doesn't have a direct StartAfter parameter, but we can use StartAfter in ListObjectsV2
	input := &oss.ListObjectsV2Request{
		Bucket:     oss.Ptr(bucket),
		Prefix:     oss.Ptr(prefix),
		StartAfter: oss.Ptr(startKey),
	}
	paginator := s.client.NewListObjectsV2Paginator(input)
	for paginator.HasNext() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, obj := range page.Contents {
			if !callback(ObjectMeta{
				Key:  *obj.Key,
				Size: obj.Size,
			}) {
				return nil
			}
		}
	}
	return nil
}

func (s *alicloudStorage) SaveObject(ctx context.Context, bucket, key string, body io.Reader) error {
	_, err := s.client.PutObject(ctx, &oss.PutObjectRequest{
		Bucket: oss.Ptr(bucket),
		Key:    oss.Ptr(key),
		Body:   body,
	})
	return err
}

func (s *alicloudStorage) Mkdir(ctx context.Context, bucket, key string, body io.Reader) error {
	if !strings.HasSuffix(key, "/") {
		key += "/"
	}
	_, err := s.client.PutObject(ctx, &oss.PutObjectRequest{
		Bucket:      oss.Ptr(bucket),
		Key:         oss.Ptr(key),
		Body:        strings.NewReader(""),
		ContentType: oss.Ptr("application/octet-stream"),
	})
	return err
}

func (s *alicloudStorage) ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]ObjectMeta, error) {
	request := &oss.ListObjectsV2Request{
		Bucket: oss.Ptr(bucket),
		Prefix: oss.Ptr(prefix),
	}
	lsRes, err := s.client.ListObjectsV2(ctx, request)
	if err != nil {
		return nil, err
	}

	objects := make([]ObjectMeta, 0, len(lsRes.Contents))
	for _, obj := range lsRes.Contents {
		objects = append(objects, ObjectMeta{
			Key:  *obj.Key,
			Size: obj.Size,
		})
	}
	return objects, nil
}

func (s *alicloudStorage) ListObject(ctx context.Context, bucket string) ([]ObjectMeta, error) {
	return s.ListPrefixObject(ctx, bucket, "")
}

func (s *alicloudStorage) ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) ([]ObjectMeta, string, error) {
	request := &oss.ListObjectsV2Request{
		Bucket:  oss.Ptr(bucket),
		Prefix:  oss.Ptr(prefix),
		MaxKeys: int32(pageSize),
	}
	if pageToken != "" {
		request.ContinuationToken = oss.Ptr(pageToken)
	}

	lsRes, err := s.client.ListObjectsV2(ctx, request)
	if err != nil {
		return nil, "", err
	}

	objects := make([]ObjectMeta, 0, len(lsRes.Contents))
	for _, obj := range lsRes.Contents {
		objects = append(objects, ObjectMeta{
			Key:  *obj.Key,
			Size: obj.Size,
		})
	}

	nextToken := ""
	if lsRes.NextContinuationToken != nil {
		nextToken = *lsRes.NextContinuationToken
	}
	return objects, nextToken, nil
}

func (s *alicloudStorage) DeleteObject(ctx context.Context, bucket, key string) error {
	_, err := s.client.DeleteObject(ctx, &oss.DeleteObjectRequest{
		Bucket: oss.Ptr(bucket),
		Key:    oss.Ptr(key),
	})
	return err
}

func (s *alicloudStorage) DeleteObjects(ctx context.Context, bucket string, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	for i := 0; i < len(keys); i += deleteObjectsBatchSize {
		end := i + deleteObjectsBatchSize
		if end > len(keys) {
			end = len(keys)
		}
		batch := keys[i:end]
		objects := make([]oss.DeleteObject, len(batch))
		for j, key := range batch {
			objects[j] = oss.DeleteObject{Key: oss.Ptr(key)}
		}
		_, err := s.client.DeleteMultipleObjects(ctx, &oss.DeleteMultipleObjectsRequest{
			Bucket:  oss.Ptr(bucket),
			Objects: objects,
			Quiet:   true,
		})
		if err != nil {
			return err
		}
	}
	return nil
}

// deleteObjectsInParallel deletes keys with bounded concurrency. Used by GCS where the
// Go client has no multi-object delete API (official recommendation is parallel deletes).
func deleteObjectsInParallel(
	ctx context.Context,
	keys []string,
	deleteFn func(ctx context.Context, key string) error,
	parallelism int,
) error {
	if len(keys) == 0 {
		return nil
	}
	if parallelism <= 0 {
		parallelism = gcpDeleteParallelism
	}
	sem := make(chan struct{}, parallelism)
	var wg sync.WaitGroup
	var firstErr error
	var mu sync.Mutex
	for _, key := range keys {
		wg.Add(1)
		go func(k string) {
			defer wg.Done()
			sem <- struct{}{}
			defer func() { <-sem }()
			if err := deleteFn(ctx, k); err != nil {
				mu.Lock()
				if firstErr == nil {
					firstErr = fmt.Errorf("delete object %s: %w", k, err)
				}
				mu.Unlock()
			}
		}(key)
	}
	wg.Wait()
	return firstErr
}

// NewAliyunOSS is a compatibility alias for data-process and legacy callers.
func NewAliyunOSS(client *oss.Client) ListStorage {
	return NewAlicloudStorage(client)
}