package oss

import (
	"context"
	"errors"
	"fmt"
	"path"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/sirupsen/logrus"

	"github.com/tidbcloud/observability/backup-worker/pkg/common"
	"github.com/tidbcloud/observability/backup-worker/pkg/configs"
)

type Client struct {
	cli      *oss.Client
	uploader *oss.Uploader
	cfg      configs.OssSink
	rp       configs.RetryPolicy
}

// New creates an Oss client with flexible credential resolution.
// Credential priority:
// 1. Explicit AccessKeyID and AccessKeySecret from config
// 2. Environment variables (ALIBABA_CLOUD_ACCESS_KEY_ID, ALIBABA_CLOUD_ACCESS_KEY_SECRET)
//
// Environment variable authentication is preferred for containerized deployments
// where credentials are injected via Kubernetes secrets or similar mechanisms.
func New(_ context.Context, cfg configs.OssSink, rp configs.RetryPolicy) (*Client, error) {
	var credProvider credentials.CredentialsProvider

	// Use explicit credentials if provided, otherwise fall back to environment.
	// This pattern supports both interactive development and production deployments.
	if cfg.AccessKeyID != "" && cfg.AccessKeySecret != "" {
		credProvider = credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.AccessKeySecret, cfg.SecurityToken)
	} else {
		credProvider = credentials.NewEnvironmentVariableCredentialsProvider()
	}

	ossCfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credProvider).
		WithRegion(cfg.Region).
		WithEndpoint(cfg.Endpoint)

	cli := oss.NewClient(ossCfg)

	logrus.Infof("Oss client created: bucket=%s, region=%s", cfg.Bucket, cfg.Region)
	return &Client{
		cli:      cli,
		uploader: oss.NewUploader(cli),
		cfg:      cfg,
		rp:       rp,
	}, nil
}

func (c *Client) Type() string {
	return common.StorageOSS
}

// CheckObject uses HeadObject to verify object existence.
//
// HeadObject retrieves only metadata without the object body,
// making it efficient for existence checks in resumable backups.
func (c *Client) CheckObject(ctx context.Context, key string) (bool, error) {
	normalizeKey := c.normalizePath(key)
	_, err := c.cli.HeadObject(ctx, &oss.HeadObjectRequest{
		Bucket: oss.Ptr(c.cfg.Bucket),
		Key:    oss.Ptr(normalizeKey),
	})
	if err == nil {
		return true, nil
	}
	var respErr *oss.ServiceError
	if errors.As(err, &respErr) && respErr.StatusCode == 404 {
		return false, nil
	}
	return false, fmt.Errorf("check oss object: %w", err)
}

// SinkObject uploads a local file to OSS with automatic retries.
//
// The uploader uses multipart upload for files larger than the threshold,
// providing better reliability and throughput for large backup files.
func (c *Client) SinkObject(ctx context.Context, key, localPath string) error {
	normalizeKey := c.normalizePath(key)
	return c.rp.Do(ctx, func() error {
		return c.sinkObject(ctx, normalizeKey, localPath)
	})
}

func (c *Client) sinkObject(ctx context.Context, key, localPath string) error {
	_, err := c.uploader.UploadFile(ctx, &oss.PutObjectRequest{
		Bucket: oss.Ptr(c.cfg.Bucket),
		Key:    oss.Ptr(key),
	}, localPath)
	if err != nil {
		return fmt.Errorf("upload oss object: %w", err)
	}
	return nil
}

// ListObjects paginates through all objects under the given prefix.
func (c *Client) ListObjects(ctx context.Context, prefix string) ([]common.ObjectInfo, error) {
	var objects []common.ObjectInfo
	var continuationToken *string
	for {
		result, err := c.cli.ListObjectsV2(ctx, &oss.ListObjectsV2Request{
			Bucket:            oss.Ptr(c.cfg.Bucket),
			Prefix:            oss.Ptr(prefix),
			ContinuationToken: continuationToken,
		})
		if err != nil {
			return nil, fmt.Errorf("list oss objects '%s': %w", prefix, err)
		}
		for _, obj := range result.Contents {
			name := ""
			if obj.Key != nil {
				name = *obj.Key
			}
			objects = append(objects, common.ObjectInfo{
				Name: name,
				Size: obj.Size,
			})
		}
		if !result.IsTruncated {
			break
		}
		continuationToken = result.NextContinuationToken
	}
	return objects, nil
}

