package eventpublisher

import (
	"fmt"
	"time"

	pb "code.justin.tv/dta/rockpaperscissors/proto"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/client"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/kinesis"
	"github.com/aws/aws-sdk-go/service/kinesis/kinesisiface"
	"github.com/golang/protobuf/proto"
	"github.com/satori/go.uuid"
)

// Environment is which RockPaperScissors environment to publish to.
type Environment int

// A list of the available RockPaperScissors environments.
const (
	UNSET Environment = iota
	PRODUCTION
	DEVELOPMENT
)

var awsRegion = "us-west-2"

var streamName = map[Environment]string{
	PRODUCTION:  "rockpaperscissors-production-events",
	DEVELOPMENT: "rockpaperscissors-development-events",
}

// Interface is a promised interface for EventPublisher objects.
//
// This is used for testing.
type Interface interface {
	PublishEvent(event *pb.Event) error
}

// Config captures EventPublisher configuration options.
//
// Source configures the default for the Source field in Event protobufs
// created by CreateEvent().
//
// Environment configures which RockPaperScissors event bus to publish events
// to. The default is PRODUCTION so, if you wish to send to the DEVELOPMENT
// environment, you will need to set this field after calling
// NewEventPublisher().
//
// AwsSession is an AWS Session object. If not provided, one will be created
// with "SharedConfigEnable" turned on so ~/.aws/config is used.
//
// KinesisSvc is an AWS Kinesis service object. If not provided, one will be
// created with the AwsSession either povided or created.
type Config struct {
	Source      string
	Environment Environment
	AwsSession  client.ConfigProvider
	KinesisSvc  kinesisiface.KinesisAPI
}

// EventPublisher contains configuration for publishing events to the
// RockPaperScissors event bus.
//
// Please use NewEventPublisher() for creating a properly initialized struct.
type EventPublisher struct {
	config *Config
}

var awsSessionFactory = func() (client.ConfigProvider, error) {
	return session.NewSessionWithOptions(session.Options{
		Config:            aws.Config{Region: aws.String(awsRegion)},
		SharedConfigState: session.SharedConfigEnable,
	})
}

var kinesisSvcFactory = func(p client.ConfigProvider) kinesisiface.KinesisAPI {
	return kinesis.New(p)
}

// NewEventPublisher creates and initializes an EventPublisher struct.
//
// "source" is a string that identifies the publisher
// (e.g. "Jenkins", "GitHub").
func NewEventPublisher(source string, cfgs ...*Config) (*EventPublisher, error) {
	ep := EventPublisher{
		&Config{
			Source:      source,
			Environment: PRODUCTION,
		},
	}

	for _, config := range cfgs {
		if config.Source != "" {
			ep.config.Source = config.Source
		}
		if config.Environment != UNSET {
			ep.config.Environment = config.Environment
		}
		if config.AwsSession != nil {
			ep.config.AwsSession = config.AwsSession
		}
		if config.KinesisSvc != nil {
			ep.config.KinesisSvc = config.KinesisSvc
		}
	}

	if ep.config.AwsSession == nil {
		sess, err := awsSessionFactory()
		if err != nil {
			return nil, err
		}
		ep.config.AwsSession = sess
	}

	if ep.config.KinesisSvc == nil {
		ep.config.KinesisSvc = kinesisSvcFactory(ep.config.AwsSession)
	}

	return &ep, nil
}

// Convert Go Time type into Unix epoch secs + fractional seconds as float64.
func convertTimestamp(timestamp time.Time) float64 {
	return float64(timestamp.UnixNano()) * 1E-9
}

// CreateEvent creates and intializes an Event protobuf.
func CreateEvent(timestamp time.Time, eventType string, body []byte, attributes map[string]string) (*pb.Event, error) {
	event := &pb.Event{
		// TODO: populate source field
		Uuid:      uuid.NewV4().Bytes(),
		Timestamp: proto.Float64(convertTimestamp(timestamp)),
		Type:      proto.String(eventType),
		Body:      body,
	}
	for key, value := range attributes {
		event.Attributes = append(event.Attributes, &pb.Event_Attribute{
			Key:   proto.String(key),
			Value: proto.String(value),
		})
	}
	return event, nil
}

// PublishEvent sends an Event protobuf to the RockPaperScissors event bus.
func (ep *EventPublisher) PublishEvent(event *pb.Event) error {
	data, err := proto.Marshal(event)
	if err != nil {
		return err
	}

	_, err = ep.config.KinesisSvc.PutRecord(&kinesis.PutRecordInput{
		Data:         data,
		PartitionKey: aws.String(fmt.Sprintf("%x", event.GetUuid())),
		StreamName:   aws.String(streamName[ep.config.Environment]),
	})
	return err
}
