package sqs

import (
	"context"
	"errors"
	"net/url"
	"strconv"
	"strings"
	"time"

	awsv2 "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"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"

	appconfig "wucher/internal/config"
	"wucher/internal/shared/pkg/metrics"
	"wucher/internal/queue"
)

type API interface {
	SendMessage(ctx context.Context, params *sqs.SendMessageInput, optFns ...func(*sqs.Options)) (*sqs.SendMessageOutput, error)
	ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error)
	DeleteMessageBatch(ctx context.Context, params *sqs.DeleteMessageBatchInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageBatchOutput, error)
	ChangeMessageVisibility(ctx context.Context, params *sqs.ChangeMessageVisibilityInput, optFns ...func(*sqs.Options)) (*sqs.ChangeMessageVisibilityOutput, error)
	GetQueueAttributes(ctx context.Context, params *sqs.GetQueueAttributesInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueAttributesOutput, error)
}

type EmailQueue struct {
	api               API
	queueURL          string
	queueName         string
	component         string
	queueType         string
	messageGroupID    string
	maxMessages       int
	waitTimeSeconds   int32
	visibilityTimeout int32
}

type QueueOption func(*EmailQueue)

func WithMetricsComponent(component string) QueueOption {
	return func(q *EmailQueue) {
		if q == nil {
			return
		}
		component = strings.TrimSpace(component)
		if component == "" {
			return
		}
		q.component = component
	}
}

func NewClient(ctx context.Context, cfg appconfig.SQSConfig) (*sqs.Client, error) {
	if err := validateConfig(cfg); err != nil {
		return nil, err
	}
	loadOptions := make([]func(*awsconfig.LoadOptions) error, 0, 1)
	if strings.TrimSpace(cfg.Region) != "" {
		loadOptions = append(loadOptions, awsconfig.WithRegion(strings.TrimSpace(cfg.Region)))
	}
	if key := strings.TrimSpace(cfg.AccessKeyID); key != "" {
		secret := strings.TrimSpace(cfg.SecretAccessKey)
		loadOptions = append(loadOptions, awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(key, secret, "")))
	}
	awsCfg, err := awsconfig.LoadDefaultConfig(ctx, loadOptions...)
	if err != nil {
		return nil, err
	}
	client := sqs.NewFromConfig(awsCfg, func(options *sqs.Options) {
		if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" {
			options.EndpointResolver = sqs.EndpointResolverFromURL(endpoint)
		}
	})
	return client, nil
}

func NewEmailQueue(api API, cfg appconfig.SQSConfig, opts ...QueueOption) (*EmailQueue, error) {
	return NewQueue(api, cfg, opts...)
}

func NewQueue(api API, cfg appconfig.SQSConfig, opts ...QueueOption) (*EmailQueue, error) {
	if err := validateConfig(cfg); err != nil {
		return nil, err
	}
	if api == nil {
		return nil, errors.New("sqs client is required")
	}
	waitTimeSeconds := cfg.WaitTimeSeconds
	if waitTimeSeconds <= 0 {
		waitTimeSeconds = 20
	}
	if waitTimeSeconds > 20 {
		waitTimeSeconds = 20
	}
	maxMessages := cfg.MaxMessages
	if maxMessages <= 0 {
		maxMessages = 10
	}
	if maxMessages > 10 {
		maxMessages = 10
	}
	visibilityTimeout := int32(cfg.VisibilityTimeout.Seconds())
	if visibilityTimeout <= 0 {
		visibilityTimeout = 60
	}
	if visibilityTimeout > 43200 {
		visibilityTimeout = 43200
	}
	queueType := strings.ToLower(strings.TrimSpace(cfg.QueueType))
	if queueType == "" {
		queueType = queue.QueueTypeStandard
	}
	queue := &EmailQueue{
		api:               api,
		queueURL:          strings.TrimSpace(cfg.QueueURL),
		queueName:         metrics.QueueNameFromURL(cfg.QueueURL),
		component:         "worker",
		queueType:         queueType,
		messageGroupID:    strings.TrimSpace(cfg.MessageGroupID),
		maxMessages:       maxMessages,
		waitTimeSeconds:   waitTimeSeconds,
		visibilityTimeout: visibilityTimeout,
	}
	for _, opt := range opts {
		if opt != nil {
			opt(queue)
		}
	}
	metrics.InitializeQueueSeries(queue.component, queue.queueName)
	return queue, nil
}

