package callee

import (
	"context"
	"encoding/json"
	"net/http"
	"sync"
	"time"

	"code.justin.tv/sse/malachai/pkg/backoff"
	"code.justin.tv/sse/malachai/pkg/config"
	"code.justin.tv/sse/malachai/pkg/internal/closer"
	"code.justin.tv/sse/malachai/pkg/log"
	"code.justin.tv/sse/policy"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/session"
	"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"
	"github.com/gofrs/uuid"
)

const (
	sqsBatchSize   = 1
	sqsPollTimeout = 20
)

type eventsConfig struct {
	Environment                string
	RoleArn                    string
	Region                     string
	CalleeCapabilitiesTopicARN string
	PubkeyInvalidationTopicARN string
	ServiceID                  string
	AWSConfigBase              *aws.Config
}

// FillDefaults fill sthe defaults
func (ec *eventsConfig) FillDefaults() (err error) {
	resources, err := config.GetResources(ec.Environment)
	if err != nil {
		return
	}

	if ec.Region == "" {
		ec.Region = "us-west-2"
	}

	if ec.PubkeyInvalidationTopicARN == "" {
		ec.PubkeyInvalidationTopicARN = resources.PubkeyInvalidationTopicARN
	}

	if ec.Region == "" {
		ec.Region = resources.Region
	}

	if ec.CalleeCapabilitiesTopicARN == "" {
		ec.CalleeCapabilitiesTopicARN = resources.CalleeCapabilitiesTopicARN
	}

	if ec.PubkeyInvalidationTopicARN == "" {
		ec.PubkeyInvalidationTopicARN = resources.PubkeyInvalidationTopicARN
	}

	return
}

func newEventsClient(cfg *eventsConfig, logger log.S2SLogger, closer closer.API) (client *eventsClient, err error) {
	err = cfg.FillDefaults()
	if err != nil {
		return
	}

	logger.Debugf("starting events client with roleArn: '%s'", cfg.RoleArn)

	sess, err := session.NewSession(config.AWSConfig(cfg.AWSConfigBase, cfg.Region, cfg.RoleArn))
	if err != nil {
		return
	}
	client = &eventsClient{
		pubkeyEvents: make(chan pubkeyEventer),
		capabilities: make(chan struct{}),
		health:       new(health),
		log:          logger,
		sns:          sns.New(sess),
		sqs:          sqs.New(sess),
		cfg:          cfg,
		closer:       closer,
		state:        &eventsClientState{},
	}
	return
}

type eventsClient struct {
	pubkeyEvents chan pubkeyEventer
	capabilities chan struct{}

	closer closer.API

	health *health
	log    log.S2SLogger
	sns    snsiface.SNSAPI
	sqs    sqsiface.SQSAPI
	cfg    *eventsConfig
	state  *eventsClientState
}

type eventsClientState struct {
	QueueARN                               string
	QueueURL                               string
	CalleeCapabilitiesTopicSubscriptionARN string
	PubkeyInvalidationTopicSubscriptionARN string
}

func (e *eventsClient) setupQueue() (err error) {
	if e.state.QueueURL == "" {
		var createQueueErr error
		err = backoff.ConstantRetry(func() error {
			err := e.createQueue()
			if err != nil {
				if awsErr, ok := err.(awserr.Error); ok {
					if awsErr.Code() == "RequestThrottled" {
						return err
					}
				}
			}
			createQueueErr = err
			return err
		}, 10, time.Second)
		if createQueueErr != nil {
			err = createQueueErr
		}
		if err != nil {
			e.log.Debugf("error creating queue: %s", err.Error())
			return
		}

		err = e.getQueueARN()
		if err != nil {
			e.log.Debugf("error getting queueARN: %s", err.Error())
			return
		}

		err = e.giveSNSPermission()
		if err != nil {
			e.log.Debugf("error giving sns: %s", err.Error())
			return
		}

		err = e.subscribeQueue()
		if err != nil {
			e.log.Debugf("error subscribing queue: %s", err.Error())
			return
		}
	}

	return
}

func (e *eventsClient) recreateQueue() (err error) {
	err = e.invalidateQueue()
	if err != nil {
		return
	}

	for {
		err = e.setupQueue()
		if err != nil {
			e.log.Error("error while recreating queue: " + err.Error())

			t := time.NewTimer(10 * time.Second)
			select {
			case <-e.closer.Done():
				e.log.Debug("exiting recreating due to context done")
				return
			case <-t.C:
				continue
			}
		}
	}
}

