// Wrapper package for SQS

package queue

import (
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/sqs"

	"code.justin.tv/foundation/history-worker/configuration"
	"code.justin.tv/foundation/history-worker/custom_aws"
	"code.justin.tv/foundation/history-worker/report"
)

const (
	maxMessages          = 10
	pollDelay            = 1 * time.Second
	maxRetries           = 5
	pollWaitTimeSeconds  = 10
	messageDeleteSuccess = "message_delete.success"
	messageDeleteFailure = "message_delete.failure"
	messageDeleteRetry   = "message_delete.retry"
)

// Returns the AWS SQS client
func getClient() *sqs.SQS {
	return sqs.New(custom_aws.Session())
}

// Main polling function to receive, process & delete messages from the queue
func Poll(callback func(string) bool) {
	client := getClient()

	for {
		messages := GetMessages(client)

		if len(messages) == 0 {
			time.Sleep(pollDelay)
			continue
		}

		var wg sync.WaitGroup
		wg.Add(len(messages)) // Going to spawn these many subroutines to process each message

		for _, msg := range messages {
			go func(message *sqs.Message) { // Process each message in parallel using subroutines
				completed := callback(*message.Body) // Process the message using the callback method
				if completed {                       // Delete the message from the Queue only if successfully processed
					DeleteMessage(client, message)
				}
				wg.Done() // Mark this subroutine as done processing the message
			}(msg)
		}

		wg.Wait() // Wait for all messages to be processed by our subroutines
	}
}

// Receive messages from SQS with a long poll
func GetMessages(client *sqs.SQS) []*sqs.Message {
	params := &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(configuration.Resolve("queueURL")),
		MaxNumberOfMessages: aws.Int64(maxMessages),
		WaitTimeSeconds:     aws.Int64(pollWaitTimeSeconds),
	}

	response, err := client.ReceiveMessage(params)

	if err != nil {
		report.RollbarError(err)
	}

	return response.Messages
}

// Delete message from SQS with retry mechanism
func DeleteMessage(client *sqs.SQS, message *sqs.Message) {
	var err error

	field := report.NewField("message", message.Body)
	params := &sqs.DeleteMessageInput{
		QueueUrl:      aws.String(configuration.Resolve("queueURL")),
		ReceiptHandle: aws.String(*message.ReceiptHandle),
	}

	for try := 1; try <= maxRetries; try++ {
		_, err = client.DeleteMessage(params)
		if err == nil {
			report.StatsdThroughput(messageDeleteSuccess)
			break
		} else {
			if try != maxRetries {
				try_field := report.NewField("try", try)
				warn_error := report.NewWarningAlert(err.Error())
				report.RollbarError(warn_error, field, try_field)
				report.StatsdThroughput(messageDeleteRetry)
			} else {
				report.RollbarError(err, field)
				report.StatsdThroughput(messageDeleteFailure)
			}
		}
	}
}
