package localdynamo

import (
	"time"

	"code.justin.tv/creator-collab/log"
	"code.justin.tv/creator-collab/log/errors"
	"github.com/aws/aws-sdk-go/aws/session"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"

	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/service/dynamodb"
	"github.com/aws/aws-sdk-go/service/dynamodb/expression"
)

const (
	// TableAutohostSettings is the name of the table used to store auto-hosting related settings.
	TableAutohostSettings = "AutohostSettings"

	// TableHost is the name of table used to store host state.
	TableHost = "Host-staging"

	tableCaches             = "Caches"
	tableChannels           = "Channels"
	tableFultonRaidSettings = "RaidSettings"
	tableFultonPastRaids    = "PastRaids"
	tableFultonRaids        = "Raids"
)

var tableNames = []string{
	tableCaches,
	tableChannels,
	tableFultonRaidSettings,
	tableFultonPastRaids,
	tableFultonRaids,
	TableAutohostSettings,
	TableHost,
}

// NewLocalDynamoDBClient creates a DynamoDB client that is configured to communicate with an instance
// of DynamoDB Local.
// This is meant to be used with local development and automated tests.
func NewLocalDynamoDBClient(localDynamoEndpoint string) (dynamodbiface.DynamoDBAPI, error) {
	// DynamoDB Local does not enforce security, but the AWS SDK requires that credentials are present.
	// We'll initialize the session with fake static credentials.
	creds := credentials.NewStaticCredentials("fake-id", "fake-secret", "fake-token")

	config := aws.NewConfig().
		WithRegion("us-west-2").
		WithCredentials(creds).
		WithEndpoint(localDynamoEndpoint)
	s, err := session.NewSession(config)
	if err != nil {
		return nil, errors.Wrap(err, "creating aws session failed")
	}

	return dynamodb.New(s), nil
}

// Tables facilitates creating and destroying the tables that AutoHost expects.
// This is meant to be used with local development and automated tests.
type Tables struct {
	client dynamodbiface.DynamoDBAPI
	logger log.Logger
}

// NewLocalDynamoTables creates a new instance of NewLocalDynamoTables configured to communicate with
// the given endpoint.
func NewLocalDynamoTables(localDynamoEndpoint string, logger log.Logger) (*Tables, error) {
	client, err := NewLocalDynamoDBClient(localDynamoEndpoint)
	if err != nil {
		return nil, err
	}

	return &Tables{
		client: client,
		logger: logger,
	}, nil
}

// Reset ensures the tables exist and are empty.
func (l *Tables) Reset() error {
	err := l.Create()
	if err != nil {
		return err
	}

	return l.TruncateTables()
}

