package ingestqueueconsumer

import (
	"context"
	"encoding/base64"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/client"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
	"github.com/golang/protobuf/proto"
	"github.com/pkg/errors"
	"google.golang.org/grpc"

	"code.justin.tv/common/config"
	"code.justin.tv/dta/rockpaperscissors/internal/sandstorm"
	pb "code.justin.tv/dta/rockpaperscissors/proto"
)

const (
	// pollSleepDuration is how long to wait between empty polls to SQS.
	defaultPollSleepDuration = 1 * time.Minute
	// loopSleepDuration is how long to wait between error retries and non-emtpy polls.
	defaultLoopSleepDuration = 1 * time.Second

	// waitTimeSeconds is how long AWS keeps the ReceiveMessage open to wait for a message.
	defaultWaitTimeSeconds = 20 // maximum is 20
	// messagesReceivedPerLoop is how many messages to receive per loop iteration.
	defaultMessagesReceivedPerLoop = 10 // maximum is 10
)

func init() {
	config.Register(map[string]string{
		"ingest-queue-name": "rockpaperscissors-development-ingest",
	})
}

type blueprintIngestorIface interface {
	Ingest(context.Context, *pb.IngestBlueprintRequest) error
}

type gitHubStatsIngestorIface interface {
	Ingest(context.Context, *pb.IngestGitHubStatsRequest) error
}

// IngestQueueConsumer is a background service to poll an SQS queue for ingest requests.
type IngestQueueConsumer struct {
	awsSession              client.ConfigProvider
	endpointAddr            string
	stopping                chan chan interface{}
	queueName               string
	queueURL                string
	sqs                     sqsiface.SQSAPI
	conn                    *grpc.ClientConn
	blueprintIngestor       blueprintIngestorIface
	gitHubStatsIngestor     gitHubStatsIngestorIface
	pollSleepDuration       time.Duration
	loopSleepDuration       time.Duration
	waitTimeSeconds         int64
	messagesReceivedPerLoop int64
}

// New creates a properly initialized IngestQueueConsumer.
func New(sess client.ConfigProvider, endpointAddr string) *IngestQueueConsumer {
	return &IngestQueueConsumer{
		awsSession:              sess,
		endpointAddr:            endpointAddr,
		stopping:                make(chan chan interface{}),
		queueName:               config.MustResolve("ingest-queue-name"),
		sqs:                     sqs.New(sess),
		pollSleepDuration:       defaultPollSleepDuration,
		loopSleepDuration:       defaultLoopSleepDuration,
		waitTimeSeconds:         defaultWaitTimeSeconds,
		messagesReceivedPerLoop: defaultMessagesReceivedPerLoop,
	}
}

func (c *IngestQueueConsumer) getGRPCClientConn() (*grpc.ClientConn, error) {
	if c.conn == nil {
		var err error
		c.conn, err = grpc.Dial(c.endpointAddr, grpc.WithInsecure())
		if err != nil {
			return nil, err
		}
	}
	return c.conn, nil
}

// Start the background consumer goroutine.
func (c *IngestQueueConsumer) Start() error {
	if c.blueprintIngestor == nil {
		conn, err := c.getGRPCClientConn()
		if err != nil {
			return err
		}
		projectMetadataClient := pb.NewProjectMetadataServiceClient(conn)
		c.blueprintIngestor = NewBlueprintIngestor(projectMetadataClient)
	}

	if c.gitHubStatsIngestor == nil {
		conn, err := c.getGRPCClientConn()
		if err != nil {
			return err
		}
		eventClient := pb.NewEventServiceClient(conn)
		sandstormClient := sandstorm.New(c.awsSession)
		c.gitHubStatsIngestor = NewGitHubStatsIngestor(eventClient, sandstormClient)
	}

	resp, err := c.sqs.GetQueueUrl(&sqs.GetQueueUrlInput{
		QueueName: aws.String(c.queueName),
	})
	if err != nil {
		return errors.Wrapf(err, "Error trying to get URL of %s queue", c.queueName)
	}
	c.queueURL = *resp.QueueUrl

	go c.loop()

	return nil
}

func (c *IngestQueueConsumer) loop() {
	log.Info("Background ingest consumer loop started.")
	sleepDuration := c.loopSleepDuration // default is to do a short poll loop.
	for {
		select {
		// check to see if we need to stop
		case stoppedChan := <-c.stopping:
			stoppedChan <- true // Value doesn't matter
			return
		// check to see if it's time to poll SQS again, only run every minute
		case <-time.After(sleepDuration):
			// Unless there are messages in the queue so we'll immediately retry.
			sleepDuration = c.loopSleepDuration

			resp, err := c.sqs.ReceiveMessage(&sqs.ReceiveMessageInput{
				QueueUrl:            aws.String(c.queueURL),
				WaitTimeSeconds:     aws.Int64(c.waitTimeSeconds),
				MaxNumberOfMessages: aws.Int64(c.messagesReceivedPerLoop),
			})
			if err != nil {
				log.Errorf("Error receiving message on %s queue: %v", c.queueName, err)
				break
			}
			if resp == nil {
				log.Error("Received nil response from sqs.ReceiveMessage")
				break
			}
			if len(resp.Messages) == 0 {
				sleepDuration = c.pollSleepDuration // Go to sleep for a while.
				break
			}

			for _, message := range resp.Messages {
				err := c.handleMessage(message)
				if err != nil {
					log.Error(err)
				}
			}
		}
	}
}

func (c *IngestQueueConsumer) handleMessage(message *sqs.Message) error {
	decoded, err := base64.StdEncoding.DecodeString(*message.Body)
	if err != nil {
		return errors.Wrapf(err,
			"Failed to decode message ID %s", *message.MessageId)
	}

	req := &pb.IngestRequest{}
	err = proto.Unmarshal(decoded, req)
	if err != nil {
		return errors.Wrapf(err,
			"Failed to unmarshal message ID %s", *message.MessageId)
	}

	if req.GetIngestBlueprintRequest() != nil {
		err = c.blueprintIngestor.Ingest(
			context.Background(), req.GetIngestBlueprintRequest())
		if err != nil {
			return err
		}
	} else if req.GetIngestGithubStatsRequest() != nil {
		err = c.gitHubStatsIngestor.Ingest(
			context.Background(), req.GetIngestGithubStatsRequest())
		if err != nil {
			return err
		}
	}

	_, err = c.sqs.DeleteMessage(&sqs.DeleteMessageInput{
		QueueUrl:      aws.String(c.queueURL),
		ReceiptHandle: message.ReceiptHandle,
	})
	if err != nil {
		return errors.Wrapf(err,
			"Failed to remove message ID %s from queue", *message.MessageId)
	}

	return nil
}

// Stop the background consumer goroutine after it's been started.
func (c *IngestQueueConsumer) Stop() {
	// Tell the loop to stop and provide a channel to let us know it happened.
	stoppedChan := make(chan interface{}, 1)
	c.stopping <- stoppedChan
	<-stoppedChan // Wait for the stop to have happened

	if c.conn != nil {
		_ = c.conn.Close() // Ignore error from close.
	}

	c.conn = nil
	c.blueprintIngestor = nil
	c.gitHubStatsIngestor = nil

	log.Info("Stopped background loop")
	return
}