func (e *eventsClient) invalidateQueue() (err error) {
	e.state.QueueARN = ""
	e.state.QueueURL = ""
	return
}

func (e *eventsClient) createQueue() (err error) {
	uid, err := uuid.NewV4()
	if err != nil {
		return err
	}

	queueName := e.cfg.ServiceID + "_" + uid.String()

	input := &sqs.CreateQueueInput{
		QueueName: aws.String(queueName),
	}

	output, err := e.sqs.CreateQueue(input)
	if err != nil {
		return
	}

	e.state.QueueURL = aws.StringValue(output.QueueUrl)
	return
}

func (e *eventsClient) getQueueARN() (err error) {
	input := &sqs.GetQueueAttributesInput{
		QueueUrl:       aws.String(e.state.QueueURL),
		AttributeNames: []*string{aws.String("QueueArn")},
	}

	output, err := e.sqs.GetQueueAttributes(input)
	if err != nil {
		return
	}

	e.state.QueueARN = aws.StringValue(output.Attributes["QueueArn"])
	return
}

func (e *eventsClient) giveSNSPermission() (err error) {
	topicArns := []string{e.cfg.PubkeyInvalidationTopicARN}
	if e.cfg.CalleeCapabilitiesTopicARN != "" {
		topicArns = append(topicArns, e.cfg.CalleeCapabilitiesTopicARN)
	}

	document := &policy.IAMPolicyDocument{
		Version: policy.DocumentVersion,
		Statement: []policy.IAMPolicyStatement{
			{
				Sid:    "allow callee sns topic",
				Effect: "Allow",
				Principal: map[string][]string{
					"AWS": {"*"},
				},
				Action:   "SQS:SendMessage",
				Resource: e.state.QueueARN,
				Condition: policy.IAMStatementCondition{
					"StringEquals": map[string]interface{}{
						"aws:SourceArn": topicArns,
					},
				},
			},
		},
	}

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

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

	// cancel when e.closer is cancelled
	ctx, cancel := context.WithCancel(e.closer)
	_, err = e.sqs.SetQueueAttributesWithContext(ctx, setParams)
	if err != nil {
		cancel()
		return
	}
	cancel()
	return
}

func (e *eventsClient) subscribeQueue() (err error) {
	output, err := e.sns.Subscribe(&sns.SubscribeInput{
		Endpoint: aws.String(e.state.QueueARN),
		Protocol: aws.String("sqs"),
		TopicArn: aws.String(e.cfg.PubkeyInvalidationTopicARN),
	})
	if err != nil {
		e.log.Debugf("error subbing to PubkeyInvalidationTopicARN: ", err.Error())
		return
	}
	e.state.PubkeyInvalidationTopicSubscriptionARN = aws.StringValue(output.SubscriptionArn)

	if e.cfg.CalleeCapabilitiesTopicARN != "" {
		output, err = e.sns.Subscribe(&sns.SubscribeInput{
			Endpoint: aws.String(e.state.QueueARN),
			Protocol: aws.String("sqs"),
			TopicArn: aws.String(e.cfg.CalleeCapabilitiesTopicARN),
		})
		if err != nil {
			e.log.Debugf("error subbing to caps: ", err.Error())
			return
		}
		e.state.CalleeCapabilitiesTopicSubscriptionARN = aws.StringValue(output.SubscriptionArn)
	}
	return
}

// event types
const (
	PubkeyGenerationEventType          = "pubkey_generation"
	PubkeyInvalidationEventType        = "pubkey_invalidation"
	InvalidateAllCapabilitiesEventType = "invalidate_all_capabilities"
)

// EventTypeMessageAttributeName is the key of the sqs message attribute map
// that identifies the type of message that was sent
const EventTypeMessageAttributeName = "TWITCH.MALACHAI.EVENTS.EventType"

func (e *eventsClient) updateLoop() (err error) {
	err = e.pollForUpdates()
	if err != nil {
		select {
		case <-e.closer.Done():
			return
		default:
		}

		e.log.Error("error while updating loop: " + err.Error())
		e.markQueueDown()

		if queueDoesNotExistError(err) {
			iqErr := e.recreateQueue()
			if iqErr != nil {
				e.log.Errorf("error while recreating queue: %s", iqErr.Error())
			}
		}
	} else {
		e.markQueueUp()
	}
	return
}

