package duckdb

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"net/url"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
	"github.com/aws/aws-sdk-go-v2/service/sts"
	_ "github.com/duckdb/duckdb-go/v2"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/utils"
	"go.uber.org/zap"
)

// AWSCredentials represents the structure of AWS credentials JSON output
type AWSCredentials struct {
	Version         int    `json:"Version"`
	AccessKeyId     string `json:"AccessKeyId"`
	SecretAccessKey string `json:"SecretAccessKey"`
	SessionToken    string `json:"SessionToken"`
	Expiration      string `json:"Expiration"`
}

const duckDBAssumeRoleSessionName = "nextgen-reporter-duckdb"

// DuckDBClient provides common DuckDB operations for Delta Lake access
type DuckDBClient struct {
	log            *zap.Logger
	db             *sql.DB
	vendor         string
	storageOptions map[string]string
}

func escapeSQLString(value string) string {
	return strings.ReplaceAll(value, "'", "''")
}

// NewDuckDBClient creates a new DuckDB client
func NewDuckDBClient(log *zap.Logger) (*DuckDBClient, error) {
	// Connect to DuckDB (use in-memory database for better performance)
	db, err := sql.Open("duckdb", ":memory:?allow_unsigned_extensions=true")
	if err != nil {
		return nil, fmt.Errorf("failed to connect to DuckDB: %w", err)
	}

	client := &DuckDBClient{
		log: log,
		db:  db,
	}

	// Setup and configure DuckDB
	if err := client.setupDuckDB(); err != nil {
		db.Close()
		return nil, fmt.Errorf("failed to setup DuckDB: %w", err)
	}

	return client, nil
}

// Close closes the DuckDB connection
func (c *DuckDBClient) Close() error {
	if c.db != nil {
		return c.db.Close()
	}
	return nil
}

// QueryMaxTimestamp queries the maximum cost_time_stamp from a Delta Lake table
func (c *DuckDBClient) QueryMaxTimestamp(tablePath string) (int64, error) {
	// nolint:gosec // G201
	query := fmt.Sprintf(`
		SELECT MAX(cost_time_stamp) as max_timestamp
		FROM delta_scan('%s')
	`, tablePath)

	c.log.Debug("executing max timestamp query",
		zap.String("table_path", tablePath),
		zap.String("sql", query))

	var maxTimestamp sql.NullInt64
	err := c.db.QueryRow(query).Scan(&maxTimestamp)

	// Check for credential errors and retry once with refreshed credentials
	if err != nil && c.isCredentialError(err) {
		c.log.Warn("Credential error detected, attempting to refresh credential chain", zap.Error(err))
		if refreshErr := c.refreshCredentialChain(); refreshErr != nil {
			c.log.Error("failed to refresh credential chain", zap.Error(refreshErr))
			return 0, fmt.Errorf("failed to query max timestamp and refresh credentials: %w", err)
		}

		// Retry the query after credential refresh
		c.log.Info("Retrying query after credential refresh")
		err = c.db.QueryRow(query).Scan(&maxTimestamp)
	}

	if err != nil {
		c.log.Error("failed to execute max timestamp query",
			zap.String("table_path", tablePath),
			zap.Error(err))
		return 0, fmt.Errorf("failed to query max timestamp: %w", err)
	}

	if !maxTimestamp.Valid {
		c.log.Info("no data found in table", zap.String("table_path", tablePath))
		return 0, nil
	}

	c.log.Info("successfully retrieved max timestamp",
		zap.String("table_path", tablePath),
		zap.Int64("max_timestamp", maxTimestamp.Int64))

	return maxTimestamp.Int64, nil
}

