package azblob

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path"
	"path/filepath"
	"strings"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"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/sas"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service"
	"github.com/sirupsen/logrus"

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

type Client struct {
	cli           *azblob.Client
	cfg           configs.AzBlobSink
	rp            configs.RetryPolicy
	sharedKeyCred *azblob.SharedKeyCredential
}

// New creates an Azure Blob client with flexible authentication.
// Authentication priority:
//  1. Connection string: Direct credentials, simplest setup
//  2. Service principal credentials file: JSON file with tenantId, clientId, clientSecret
//  3. Default credential chain: Supports managed identity, environment variables,
//     and CLI credentials - recommended for production Azure deployments
//
// The default URL format assumes Azure public cloud. For sovereign clouds
// (China, Government), AccountURL must be explicitly configured.
func New(_ context.Context, cfg configs.AzBlobSink, rp configs.RetryPolicy) (*Client, error) {
	var cli *azblob.Client
	var err error

	// Connection string authentication is simplest for development.
	// It contains account name, key, and endpoint in a single string.
	if cfg.ConnectionString != "" {
		cli, err = azblob.NewClientFromConnectionString(cfg.ConnectionString, nil)
		if err != nil {
			return nil, fmt.Errorf("create azure client from connection string: %w", err)
		}
		// Extract account key for SAS URL generation during copy operations.
		var accountKey string
		var sharedKeyCred *azblob.SharedKeyCredential
		accountName, accountKey := parseConnectionString(cfg.ConnectionString)
		if accountName != "" && accountKey != "" {
			sharedKeyCred, err = azblob.NewSharedKeyCredential(accountName, accountKey)
			if err != nil {
				return nil, fmt.Errorf("create shared key credential: %w", err)
			}
		}
		logrus.Infof("Azure Blob client created: container=%s", cfg.Container)
		return &Client{
			cli:           cli,
			cfg:           cfg,
			rp:            rp,
			sharedKeyCred: sharedKeyCred,
		}, nil
	}

	// Construct account URL from account name if not explicitly provided.
	// This uses the standard Azure public cloud endpoint format.
	accountURL := cfg.AccountURL
	if accountURL == "" {
		accountURL = fmt.Sprintf("https://%s.blob.core.windows.net/", cfg.AccountName)
	}

	// Use DefaultAzureCredential for production scenarios.
	// This credential chain attempts multiple authentication methods in order:
	// 1. Environment variables (service principal)
	// 2. Managed Identity (recommended for Azure resources)
	// 3. Azure CLI credentials (for local development)
	var cred azcore.TokenCredential
	if cfg.CredentialsFile != "" {
		sp, loadErr := loadServicePrincipal(cfg.CredentialsFile)
		if loadErr != nil {
			return nil, fmt.Errorf("load azure credentials from %s: %w", cfg.CredentialsFile, loadErr)
		}
		cred, err = azidentity.NewClientSecretCredential(sp.TenantID, sp.ClientID, sp.ClientSecret, nil)
		if err != nil {
			return nil, fmt.Errorf("create azure client secret credential: %w", err)
		}
	} else {
		cred, err = azidentity.NewDefaultAzureCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("create default azure credential: %w", err)
		}
	}

	cli, err = azblob.NewClient(accountURL, cred, nil)
	if err != nil {
		return nil, fmt.Errorf("create azure client: %w", err)
	}

	logrus.Infof("Azure Blob client created: container=%s", cfg.Container)
	return &Client{
		cli: cli,
		cfg: cfg,
		rp:  rp,
	}, nil
}

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

// CheckObject uses GetProperties to verify blob existence.
//
// GetProperties is a HEAD-like operation that retrieves metadata without
// downloading the blob content, making it efficient for existence checks.
func (c *Client) CheckObject(ctx context.Context, key string) (bool, error) {
	normalizeKey := c.normalizePath(key)
	_, err := c.cli.ServiceClient().NewContainerClient(c.cfg.Container).NewBlobClient(normalizeKey).GetProperties(ctx, nil)
	if err == nil {
		return true, nil
	}
	var respErr *azcore.ResponseError
	if errors.As(err, &respErr) && respErr.StatusCode == 404 {
		return false, nil
	}
	return false, fmt.Errorf("check azure blob: %w", err)
}