func (e *eventsClient) pollForUpdates() (err error) {
	e.log.Debug("running queue update loop")
	input := &sqs.ReceiveMessageInput{
		MaxNumberOfMessages: aws.Int64(sqsBatchSize),
		QueueUrl:            aws.String(e.state.QueueURL),
		WaitTimeSeconds:     aws.Int64(sqsPollTimeout),
	}
	// cancel this context when e.closer is closed
	ctx, cancel := context.WithCancel(e.closer)
	defer cancel()
	output, err := e.sqs.ReceiveMessageWithContext(ctx, input)
	if err != nil {
		return
	}

	if len(output.Messages) > 0 {
		var f int
		var handledCapabilitiesEvent bool

		for _, message := range output.Messages {
			select {
			case <-e.closer.Done():
				e.log.Info("events client closed, stopped processing messages")
				return
			default:
			}

			var msg snsMessage
			err = json.Unmarshal([]byte(aws.StringValue(message.Body)), &msg)
			av, ok := msg.MessageAttributes[EventTypeMessageAttributeName]
			if !ok {
				// TODO: send message with missing event type to dlq
				e.log.Error("message missing event type")
				continue
			}

			e.log.Debugf("received event: %s", av.Value)

			switch av.Value {
			case PubkeyGenerationEventType:
				err = e.handlePubkeyGenerationEvent(msg.Message)
				if err != nil {
					// TODO: send message to dlq
					e.log.Error("error handling pubkey generation event: " + err.Error())
					continue
				}
			case PubkeyInvalidationEventType:
				err = e.handlePubkeyInvalidationEvent(msg.Message)
				if err != nil {
					// TODO: send message to dlq
					e.log.Error("error handling pubkey invalidation event: " + err.Error())
					continue
				}
			case InvalidateAllCapabilitiesEventType:
				if handledCapabilitiesEvent {
					// already handled caps event, dont do it this loop,
					// but still delete the message
					break
				}
				err = e.handleCapabilitiesEvent(msg.Message)
				if err != nil {
					// TODO: send message to dlq
					e.log.Error("error handling capabilities event: " + err.Error())
					continue
				}
				f++
				handledCapabilitiesEvent = true
			default:
				// TODO: send messages with unhandled type to dead letter queue
				e.log.Error("invalid event type: " + av.Value)
				continue
			}

			_, err = e.sqs.DeleteMessageWithContext(ctx, &sqs.DeleteMessageInput{
				QueueUrl:      aws.String(e.state.QueueURL),
				ReceiptHandle: message.ReceiptHandle,
			})
			if err != nil {
				return
			}
		}
	} else {
		e.log.Debug("received no events. polling.")
	}
	return
}

func (e *eventsClient) start() (err error) {
	err = e.setupQueue()
	if err != nil {
		e.log.Debugf("setupQueue failed: %s", err.Error())
		return
	}

	go e.main()
	return
}

func (e *eventsClient) main() {
	for {
		updateErr := e.updateLoop()
		if updateErr == nil {
			continue
		}

		e.log.Errorf("error updating loop: %s", updateErr.Error())

		timer := time.NewTimer(10 * time.Second)
		select {
		case <-e.closer.Done():
			return
		case <-timer.C:
		}
	}
}

// health represents the health of the events client
type health struct {
	queueDownSince time.Time
	updateLock     sync.Mutex
}

// QueueDownSince returns the time that the queue was down
func (h *health) QueueDownSince() time.Time {
	h.updateLock.Lock()
	t := h.queueDownSince
	h.updateLock.Unlock()
	return t
}

// MarkQueueDown marks the queue as down
func (h *health) MarkQueueDown() {
	h.updateLock.Lock()
	h.queueDownSince = time.Now().UTC()
	h.updateLock.Unlock()
}

// MarkQueueUp marks the queue as up
func (h *health) MarkQueueUp() {
	h.updateLock.Lock()
	h.queueDownSince = time.Time{}
	h.updateLock.Unlock()
}

func (e *eventsClient) markQueueDown() {
	e.log.Debug("marking queue as down")
	e.health.MarkQueueDown()
}

func (e *eventsClient) markQueueUp() {
	e.log.Debug("marking queue as up")
	e.health.MarkQueueUp()
}

func (e *eventsClient) Health() (h *health) {
	return e.health
}

func queueDoesNotExistError(err error) bool {
	awsErr, ok := err.(awserr.RequestFailure)
	if !ok {
		return false
	}

	return awsErr.StatusCode() == http.StatusNotFound
}