// QueryMaxTimestampWithPartitionFilter queries the maximum cost_time_stamp from a Delta Lake table with partition filter for better performance
func (c *DuckDBClient) QueryMaxTimestampWithPartitionFilter(tablePath string, startTime int64) (int64, error) {
	// Generate hour dimension for partition pruning - use 30 days before startTime to ensure we capture data
	// even when there's no metering data at the exact startTime
	startHourDimension := utils.GenerateHourDimensionWithOffset(startTime, -30)
	// nolint:gosec // G201
	query := fmt.Sprintf(`
		SELECT MAX(cost_time_stamp) as max_timestamp
		FROM delta_scan('%s')
		WHERE hour_dimension >= '%s'
	`, tablePath, startHourDimension)

	c.log.Debug("executing max timestamp query with partition filter",
		zap.String("table_path", tablePath),
		zap.String("start_hour_dimension", startHourDimension),
		zap.String("sql", query))

	var maxTimestamp sql.NullInt64
	err := c.db.QueryRow(query).Scan(&maxTimestamp)

	// Check for credential errors and retry once with refreshed credentials
	if err != nil && c.isCredentialError(err) {
		c.log.Warn("Credential error detected, attempting to refresh credential chain", zap.Error(err))
		if refreshErr := c.refreshCredentialChain(); refreshErr != nil {
			c.log.Error("failed to refresh credential chain", zap.Error(refreshErr))
			return 0, fmt.Errorf("failed to query max timestamp and refresh credentials: %w", err)
		}

		// Retry the query after credential refresh
		c.log.Info("Retrying query after credential refresh")
		err = c.db.QueryRow(query).Scan(&maxTimestamp)
	}

	if err != nil {
		c.log.Error("failed to execute max timestamp query with partition filter",
			zap.String("table_path", tablePath),
			zap.Error(err))
		return 0, fmt.Errorf("failed to query max timestamp: %w", err)
	}

	if !maxTimestamp.Valid {
		c.log.Info("no data found in table with partition filter",
			zap.String("table_path", tablePath),
			zap.String("start_hour_dimension", startHourDimension))
		return 0, nil
	}

	c.log.Info("successfully retrieved max timestamp with partition filter",
		zap.String("table_path", tablePath),
		zap.String("start_hour_dimension", startHourDimension),
		zap.Int64("max_timestamp", maxTimestamp.Int64))

	return maxTimestamp.Int64, nil
}

// QueryData executes a custom query on a Delta Lake table
func (c *DuckDBClient) QueryData(tablePath, customSQL string) (*sql.Rows, error) {
	query := strings.ReplaceAll(customSQL, "{table_path}", fmt.Sprintf("delta_scan('%s')", tablePath))

	c.log.Debug("executing custom query",
		zap.String("table_path", tablePath),
		zap.String("sql", query))

	rows, err := c.db.Query(query)

	// Check for credential errors and retry once with refreshed credentials
	if err != nil && c.isCredentialError(err) {
		c.log.Warn("Credential error detected, attempting to refresh credential chain", zap.Error(err))
		if refreshErr := c.refreshCredentialChain(); refreshErr != nil {
			c.log.Error("failed to refresh credential chain", zap.Error(refreshErr))
			return nil, fmt.Errorf("failed to execute query and refresh credentials: %w", err)
		}

		// Retry the query after credential refresh
		c.log.Info("Retrying query after credential refresh")
		rows, err = c.db.Query(query)
	}

	if err != nil {
		c.log.Error("failed to execute custom query",
			zap.String("table_path", tablePath),
			zap.String("sql", query),
			zap.Error(err))
		return nil, fmt.Errorf("failed to execute query: %w", err)
	}

	return rows, nil
}

// ConfigureForStorageOptions configures DuckDB for specific cloud provider
func (c *DuckDBClient) ConfigureForStorageOptions(vendor string, storageOptions map[string]string) error {
	if storageOptions == nil {
		c.log.Info("No storage options provided, skipping cloud access configuration")
		return nil
	}

	// Store configuration for potential refreshes
	c.vendor = vendor
	c.storageOptions = storageOptions

	// Get cloud service provider
	switch vendor {
	case constants.VendorAWS:
		return c.configureAWSAccess(storageOptions)
	case constants.VendorAlicloud:
		return c.configureAlicloudAccess(storageOptions)
	case constants.VendorGCP:
		return c.configureGCPAccess(storageOptions)
	case constants.VendorAzure:
		return c.configureAzureAccess(storageOptions)
	default:
		c.log.Warn("Unsupported cloud vendor", zap.String("vendor", vendor))
		return nil
	}
}