// SinkObject uploads a local file to Azure Blob Storage with automatic retries.
//
// The Azure SDK handles chunked uploads for large files automatically.
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 {
	f, err := os.Open(localPath)
	if err != nil {
		return fmt.Errorf("open local file: %w", err)
	}
	defer f.Close()

	_, err = c.cli.UploadFile(ctx, c.cfg.Container, key, f, nil)
	if err != nil {
		return fmt.Errorf("upload azure blob: %w", err)
	}
	return nil
}

// ListObjects paginates through all blobs under the given prefix.
func (c *Client) ListObjects(ctx context.Context, prefix string) ([]common.ObjectInfo, error) {
	var objects []common.ObjectInfo
	pager := c.cli.NewListBlobsFlatPager(c.cfg.Container, &azblob.ListBlobsFlatOptions{
		Prefix: &prefix,
	})
	for pager.More() {
		page, err := pager.NextPage(ctx)
		if err != nil {
			return nil, fmt.Errorf("list azure blobs '%s': %w", prefix, err)
		}
		for _, item := range page.Segment.BlobItems {
			var size int64
			if item.Properties != nil && item.Properties.ContentLength != nil {
				size = *item.Properties.ContentLength
			}
			objects = append(objects, common.ObjectInfo{
				Name: *item.Name,
				Size: size,
			})
		}
	}
	return objects, nil
}

// CopyObject copies a blob from another container into this client's container.
// Uses StartCopyFromURL (async server-side copy) and polls for completion.
func (c *Client) CopyObject(ctx context.Context, srcContainer, srcKey, destKey string, srcCli *Client) error {
	return c.rp.Do(ctx, func() error {
		return c.copyObject(ctx, srcContainer, srcKey, destKey, srcCli)
	})
}

// sourceSASURL generates a short-lived SAS URL for the source blob.
//
// StartCopyFromURL is a server-side copy where Azure Storage independently
// accesses the source URL. AAD/Token credentials used by the client cannot be
// forwarded, so a SAS token must be embedded in the source URL.
//
// This method is always called on the SOURCE client (via srcCli in copyObject),
// so the SAS is signed with the source account's credentials — either shared key
// (from connection string) or user delegation (from AAD). This ensures the dest
// account receives a URL it can present to Azure Storage for the copy operation.
func (c *Client) sourceSASURL(ctx context.Context, srcContainer, srcKey string) (string, error) {
	srcBlobURL := c.cli.ServiceClient().NewContainerClient(srcContainer).NewBlobClient(srcKey).URL()

	now := time.Now().UTC()
	expiry := now.Add(1 * time.Hour)

	if c.sharedKeyCred != nil {
		sasQP, err := sas.BlobSignatureValues{
			Protocol:      sas.ProtocolHTTPS,
			StartTime:     now.Add(-10 * time.Second),
			ExpiryTime:    expiry,
			Permissions:   (&sas.BlobPermissions{Read: true}).String(),
			ContainerName: srcContainer,
			BlobName:      srcKey,
		}.SignWithSharedKey(c.sharedKeyCred)
		if err != nil {
			return "", fmt.Errorf("sign sas with shared key: %w", err)
		}
		return srcBlobURL + "?" + sasQP.Encode(), nil
	}

	// AAD auth: use user delegation SAS
	info := service.KeyInfo{
		Start:  to.Ptr(now.Format(sas.TimeFormat)),
		Expiry: to.Ptr(expiry.Format(sas.TimeFormat)),
	}
	udc, err := c.cli.ServiceClient().GetUserDelegationCredential(ctx, info, nil)
	if err != nil {
		return "", fmt.Errorf("get user delegation credential: %w", err)
	}

	sasQP, err := sas.BlobSignatureValues{
		Protocol:      sas.ProtocolHTTPS,
		StartTime:     now.Add(-10 * time.Second),
		ExpiryTime:    expiry,
		Permissions:   (&sas.BlobPermissions{Read: true}).String(),
		ContainerName: srcContainer,
		BlobName:      srcKey,
	}.SignWithUserDelegation(udc)
	if err != nil {
		return "", fmt.Errorf("sign sas with user delegation: %w", err)
	}
	return srcBlobURL + "?" + sasQP.Encode(), nil
}

