package queue

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"

	"code.justin.tv/sse/policy"
	"code.justin.tv/systems/sandstorm/logging"

	"os"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/sns"
	"github.com/aws/aws-sdk-go/service/sns/snsiface"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
)

// Message describes an AWS SQS queue message
type Message struct {
	Type             string
	MessageID        string
	RawSecret        string `json:"message"`
	Timestamp        string
	SignatureVersion string
	Signature        string
	SigningCertURL   string
	UnsubscribeURL   string
}

// Secret describes the JSON message payload sent from the lambda
// function over SNS/SQS
type Secret struct {
	UpdatedAt struct {
		S string
	}
	Name struct {
		S string
	}
}

// Queuer describes the sandstorm queue api
type Queuer interface {
	ApproximateNumberOfMessages() (int64, error)
	PollForSecret(ctx context.Context) (*Secret, error)
	Delete() (err error)
	Setup() (err error)
	Status() (*Status, error)
}

// Queue contains aws SQS and the Config struct containing configuration values
type Queue struct {
	SQS    sqsiface.SQSAPI
	SNS    snsiface.SNSAPI
	Config Config
	logger logging.Logger
}

// New returns intialized queue ready to poll SQS
// It will attempt to load an existing queue first, if that errors, will create a new one.
func New(sqsClient sqsiface.SQSAPI, snsClient snsiface.SNSAPI, config Config, logger logging.Logger) (q *Queue) {
	mergeConfig(&config, DefaultConfig())

	logger.Debugf("sandstorm queue config: %#v", config)

	// We could not read queueARN and queueURL from the file. create a new queue.
	q = &Queue{
		Config: config,
		logger: logger,
		SQS:    sqsClient,
		SNS:    snsClient,
	}
	return
}

// Setup creates a new SQS queue and subscribes it to the SNS topic
func (q *Queue) Setup() (err error) {
	// check to see if the queue is available
	_, err = q.ApproximateNumberOfMessages()
	if err != nil {
		err = q.createQueue()
		return
	}
	return
}

func (q *Queue) generateQueuePolicy() (document string, err error) {
	iamPolicy := policy.IAMPolicyDocument{
		Version: policy.DocumentVersion,
		Statement: []policy.IAMPolicyStatement{
			{
				Effect: "Allow",
				Principal: map[string][]string{
					"AWS": {"*"},
				},
				Action:   "SQS:SendMessage",
				Resource: q.Config.QueueArn,
				Condition: policy.IAMStatementCondition{
					"ArnEquals": map[string]interface{}{
						"aws:SourceArn": q.Config.TopicArn,
					},
				},
			},
		},
	}

	bs, err := json.Marshal(iamPolicy)
	if err != nil {
		return
	}
	document = string(bs)
	return
}

// createQueue builds a new Queue with a corresponding AWS IAM policy and saves
// the resulting configuration to file for use even after a restart.
func (q *Queue) createQueue() error {
	hostname, err := os.Hostname()
	if err != nil {
		return err
	}

	queueName, err := buildQueueIdentifier(q.Config.QueueNamePrefix, hostname)
	if err != nil {
		return err
	}
	q.Config.QueueArn = buildQueueArn(q.Config.Region, q.Config.AccountID, queueName)

	policy, err := q.generateQueuePolicy()
	if err != nil {
		return err
	}

	params := &sqs.CreateQueueInput{
		QueueName: aws.String(queueName), // Required
		Attributes: map[string]*string{
			"Policy": aws.String(policy), // Required
		},
	}

	resp, err := q.SQS.CreateQueue(params)
	if err != nil {
		return err
	}

	q.Config.QueueURL = aws.StringValue(resp.QueueUrl)
	if err != nil {
		return err
	}

	err = q.subscribe()
	if err != nil {
		return errors.New("error subscribing to SQS: " + err.Error())
	}

	return nil
}

// getQueueAttributes returns the requested `attribute` for the queue.
// See http://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_GetQueueAttributes.html
// for api reference.
func (q *Queue) getQueueAttributes(attribute string) (string, error) {
	params := &sqs.GetQueueAttributesInput{
		QueueUrl: aws.String(q.Config.QueueURL), // Required
		AttributeNames: []*string{
			aws.String("All"), // Required
		},
	}

	resp, err := q.SQS.GetQueueAttributes(params)
	if err != nil {
		return "", err
	}

	value, ok := resp.Attributes[attribute]
	if !ok {
		return "", nil
	}
	return aws.StringValue(value), nil
}