// setupDuckDB configures DuckDB settings and extensions
func (c *DuckDBClient) setupDuckDB() error {
	// 1. Set memory limits
	if err := c.configureDuckDBMemory(); err != nil {
		return fmt.Errorf("failed to configure DuckDB memory: %w", err)
	}

	// 2. Install and load necessary extensions
	if err := c.installExtensions(); err != nil {
		return fmt.Errorf("failed to install extensions: %w", err)
	}

	// 3. Configure caching settings
	if err := c.configureCaching(); err != nil {
		return fmt.Errorf("failed to configure caching: %w", err)
	}

	return nil
}

// configureDuckDBMemory configures DuckDB memory limits
func (c *DuckDBClient) configureDuckDBMemory() error {
	podMemLimitMBStr := os.Getenv("POD_MEMORY_LIMITS_MB")
	podMemLimitMB, parseErr := strconv.Atoi(podMemLimitMBStr)
	if parseErr != nil || podMemLimitMB <= 0 {
		c.log.Warn("Failed to parse POD_MEMORY_LIMITS_MB env var",
			zap.Error(parseErr),
			zap.String("POD_MEMORY_LIMITS_MB", podMemLimitMBStr))
		podMemLimitMB = 2048
	}

	duckdbMemoryPercentStr := os.Getenv("DUCKDB_MEMORY_PERCENTAGE")
	duckdbMemoryPercent, parseErr := strconv.Atoi(duckdbMemoryPercentStr)
	if parseErr != nil || duckdbMemoryPercent <= 0 || duckdbMemoryPercent > 100 {
		c.log.Warn("Failed to parse DUCKDB_MEMORY_PERCENTAGE env var or invalid value, using default 80%",
			zap.Error(parseErr),
			zap.String("DUCKDB_MEMORY_PERCENTAGE", duckdbMemoryPercentStr))
		duckdbMemoryPercent = 80
	}

	duckdbMemoryLimitMB := int(float64(podMemLimitMB) * float64(duckdbMemoryPercent) / 100.0)

	if duckdbMemoryLimitMB < 2048 {
		duckdbMemoryLimitMB = 2048
		c.log.Info("Calculated DuckDB memory limit too low, adjusting to minimum",
			zap.Int("calculatedMB", duckdbMemoryLimitMB))
	}

	duckdbMemoryLimit := fmt.Sprintf("%dMB", duckdbMemoryLimitMB)

	_, err := c.db.Exec(fmt.Sprintf("SET memory_limit='%s';", duckdbMemoryLimit))
	if err != nil {
		return fmt.Errorf("failed to set DuckDB memory_limit to '%s': %w", duckdbMemoryLimit, err)
	}

	c.log.Info("Set DuckDB memory limit",
		zap.String("limit", duckdbMemoryLimit),
		zap.Int("pod_memory_limit_mb", podMemLimitMB),
		zap.Int("duckdbMemoryPercent", duckdbMemoryPercent))

	return nil
}

