package sqsclient

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/service/sts/stsiface"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/client"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/endpoints"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
	"github.com/aws/aws-sdk-go/service/sts"
	"github.com/pkg/errors"

	eventbus "code.justin.tv/eventbus/client"
	sess "code.justin.tv/eventbus/client/internal/session"
	"code.justin.tv/eventbus/client/internal/sqspoller"
	"code.justin.tv/eventbus/client/internal/stsregion"
	"code.justin.tv/eventbus/client/lowlevel/cfnversion"
)

type SQSClient struct {
	poller     *sqspoller.SQSPoller
	dispatcher eventbus.Dispatcher
	fineTuning FineTuning
	callback   func(HookEvent)

	deliverGroup sync.WaitGroup
	ctxRequests  context.Context
	stopRequests func()
	deliver      chan *sqs.Message
	acks         chan string
}

// New creates an SQS client that will poll SQS and dispatch messages.
//
// See package documentation for overall description and usage.
func New(config Config) (*SQSClient, error) {
	if config.Session == nil && config.OverrideClient == nil {
		return nil, errors.New("no client or session was found")
	}

	config.setDefaults()
	var sqsClient sqsiface.SQSAPI

	if config.Session != nil {
		// The default setting will use the deprecated STS global endpoint. If
		// we detect that the subscriber has not correctly set this value to a
		// valid endpoint, we default to the STS regional endpoint.
		stsEndpoint := stsregion.ConfiguredEndpoint(config.Session)
		if stsEndpoint == endpoints.UnsetSTSEndpoint || stsEndpoint == endpoints.LegacySTSEndpoint {
			correctedSession, err := sess.Corrected(config.Session)
			if err != nil {
				return nil, errors.New("failed to copy session configuration for setting sts regional endpoint in subscriber")
			}
			config.Session = correctedSession
		}

		stsClient := sts.New(config.Session, &aws.Config{STSRegionalEndpoint: stsEndpoint})

		var err error
		sqsClient, err = getSQSClient(config.Session, stsClient, cfnversion.NewFetcher())
		if err != nil {
			return nil, errors.Wrap(err, "could not create sqs client")
		}
	} else {
		sqsClient = config.OverrideClient
	}

	deliver := make(chan *sqs.Message, config.DeliverChannelDepth)
	ctxRequests, stopRequests := context.WithCancel(context.Background())

	poller := sqspoller.New(
		sqsClient, config.QueueURL, int64(config.VisibilityTimeout.Seconds()),
		config.MinPollers, deliver, wrapCallback(config.HookCallback),
	)

	c := &SQSClient{
		poller:     poller,
		dispatcher: config.Dispatcher,
		fineTuning: config.FineTuning,
		callback:   config.HookCallback,

		ctxRequests:  ctxRequests,
		stopRequests: stopRequests,
		deliver:      deliver,
		acks:         make(chan string, config.AckChannelDepth),
	}

	c.deliverGroup.Add(config.DeliverConcurrency)
	for i := 0; i < config.DeliverConcurrency; i++ {
		go func() {
			defer c.deliverGroup.Done()
			c.deliverLoop()
		}()
	}

	return c, nil
}

// Shutdown this SQS client in stages, allowing tasks to finish cleanly.
//
// This method will block until all running goroutines have completed.
// Roughly, the shutdown sequence looks like this:
//
//   1. Stop goroutines polling SQS for messages
//   2. Finish processing any in-flight messages
//   3. Acknowledge any successes (DeleteMessage from the queue)
//
// See https://git-aws.internal.justin.tv/eventbus/client/blob/master/internal/sqspoller/DESIGN.md#shutdown
// for detailed design of the shutdown process.
func (c *SQSClient) Shutdown() error {
	go func() {
		select {
		case <-time.After(c.fineTuning.ShutdownCancelRequests):
			c.stopRequests()
		case <-c.ctxRequests.Done():
		}
	}()
	c.poller.StopPollers()
	close(c.deliver)
	c.deliverGroup.Wait()
	c.poller.StopAcking()
	c.stopRequests()
	return nil
}

func getSQSClient(sess client.ConfigProvider, stsClient stsiface.STSAPI, versionChecker cfnversion.Fetcher) (*sqs.SQS, error) {
	meetsNoAssumeRequirement, err := versionChecker.Require(context.Background(), sess, "v1.3.0")
	if err != nil {
		return nil, errors.Wrap(err, "unable to determine cross account cloudformation version")
	}

	var sqsClient *sqs.SQS
	if meetsNoAssumeRequirement {
		sqsClient = sqs.New(sess)
	} else {
		callerIdentity, err := stsClient.GetCallerIdentity(&sts.GetCallerIdentityInput{})
		if err != nil {
			return nil, errors.Wrap(err, "could not request caller identity (GetCallerIdentity) for accessing aws account id, please ensure you have followed the instructions here: https://git-aws.internal.justin.tv/pages/eventbus/docs/getting_started/#run-the-cloudformation-template")
		}

		accountID := aws.StringValue(callerIdentity.Account)
		if accountID == "" {
			return nil, errors.New("no aws account id was found, please ensure you have followed the instructions here: https://git-aws.internal.justin.tv/pages/eventbus/docs/getting_started/#run-the-cloudformation-template")
		}
		creds := stscreds.NewCredentials(sess, assumeRoleARNSubscriber(accountID))
		sqsClient = sqs.New(sess, &aws.Config{Credentials: creds})
	}

	return sqsClient, nil
}

func assumeRoleARNSubscriber(accountID string) string {
	return fmt.Sprintf("arn:aws:iam::%s:role/EventBus/EventBusSubscriberV1", accountID)
}
