package gdpr

import (
	"context"
	"encoding/csv"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"sync"
	"time"

	"code.justin.tv/foundation/history-service/configuration"
	"code.justin.tv/foundation/history-service/internal/awssig"
	"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/dynamodbattribute"
	"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/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3iface"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/aws/aws-sdk-go/service/s3/s3manager/s3manageriface"
	"github.com/sirupsen/logrus"
	"golang.org/x/sync/errgroup"
	elastic "gopkg.in/olivere/elastic.v5"
)

const writeBatchSize = 25

type queryOutputHandler func(context.Context, *dynamodb.QueryOutput) error

// GenerateReportOutput is a struct represents the response of GenerateReport
type GenerateReportOutput struct {
	UserID     string
	Timestamp  time.Time
	Key        string
	Bucket     string
	Expiration time.Time
}

// DDBPartitionKey parition key for audits ddb table.
type DDBPartitionKey struct {
	UserID          string `dynamodbav:"user_id"`
	ActionCreatedAt string `dynamodbav:"action_created_at"`
}

// UserReportItem represent row exported in the user report
type UserReportItem struct {
	UserID      string `dynamodbav:"user_id"`
	CreatedAt   int64  `dynamodbav:"created_at"`
	Action      string `dynamodbav:"action"`
	Description string `dynamodbav:"description"`
}

// ClientAPI represent gdpr client is expected to implement
type ClientAPI interface {
	DeleteAudits(ctx context.Context, userID string) error
	GenerateUserReport(ctx context.Context, userID string) (*GenerateReportOutput, error)
}

// Client implements ClientAPI and provides gdpr functionality
type Client struct {
	Environment string
	Logger      logrus.FieldLogger

	ddbClient          dynamodbiface.DynamoDBAPI
	ddbAuditsTableName string

	esClient *elastic.Client

	s3             s3iface.S3API
	uploader       s3manageriface.UploaderAPI
	uploadLocation string

	initOnce  sync.Once
	initError error
}

func (c *Client) init() error {

	c.initOnce.Do(func() {
		if c.Environment == "" {
			c.Environment = "prod"
		}

		if c.Logger == nil {
			lgr := logrus.New()
			lgr.Formatter = new(logrus.JSONFormatter)
			c.Logger = lgr
		}

		sess := session.Must(session.NewSession(&aws.Config{
			Region: aws.String("us-west-2"),
		}))

		c.ddbClient = dynamodb.New(sess)
		c.s3 = s3.New(sess)
		c.ddbAuditsTableName = configuration.Resolve("auditsTableName")

		c.uploader = s3manager.NewUploaderWithClient(c.s3, func(u *s3manager.Uploader) {
			u.MaxUploadParts = 1
		})

		c.uploadLocation = configuration.Resolve("userReportsS3Bucket")

		if configuration.Resolve("elasticSearchUrl") == "" {
			c.initError = fmt.Errorf("elasticSearchURL is not set in the config for env: %s", c.Environment)
			return
		}

		c.esClient, c.initError = elastic.NewSimpleClient(
			elastic.SetURL(configuration.Resolve("elasticSearchUrl")),
			elastic.SetHttpClient(&http.Client{
				Transport: &awssig.RoundTripper{
					AWSService:  "es",
					Credentials: sess.Config.Credentials,
				},
			}),
		)
	})
	return c.initError
}

// DeleteAudits delete user data from dyanamodb and elastic search
func (c *Client) DeleteAudits(ctx context.Context, userID string) error {
	if err := c.init(); err != nil {
		return err
	}

	if userID == "" {
		return errors.New("please provide user id to delete data")
	}

	var g errgroup.Group

	//Delete data from elastic search
	g.Go(func() error {
		c.Logger.Info("delete audits from elastic search auditor index")
		return c.deleteAuditsFromElasticSearch(ctx, "auditor*", userID)
	})

	g.Go(func() error {
		c.Logger.Info("delete audits from elastic search history index")
		return c.deleteAuditsFromElasticSearch(ctx, "history*", userID)
	})

	// delete data from dynamoDB
	g.Go(func() error {
		c.Logger.Info("delete audits from dynamodb")
		return c.deleteAuditsFromDynamoDB(ctx, userID)
	})

	// Wait for all go routiner to complete.
	return g.Wait()
}

