package queue

import (
	"encoding/json"
	"fmt"
	"path"

	"os"

	log "github.com/Sirupsen/logrus"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/sns"
	"github.com/aws/aws-sdk-go/service/sqs"
)

// 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
	}
}

// SQS describes the aws SQS service
type SQS interface {
	ReceiveMessage(input *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)
	CreateQueue(input *sqs.CreateQueueInput) (*sqs.CreateQueueOutput, error)
	SetQueueAttributes(input *sqs.SetQueueAttributesInput) (*sqs.SetQueueAttributesOutput, error)
	GetQueueAttributes(input *sqs.GetQueueAttributesInput) (*sqs.GetQueueAttributesOutput, error)
	DeleteQueue(input *sqs.DeleteQueueInput) (*sqs.DeleteQueueOutput, error)
	PurgeQueue(input *sqs.PurgeQueueInput) (*sqs.PurgeQueueOutput, error)
	DeleteMessage(input *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)
}

// Queue contains aws SQS and the Config struct containing configuration values
type Queue struct {
	SQS    SQS
	Config Config
}

// 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(config Config) (q *Queue, err error) {
	defaultConfig := DefaultConfig()
	mergeConfig(&config, defaultConfig)

	q = &Queue{
		Config: config,
	}

	q.SQS = sqs.New(session.New(q.Config.AWSConfig))

	if q.Config.QueueArn == "" || q.Config.QueueURL == "" {
		err = q.loadQueueConfig()
		if err != nil {
			log.Error(err)
		}
	}

	err = q.checkQueue()
	if err != nil {
		queueConfigPath := path.Join(q.Config.QueueConfigPath, q.Config.QueueFileName)
		err = q.createQueue(queueConfigPath)
		if err != nil {
			return
		}
	}

	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(queueConfigPath string) error {
	if q.SQS == nil {
		return fmt.Errorf("AWS SQS has not yet been set initialized")
	}

	id, fqdn, err := buildQueueIdentifiers(q.Config.QueueNamePrefix)
	if err != nil {
		return err
	}

	params := &sqs.CreateQueueInput{
		QueueName: aws.String(fqdn), // Required
	}

	resp, err := q.SQS.CreateQueue(params)
	log.Infof("Attempting to create queue: %s", fqdn)
	if err != nil {
		return err
	}

	q.Config.QueueURL = *resp.QueueUrl
	q.Config.QueueArn, err = q.GetQueueAttributes("QueueArn")
	if err != nil {
		return err
	}

	policy := map[string]interface{}{
		"Version": "2012-10-17",
		"Id":      fqdn,
		"Statement": []map[string]interface{}{map[string]interface{}{
			"Sid":    id,
			"Effect": "Allow",
			"Principal": map[string]interface{}{
				"AWS": "*",
			},
			"Action":   "SQS:SendMessage",
			"Resource": q.Config.QueueArn,
			"Condition": map[string]interface{}{
				"ArnEquals": map[string]interface{}{
					"aws:SourceArn": q.Config.TopicArn,
				},
			},
		}},
	}

	str, err := json.Marshal(policy)
	if err != nil {
		return err
	}

	setParams := &sqs.SetQueueAttributesInput{
		Attributes: map[string]*string{
			"Policy": aws.String(string(str)), // Required
		},
		QueueUrl: aws.String(*resp.QueueUrl), // Required
	}

	_, err = q.SQS.SetQueueAttributes(setParams)
	if err != nil {
		log.Errorf("Error setting queue attributes: %s", err.Error())
		return err
	}

	err = q.subscribe()
	if err != nil {
		return fmt.Errorf("Error subscribing to SQS: %s", err.Error())
	}

	f, err := os.Create(queueConfigPath)
	if err != nil {
		return err
	}

	defer func() {
		err := f.Close()
		if err != nil {
			log.Error(fmt.Errorf("Error closing queue config file: %s", err.Error()))
		}
	}()

	_, err = f.WriteString("[sandstorm-agent]\nQueueArn=" + q.Config.QueueArn + "\nQueueURL=" + q.Config.QueueURL)
	if err != nil {
		return err
	}

	// If all goes well, flush changes to the file
	return f.Sync()
}

// 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
	}

	attr := resp.Attributes
	return *attr[attribute], nil
}

func (q *Queue) subscribe() error {
	svc := sns.New(session.New(q.Config.AWSConfig))

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

	resp, err := svc.Subscribe(params)
	if err != nil {
		return err
	}

	log.Infof("SQS Subscribe Response: %s", resp)
	return nil
}

// Checks queue for healthiness, retries on network error
func (q *Queue) checkQueue() error {
	_, err := q.GetQueueAttributes("ApproximateNumberOfMessages")
	return err
}

// PopQueueSecret polls the queue for new messages in the queue and unmarshals them as
// type QueueSecret.
func (q *Queue) PopQueueSecret() (*Secret, error) {
	log.Debug("Polling for SQS Queue Messages")

	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.ReceiveMessage(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, fmt.Errorf("Error parsing SQS message: %s", err.Error())
	}

	queueSecret := &Secret{}
	err = json.Unmarshal([]byte(message.RawSecret), queueSecret)
	if err != nil {
		return nil, fmt.Errorf("Error marshaling SQS message: %s", err.Error())
	}

	secretName := queueSecret.Name.S
	if secretName == "" {
		return nil, fmt.Errorf(
			"Unable to marshal name value from QueueMessage: %+v",
			message,
		)
	}

	log.Infof("Received SQS update for secret: %s", secretName)

	go func() {
		err := q.deleteMessage(*sqsMessage.ReceiptHandle)
		if err != nil {
			log.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
}

func (q *Queue) deleteQueue(queueConfigPath string) error {

	params := &sqs.DeleteQueueInput{
		QueueUrl: aws.String(q.Config.QueueURL),
	}
	_, err := q.SQS.DeleteQueue(params)
	if err != nil {
		return err
	}

	return os.Remove(queueConfigPath)
}

func (q *Queue) purgeQueue() error {
	params := &sqs.PurgeQueueInput{
		QueueUrl: aws.String(q.Config.QueueURL),
	}

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