// Create creates the tables that AutoHost requests.
// It skips creating tables that already exist.
func (l *Tables) Create() error {
	tableParams := map[string]*dynamodb.CreateTableInput{
		TableAutohostSettings: {
			TableName: aws.String(TableAutohostSettings),
			KeySchema: []*dynamodb.KeySchemaElement{
				{
					AttributeName: aws.String("ChannelID"),
					KeyType:       aws.String("HASH"),
				},
			},
			AttributeDefinitions: []*dynamodb.AttributeDefinition{
				{
					AttributeName: aws.String("ChannelID"),
					AttributeType: aws.String("S"),
				},
			},
			BillingMode: aws.String(dynamodb.BillingModePayPerRequest),
		},
		tableChannels: {
			TableName: aws.String(tableChannels),
			KeySchema: []*dynamodb.KeySchemaElement{
				{
					AttributeName: aws.String("ID"),
					KeyType:       aws.String("HASH"),
				},
				{
					AttributeName: aws.String("Resource"),
					KeyType:       aws.String("RANGE"),
				},
			},
			AttributeDefinitions: []*dynamodb.AttributeDefinition{
				{
					AttributeName: aws.String("ID"),
					AttributeType: aws.String("S"),
				},
				{
					AttributeName: aws.String("Resource"),
					AttributeType: aws.String("S"),
				},
			},
			ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
				ReadCapacityUnits:  aws.Int64(1000),
				WriteCapacityUnits: aws.Int64(1000),
			},
		},
		tableCaches: {
			TableName: aws.String(tableCaches),
			KeySchema: []*dynamodb.KeySchemaElement{
				{
					AttributeName: aws.String("Key"),
					KeyType:       aws.String("HASH"),
				},
				{
					AttributeName: aws.String("Bucket"),
					KeyType:       aws.String("RANGE"),
				},
			},
			AttributeDefinitions: []*dynamodb.AttributeDefinition{
				{
					AttributeName: aws.String("Key"),
					AttributeType: aws.String("S"),
				},
				{
					AttributeName: aws.String("Bucket"),
					AttributeType: aws.String("N"),
				},
			},
			ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
				ReadCapacityUnits:  aws.Int64(1000),
				WriteCapacityUnits: aws.Int64(1000),
			},
		},
		TableHost: {
			TableName: aws.String(TableHost),
			KeySchema: []*dynamodb.KeySchemaElement{
				{
					AttributeName: aws.String("ChannelID"),
					KeyType:       aws.String("HASH"),
				},
			},
			AttributeDefinitions: []*dynamodb.AttributeDefinition{
				{
					AttributeName: aws.String("ChannelID"),
					AttributeType: aws.String("S"),
				},
				{
					AttributeName: aws.String("HostingTargetID"),
					AttributeType: aws.String("S"),
				},
			},
			ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
				ReadCapacityUnits:  aws.Int64(1000),
				WriteCapacityUnits: aws.Int64(1000),
			},
			GlobalSecondaryIndexes: []*dynamodb.GlobalSecondaryIndex{
				{
					IndexName: aws.String("HostingTargetIDIndex"),
					KeySchema: []*dynamodb.KeySchemaElement{
						{
							AttributeName: aws.String("HostingTargetID"),
							KeyType:       aws.String("HASH"),
						},
					},
					Projection: &dynamodb.Projection{
						ProjectionType: aws.String("KEYS_ONLY"),
					},
					ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
						ReadCapacityUnits:  aws.Int64(1000),
						WriteCapacityUnits: aws.Int64(1000),
					},
				},
			},
		},
		tableFultonRaidSettings: {
			TableName: aws.String(tableFultonRaidSettings),
			KeySchema: []*dynamodb.KeySchemaElement{
				{
					AttributeName: aws.String("ChannelID"),
					KeyType:       aws.String("HASH"),
				},
			},
			AttributeDefinitions: []*dynamodb.AttributeDefinition{
				{
					AttributeName: aws.String("ChannelID"),
					AttributeType: aws.String("S"),
				},
			},
			ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
				ReadCapacityUnits:  aws.Int64(1000),
				WriteCapacityUnits: aws.Int64(1000),
			},
		},
		tableFultonPastRaids: {
			TableName: aws.String(tableFultonPastRaids),
			KeySchema: []*dynamodb.KeySchemaElement{
				{
					AttributeName: aws.String("RaidID"),
					KeyType:       aws.String("HASH"),
				},
			},
			AttributeDefinitions: []*dynamodb.AttributeDefinition{
				{
					AttributeName: aws.String("RaidID"),
					AttributeType: aws.String("S"),
				},
				{
					AttributeName: aws.String("TargetID"),
					AttributeType: aws.String("S"),
				},
				{
					AttributeName: aws.String("StartedAt"),
					AttributeType: aws.String("S"),
				},
			},
			GlobalSecondaryIndexes: []*dynamodb.GlobalSecondaryIndex{
				{
					IndexName: aws.String("TargetIDIndex"),
					KeySchema: []*dynamodb.KeySchemaElement{
						{
							AttributeName: aws.String("TargetID"),
							KeyType:       aws.String("HASH"),
						},
						{
							AttributeName: aws.String("StartedAt"),
							KeyType:       aws.String("RANGE"),
						},
					},
					Projection: &dynamodb.Projection{
						ProjectionType: aws.String("ALL"),
					},
					ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
						ReadCapacityUnits:  aws.Int64(1000),
						WriteCapacityUnits: aws.Int64(1000),
					},
				},
			},
			ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
				ReadCapacityUnits:  aws.Int64(1000),
				WriteCapacityUnits: aws.Int64(1000),
			},
		},
		tableFultonRaids: {
			TableName: aws.String(tableFultonRaids),
			KeySchema: []*dynamodb.KeySchemaElement{
				{
					AttributeName: aws.String("SourceID"),
					KeyType:       aws.String("HASH"),
				},
			},
			AttributeDefinitions: []*dynamodb.AttributeDefinition{
				{
					AttributeName: aws.String("SourceID"),
					AttributeType: aws.String("S"),
				},
				{
					AttributeName: aws.String("RaidID"),
					AttributeType: aws.String("S"),
				},
			},
			GlobalSecondaryIndexes: []*dynamodb.GlobalSecondaryIndex{
				{
					IndexName: aws.String("RaidIDIndex"),
					KeySchema: []*dynamodb.KeySchemaElement{
						{
							AttributeName: aws.String("RaidID"),
							KeyType:       aws.String("HASH"),
						},
					},
					Projection: &dynamodb.Projection{
						ProjectionType: aws.String("ALL"),
					},
					ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
						ReadCapacityUnits:  aws.Int64(1000),
						WriteCapacityUnits: aws.Int64(1000),
					},
				},
			},
			ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
				ReadCapacityUnits:  aws.Int64(1000),
				WriteCapacityUnits: aws.Int64(1000),
			},
		},
	}

	for tableName, params := range tableParams {
		err := l.createTable(tableName, params)
		if err != nil {
			return err
		}
	}

	deadline := time.Now().Add(10 * time.Second)
	for _, tableName := range tableNames {
		err := l.waitOnTableToBeCreated(tableName, deadline)
		if err != nil {
			return err
		}
	}

	return nil
}

