package manager

import (
	"fmt"
	"time"

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

// CreateTable createst the table in m.Config.TableName with our
// schema, with capacity as specified in
// m.Config.ProvisionedThroughputRead and
// m.Config.ProvisionedThroughputWrite.
//
// The schema's primary key is:
// HashKey name S
// RangeKey updated_at N
//
// updated_at is a unix timestamp
//
// We add a Local Secondary Index over:
// HashKey  name S
// RangeKey generation N
//
// The secondary index is to allow us to query the latest versions of
// keys, since you can't do something like:
// SELECT name, MAX(updated_at) GROUP BY name
// in DynamoDB
func (m *Manager) CreateTable() error {
	mainTableExists, err := m.tableExists(m.TableName())
	if err != nil {
		return err
	}
	nsTableExists, err := m.tableExists(m.NamespaceTableName())
	if err != nil {
		return err
	}
	auditTableExists, err := m.tableExists(m.AuditTableName())
	if err != nil {
		return err
	}

	if mainTableExists {
		return fmt.Errorf("Table %s already exists", m.TableName())
	}
	if nsTableExists {
		return fmt.Errorf("Table %s already exists", m.NamespaceTableName())
	}
	if auditTableExists {
		return fmt.Errorf("Table %s already exists", m.AuditTableName())
	}

	mainCreateTableInput := m.inputForCreateTable()
	_, err = m.DynamoDB.CreateTable(mainCreateTableInput)
	if err != nil {
		return err
	}

	nsCreateTableInput := m.inputForCreateNamespaceTable()
	_, err = m.DynamoDB.CreateTable(nsCreateTableInput)
	if err != nil {
		return err
	}

	auditCreateTableInput := m.inputForCreateAuditTable()
	_, err = m.DynamoDB.CreateTable(auditCreateTableInput)
	if err != nil {
		return err
	}

	var timeout time.Duration
	timeout = 1
	for {
		mainInput := m.inputForDescribeTable(m.TableName())
		mainOutput, err := m.DynamoDB.DescribeTable(mainInput)
		if err != nil {
			return err
		}
		nsInput := m.inputForDescribeTable(m.NamespaceTableName())
		nsOutput, err := m.DynamoDB.DescribeTable(nsInput)
		if err != nil {
			return err
		}

		auditInput := m.inputForDescribeTable(m.AuditTableName())
		auditOutput, err := m.DynamoDB.DescribeTable(auditInput)
		if err != nil {
			return err
		}

		if *mainOutput.Table.TableStatus == "ACTIVE" && *nsOutput.Table.TableStatus == "ACTIVE" && *auditOutput.Table.TableStatus == "ACTIVE" {
			return nil
		}

		time.Sleep(timeout * time.Second)
		if timeout < 8 {
			timeout = timeout * 2
		}
	}
}

func (m *Manager) inputForCreateTable() *dynamodb.CreateTableInput {
	return &dynamodb.CreateTableInput{
		AttributeDefinitions: []*dynamodb.AttributeDefinition{
			{
				AttributeName: aws.String("name"),
				AttributeType: aws.String("S"),
			},
			{
				AttributeName: aws.String("namespace"),
				AttributeType: aws.String("S"),
			},
		},
		GlobalSecondaryIndexes: []*dynamodb.GlobalSecondaryIndex{
			{
				IndexName: aws.String("namespace_name"),
				KeySchema: []*dynamodb.KeySchemaElement{
					{
						AttributeName: aws.String("namespace"),
						KeyType:       aws.String("HASH"),
					},
					{
						AttributeName: aws.String("name"),
						KeyType:       aws.String("RANGE"),
					},
				},
				Projection: &dynamodb.Projection{
					ProjectionType: aws.String("ALL"),
				},
				ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
					ReadCapacityUnits:  aws.Int64(m.Config.ProvisionedThroughputRead),
					WriteCapacityUnits: aws.Int64(m.Config.ProvisionedThroughputWrite),
				},
			},
		},
		KeySchema: []*dynamodb.KeySchemaElement{
			{
				AttributeName: aws.String("name"),
				KeyType:       aws.String("HASH"),
			},
		},
		ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
			ReadCapacityUnits:  aws.Int64(m.Config.ProvisionedThroughputRead),
			WriteCapacityUnits: aws.Int64(m.Config.ProvisionedThroughputWrite),
		},
		StreamSpecification: &dynamodb.StreamSpecification{
			StreamEnabled:  aws.Bool(true),
			StreamViewType: aws.String("NEW_AND_OLD_IMAGES"),
		},
		TableName: aws.String(m.Config.TableName),
	}
}