// configureCABundle sets the CA certificate file for DuckDB's internal HTTP client.
// DuckDB uses a bundled libcurl that may not find the system CA bundle automatically,
// especially in container environments.
func (c *DuckDBClient) configureCABundle() error {
	// Allow explicit skip via environment variable
	if os.Getenv("DUCKDB_SKIP_CA_BUNDLE") == "true" {
		c.log.Debug("Skipping CA bundle configuration (DUCKDB_SKIP_CA_BUNDLE=true)")
		return nil
	}

	// Allow explicit override via environment variable
	caFile := os.Getenv("DUCKDB_CA_CERT_FILE")
	if caFile == "" {
		// Common CA bundle locations across Linux distributions
		candidates := []string{
			"/etc/ssl/certs/ca-certificates.crt", // Debian/Ubuntu
			"/etc/pki/tls/certs/ca-bundle.crt",   // RHEL/CentOS
			"/etc/ssl/ca-bundle.pem",             // openSUSE
		}
		for _, path := range candidates {
			if _, err := os.Stat(path); err == nil {
				caFile = path
				break
			}
		}
	} else {
		// Verify the explicitly configured path exists
		if _, err := os.Stat(caFile); err != nil {
			c.log.Warn("DUCKDB_CA_CERT_FILE path does not exist, skipping CA bundle configuration",
				zap.String("path", caFile), zap.Error(err))
			caFile = ""
		}
	}

	if caFile == "" {
		c.log.Warn("No system CA bundle found, DuckDB HTTPS requests may fail")
		return nil
	}

	_, err := c.db.Exec(fmt.Sprintf("SET ca_cert_file='%s';", escapeSQLString(caFile)))
	if err != nil {
		return fmt.Errorf("failed to set ca_cert_file to '%s': %w", caFile, err)
	}

	// Also set the Azure extension's transport CA info (it uses its own libcurl instance)
	_, err = c.db.Exec(fmt.Sprintf("SET azure_transport_option_ca_info='%s';", escapeSQLString(caFile)))
	if err != nil {
		c.log.Warn("failed to set azure_transport_option_ca_info", zap.Error(err))
	}

	_, err = c.db.Exec("SET azure_transport_option_type='curl';")
	if err != nil {
		c.log.Warn("failed to set azure_transport_option_type", zap.Error(err))
	}

	c.log.Info("Set DuckDB CA certificate file", zap.String("ca_cert_file", caFile))
	return nil
}

// installExtensions installs required DuckDB extensions
func (c *DuckDBClient) installExtensions() error {
	// Install and load httpfs extension (for S3 and GCS access)
	_, err := c.db.Exec("INSTALL httpfs; LOAD httpfs;")
	if err != nil {
		return fmt.Errorf("failed to install or load httpfs: %w", err)
	}

	// Install and load azure extension (for Azure Blob Storage)
	_, err = c.db.Exec("INSTALL azure; LOAD azure;")
	if err != nil {
		return fmt.Errorf("failed to install or load azure extension: %w", err)
	}

	// Install and configure cache extension
	cacheDir := os.Getenv("DUCKDB_CACHE_DIR")
	if cacheDir == "" {
		cacheDir = "/tmp/duckdb_cache"
	}

	_, err = c.db.Exec(fmt.Sprintf("INSTALL cache_httpfs FROM community; LOAD cache_httpfs; SET cache_httpfs_cache_directory='%s'; SET cache_httpfs_type='on_disk';", cacheDir))
	if err != nil {
		c.log.Warn("failed to install or load cache_httpfs extension", zap.Error(err))
		// Cache extension failure should not block the entire process
	}

	c.log.Info("Successfully installed DuckDB extensions")
	return nil
}

// configureCaching configures DuckDB caching settings
func (c *DuckDBClient) configureCaching() error {
	_, err := c.db.Exec("SET enable_object_cache=true; SET enable_http_metadata_cache=true; SET enable_external_file_cache=true; SET parquet_metadata_cache=true;")
	if err != nil {
		return fmt.Errorf("failed to set cache configuration: %w", err)
	}

	c.log.Info("Configured DuckDB caching settings")
	return nil
}

func (c *DuckDBClient) installAdditionalExtensions(storageOptions map[string]string) error {
	// Install and load delta extension based on configuration

	deltaExtensionPath := strings.TrimSpace(storageOptions[constants.StorageOptionPatchDeltaSo])

	if deltaExtensionPath != "" {
		if _, err := c.db.Exec(fmt.Sprintf("LOAD '%s';", escapeSQLString(deltaExtensionPath))); err != nil {
			return fmt.Errorf("failed to load delta extension from %q: %w", deltaExtensionPath, err)
		}
		c.log.Info("Loaded delta extension from provided path",
			zap.String("path", deltaExtensionPath))
	} else {
		if _, err := c.db.Exec("INSTALL delta; LOAD delta;"); err != nil {
			return fmt.Errorf("failed to install or load delta: %w", err)
		}
		c.log.Info("Installed delta extension using default loader")
	}
	return nil
}

