package statsrunner

import (
	"context"
	"errors"
	"sync"
	"time"

	"code.justin.tv/sse/malachai/pkg/internal/stats"
	"code.justin.tv/sse/malachai/pkg/log"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/cloudwatch"
	"github.com/aws/aws-sdk-go/service/cloudwatch/cloudwatchiface"
)

// RunnerAPI is the API for Runner
type RunnerAPI interface {
	IncrementMetric(*IncrementableMetric) error
	Run()
	Stop(time.Duration) error
}

// Config is the input to Runner
type Config struct {
	FlushRate time.Duration
}

// Runner pushes up stats at a rate
type Runner struct {
	Config     Config
	CloudWatch cloudwatchiface.CloudWatchAPI
	Logger     log.S2SLogger

	// created in Run
	startedLock  sync.Mutex
	started      bool
	stop         chan interface{}
	metricValues *incrementableMetricValues
}

// Run is the loop that pushes up stats
func (r *Runner) Run() {
	if r.isStarted() {
		r.Logger.Warn("stats: runner can only be started once")
		return
	}

	r.Logger.Debug("stats: starting runner")

	r.stop = make(chan interface{})
	r.metricValues = &incrementableMetricValues{
		values:     make(map[string]*stats.IncrementableMetricCount),
		valuesLock: &sync.Mutex{},
	}
	r.setStarted()

	for {
		select {
		case <-r.stop:
			r.Logger.Debug("stats: exiting runner - flushing once")
			if err := r.flushData(); err != nil {
				r.Logger.Errorf("error flushing data to cloudwatch: %s", err.Error())
			}
			return
		case <-time.NewTimer(r.Config.FlushRate).C:
			if err := r.flushData(); err != nil {
				r.Logger.Errorf("error flushing data to cloudwatch: %s", err.Error())
			}
		}
	}
}

// IncrementMetric increments a metric by i
func (r *Runner) IncrementMetric(i *IncrementableMetric) (err error) {
	if !r.isStarted() {
		err = errors.New("stats runner not running")
		return
	}

	err = r.metricValues.Increment(i)
	return
}

// Stop stops Run
func (r *Runner) Stop(timeout time.Duration) (err error) {
	if !r.isStarted() {
		return
	}

	r.Logger.Debug("stats: stopping runner")
	timer := time.NewTimer(timeout)

	select {
	case r.stop <- struct{}{}:
		r.Logger.Debug("stats: runner stopped")
	case <-timer.C:
		err = errors.New("failed to stop statter runner")
	}

	return
}

func (r *Runner) flushData() (err error) {
	for _, value := range r.metricValues.PopValues() {
		dimensions := make([]*cloudwatch.Dimension, len(value.Metric.Dimensions))
		for nDimension, dimension := range value.Metric.Dimensions {
			dimensions[nDimension] = &cloudwatch.Dimension{
				Name:  aws.String(dimension.Name),
				Value: aws.String(dimension.Value),
			}
		}

		now := aws.Time(time.Now().UTC())

		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		_, err = r.CloudWatch.PutMetricDataWithContext(ctx, &cloudwatch.PutMetricDataInput{
			MetricData: []*cloudwatch.MetricDatum{
				{
					Dimensions:        dimensions,
					MetricName:        aws.String(value.Metric.MetricName),
					StorageResolution: aws.Int64(60),
					Timestamp:         now,
					Unit:              aws.String(cloudwatch.StandardUnitCount),
					Value:             aws.Float64(float64(value.Count)),
				},
			},
			Namespace: aws.String("Malachai"),
		})
		if err != nil {
			return
		}
	}
	return
}

func (r *Runner) isStarted() bool {
	r.startedLock.Lock()
	started := r.started
	r.startedLock.Unlock()
	return started
}

func (r *Runner) setStarted() {
	r.startedLock.Lock()
	r.started = true
	r.startedLock.Unlock()
}