func (q *EmailQueue) Backend() string {
	return queue.BackendSQS
}

func (q *EmailQueue) Publish(ctx context.Context, message queue.OutboundMessage) error {
	input := &sqs.SendMessageInput{
		QueueUrl:    awsv2.String(q.queueURL),
		MessageBody: awsv2.String(string(message.Body)),
	}
	if len(message.Attributes) > 0 {
		input.MessageAttributes = make(map[string]sqstypes.MessageAttributeValue, len(message.Attributes))
		for key, value := range message.Attributes {
			v := value
			input.MessageAttributes[key] = sqstypes.MessageAttributeValue{
				DataType:    awsv2.String("String"),
				StringValue: awsv2.String(v),
			}
		}
	}
	if q.queueType == queue.QueueTypeFIFO {
		groupID := strings.TrimSpace(message.MessageGroupID)
		if groupID == "" {
			groupID = q.messageGroupID
		}
		if groupID == "" {
			groupID = "default"
		}
		deduplicationID := strings.TrimSpace(message.DeduplicationID)
		if deduplicationID == "" {
			deduplicationID = message.ID
		}
		input.MessageGroupId = awsv2.String(groupID)
		input.MessageDeduplicationId = awsv2.String(deduplicationID)
	}
	started := time.Now()
	_, err := q.api.SendMessage(ctx, input)
	metrics.ObserveQueueSend(q.component, q.queueName, int64(len(message.Body)), err, started)
	return err
}

func (q *EmailQueue) Receive(ctx context.Context, maxMessages int) ([]*queue.Delivery, error) {
	if maxMessages <= 0 {
		maxMessages = 1
	}
	if q.maxMessages > 0 && maxMessages > q.maxMessages {
		maxMessages = q.maxMessages
	}
	if maxMessages > 10 {
		maxMessages = 10
	}
	started := time.Now()
	output, err := q.api.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
		QueueUrl:            awsv2.String(q.queueURL),
		MaxNumberOfMessages: int32(maxMessages),
		WaitTimeSeconds:     q.waitTimeSeconds,
		VisibilityTimeout:   q.visibilityTimeout,
		MessageSystemAttributeNames: []sqstypes.MessageSystemAttributeName{
			sqstypes.MessageSystemAttributeNameApproximateReceiveCount,
			sqstypes.MessageSystemAttributeNameSentTimestamp,
		},
		MessageAttributeNames: []string{"All"},
	})
	if err != nil {
		metrics.ObserveQueueReceive(q.component, q.queueName, 0, 0, err, started)
		return nil, err
	}
	var totalMessageBytes int64
	for i := range output.Messages {
		totalMessageBytes += int64(len(awsv2.ToString(output.Messages[i].Body)))
	}
	metrics.ObserveQueueReceive(q.component, q.queueName, len(output.Messages), totalMessageBytes, nil, started)

	deliveries := make([]*queue.Delivery, 0, len(output.Messages))
	for _, message := range output.Messages {
		attributes := make(map[string]string, len(message.MessageAttributes))
		for key, value := range message.MessageAttributes {
			if value.StringValue != nil {
				attributes[key] = *value.StringValue
			}
		}
		receiveCount, _ := strconv.Atoi(message.Attributes[string(sqstypes.MessageSystemAttributeNameApproximateReceiveCount)])
		sentAt := time.Now().UTC()
		if raw := message.Attributes[string(sqstypes.MessageSystemAttributeNameSentTimestamp)]; raw != "" {
			if millis, convErr := strconv.ParseInt(raw, 10, 64); convErr == nil {
				sentAt = time.UnixMilli(millis).UTC()
			}
		}
		body := awsv2.ToString(message.Body)
		deliveries = append(deliveries, &queue.Delivery{
			ID:            awsv2.ToString(message.MessageId),
			ReceiptHandle: awsv2.ToString(message.ReceiptHandle),
			Body:          []byte(body),
			Attributes:    attributes,
			ReceiveCount:  receiveCount,
			EnqueuedAt:    sentAt,
		})
	}
	return deliveries, nil
}

