package migration

import (
	"fmt"

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

func createConsumerGroup(awsConfig *aws.Config, streamName string, consumerGroupName string) *consumerGroup {
	return &consumerGroup{
		connection:    dynamodb.New(session.New(), awsConfig),
		streamName:    streamName,
		consumerGroup: consumerGroupName,
	}
}

type consumerGroup struct {
	connection    *dynamodb.DynamoDB
	streamName    string
	consumerGroup string
}

func (self *consumerGroup) SaveProgress(shardId string, progress string) error {
	tableName := fmt.Sprintf("%s_progress", self.streamName)

	_, err := self.connection.PutItem(&dynamodb.PutItemInput{
		TableName: &tableName,
		Item: map[string]*dynamodb.AttributeValue{
			"shard_id": &dynamodb.AttributeValue{
				S: aws.String(shardId),
			},
			"consumer_group": &dynamodb.AttributeValue{
				S: aws.String(self.consumerGroup),
			},
			"progress": &dynamodb.AttributeValue{
				S: aws.String(progress),
			},
		},
	})

	return err
}

func (self *consumerGroup) LoadProgress(shardId string) (string, error) {
	tableName := fmt.Sprintf("%s_progress", self.streamName)

	response, err := self.connection.GetItem(&dynamodb.GetItemInput{
		TableName:      &tableName,
		ConsistentRead: aws.Bool(true),
		Key: map[string]*dynamodb.AttributeValue{
			"shard_id": &dynamodb.AttributeValue{
				S: aws.String(shardId),
			},
			"consumer_group": &dynamodb.AttributeValue{
				S: aws.String(self.consumerGroup),
			},
		},
	})

	if err != nil {
		return "", err
	}

	val, ok := response.Item["progress"]
	if !ok {
		return "", nil
	}

	str := val.S
	if str == nil {
		return "", nil
	}

	return *str, nil
}