// configureAWSAccess configures AWS access
func (c *DuckDBClient) configureAWSAccess(storageOptions map[string]string) error {
	region, ok := storageOptions["region"]
	if !ok {
		region = "us-east-1" // Default region
	}

	_, err := c.db.Exec(fmt.Sprintf("SET s3_region='%s';", region))
	if err != nil {
		return fmt.Errorf("failed to set s3_region: %w", err)
	}

	roleArn := strings.TrimSpace(storageOptions["AWS_IAM_ROLE_ARN"])

	// Check if AWS_PROFILE is specified in storage options
	awsProfile, hasProfile := storageOptions["AWS_PROFILE"]

	if hasProfile && awsProfile != "" {
		// Use AWS CLI approach for specific profile
		if err := c.exportAndSetAWSCredentials(awsProfile); err != nil {
			return fmt.Errorf("failed to export AWS credentials for profile %s: %w", awsProfile, err)
		}

		// Configure with environment credentials
		awsSecret := fmt.Sprintf(`
			CREATE OR REPLACE SECRET aws_secret (
				TYPE S3,
				PROVIDER credential_chain,
				CHAIN 'env',
				REGION '%s'
			);
		`, region)

		_, err = c.db.Exec(awsSecret)
		if err != nil {
			return fmt.Errorf("failed to set AWS secret: %w", err)
		}

		c.log.Info("Configured AWS access with exported profile credentials",
			zap.String("region", region),
			zap.String("profile", awsProfile))
	} else {
		// Use AWS SDK to get credentials from default credential chain
		// This handles EKS Service Account and optional AssumeRole correctly
		credentials, err := c.getAWSCredentialsFromSDK(region, roleArn)
		if err != nil {
			return fmt.Errorf("failed to get AWS credentials from SDK: %w", err)
		}

		// Create DuckDB secret with explicit credentials instead of credential_chain
		var awsSecret string
		if credentials.SessionToken != "" {
			// With session token (for IAM roles, STS tokens, etc.)
			awsSecret = fmt.Sprintf(`
				CREATE OR REPLACE SECRET aws_secret (
					TYPE S3,
					KEY_ID '%s',
					SECRET '%s',
					SESSION_TOKEN '%s',
					REGION '%s'
				);
			`, credentials.AccessKeyID, credentials.SecretAccessKey, credentials.SessionToken, region)
		} else {
			// Without session token (for IAM users)
			awsSecret = fmt.Sprintf(`
				CREATE OR REPLACE SECRET aws_secret (
					TYPE S3,
					KEY_ID '%s',
					SECRET '%s',
					REGION '%s'
				);
			`, credentials.AccessKeyID, credentials.SecretAccessKey, region)
		}

		_, err = c.db.Exec(awsSecret)
		if err != nil {
			return fmt.Errorf("failed to set AWS secret: %w", err)
		}

		maskedAccessKey := credentials.AccessKeyID
		if len(maskedAccessKey) > 8 {
			maskedAccessKey = maskedAccessKey[:8] + "..."
		}
		c.log.Info("Configured AWS access with SDK credentials",
			zap.String("region", region),
			zap.String("assume_role_arn", roleArn),
			zap.String("access_key_id", maskedAccessKey),
			zap.Bool("has_session_token", credentials.SessionToken != ""))
	}

	return nil
}