func (q *EmailQueue) Ack(ctx context.Context, deliveries []*queue.Delivery) error {
	if len(deliveries) == 0 {
		return nil
	}
	entries := make([]sqstypes.DeleteMessageBatchRequestEntry, 0, len(deliveries))
	for i, delivery := range deliveries {
		if delivery == nil || strings.TrimSpace(delivery.ReceiptHandle) == "" {
			continue
		}
		entries = append(entries, sqstypes.DeleteMessageBatchRequestEntry{
			Id:            awsv2.String(strconv.Itoa(i)),
			ReceiptHandle: awsv2.String(delivery.ReceiptHandle),
		})
	}
	if len(entries) == 0 {
		return nil
	}
	started := time.Now()
	output, err := q.api.DeleteMessageBatch(ctx, &sqs.DeleteMessageBatchInput{
		QueueUrl: awsv2.String(q.queueURL),
		Entries:  entries,
	})
	if err != nil {
		metrics.ObserveQueueDeleteBatch(q.component, q.queueName, 0, err, started)
		return err
	}
	if len(output.Failed) > 0 {
		batchErr := errors.New("sqs batch delete reported failures")
		metrics.ObserveQueueDeleteBatch(q.component, q.queueName, len(output.Successful), batchErr, started)
		return batchErr
	}
	metrics.ObserveQueueDeleteBatch(q.component, q.queueName, len(output.Successful), nil, started)
	return nil
}

func (q *EmailQueue) Retry(ctx context.Context, delivery *queue.Delivery, delay time.Duration) error {
	if delivery == nil {
		return nil
	}
	seconds := int32(delay.Seconds())
	if seconds < 0 {
		seconds = 0
	}
	if seconds > 43200 {
		seconds = 43200
	}
	started := time.Now()
	_, err := q.api.ChangeMessageVisibility(ctx, &sqs.ChangeMessageVisibilityInput{
		QueueUrl:          awsv2.String(q.queueURL),
		ReceiptHandle:     awsv2.String(delivery.ReceiptHandle),
		VisibilityTimeout: seconds,
	})
	metrics.ObserveQueueVisibilityChange(q.component, q.queueName, err, started)
	return err
}

func (q *EmailQueue) DeadLetter(ctx context.Context, delivery *queue.Delivery, reason string, cause error) error {
	// Leave the message unacked so SQS redrive policy can move it to the queue DLQ.
	_ = ctx
	_ = delivery
	_ = reason
	_ = cause
	return nil
}

func (q *EmailQueue) ApproximateDepth(ctx context.Context) (int64, error) {
	started := time.Now()
	output, err := q.api.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
		QueueUrl: awsv2.String(q.queueURL),
		AttributeNames: []sqstypes.QueueAttributeName{
			sqstypes.QueueAttributeNameApproximateNumberOfMessages,
			sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible,
		},
	})
	if err != nil {
		metrics.ObserveQueueDepth(q.component, q.queueName, 0, err, started)
		return 0, err
	}
	visible, _ := strconv.ParseInt(output.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessages)], 10, 64)
	inFlight, _ := strconv.ParseInt(output.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible)], 10, 64)
	depth := visible + inFlight
	metrics.ObserveQueueDepth(q.component, q.queueName, depth, nil, started)
	return depth, nil
}

func (q *EmailQueue) Close() error {
	return nil
}

func validateConfig(cfg appconfig.SQSConfig) error {
	if strings.TrimSpace(cfg.QueueURL) == "" {
		return errors.New("sqs queue url is required")
	}
	if strings.TrimSpace(cfg.Region) == "" {
		return errors.New("aws region is required")
	}
	if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" {
		parsed, err := url.Parse(endpoint)
		if err != nil {
			return err
		}
		if parsed.Scheme != "https" && !cfg.AllowInsecureEndpoint {
			return errors.New("insecure sqs endpoint is disabled")
		}
	}
	queueType := strings.ToLower(strings.TrimSpace(cfg.QueueType))
	if queueType == "" {
		queueType = queue.QueueTypeStandard
	}
	switch queueType {
	case queue.QueueTypeStandard:
	case queue.QueueTypeFIFO:
		if strings.TrimSpace(cfg.MessageGroupID) == "" {
			return errors.New("sqs message group id is required for fifo queues")
		}
	default:
		return errors.New("unsupported sqs queue type")
	}
	return nil
}