package cmd

import (
	"context"
	"fmt"
	"log"
	"time"

	"code.justin.tv/eventbus/express/msgfilter"
	"code.justin.tv/eventbus/express/parser"
	"code.justin.tv/eventbus/express/sqsutil"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/arn"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
	"github.com/pkg/errors"
	"github.com/spf13/cobra"
)

var (
	action        string
	number        uint64
	inputQueueURL string
	noDLQ         bool
	expression    string
	eventType     string
	interactive   bool

	// TODO
	visibilityTimeout time.Time
	wait              bool
)

const (
	actionInteractive = "interactive" // when action is omitted
	actionRequeue     = "requeue"
	actionList        = "list"
	actionDelete      = "delete"
)

func init() {
	rootCmd.AddCommand(receiveCmd)

	receiveCmd.Flags().StringVarP(&action, "action", "a", "", "action to take on received messages (options: list, requeue, delete; default: list)")
	receiveCmd.Flags().Uint64VarP(&number, "number", "n", 25, "the first N messages to take action on")
	receiveCmd.Flags().StringVarP(&inputQueueURL, "queue-url", "u", "", "the queue to operate on")
	receiveCmd.Flags().BoolVar(&noDLQ, "no-dlq", false, "when enabled, reads direct from provided queue instead of its deadletter queue")
	receiveCmd.Flags().BoolVarP(&interactive, "interactive", "i", false, "When enabled, asks for action on per-message basis")
	receiveCmd.Flags().StringVarP(&expression, "expr", "e", "", "filter expression")
	receiveCmd.Flags().StringVarP(&eventType, "event-type", "t", "", "when provided, restrict matches to this event type and allow filters to filter on event properties")
	receiveCmd.Flags().BoolVarP(&wait, "wait", "w", false, "whether or not to wait for more messages in the case that the queue is empty")

}