func (c *DuckDBClient) configureAlicloudAccess(storageOptions map[string]string) error {
	accessKey := strings.TrimSpace(os.Getenv(constants.AlicloudAccessKeyIDEnv))
	secretKey := strings.TrimSpace(os.Getenv(constants.AlicloudAccessKeySecretEnv))
	if accessKey == "" || secretKey == "" {
		return fmt.Errorf("%s and %s environment variables are required for Alicloud configuration",
			constants.AlicloudAccessKeyIDEnv, constants.AlicloudAccessKeySecretEnv)
	}

	endpoint := strings.TrimSpace(storageOptions["BucketEndpoint"])
	if endpoint == "" {
		endpoint = strings.TrimSpace(storageOptions["AWS_ENDPOINT_URL"])
	}
	if endpoint == "" {
		return fmt.Errorf("BucketEndpoint is required for Alicloud configuration")
	}

	region := strings.TrimSpace(storageOptions["region"])
	if region == "" {
		region = strings.TrimSpace(storageOptions["AWS_REGION"])
	}
	if region == "" {
		return fmt.Errorf("region or AWS_REGION is required for Alicloud configuration")
	}

	bucket := strings.TrimSpace(storageOptions["bucket"])
	if bucket == "" {
		return fmt.Errorf("bucket is required in storage options for Alicloud configuration")
	}

	parsedURL, err := url.Parse(endpoint)
	if err != nil {
		return fmt.Errorf("invalid BucketEndpoint %q: %w", endpoint, err)
	}

	if parsedURL.Host == "" {
		return fmt.Errorf("BucketEndpoint %q missing host", endpoint)
	}

	host := parsedURL.Host
	if !strings.HasPrefix(host, bucket+".") {
		host = bucket + "." + host
	}

	parsedEndpoint := host

	secretTemplate := `
		CREATE OR REPLACE SECRET alicloud_secret (
			TYPE S3,
			URL_STYLE 'vhost',
			KEY_ID '%s',
			SECRET '%s',
			REGION '%s',
			ENDPOINT '%s'
		);
	`

	if _, err := c.db.Exec(fmt.Sprintf(secretTemplate,
		escapeSQLString(accessKey),
		escapeSQLString(secretKey),
		escapeSQLString(region),
		escapeSQLString(parsedEndpoint),
	)); err != nil {
		return fmt.Errorf("failed to set Alicloud secret: %w", err)
	}

	return nil
}

// getAWSCredentialsFromSDK gets AWS credentials using AWS SDK v2 default credential chain
func (c *DuckDBClient) getAWSCredentialsFromSDK(region, roleArn string) (*aws.Credentials, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Load AWS config using default credential chain
	cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
	if err != nil {
		c.log.Error("failed to load AWS config", zap.Error(err))
		return nil, fmt.Errorf("failed to load AWS config: %w", err)
	}

	if roleArn = strings.TrimSpace(roleArn); roleArn != "" {
		provider := stscreds.NewAssumeRoleProvider(sts.NewFromConfig(cfg), roleArn, func(o *stscreds.AssumeRoleOptions) {
			o.RoleSessionName = duckDBAssumeRoleSessionName
		})
		credCache := aws.NewCredentialsCache(provider)
		credentials, err := credCache.Retrieve(ctx)
		if err != nil {
			c.log.Error("failed to assume AWS role for DuckDB", zap.String("role_arn", roleArn), zap.Error(err))
			return nil, fmt.Errorf("failed to retrieve assumed credentials: %w", err)
		}

		maskedAccessKey := credentials.AccessKeyID
		if len(maskedAccessKey) > 8 {
			maskedAccessKey = maskedAccessKey[:8] + "..."
		}
		c.log.Info("Successfully assumed AWS role for DuckDB",
			zap.String("role_arn", roleArn),
			zap.String("access_key_id", maskedAccessKey),
			zap.Bool("has_session_token", credentials.SessionToken != ""),
			zap.Time("expires", credentials.Expires))

		return &credentials, nil
	}

	// Retrieve credentials from the credential chain
	credentials, err := cfg.Credentials.Retrieve(ctx)
	if err != nil {
		c.log.Error("failed to retrieve AWS credentials from credential chain", zap.Error(err))
		return nil, fmt.Errorf("failed to retrieve AWS credentials: %w", err)
	}

	maskedAccessKey := credentials.AccessKeyID
	if len(maskedAccessKey) > 8 {
		maskedAccessKey = maskedAccessKey[:8] + "..."
	}
	c.log.Info("Successfully retrieved AWS credentials from SDK",
		zap.String("access_key_id", maskedAccessKey),
		zap.Bool("has_session_token", credentials.SessionToken != ""),
		zap.Time("expires", credentials.Expires))

	return &credentials, nil
}

