package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"time"

	"code.justin.tv/foundation/history-service/internal/gdpr"
	"code.justin.tv/foundation/history-service/internal/queue"
	"code.justin.tv/foundation/history-service/rpc/history"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/dynamodb"
	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
	"github.com/aws/aws-sdk-go/service/dynamodb/expression"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
)

func main() {
	region, ok := os.LookupEnv("REGION")
	if !ok {
		log.Fatal("environment variable 'REGION' is not set")
	}

	environment, ok := os.LookupEnv("ENVIRONMENT")
	if !ok {
		log.Fatal("environment variable 'ENVIRONMENT' is not set")
	}

	queueURL, ok := os.LookupEnv("QUEUE_URL")
	if !ok {
		log.Fatal("environment variable 'QUEUE_URL' is not set")
	}

	reportStatusTableName, ok := os.LookupEnv("REPORT_STATUS_TABLE_NAME")
	if !ok {
		log.Fatal("environment variable 'REPORT_STATUS_TABLE_NAME' is not set")
	}

	sess := session.Must(session.NewSession(&aws.Config{
		Region: aws.String(region),
	}))

	gr := &generateReport{
		queue:    sqs.New(sess),
		queueURL: queueURL,
		gdprClient: &gdpr.Client{
			Environment: environment,
		},
		ddb:                   dynamodb.New(sess),
		reportStatusTableName: reportStatusTableName,
	}
	err := gr.GenerateWithContext(context.Background())
	if err != nil {
		log.Fatal(err)
	}

	log.Println("finished processsing generate report requests")
}

type generateReport struct {
	ddb                   dynamodbiface.DynamoDBAPI
	reportStatusTableName string
	queue                 sqsiface.SQSAPI
	gdprClient            gdpr.ClientAPI
	queueURL              string
}

func (gr *generateReport) GenerateWithContext(ctx context.Context) error {

	for {
		log.Printf("retrieving generate report request from queue: %s", gr.queueURL)
		resp, err := gr.queue.ReceiveMessageWithContext(
			ctx,
			&sqs.ReceiveMessageInput{
				QueueUrl:            aws.String(gr.queueURL),
				MaxNumberOfMessages: aws.Int64(1),
				WaitTimeSeconds:     aws.Int64(20),
			},
		)
		if err != nil {
			return err
		}

		if len(resp.Messages) == 0 {
			log.Printf("no message found in sqs: %s, exiting.... \n", gr.queueURL)
			return nil
		}

		sqsMessage := resp.Messages[0]
		msg := &queue.GenerateUserReportMessage{}
		err = json.Unmarshal([]byte(*sqsMessage.Body), msg)
		if err != nil {
			return errors.New("error parsing SQS message: " + err.Error())
		}

		log.Printf("generating report for user: %s. this may take a while", msg.UserID)
		result, err := gr.gdprClient.GenerateUserReport(ctx, msg.UserID)
		if err != nil {
			return err
		}

		if result.Expiration.IsZero() {
			return errors.New("Cannot store user report without expiration time")
		}

		// set status to available
		log.Println("updating report status to 'available'")
		err = gr.updateReportStatus(ctx, &updateStatusReportInput{
			UserID:         msg.UserID,
			Timestamp:      msg.Timestamp,
			ExpirationTime: result.Expiration,
			Status:         int(history.Status_Available),
			Key:            result.Key,
			Bucket:         result.Bucket,
		})
		if err != nil {
			return err
		}

		//delete the message when done.
		log.Println("deleting processed request from queue.")
		_, err = gr.queue.DeleteMessageWithContext(ctx, &sqs.DeleteMessageInput{
			ReceiptHandle: sqsMessage.ReceiptHandle,
			QueueUrl:      aws.String(gr.queueURL),
		})
		if err != nil {
			return fmt.Errorf("Error deleting SQS Message(%s): %s", sqsMessage, err.Error())
		}
	}
}

type updateStatusReportInput struct {
	UserID         string    //user id for which report is being generated
	Timestamp      time.Time // report request time
	ExpirationTime time.Time // Report will not longer be available after this time
	Status         int       // What is the current status of report `requested`, available, etc.
	Bucket         string
	Key            string
}

func (gr *generateReport) updateReportStatus(ctx context.Context, input *updateStatusReportInput) error {

	update := expression.Set(expression.Name("updated_at"), expression.Value(time.Now().Unix()))

	taskMetadata, err := gr.getTaskMetadata()
	if err != nil {
		return err
	}
	update.Set(expression.Name("ecs-task-metadata"), expression.Value(taskMetadata)).
		Set(expression.Name("status"), expression.Value(input.Status)).
		Set(expression.Name("expires_at"), expression.Value(input.ExpirationTime.Unix()))

	update = update.
		Set(expression.Name("bucket"), expression.Value(input.Bucket)).
		Set(expression.Name("key"), expression.Value(input.Key))

	expr, err := expression.NewBuilder().
		WithUpdate(update).
		Build()
	if err != nil {
		return err
	}

	uiInput := &dynamodb.UpdateItemInput{
		TableName:                 aws.String(gr.reportStatusTableName),
		ExpressionAttributeValues: expr.Values(),
		ExpressionAttributeNames:  expr.Names(),
		UpdateExpression:          expr.Update(),
		Key: map[string]*dynamodb.AttributeValue{
			"user_id":   {S: aws.String(input.UserID)},
			"timestamp": {N: aws.String(fmt.Sprintf("%d", input.Timestamp.Unix()))},
		},
	}

	_, err = gr.ddb.UpdateItemWithContext(ctx, uiInput)
	return err
}

func (gr *generateReport) getTaskMetadata() (string, error) {

	taskMetadatURL, ok := os.LookupEnv("ECS_CONTAINER_METADATA_URI")
	if !ok {
		return "", errors.New("ecs container not configured with container metadata. env ECS_CONTAINER_METADATA_URI is unset")
	}

	resp, err := http.Get(taskMetadatURL)
	if err != nil {
		return "", fmt.Errorf("failed to get task metadata, uri: %s, err: %s", taskMetadatURL, err.Error())
	}

	defer func() {
		err := resp.Body.Close()
		if err != nil {
			log.Println("failed to close metadata response body, err: ", err.Error())
		}
	}()
	metadata, err := ioutil.ReadAll(resp.Body)
	return string(metadata), err
}