var receiveCmd = &cobra.Command{
	Use:   "receive",
	Short: "Iterates through dead letter queue messages and takes action on them",
	Long: `
Allows for a variety of actions to be taken on the queue in question.

Actions:
	- 'requeue' will replay the event onto the source queue
	- 'skip' will do nothing, only display the message in the output
	- 'delete' will delete the event from the dead letter queue

If no action is provided, interactive mode will be enabled, and every event will prompt for an explicit action.

Use --no-dlq to indicate that messages should be pulled directly from the main queue and not the dead letter queue. When using this mode, the action 'requeue' is not allowed.`,
	Run: func(cmd *cobra.Command, args []string) {
		ctx := context.Background()
		queueURL := inputQueueURL // reassign so we can maintain both original queue and DLQ, if necessary

		if queueURL == "" {
			log.Fatal("--queue-url is a required parameter")
		}

		if action == actionRequeue && noDLQ {
			log.Fatal("Cannot specify action 'requeue' and --no-dlq")
		}

		if action != "" && interactive {
			log.Fatal("Cannot use --action [action] and --interactive flags simultaneously")
		} else if interactive {
			action = actionInteractive
		}

		if action == "" {
			action = actionList
		}

		var f *msgfilter.MessageFilter
		if expression != "" {
			var err error
			f, err = msgfilter.NewFilter(expression, eventType)
			if err != nil {
				log.Fatalf("Error parsing message filter: %s", err.Error())
			}
		}

		// TODO: region config support
		sess := session.Must(session.NewSession(aws.NewConfig().WithRegion("us-west-2")))

		sqsClient := sqs.New(sess)

		if !noDLQ {
			output, err := sqsClient.GetQueueAttributesWithContext(ctx, &sqs.GetQueueAttributesInput{
				AttributeNames: aws.StringSlice([]string{sqsutil.KeyRedrivePolicy}),
				QueueUrl:       aws.String(queueURL),
			})
			if err != nil {
				log.Fatal("Error fetching deadletter queue information: ", err)
			}

			redrivePolicy, err := sqsutil.ParseRedrivePolicy(aws.StringValue(output.Attributes[sqsutil.KeyRedrivePolicy]))
			if err != nil {
				log.Fatal("Error parsing queue redrive policy: ", err)
			}

			arnParts, err := arn.Parse(redrivePolicy.DeadletterTargetARN)
			if err != nil {
				log.Fatal("Error parsing deadletter queue ARN: ", err)
			}

			dlqOutput, err := sqsClient.GetQueueUrlWithContext(ctx, &sqs.GetQueueUrlInput{
				QueueName: aws.String(arnParts.Resource),
			})
			if err != nil {
				log.Fatal("Error fetching deadletter queue URL: ", err)
			}

			queueURL = aws.StringValue(dlqOutput.QueueUrl)
		}

		var actionTaker actionTaker
		if action == actionInteractive {
			actionTaker = &actionInteractor{
				queueURL:           inputQueueURL,
				deadletterQueueURL: queueURL,
				sqsClient:          sqsClient,
				noDLQ:              noDLQ,
			}
		} else if action == actionList {
			actionTaker = &actionNooper{}
		} else if action == actionDelete {
			confirmDelete(queueURL, number)
			actionTaker = &actionDeleter{
				queueURL:  queueURL,
				sqsClient: sqsClient,
			}
		} else if action == actionRequeue {
			confirmRequeue(inputQueueURL, queueURL, number)
			actionTaker = &actionRequeuer{
				queueURL:           inputQueueURL,
				deadletterQueueURL: queueURL,
				sqsClient:          sqsClient,
			}

		} else {
			log.Fatal("Unsupported action:", action)
		}

		processed := uint64(0)

	ProcessLoop:
		for {
			receiveOutput, err := sqsClient.ReceiveMessageWithContext(ctx, &sqs.ReceiveMessageInput{
				QueueUrl:              aws.String(queueURL),
				AttributeNames:        aws.StringSlice([]string{"All"}),
				MessageAttributeNames: aws.StringSlice([]string{"All"}),
				MaxNumberOfMessages:   aws.Int64(10),
				WaitTimeSeconds:       aws.Int64(5),
			})
			if err != nil {
				log.Fatal("Error fetching messages from queue: ", err)
			}

			if len(receiveOutput.Messages) == 0 {
				if !wait {
					log.Println("Ran out of messages to process")
					break ProcessLoop
				} else {
					log.Println("No messages available yet. Waiting...")
					continue
				}

			}

			for _, msg := range receiveOutput.Messages {
				parseResult, err := parser.Parse(msg)
				if err != nil {
					log.Fatal("Failed to parse SQS message: ", err)
				}

				if eventType != "" && parseResult.EventBusType != eventType {
					log.Printf("Skipping, event type %+v != %+v", parseResult.EventBusType, eventType)
					continue
				}

				if f != nil {
					result, err := f.FilterOne(parseResult)
					if err != nil {
						log.Fatalf("error running filter: %s", err.Error())
					}
					if !result {
						log.Print("Skipping due to filter")
						continue
					}
				}

				log.Println("Message", processed+1, "of", number)
				log.Println("===========================")

				log.Println(parseResult.String())

				err = actionTaker.Do(ctx, parseResult)
				if err != nil {
					log.Fatal("Error handling SQS message: ", err)
				}

				log.Println("")

				processed += 1
				if processed >= number {
					break ProcessLoop
				}
			}
		}

	},
}

type actionTaker interface {
	Do(context.Context, *parser.ParseResult) error
}

type actionDeleter struct {
	queueURL  string
	sqsClient sqsiface.SQSAPI
}

func (a *actionDeleter) Do(ctx context.Context, pr *parser.ParseResult) error {
	_, err := a.sqsClient.DeleteMessageWithContext(ctx, &sqs.DeleteMessageInput{
		QueueUrl:      aws.String(a.queueURL),
		ReceiptHandle: pr.SQS.ReceiptHandle,
	})
	if err != nil {
		return nil
	}

	log.Println("Message successfully deleted")

	return nil
}