// CopyObject copies an object from another bucket into this client's bucket.
// Retries on failure and aborts any in-progress multipart upload to avoid
// orphaned parts that would consume storage and incur costs.
//
// Small objects use CopyObject directly. Large objects use multipart copy
// via InitiateMultipartUpload + UploadPartCopy (5 GB parts).
func (c *Client) CopyObject(ctx context.Context, srcBucket, srcKey, destKey string, size int64) error {
	return c.rp.Do(ctx, func() error {
		uploadId, err := c.copyObject(ctx, srcBucket, srcKey, destKey, size)
		if err != nil {
			if uploadId != "" {
				c.cli.AbortMultipartUpload(ctx, &oss.AbortMultipartUploadRequest{
					Bucket:   oss.Ptr(c.cfg.Bucket),
					Key:      oss.Ptr(destKey),
					UploadId: oss.Ptr(uploadId),
				})
			}
			return err
		}
		return nil
	})
}

func (c *Client) copyObject(ctx context.Context, srcBucket, srcKey, destKey string, size int64) (string, error) {
	const partSize int64 = 5 * 1024 * 1024 * 1024 // 5 GB per part

	if size <= partSize {
		_, err := c.cli.CopyObject(ctx, &oss.CopyObjectRequest{
			Bucket:       oss.Ptr(c.cfg.Bucket),
			Key:          oss.Ptr(destKey),
			SourceBucket: oss.Ptr(srcBucket),
			SourceKey:    oss.Ptr(srcKey),
		})
		if err != nil {
			return "", fmt.Errorf("copy oss object: %w", err)
		}
		return "", nil
	}

	mpu, err := c.cli.InitiateMultipartUpload(ctx, &oss.InitiateMultipartUploadRequest{
		Bucket: oss.Ptr(c.cfg.Bucket),
		Key:    oss.Ptr(destKey),
	})
	if err != nil {
		return "", fmt.Errorf("initiate multipart upload: %w", err)
	}

	var completedParts []oss.UploadPart
	var partNumber int32 = 1

	for offset := int64(0); offset < size; offset += partSize {
		end := offset + partSize - 1
		if end >= size {
			end = size - 1
		}
		rangeStr := fmt.Sprintf("bytes=%d-%d", offset, end)

		partResult, err := c.cli.UploadPartCopy(ctx, &oss.UploadPartCopyRequest{
			Bucket:       oss.Ptr(c.cfg.Bucket),
			Key:          oss.Ptr(destKey),
			UploadId:     mpu.UploadId,
			PartNumber:   partNumber,
			SourceBucket: oss.Ptr(srcBucket),
			SourceKey:    oss.Ptr(srcKey),
			Range:        oss.Ptr(rangeStr),
		})
		if err != nil {
			return *mpu.UploadId, fmt.Errorf("upload part copy %d: %w", partNumber, err)
		}

		completedParts = append(completedParts, oss.UploadPart{
			PartNumber: partNumber,
			ETag:       partResult.ETag,
		})
		partNumber++
	}

	_, err = c.cli.CompleteMultipartUpload(ctx, &oss.CompleteMultipartUploadRequest{
		Bucket:   oss.Ptr(c.cfg.Bucket),
		Key:      oss.Ptr(destKey),
		UploadId: mpu.UploadId,
		CompleteMultipartUpload: &oss.CompleteMultipartUpload{
			Parts: completedParts,
		},
	})
	return *mpu.UploadId, err
}

func (c *Client) Close() error {
	return nil
}

func (c *Client) normalizePath(key string) string {
	return path.Join(c.cfg.PathPrefix, key)
}