func (m *Manager) inputForDescribeTable(tablename string) *dynamodb.DescribeTableInput {
	return &dynamodb.DescribeTableInput{
		TableName: aws.String(tablename),
	}
}

func (m *Manager) tableExists(tableName string) (bool, error) {
	var lastTable *string
	for {
		params := &dynamodb.ListTablesInput{}
		if lastTable != nil {
			params.ExclusiveStartTableName = lastTable
		}
		out, err := m.DynamoDB.ListTables(params)
		if err != nil {
			return false, err
		}
		for _, table := range out.TableNames {
			if *table == tableName {
				return true, nil
			}
		}
		if out.LastEvaluatedTableName == nil {
			return false, nil
		}
		lastTable = out.LastEvaluatedTableName
	}
}

func (m *Manager) inputForCreateNamespaceTable() *dynamodb.CreateTableInput {
	return &dynamodb.CreateTableInput{
		AttributeDefinitions: []*dynamodb.AttributeDefinition{
			{
				AttributeName: aws.String("namespace"),
				AttributeType: aws.String("S"),
			},
		},
		KeySchema: []*dynamodb.KeySchemaElement{
			{
				AttributeName: aws.String("namespace"),
				KeyType:       aws.String("HASH"),
			},
		},
		ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
			ReadCapacityUnits:  aws.Int64(m.Config.ProvisionedThroughputRead),
			WriteCapacityUnits: aws.Int64(m.Config.ProvisionedThroughputWrite),
		},
		TableName: aws.String(m.NamespaceTableName()),
	}
}

func (m *Manager) inputForCreateAuditTable() *dynamodb.CreateTableInput {
	return &dynamodb.CreateTableInput{
		AttributeDefinitions: []*dynamodb.AttributeDefinition{
			{
				AttributeName: aws.String("name"),
				AttributeType: aws.String("S"),
			},
			{
				AttributeName: aws.String("namespace"),
				AttributeType: aws.String("S"),
			},
			{
				AttributeName: aws.String("updated_at"),
				AttributeType: aws.String("N"),
			},
		},
		GlobalSecondaryIndexes: []*dynamodb.GlobalSecondaryIndex{
			{
				IndexName: aws.String("namespace_name"),
				KeySchema: []*dynamodb.KeySchemaElement{
					{
						AttributeName: aws.String("namespace"),
						KeyType:       aws.String("HASH"),
					},
					{
						AttributeName: aws.String("name"),
						KeyType:       aws.String("RANGE"),
					},
				},
				Projection: &dynamodb.Projection{
					ProjectionType: aws.String("INCLUDE"),
					NonKeyAttributes: []*string{
						aws.String("updated_at"),
					},
				},
				ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
					ReadCapacityUnits:  aws.Int64(m.Config.ProvisionedThroughputRead),
					WriteCapacityUnits: aws.Int64(m.Config.ProvisionedThroughputWrite),
				},
			},
		},
		KeySchema: []*dynamodb.KeySchemaElement{
			{
				AttributeName: aws.String("name"),
				KeyType:       aws.String("HASH"),
			},
			{
				AttributeName: aws.String("updated_at"),
				KeyType:       aws.String("RANGE"),
			},
		},
		ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
			ReadCapacityUnits:  aws.Int64(m.Config.ProvisionedThroughputRead),
			WriteCapacityUnits: aws.Int64(m.Config.ProvisionedThroughputWrite),
		},
		TableName: aws.String(m.AuditTableName()),
	}
}