func (l *Tables) createTable(tableName string, params *dynamodb.CreateTableInput) error {
	_, err := l.client.CreateTable(params)
	if err != nil && !isResourceInUseError(err) {
		wrappedErr := errors.Wrap(err, "creating table failed", errors.Fields{
			"table_name": tableName,
		})
		return wrappedErr
	}

	return nil
}

func (l *Tables) waitOnTableToBeCreated(tableName string, deadline time.Time) error {
	for {
		desc, err := l.client.DescribeTable(&dynamodb.DescribeTableInput{
			TableName: aws.String(tableName),
		})
		if err != nil && !isResourceNotFoundError(err) {
			return errors.Wrap(err, "describing table failed", errors.Fields{
				"table_name": tableName,
			})
		}

		tableCreated := err == nil &&
			desc != nil &&
			desc.Table != nil &&
			desc.Table.TableStatus != nil &&
			*desc.Table.TableStatus == "ACTIVE"
		if tableCreated {
			return nil
		}

		if time.Now().After(deadline) {
			return errors.New("passed deadline waiting for DynamoDB table to be created", errors.Fields{
				"table_name": tableName,
			})
		}

		time.Sleep(time.Second)
	}
}

// Delete removes the tables that AutoHost requires.
func (l *Tables) Delete() error {
	for _, tableName := range tableNames {
		err := l.deleteTable(tableName)
		if err != nil {
			return err
		}
	}

	deadline := time.Now().Add(10 * time.Second)
	for _, tableName := range tableNames {
		err := l.waitOnTableToBeDeleted(tableName, deadline)
		if err != nil {
			return err
		}
	}

	return nil
}

