package event

import (
	"context"
	"encoding/base64"
	"fmt"
	"strconv"
	"strings"

	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/dynamodb"
	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
	"github.com/golang/protobuf/proto"
	"github.com/pkg/errors"

	"code.justin.tv/common/config"

	"code.justin.tv/dta/rockpaperscissors/internal/api/contextlogger"
	pb "code.justin.tv/dta/rockpaperscissors/proto"
)

// Datastore is a simple interface for an Event datastore.
type Datastore interface {
	Put(context.Context, *pb.Event) error
	Get(context.Context, []byte) (*pb.Event, error)
	Query(context.Context, *pb.EventDatastoreQuery) (*pb.EventDatastoreQueryResults, error)
}

// DynamoDBDatastore is an implementation of Datastore using AWS DynamoDB.
type DynamoDBDatastore struct {
	dynamoDBSvc   dynamodbiface.DynamoDBAPI
	dynamoDBTable string
	EnableWrites  bool
}

type eventRecord struct {
	UUID       []byte            `dynamodbav:"uuid"`
	Timestamp  float64           `dynamodbav:"timestamp"`
	Type       string            `dynamodbav:"type"`
	Body       []byte            `dynamodbav:"body"`
	Attributes map[string]string `dynamodbav:"attributes"`
}

type queryValues struct {
	StartTime int64  `dynamodbav:":starttime"`
	EndTime   int64  `dynamodbav:":endtime"`
	Type      string `dynamodbav:":type,omitempty"`
}

func init() {
	config.Register(map[string]string{
		"enable-events-datastore-writes": "true",
	})
}

// NewDynamoDBDatastore constructs a configured DynamoDBDatastore.
func NewDynamoDBDatastore(sess client.ConfigProvider, table string) *DynamoDBDatastore {
	enableWrites, err := strconv.ParseBool(
		config.MustResolve("enable-events-datastore-writes"))
	if err != nil {
		log.Fatalf("Can't parse enable-events-datastore-writes config: %v", err)
	}

	return &DynamoDBDatastore{
		dynamoDBSvc:   dynamodb.New(sess),
		dynamoDBTable: table,
		EnableWrites:  enableWrites,
	}
}

// Put the event into the datastore.
func (d *DynamoDBDatastore) Put(ctx context.Context, event *pb.Event) error {
	item, err := d.marshalEventToItem(event)
	if err != nil {
		return err
	}

	params := &dynamodb.PutItemInput{
		Item:      item,
		TableName: aws.String(d.dynamoDBTable),
	}

	if d.EnableWrites {
		if _, err = d.dynamoDBSvc.PutItemWithContext(ctx, params); err != nil {
			return err
		}
	}

	return nil
}

// Get an event from the datastore.
func (d *DynamoDBDatastore) Get(ctx context.Context, uuid []byte) (*pb.Event, error) {
	av, err := dynamodbattribute.Marshal(uuid)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to marshal uuid to datastore key")
	}

	resp, err := d.dynamoDBSvc.QueryWithContext(ctx, &dynamodb.QueryInput{
		TableName:              aws.String(d.dynamoDBTable),
		KeyConditionExpression: aws.String("#U = :uuid"),
		ProjectionExpression:   aws.String("#U, body"),
		ExpressionAttributeNames: map[string]*string{
			"#U": aws.String("uuid"),
		},
		ExpressionAttributeValues: map[string]*dynamodb.AttributeValue{":uuid": av},
	})
	if err != nil {
		return nil, errors.Wrap(err, "Failed to GetItem from DynamoDB table")
	}

	if len(resp.Items) == 0 {
		return nil, ErrEventNotFound
	}

	event, err := d.unmarshalItemToEvent(resp.Items[0])
	if err != nil {
		return nil, errors.Wrap(err, "Error unmarshaling DynamoDb item to Event")
	}

	return event, nil
}