func (c *Client) copyObject(ctx context.Context, srcContainer, srcKey, destKey string, srcCli *Client) error {
	// Use source client to generate SAS URL for cross-account scenarios.
	// Even within the same account, if the source client uses AAD credentials,
	// a SAS URL is needed to delegate read access to Azure Storage for the copy operation.
	srcBlobURL, err := srcCli.sourceSASURL(ctx, srcContainer, srcKey)
	if err != nil {
		return fmt.Errorf("generate source SAS URL: %w", err)
	}

	destBlobClient := c.cli.ServiceClient().NewContainerClient(c.cfg.Container).NewBlobClient(destKey)
	copyResp, err := destBlobClient.StartCopyFromURL(ctx, srcBlobURL, nil)
	if err != nil {
		return fmt.Errorf("start copy from URL: %w", err)
	}

	maxWait := time.Hour
	deadline := time.Now().Add(maxWait)
	pollInterval := 2 * time.Second

	// Poll for copy completion
	for time.Now().Before(deadline) {
		props, err := destBlobClient.GetProperties(ctx, nil)
		if err != nil {
			return fmt.Errorf("get copy status: %w", err)
		}
		if props.CopyStatus == nil {
			return fmt.Errorf("copy status not available")
		}
		switch *props.CopyStatus {
		case blob.CopyStatusTypeSuccess:
			return nil
		case blob.CopyStatusTypeFailed:
			return fmt.Errorf("copy failed: %s", ptrOrEmpty(props.CopyStatusDescription))
		case blob.CopyStatusTypeAborted:
			return fmt.Errorf("copy aborted: %s", ptrOrEmpty(props.CopyStatusDescription))
		case blob.CopyStatusTypePending:
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(pollInterval):
			}
		}
	}
	return fmt.Errorf("copy timed out after %v, copy ID: %s", maxWait, ptrOrEmpty(copyResp.CopyID))
}

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

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

func ptrOrEmpty(s *string) string {
	if s == nil {
		return ""
	}
	return *s
}

// servicePrincipal holds Azure AD app registration credentials.
// The JSON file format matches the Azure CLI's service principal output:
//
//	{"tenantId": "...", "clientId": "...", "clientSecret": "..."}
type servicePrincipal struct {
	TenantID     string `json:"tenantId"`
	ClientID     string `json:"clientId"`
	ClientSecret string `json:"clientSecret"`
}

// loadServicePrincipal reads and parses Azure AD service principal credentials from a JSON file.
func loadServicePrincipal(path string) (*servicePrincipal, error) {
	b, err := os.ReadFile(filepath.Clean(path))
	if err != nil {
		return nil, fmt.Errorf("read credentials file: %w", err)
	}
	var sp servicePrincipal
	if err := json.Unmarshal(b, &sp); err != nil {
		return nil, fmt.Errorf("parse credentials file: %w", err)
	}
	if sp.TenantID == "" || sp.ClientID == "" || sp.ClientSecret == "" {
		return nil, fmt.Errorf("credentials file must contain tenantId, clientId, and clientSecret")
	}
	return &sp, nil
}

// parseConnectionString extracts AccountName and AccountKey from an Azure Storage connection string.
func parseConnectionString(cs string) (string, string) {
	var accountName, accountKey string
	for _, part := range strings.Split(cs, ";") {
		part = strings.TrimSpace(part)
		if key, val, ok := strings.Cut(part, "="); ok {
			switch key {
			case "AccountName":
				accountName = val
			case "AccountKey":
				accountKey = val
			}
		}
	}
	return accountName, accountKey
}