type actionRequeuer struct {
	queueURL           string
	deadletterQueueURL string
	sqsClient          sqsiface.SQSAPI
}

func (a *actionRequeuer) Do(ctx context.Context, pr *parser.ParseResult) error {
	_, err := a.sqsClient.SendMessageWithContext(ctx, &sqs.SendMessageInput{
		QueueUrl:          aws.String(a.queueURL),
		MessageAttributes: pr.SQS.MessageAttributes,
		MessageBody:       pr.SQS.Body,
	})
	if err != nil {
		return errors.Wrap(err, "could not requeue sqs message")
	}

	_, err = a.sqsClient.DeleteMessageWithContext(ctx, &sqs.DeleteMessageInput{
		QueueUrl:      aws.String(a.deadletterQueueURL),
		ReceiptHandle: pr.SQS.ReceiptHandle,
	})
	if err != nil {
		return errors.Wrap(err, "could not delete message from deadletter queue")
	}

	log.Println("Message successfully requeued")

	return err
}

type actionNooper struct{}

func (a *actionNooper) Do(ctx context.Context, pr *parser.ParseResult) error {
	return nil
}

// actionInteractor asks the user which of the above actions to take and acts accordingly
type actionInteractor struct {
	queueURL           string
	deadletterQueueURL string
	sqsClient          sqsiface.SQSAPI
	noDLQ              bool
}

func (a *actionInteractor) Do(ctx context.Context, pr *parser.ParseResult) error {
	var prompt string
	if noDLQ {
		prompt = `
Actions:
	(s) Skip
	(d) Delete

`
	} else {
		prompt = `
Actions:
	(r) Requeue
	(s) Skip
	(d) Delete
		 
`
	}

	log.Println(prompt)

	var actionChoice actionTaker

	for {
		fmt.Print("Choice: ")
		// choice, _ := reader.ReadString('\n')
		// choice = strings.TrimSpace(choice)
		var choice string
		_, err := fmt.Scanf("%s", &choice)
		if err != nil {
			log.Fatal("Error getting user input: ", err)
		}

		switch choice {
		case "r":
			if !noDLQ { // cannot requeue when acting on a non-DLQ
				actionChoice = &actionRequeuer{
					queueURL:           a.queueURL,
					deadletterQueueURL: a.deadletterQueueURL,
					sqsClient:          a.sqsClient,
				}
			}
		case "s":
			actionChoice = &actionNooper{}
		case "d":
			var queueURL string
			if noDLQ {
				queueURL = a.queueURL
			} else {
				queueURL = a.deadletterQueueURL
			}
			actionChoice = &actionDeleter{
				queueURL:  queueURL,
				sqsClient: a.sqsClient,
			}
		}

		if actionChoice != nil {
			break
		}

		log.Println("Invalid option...")
		log.Println("")
	}

	return actionChoice.Do(ctx, pr)
}

func confirmDelete(queueURL string, n uint64) {
	fmt.Printf(`
WARNING! You are about to delete up to %d messages from the following queue: %s\n`,
		n, queueURL)
	confirm()
}

func confirmRequeue(mainURL, dlqURL string, n uint64) {
	fmt.Printf(`
WARNING! You are about to requeue up to %d messages from a deadletter queue to its primary queue.
Primary queue URL:    %s
Deadletter queue URL: %s
`, n, mainURL, dlqURL)
	confirm()
}

func confirm() {
	for {
		fmt.Print("Proceed? (y/n): ")
		var choice string
		_, err := fmt.Scanf("%s", &choice)
		if err != nil {
			log.Fatal("Error getting user input: ", err)
		}
		switch choice {
		case "y":
			return
		case "n":
			log.Fatal("Aborting.")
		}
	}
}
