package sqsclient

import (
	"context"
	"encoding/json"
	"strconv"
	"time"

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

	eventbus "code.justin.tv/eventbus/client"
	"code.justin.tv/eventbus/client/internal/retrycount"
	"code.justin.tv/eventbus/client/lowlevel/snsmarshal"
)

func (c *SQSClient) deliverLoop() {
	shutdown := c.ctxRequests.Done()

	for {
		select {
		case <-shutdown:
			return
		case message, ok := <-c.deliver:
			if !ok {
				return
			}
			err := c.doDeliver(message)
			if err == nil {
				c.poller.Ack(*message.ReceiptHandle)
			}
		}
	}
}

func (c *SQSClient) doDeliver(message *sqs.Message) error {
	var target snsMessage
	err := json.Unmarshal([]byte(*message.Body), &target)
	if err != nil {
		err = errors.Wrap(err, "could not parse SNS message body from SQS message")
		c.fireDecodeError(err, message)
		return err
	}

	buf, err := snsmarshal.DecodeString(target.Message)
	if err != nil {
		err = errors.Wrap(err, "could not decode event bus payload in SNS message")
		c.fireDecodeError(err, message)
		return err
	}

	msgCtx, msgCancel := context.WithTimeout(c.ctxRequests, time.Duration(c.poller.VisibilitySeconds)*time.Second)
	receiveCount, ok := message.Attributes["ApproximateReceiveCount"]
	if ok {
		i, err := strconv.Atoi(aws.StringValue(receiveCount))
		if err != nil {
			err = errors.Wrap(err, "could not parse receive count in SQS attributes")
			c.fireDecodeError(err, message)
			return err
		}
		msgCtx = retrycount.Set(msgCtx, i)
	}

	defer msgCancel()

	err = c.dispatcher.Dispatch(msgCtx, buf)
	if err != nil {
		if _, ok := err.(eventbus.DecodeError); ok {
			c.fireDecodeError(err, message)
		} else {
			err = errors.Wrap(err, "could not dispatch")
			c.callback(&HookEventDispatchError{
				Err:        err,
				RawMessage: message,
			})
		}
		return err
	}
	return nil
}

func (c *SQSClient) fireDecodeError(err error, message *sqs.Message) {
	c.callback(&HookEventDecodeError{
		Err:        err,
		RawMessage: message,
	})
}

// snsMessage is the embedded payload inside the SQS message.
// It is a subset of fields we actually use from:
// https://docs.aws.amazon.com/sns/latest/dg/sns-message-and-json-formats.html#http-notification-json
type snsMessage struct {
	Message string `json:"Message"`
}