func (q *Queue) subscribe() error {
	params := &sns.SubscribeInput{
		Protocol: aws.String("sqs"),             // Required
		TopicArn: aws.String(q.Config.TopicArn), // Required
		Endpoint: aws.String(q.Config.QueueArn),
	}

	_, err := q.SNS.Subscribe(params)
	if err != nil {
		return err
	}

	return nil
}

// ApproximateNumberOfMessages returns the approximate number of messaages in the queue
func (q *Queue) ApproximateNumberOfMessages() (messages int64, err error) {
	val, err := q.getQueueAttributes("ApproximateNumberOfMessages")
	if err != nil {
		return 0, err
	}
	messages, err = strconv.ParseInt(val, 10, 64)
	if err != nil {
		return 0, err
	}
	return messages, nil
}

// PollForSecret polls the queue for new messages in the queue and unmarshals them as
// type QueueSecret.
func (q *Queue) PollForSecret(ctx context.Context) (*Secret, error) {
	params := &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(q.Config.QueueURL), // Required
		MaxNumberOfMessages: aws.Int64(10),
		VisibilityTimeout:   aws.Int64(1),
		WaitTimeSeconds:     aws.Int64(1),
	}
	resp, err := q.SQS.ReceiveMessageWithContext(ctx, params)
	if err != nil {
		return nil, err
	}

	if len(resp.Messages) == 0 {
		return nil, nil
	}
	sqsMessage := resp.Messages[0]
	message := &Message{}
	err = json.Unmarshal([]byte(*sqsMessage.Body), message)
	if err != nil {
		return nil, errors.New("error parsing SQS message: " + err.Error())
	}

	queueSecret := &Secret{}
	err = json.Unmarshal([]byte(message.RawSecret), queueSecret)
	if err != nil {
		return nil, errors.New("error marshaling SQS message: " + err.Error())
	}

	secretName := queueSecret.Name.S
	if secretName == "" {
		return nil, fmt.Errorf(
			"unable to marshal name value from QueueMessage: %+v",
			message,
		)
	}
	if q.logger != nil {
		q.logger.Infof("Received SQS update for secret: %s", secretName)
	}
	go func() {
		err := q.deleteMessage(*sqsMessage.ReceiptHandle)
		if err != nil && q.logger != nil {
			q.logger.Errorf(
				"Error deleting SQS Message(%s): %s",
				sqsMessage,
				err.Error(),
			)
		}
	}()

	return queueSecret, nil
}

// DeleteMessage removes a message from the agent's queue
func (q *Queue) deleteMessage(handle string) error {
	params := &sqs.DeleteMessageInput{
		QueueUrl:      aws.String(q.Config.QueueURL), // Required
		ReceiptHandle: aws.String(handle),            // Required
	}

	_, err := q.SQS.DeleteMessage(params)
	return err
}

// Delete removes the SQS queue and the queue state file
func (q *Queue) Delete() error {
	// if no configured queue url, can't delete anything
	if q.Config.QueueURL == "" {
		return nil
	}

	params := &sqs.DeleteQueueInput{
		QueueUrl: aws.String(q.Config.QueueURL),
	}
	_, err := q.SQS.DeleteQueue(params)
	if err != nil {
		return fmt.Errorf("Error deleting queue: %s", err.Error())
	}

	err = os.Remove(q.Config.QueueConfigFullPath())
	if err != nil {
		// We dont care if could not delete the local queue config file.
	}
	return nil
}

// Status is returned by Describe
type Status struct {
	ApproximateNumberOfMessages int64
	QueueArn                    string
	QueueURL                    string
}

// Status returns information about the queue being used by this manager
func (q *Queue) Status() (*Status, error) {
	approximateNumberOfMessages, err := q.ApproximateNumberOfMessages()
	if err != nil {
		return nil, err
	}

	return &Status{
		ApproximateNumberOfMessages: approximateNumberOfMessages,
		QueueArn:                    q.Config.QueueArn,
		QueueURL:                    q.Config.QueueURL,
	}, nil
}