// exportAndSetAWSCredentials exports AWS credentials using AWS CLI and sets environment variables
func (c *DuckDBClient) exportAndSetAWSCredentials(profile string) error {
	// Execute AWS CLI command to export credentials
	cmd := exec.Command("aws", "configure", "export-credentials")
	cmd.Env = append(os.Environ(), fmt.Sprintf("AWS_PROFILE=%s", profile))

	output, err := cmd.Output()
	if err != nil {
		c.log.Error("failed to execute aws configure export-credentials",
			zap.String("profile", profile),
			zap.Error(err))
		return fmt.Errorf("failed to execute aws configure export-credentials: %w", err)
	}

	// Parse JSON output
	var credentials AWSCredentials
	if err := json.Unmarshal(output, &credentials); err != nil {
		c.log.Error("failed to parse AWS credentials JSON",
			zap.String("output", string(output)),
			zap.Error(err))
		return fmt.Errorf("failed to parse AWS credentials JSON: %w", err)
	}

	// Set environment variables for DuckDB to use
	os.Setenv("AWS_ACCESS_KEY_ID", credentials.AccessKeyId)
	os.Setenv("AWS_SECRET_ACCESS_KEY", credentials.SecretAccessKey)
	if credentials.SessionToken != "" {
		os.Setenv("AWS_SESSION_TOKEN", credentials.SessionToken)
	}

	c.log.Info("Successfully exported and set AWS credentials",
		zap.String("profile", profile),
		zap.String("access_key_id", credentials.AccessKeyId[:8]+"..."), // Only log first 8 chars for security
		zap.String("expiration", credentials.Expiration))

	return nil
}

// isCredentialError checks if the error indicates a credential issue
func (c *DuckDBClient) isCredentialError(err error) bool {
	if err == nil {
		return false
	}

	errMsg := err.Error()
	return strings.Contains(errMsg, "The provided token has expired.") ||
		strings.Contains(errMsg, "AuthenticationFailed") ||
		strings.Contains(errMsg, "Credential should be updated") ||
		strings.Contains(errMsg, "InvalidAccessKeyId") ||
		strings.Contains(errMsg, "SignatureDoesNotMatch") ||
		strings.Contains(errMsg, "TokenRefreshRequired") ||
		strings.Contains(errMsg, "InvalidToken") ||
		strings.Contains(errMsg, "ExpiredToken") ||
		strings.Contains(errMsg, "Request has expired") ||
		strings.Contains(errMsg, "Unable to locate credentials") ||
		strings.Contains(errMsg, "NoCredentialProviders") ||
		strings.Contains(errMsg, "HTTP 403") ||
		strings.Contains(errMsg, "Forbidden") ||
		strings.Contains(errMsg, "Access Denied")
}

// refreshCredentialChain refreshes the credential chain for the configured cloud provider
func (c *DuckDBClient) refreshCredentialChain() error {
	if c.vendor == "" || c.storageOptions == nil {
		return fmt.Errorf("no vendor or storage options configured for credential refresh")
	}

	c.log.Info("Refreshing credential chain", zap.String("vendor", c.vendor))

	switch c.vendor {
	case constants.VendorAWS:
		return c.configureAWSAccess(c.storageOptions)
	case constants.VendorAlicloud:
		return c.configureAlicloudAccess(c.storageOptions)
	case constants.VendorGCP:
		return c.configureGCPAccess(c.storageOptions)
	case constants.VendorAzure:
		return c.configureAzureAccess(c.storageOptions)
	default:
		return fmt.Errorf("unsupported vendor for credential refresh: %s", c.vendor)
	}
}

// configureGCPAccess configures GCP access
func (c *DuckDBClient) configureGCPAccess(storageOptions map[string]string) error {
	gcsSecret := `
		CREATE OR REPLACE SECRET gcs_secret (
			TYPE GCS,
			PROVIDER credential_chain
		);
	`

	_, err := c.db.Exec(gcsSecret)
	if err != nil {
		return fmt.Errorf("failed to set GCP secret: %w", err)
	}

	c.log.Info("Configured GCP access")
	return nil
}