func (l *Tables) deleteTable(tableName string) error {
	l.logger.Debug("starting to delete table", log.Fields{
		"table_name": tableName,
	})

	_, err := l.client.DeleteTable(&dynamodb.DeleteTableInput{
		TableName: aws.String(tableName),
	})
	if err != nil && !isResourceNotFoundError(err) {
		return errors.Wrap(err, "DeleteTable failed", errors.Fields{
			"table_name": tableName,
		})
	}

	return nil
}

func (l *Tables) waitOnTableToBeDeleted(tableName string, deadline time.Time) error {
	for {
		_, err := l.client.DescribeTable(&dynamodb.DescribeTableInput{
			TableName: aws.String(tableName),
		})
		if isResourceNotFoundError(err) {
			l.logger.Debug("table deleted", log.Fields{
				"table_name": tableName,
			})
			return nil
		}

		if err != nil {
			return errors.Wrap(err, "describing table failed", errors.Fields{
				"table_name": tableName,
			})
		}

		if time.Now().After(deadline) {
			return errors.New("deadline expired waiting for DynamoDB table to be deleted", errors.Fields{
				"table_name": tableName,
			})
		}

		time.Sleep(1 * time.Second)
	}
}

func (l *Tables) TruncateTables() error {
	params := []struct {
		tableName string
		hashKey   string
		rangeKey  string
	}{
		{
			tableName: TableAutohostSettings,
			hashKey:   "ChannelID",
		},
		{
			tableName: tableChannels,
			hashKey:   "ID",
			rangeKey:  "Resource",
		},
		{
			tableName: tableCaches,
			hashKey:   "Key",
			rangeKey:  "Bucket",
		},
		{
			tableName: TableHost,
			hashKey:   "ChannelID",
		},
		{
			tableName: tableFultonRaidSettings,
			hashKey:   "ChannelID",
		},
		{
			tableName: tableFultonPastRaids,
			hashKey:   "RaidID",
		},
		{
			tableName: tableFultonRaids,
			hashKey:   "SourceID",
		},
	}

	for _, param := range params {
		err := l.truncateTable(param.tableName, param.hashKey, param.rangeKey)
		if err != nil {
			return err
		}
	}

	return nil
}

func (l *Tables) truncateTable(tableName string, haskKey string, rangeKey string) error {
	proj := expression.NamesList(expression.Name(haskKey))
	if rangeKey != "" {
		proj = proj.AddNames(expression.Name(rangeKey))
	}

	exp, err := expression.NewBuilder().WithProjection(proj).Build()
	if err != nil {
		return errors.Wrap(err, "building projection expression failed")
	}

	var exclusiveStartKey map[string]*dynamodb.AttributeValue
	firstIteration := true

	for exclusiveStartKey != nil || firstIteration {
		firstIteration = false

		scanOutput, err := l.client.Scan(&dynamodb.ScanInput{
			ExpressionAttributeNames:  exp.Names(),
			ExpressionAttributeValues: exp.Values(),
			ConsistentRead:            aws.Bool(true),
			ExclusiveStartKey:         exclusiveStartKey,
			ProjectionExpression:      exp.Projection(),
			Select:                    aws.String("SPECIFIC_ATTRIBUTES"),
			TableName:                 aws.String(tableName),
		})
		if err != nil {
			return errors.Wrap(err, "scan failed")
		}

		for _, item := range scanOutput.Items {
			_, err = l.client.DeleteItem(&dynamodb.DeleteItemInput{
				Key:       item,
				TableName: aws.String(tableName),
			})
			if err != nil {
				return errors.Wrap(err, "deleting item failed", errors.Fields{
					"table_name": tableName,
				})
			}
		}

		exclusiveStartKey = scanOutput.LastEvaluatedKey
	}

	return nil
}

func isResourceNotFoundError(err error) bool {
	awsErr, ok := err.(awserr.Error)
	return ok && awsErr.Code() == dynamodb.ErrCodeResourceNotFoundException
}

func isResourceInUseError(err error) bool {
	awsErr, ok := err.(awserr.Error)
	return ok && awsErr.Code() == dynamodb.ErrCodeResourceInUseException
}
