package s3

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
)

var (
	s3ClientInstance *S3Client
	once             sync.Once
)

// S3Client wraps the AWS S3 client
type S3Client struct {
	client *s3.S3
	bucket string
}

// GetS3Client returns a singleton S3 client instance.
func GetS3Client(cfg *config.Config) (*S3Client, error) {
	var err error
	once.Do(func() {
		if cfg.S3.AccessKeyID != "" && cfg.S3.SecretAccessKey != "" {
			s3ClientInstance, err = newS3ClientWithCredentials(
				cfg.S3.Region,
				cfg.S3.Bucket,
				cfg.S3.AccessKeyID,
				cfg.S3.SecretAccessKey,
				cfg.S3.SessionToken,
			)
		} else {
			s3ClientInstance, err = newS3Client(cfg.S3.Region, cfg.S3.Bucket)
		}
	})
	if err != nil {
		return nil, err
	}
	return s3ClientInstance, nil
}

// newS3Client creates a new S3 client
func newS3Client(region, bucket string) (*S3Client, error) {
	sess, err := session.NewSession(&aws.Config{
		Region: aws.String(region),
	})
	if err != nil {
		return nil, fmt.Errorf("unable to create AWS session: %w", err)
	}

	client := s3.New(sess)
	return &S3Client{
		client: client,
		bucket: bucket,
	}, nil
}

// newS3ClientWithCredentials creates a new S3 client with explicit credentials
func newS3ClientWithCredentials(region, bucket, accessKeyID, secretAccessKey, sessionToken string) (*S3Client, error) {
	creds := credentials.NewStaticCredentials(accessKeyID, secretAccessKey, sessionToken)
	sess, err := session.NewSession(&aws.Config{
		Region:      aws.String(region),
		Credentials: creds,
	})
	if err != nil {
		return nil, fmt.Errorf("unable to create AWS session: %w", err)
	}

	client := s3.New(sess)
	return &S3Client{
		client: client,
		bucket: bucket,
	}, nil
}

// ReadFile reads a file from S3
func (c *S3Client) ReadFile(key string) ([]byte, error) {
	result, err := c.client.GetObject(&s3.GetObjectInput{
		Bucket: aws.String(c.bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		return nil, fmt.Errorf("failed to get object %s: %w", key, err)
	}
	defer result.Body.Close()

	body, err := io.ReadAll(result.Body)
	if err != nil {
		return nil, fmt.Errorf("failed to read object body: %w", err)
	}

	return body, nil
}

// ReadFileStream returns a reader for streaming file content from S3
func (c *S3Client) ReadFileStream(key string) (io.ReadCloser, error) {
	result, err := c.client.GetObject(&s3.GetObjectInput{
		Bucket: aws.String(c.bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		return nil, fmt.Errorf("failed to get object %s: %w", key, err)
	}

	return result.Body, nil
}

// ProcessFileStream processes S3 file content line by line using a callback function
func (c *S3Client) ProcessFileStream(key string, lineProcessor func(line string) error) error {
	reader, err := c.ReadFileStream(key)
	if err != nil {
		return err
	}
	defer reader.Close()

	scanner := bufio.NewScanner(reader)
	// Set a larger buffer to handle long lines
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, 1024*1024) // 1MB max token size

	lineNum := 0
	for scanner.Scan() {
		lineNum++
		line := scanner.Text()
		if err := lineProcessor(line); err != nil {
			return fmt.Errorf("error processing line %d: %w", lineNum, err)
		}
	}

	if err := scanner.Err(); err != nil {
		return fmt.Errorf("error reading file stream: %w", err)
	}

	return nil
}

// ListFiles lists files in S3 with the given prefix
func (c *S3Client) ListFiles(prefix string) ([]string, error) {
	result, err := c.client.ListObjectsV2(&s3.ListObjectsV2Input{
		Bucket: aws.String(c.bucket),
		Prefix: aws.String(prefix),
	})
	if err != nil {
		// log.PlainError("Failed to list objects", zap.String("prefix", prefix), zap.Error(err))
		return nil, err
	}

	var files []string
	for _, obj := range result.Contents {
		files = append(files, *obj.Key)
	}

	return files, nil
}

// DownloadFile downloads a file from S3 to local disk
func (c *S3Client) DownloadFile(ctx context.Context, s3Key string, localPath string) error {
	// Create local directory (if not exists)
	dir := filepath.Dir(localPath)
	if err := os.MkdirAll(dir, 0755); err != nil {
		return fmt.Errorf("failed to create directory %s: %w", dir, err)
	}

	// Get object from S3
	resp, err := c.client.GetObject(&s3.GetObjectInput{
		Bucket: aws.String(c.bucket),
		Key:    aws.String(s3Key),
	})
	if err != nil {
		return fmt.Errorf("failed to get object from S3: %w", err)
	}
	defer resp.Body.Close()

	// Create local file
	file, err := os.Create(localPath)
	if err != nil {
		return fmt.Errorf("failed to create local file %s: %w", localPath, err)
	}
	defer file.Close()

	// Copy data
	_, err = io.Copy(file, resp.Body)
	if err != nil {
		return fmt.Errorf("failed to copy data to local file: %w", err)
	}

	return nil
}

// FileExists checks if a file exists in S3
func (c *S3Client) FileExists(key string) (bool, error) {
	_, err := c.client.HeadObject(&s3.HeadObjectInput{
		Bucket: aws.String(c.bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		// Check if the error is "not found"
		if strings.Contains(err.Error(), "NotFound") {
			return false, nil
		}
		return false, fmt.Errorf("failed to check object existence: %w", err)
	}

	return true, nil
}