// GenerateUserReport ....
func (c *Client) GenerateUserReport(ctx context.Context, userID string) (*GenerateReportOutput, error) {

	if err := c.init(); err != nil {
		return nil, err
	}

	keyCondition := expression.Key("user_id").Equal(expression.Value(userID))
	proj := expression.NamesList(
		expression.Name("user_id"),
		expression.Name("created_at"),
		expression.Name("action"),
		expression.Name("description"),
	)
	expr, err := expression.NewBuilder().
		WithKeyCondition(keyCondition).
		WithProjection(proj).
		Build()

	if err != nil {
		return nil, err
	}

	reportTime := time.Now()
	key := fmt.Sprintf("report-%s-%d.csv", userID, reportTime.Unix())

	g, ctx := errgroup.WithContext(ctx)

	rPipe, wPipe := io.Pipe()

	w := csv.NewWriter(wPipe)

	handler := func(ctx context.Context, output *dynamodb.QueryOutput) error {
		for _, item := range output.Items {
			i := new(UserReportItem)
			err := dynamodbattribute.UnmarshalMap(item, i)
			if err != nil {
				return err
			}
			record := []string{
				i.UserID,
				i.Action,
				i.Description,
				fmt.Sprintf("%d", i.CreatedAt),
			}
			if err := w.Write(record); err != nil {
				return err
			}
		}
		return nil
	}

	// Read data from dynamodb and write to pipe
	// returning named error so that err can be modified in defer
	// to return appropriate error message
	g.Go(func() (err error) {

		// bubble error if we failed to w.Flush or wPipe.Close
		defer func() {

			// flush content
			w.Flush()
			if fErr := w.Error(); fErr != nil {
				if err != nil {
					c.Logger.Errorf("failed to flush data to file. err: %s", fErr.Error())
				} else {
					err = fErr
				}
			}

			// Signal reader with io.EOF
			if wErr := wPipe.Close(); wErr != nil {
				if err != nil {
					c.Logger.Errorf("failed to close pipe, err: %s", wErr.Error())
				} else {
					err = wErr
				}
			}
		}()

		header := []string{
			"UserID",
			"Action",
			"Description",
			"Timestamp",
		}
		if err := w.Write(header); err != nil {
			return err
		}

		return c.queryDynamodb(ctx, expr, handler)
	})

	g.Go(func() error {
		_, err := c.uploader.UploadWithContext(ctx, &s3manager.UploadInput{
			Bucket: aws.String(c.uploadLocation),
			Key:    aws.String(key),
			Body:   rPipe,
		})
		return err
	})

	if err := g.Wait(); err != nil {
		return nil, err
	}

	// Reports will only be available for 7 days, after which the signed url will not work.
	ttl := time.Hour * 24 * 7

	return &GenerateReportOutput{
		UserID:     userID,
		Timestamp:  reportTime,
		Bucket:     c.uploadLocation,
		Key:        key,
		Expiration: reportTime.Add(ttl),
	}, nil
}

func (c *Client) deleteAuditsFromDynamoDB(ctx context.Context, userID string) error {

	keyCondition := expression.Key("user_id").Equal(expression.Value(userID))
	proj := expression.NamesList(expression.Name("user_id"), expression.Name("action_created_at"))
	expr, err := expression.NewBuilder().
		WithKeyCondition(keyCondition).
		WithProjection(proj).
		Build()
	if err != nil {
		return err
	}

	return c.queryDynamodb(ctx, expr, c.batchDeleteItems)
}

func (c *Client) batchDeleteItems(ctx context.Context, output *dynamodb.QueryOutput) error {

	var keys []*DDBPartitionKey
	for _, item := range output.Items {
		k := new(DDBPartitionKey)
		err := dynamodbattribute.UnmarshalMap(item, k)
		if err != nil {
			return nil
		}
		keys = append(keys, k)
	}

	for i := 0; i < len(keys); i += writeBatchSize {
		end := i + writeBatchSize
		if end > len(keys) {
			end = len(keys)
		}
		chunk := keys[i:end]

		var writeRequests []*dynamodb.WriteRequest
		for _, item := range chunk {
			wr := &dynamodb.WriteRequest{
				DeleteRequest: &dynamodb.DeleteRequest{
					Key: map[string]*dynamodb.AttributeValue{
						"user_id": {
							S: aws.String(item.UserID),
						},
						"action_created_at": {
							S: aws.String(item.ActionCreatedAt),
						},
					},
				},
			}
			writeRequests = append(writeRequests, wr)
		}

		wr := map[string][]*dynamodb.WriteRequest{
			c.ddbAuditsTableName: writeRequests,
		}
		for {
			if len(wr) == 0 {
				break
			}

			input := &dynamodb.BatchWriteItemInput{
				RequestItems: wr,
			}

			result, err := c.ddbClient.BatchWriteItemWithContext(ctx, input)
			if err != nil {
				return err
			}
			wr = result.UnprocessedItems
		}
	}
	return nil
}

func (c *Client) deleteAuditsFromElasticSearch(ctx context.Context, indexName string, userID string) error {

	//Delete audits docs from index
	query := elastic.NewQueryStringQuery(userID)
	query = query.Field("user_id")
	source, err := query.Source()
	if err != nil {
		return err
	}
	c.Logger.Infof("running delete query: '%s' on index name: %s", source, indexName)

	out, err := c.esClient.DeleteByQuery().
		ScrollSize(1000).
		Index(indexName).
		Type("audits").
		Query(query).
		Conflicts("proceed").
		Do(ctx)

	if err != nil {
		return err
	}
	// len(out.Failures) should always be zero,
	// conflicts should not abort the opertion.
	if len(out.Failures) != 0 {
		bs, err := json.Marshal(out.Failures)
		if err != nil {
			return err
		}
		return fmt.Errorf("failure reported by elastic search delete by query, err: %s", string(bs))
	}

	c.Logger.Infof("finished deleteing document from index '%s' for user '%v', from elastic search, response: %+v", indexName, userID, out)
	return nil
}

func (c *Client) queryDynamodb(ctx context.Context, expr expression.Expression, handler queryOutputHandler) error {

	input := &dynamodb.QueryInput{
		TableName:                 aws.String(c.ddbAuditsTableName),
		ExpressionAttributeValues: expr.Values(),
		ExpressionAttributeNames:  expr.Names(),
		KeyConditionExpression:    expr.KeyCondition(),
		ProjectionExpression:      expr.Projection(),
	}

	for {
		var queryOutput *dynamodb.QueryOutput
		queryOutput, err := c.ddbClient.QueryWithContext(ctx, input)
		if err != nil {
			return err
		}
		err = handler(ctx, queryOutput)
		if err != nil {
			return err
		}

		if queryOutput.LastEvaluatedKey == nil {
			break
		} else {
			input.ExclusiveStartKey = queryOutput.LastEvaluatedKey
		}
	}

	return nil
}