// configureAzureAccess configures Azure access by acquiring an OAuth2 access token
// via DefaultAzureCredential (managed identity, environment variables, Azure CLI, etc.)
// and passing it to DuckDB's access_token provider.
func (c *DuckDBClient) configureAzureAccess(storageOptions map[string]string) error {
	accountName := strings.TrimSpace(storageOptions[constants.AzureStorageAccountNameEnv])
	if accountName == "" {
		accountName = strings.TrimSpace(os.Getenv(constants.AzureStorageAccountNameEnv))
	}
	if accountName == "" {
		return fmt.Errorf("%s is required for Azure configuration", constants.AzureStorageAccountNameEnv)
	}

	// Acquire an OAuth2 access token for Azure Storage using DefaultAzureCredential
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	cred, err := azidentity.NewDefaultAzureCredential(nil)
	if err != nil {
		return fmt.Errorf("failed to create default Azure credential: %w", err)
	}

	token, err := cred.GetToken(ctx, policy.TokenRequestOptions{
		Scopes: []string{"https://storage.azure.com/.default"},
	})
	if err != nil {
		return fmt.Errorf("failed to acquire Azure access token: %w", err)
	}

	azureSecret := fmt.Sprintf(`
		CREATE OR REPLACE SECRET azure_secret (
			TYPE AZURE,
			PROVIDER access_token,
			ACCOUNT_NAME '%s',
			ACCESS_TOKEN '%s'
		);
	`, escapeSQLString(accountName), escapeSQLString(token.Token))
	_, err = c.db.Exec(azureSecret)
	if err != nil {
		return fmt.Errorf("failed to set Azure secret: %w", err)
	}

	// Configure CA certificate path for DuckDB's internal HTTP client
	if err := c.configureCABundle(); err != nil {
		return fmt.Errorf("failed to configure CA bundle: %w", err)
	}

	c.log.Info("Configured Azure access with access_token",
		zap.String("account_name", accountName),
		zap.Time("token_expires", token.ExpiresOn))
	return nil
}

// NewDuckDBClientWithConfig creates a new DuckDB client with service config
func NewDuckDBClientWithConfig(conf *config.ServiceConfig, log *zap.Logger) (*DuckDBClient, error) {
	client, err := NewDuckDBClient(log)
	if err != nil {
		return nil, err
	}

	// Configure for specific storage if NextGen config exists
	if conf.NextGenMeteringConfig != nil && conf.NextGenMeteringConfig.StorageOptions != nil {
		conf.NextGenMeteringConfig.StorageOptions["BucketEndpoint"] = conf.NextGenMeteringConfig.BucketEndpoint
		if conf.Bucket != "" {
			conf.NextGenMeteringConfig.StorageOptions["bucket"] = conf.Bucket
		}

		// For Azure, extract container name from delta table path if not explicitly set
		if conf.Vendor == constants.VendorAzure {
			if _, ok := conf.NextGenMeteringConfig.StorageOptions[constants.AzureContainerNameKey]; !ok {
				containerName := extractContainerFromAzurePath(conf.NextGenMeteringConfig.DeltaTablePath)
				if containerName != "" {
					conf.NextGenMeteringConfig.StorageOptions[constants.AzureContainerNameKey] = containerName
				}
			}
		}

		if err := client.installAdditionalExtensions(conf.NextGenMeteringConfig.StorageOptions); err != nil {
			client.Close()
			return nil, fmt.Errorf("failed to install additional extensions: %w", err)
		}
		if err := client.ConfigureForStorageOptions(conf.Vendor, conf.NextGenMeteringConfig.StorageOptions); err != nil {
			client.Close()
			return nil, fmt.Errorf("failed to configure storage options: %w", err)
		}
	}

	return client, nil
}

// extractContainerFromAzurePath extracts the container name from an Azure blob storage path.
// Expected format: az://account.blob.core.windows.net/container/path/...
// or: https://account.blob.core.windows.net/container/path/...
func extractContainerFromAzurePath(azurePath string) string {
	if azurePath == "" {
		return ""
	}

	parsed, err := url.Parse(azurePath)
	if err != nil {
		return ""
	}

	path := strings.TrimPrefix(parsed.Path, "/")
	if path == "" {
		return ""
	}

	parts := strings.SplitN(path, "/", 2)
	if len(parts) > 0 && parts[0] != "" {
		return parts[0]
	}

	return ""
}