// Query for events from the datastore.
func (d *DynamoDBDatastore) Query(ctx context.Context, q *pb.EventDatastoreQuery) (*pb.EventDatastoreQueryResults, error) {
	av, err := dynamodbattribute.MarshalMap(&queryValues{
		StartTime: q.StartSeconds,
		EndTime:   q.EndSeconds,
		Type:      q.Type,
	})
	if err != nil {
		return nil, errors.Wrap(err, "Failed to marshal query values")
	}
	params := &dynamodb.QueryInput{
		TableName:            aws.String(d.dynamoDBTable),
		IndexName:            aws.String("EventTypeIndex"),
		ProjectionExpression: aws.String("#U, #B"),
		ExpressionAttributeNames: map[string]*string{
			"#U":  aws.String("uuid"),
			"#TS": aws.String("timestamp"),
			"#T":  aws.String("type"),
			"#B":  aws.String("body"),
		},
		KeyConditionExpression: aws.String(
			"#T = :type AND #TS BETWEEN :starttime AND :endtime"),
		ExpressionAttributeValues: av,
		ReturnConsumedCapacity:    aws.String("TOTAL"),
	}

	var filterExpressions []string
	var i int
	for k, v := range q.Filter {
		filterExpressions = append(filterExpressions, fmt.Sprintf("#A.#A%d = :%d", i, i))
		params.ExpressionAttributeNames[fmt.Sprintf("#A%d", i)] = aws.String(k)
		params.ExpressionAttributeValues[fmt.Sprintf(":%d", i)] = (&dynamodb.AttributeValue{}).SetS(v)
		i++
	}
	if len(filterExpressions) > 0 {
		params.ExpressionAttributeNames["#A"] = aws.String("attributes")
		params.FilterExpression = aws.String(
			strings.Join(filterExpressions, " AND "))
	}

	results := &pb.EventDatastoreQueryResults{}
	var innerError error
	err = d.dynamoDBSvc.QueryPagesWithContext(ctx, params,
		func(queryResponse *dynamodb.QueryOutput, lastPage bool) bool {
			// TODO: log this error?
			_ = contextlogger.IncContextLogField(ctx, "event_queries")
			_ = contextlogger.AddContextLogField(
				ctx, "event_query_consumed_capacity", *queryResponse.ConsumedCapacity.CapacityUnits)
			for _, item := range queryResponse.Items {
				var event *pb.Event
				event, innerError = d.unmarshalItemToEvent(item)
				if innerError != nil {
					innerError = errors.Wrap(err,
						"Error unmarshaling DynamoDb item to Event")
					return false
				}
				results.Events = append(results.Events, event)
			}
			return true
		})
	if err != nil {
		return nil, errors.Wrap(err, "Failed to query DynamoDB table")
	}
	if innerError != nil {
		return nil, innerError
	}

	return results, nil
}

func (d *DynamoDBDatastore) marshalEventToItem(event *pb.Event) (map[string]*dynamodb.AttributeValue, error) {
	data, err := proto.Marshal(event)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to marshal item")
	}
	record := eventRecord{
		UUID:       event.GetUuid(),
		Timestamp:  event.GetTimestamp(),
		Type:       event.GetType(),
		Body:       data,
		Attributes: make(map[string]string),
	}
	for _, attribute := range event.GetAttributes() {
		record.Attributes[attribute.GetKey()] = attribute.GetValue()
	}
	item, err := dynamodbattribute.MarshalMap(record)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to marshal item")
	}

	// Work around too-short floats and not matching the existing timestamps in
	// dynamodb that were created in Python.
	// This is due to how Go marshals floats into JSON in encode/json.
	// See also https://go-review.googlesource.com/#/c/30371/ and the issues
	// linked from there.
	// TODO: re-architect timestamp to use a more canonical format.
	f := strconv.FormatFloat(record.Timestamp, 'f', 53, 64)
	item["timestamp"].SetN(strings.TrimRight(f, "0"))

	return item, nil
}

func (d *DynamoDBDatastore) unmarshalItemToEvent(item map[string]*dynamodb.AttributeValue) (*pb.Event, error) {
	record := eventRecord{}
	err := dynamodbattribute.UnmarshalMap(item, &record)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to unmarshal item")
	}

	event := &pb.Event{}
	err = proto.Unmarshal(record.Body, event)
	if err != nil {
		encoded := base64.StdEncoding.EncodeToString(record.UUID)
		return nil, errors.Wrapf(err, "Failed to unmarshal item with uuid '%s'", encoded)
	}

	return event, nil
}

// Delete an event from the datastore.
func (d *DynamoDBDatastore) Delete(ctx context.Context, uuid []byte) error {
	av, err := dynamodbattribute.Marshal(uuid)
	if err != nil {
		return errors.Wrap(err, "Failed to marshal uuid to datastore key")
	}

	// We have to do a query in order to find out the timestamp which is
	// part of the key so we need that in order to delete an item.
	resp, err := d.dynamoDBSvc.QueryWithContext(ctx, &dynamodb.QueryInput{
		TableName:              aws.String(d.dynamoDBTable),
		KeyConditionExpression: aws.String("#U = :uuid"),
		ProjectionExpression:   aws.String("#ts"),
		ExpressionAttributeNames: map[string]*string{
			"#U":  aws.String("uuid"),
			"#ts": aws.String("timestamp"),
		},
		ExpressionAttributeValues: map[string]*dynamodb.AttributeValue{
			":uuid": av,
		},
	})
	if err != nil {
		return errors.Wrap(err, "Failed to GetItem from DynamoDB table")
	}

	if len(resp.Items) == 0 {
		return ErrEventNotFound
	}

	if d.EnableWrites {
		_, err = d.dynamoDBSvc.DeleteItemWithContext(ctx,
			&dynamodb.DeleteItemInput{
				TableName: aws.String(d.dynamoDBTable),
				Key: map[string]*dynamodb.AttributeValue{
					"uuid":      av,
					"timestamp": resp.Items[0]["timestamp"],
				},
			})
		if err != nil {
			return errors.Wrap(err, "Failed to DeleteItem in DynamoDB table")
		}
	}
	return nil
}
