package statsd

import (
	"math"
	"sync"
	"time"

	logging "code.justin.tv/amzn/TwitchLogging"
	"code.justin.tv/amzn/TwitchProcessIdentifier"
	telemetry "code.justin.tv/amzn/TwitchTelemetry"
	"github.com/cactus/go-statsd-client/statsd"
)

// Notes
// This is the simplest possible implementation for a statsd sender
// for the Twitch Telemetry library. There is a huge amount of room
// for optimizations and improvements
//
// process identifier hierarchy in statsd naming:
// <service>.<stage>.<substage>.<region>.<everything_else_sorted_on_dimension_name>
//
// if some part of service identifier is empty, use "all" as the value

// TODO
// - sanitize names: remove/rename characters ":", "."

const (
	// DefaultStatsdEndpoint is the standard place to send statsd events
	DefaultStatsdEndpoint = "statsd.internal.justin.tv:8125"

	// the string to use for rollup purposes
	allString = "all"
	// the string to use for unset process identifier dimensions
	noneString = "none"
)

// Client holds a statsd.Statter and some extra state to send events efficiently
type Client struct {
	*CommonSender
	aggregator *telemetry.Aggregator
	stopSignal chan bool
	stopWait   sync.WaitGroup
}

// Unbuffered represents an unbuffered Statsd sender (commonly used with a Telemetry.BufferedAggregator)
type Unbuffered struct {
	*CommonSender
}

// CommonSender contains the common struct components and functions for statsd senders
type CommonSender struct {
	endpoint          string
	statter           statsd.Statter
	processIdentifier *identifier.ProcessIdentifier
	logger            logging.Logger
	mux               sync.Mutex
}

// New returns a new StatsdClient
func New(processIdentifier *identifier.ProcessIdentifier, endpoint string, logger logging.Logger) (telemetry.SampleObserver, error) {
	common, err := createCommonSender(processIdentifier, endpoint, logger)
	if err != nil {
		return nil, err
	}

	c := &Client{
		CommonSender: common,
		// aggregation period is zero/disabled because the sample timestamps are not used.
		aggregator: telemetry.NewAggregator(0),
		stopSignal: make(chan bool),
	}
	c.run()
	return c, nil
}

// sleeps for 2 seconds at a time, then flushes aggregated sample data to statsd
func (s *Client) run() {
	go func() {
		ticker := time.NewTicker(2 * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				s.Flush()
			case <-s.stopSignal:
				s.Flush()
				s.stopWait.Done()
				return
			}
		}
	}()
}

// Flush flushes the metrics and re-instantiates the statsd client (which will re-resolve DNS!)
func (s *Client) Flush() {
	s.mux.Lock()
	defer s.mux.Unlock()
	if s.aggregator != nil {
		// Get the distributions from the aggregator
		distributions := s.aggregator.Flush()
		// flush them to the underlying statsd.Statter
		s.flushDistributions(distributions)
	}
	// close the statsd.Statter so the statsd events flush to the network
	s.statter.Close()
	// create a fresh statsd.Statter
	statter, err := s.freshStatter(s.endpoint)
	if err != nil {
		s.log("Failed to publish metrics", "err", err.Error())
	}
	s.statter = statter
}

// Stop will flush the buffered stats and close the statter
// TODO: check if the client is stopped elsewhere in the code
func (s *Client) Stop() {
	s.mux.Lock()
	defer s.mux.Unlock()
	s.statter.Close()
}

// ObserveSample takes an internal collection of samples and sends it to statsd
// TODO: use *telemetry.Sample when the upstream changes
func (s *Client) ObserveSample(sample *telemetry.Sample) {
	if s.aggregator != nil {
		s.aggregator.AggregateSample(sample)
	}
}

// NewUnbuffered returns a new StatsdClient that needs to be run within a separate BufferedAggregator as it does not
// main a buffer/ any aggregation or flushing logic on its own
func NewUnbuffered(processIdentifier *identifier.ProcessIdentifier, endpoint string, logger logging.Logger) (telemetry.SampleUnbufferedObserver, error) {
	common, err := createCommonSender(processIdentifier, endpoint, logger)
	if err != nil {
		return nil, err
	}

	c := &Unbuffered{
		CommonSender: common,
	}
	return c, nil
}

// FlushWithoutBuffering flushes without buffering the flush. This additionally re-instantiates the statsd client
// (which will re-resolve DNS!)
func (s *Unbuffered) FlushWithoutBuffering(distributions []*telemetry.Distribution) {
	s.mux.Lock()
	defer s.mux.Unlock()

	// flush them to the underlying statsd.Statter
	s.flushDistributions(distributions)
	// close the statsd.Statter so the statsd events flush to the network
	s.statter.Close()
	// create a fresh statsd.Statter
	statter, err := s.freshStatter(s.endpoint)
	if err != nil {
		s.log("Failed to publish metrics", "err", err.Error())
	}
	s.statter = statter
}

func (s *CommonSender) flushDistributions(distributions []*telemetry.Distribution) {

	for _, distribution := range distributions {

		// Use a metricBuilder to generate metric names for this distribution
		metricBuilder := newMetricBuilder(distribution, s.logger)

		// Send the value for each generated metric name
		for _, metricName := range metricBuilder.expandMetricNames() {
			// - in the case of a counter, just send Sum
			// - in the case of a timer, send one event per histogram bucket,
			//   with adjusted sampling rate to be hyper-efficient!
			// TODO: handle errors
			switch distribution.Unit {
			case telemetry.UnitCount:
				s.statter.Inc(metricName, int64(distribution.Sum), 1)
			case telemetry.UnitSeconds:
				for bucket, count := range distribution.SEH1.Histogram {
					approxValue := distribution.SEH1.ApproximateOriginalValue(bucket)
					// statsd.Statter wants an int64, sadly... so morph to milliseconds and cast to int64
					approxDuration := durationFromFloat64(approxValue)
					sampleRate := float32(1) / float32(count)
					s.statter.TimingDuration(metricName, approxDuration, sampleRate)
				}
				// don't forget about the zero bucket!
				// only send if it actually has values
				if distribution.SEH1.ZeroBucket > 0 {
					zeroValue := int64(0)
					zeroSampleRate := float32(1) / float32(distribution.SEH1.ZeroBucket)
					s.statter.Timing(metricName, zeroValue, zeroSampleRate)
				}
			}

		}
	}
}

// withStatter is a convenience function that makes testing easier by allowing
// the client's statter to be mocked out with some fake statsd client
func (s *CommonSender) withStatter(statter statsd.Statter) {
	s.statter = statter
}

// log logs if the logger is not nil
func (s *CommonSender) log(msg string, keyvals ...interface{}) {
	if s.logger != nil {
		s.logger.Log(msg, keyvals...)
	}
}

// generates a new statsd.Statter to use in this client. Very useful because it will
// re-resolve DNS, thus serving the all-important purpose of balancing traffic to the
// central statsd edge relay cluster!
func (s *CommonSender) freshStatter(endpoint string) (statsd.Statter, error) {
	client, err := statsd.NewBufferedClient(endpoint, "", 1*time.Second, 1432)
	if err != nil {
		s.log("failed to create fresh statsd client", "err", err.Error())
		return nil, err
	}
	// we control the sampling, so just always say "true" so samples always send
	client.(*statsd.Client).SetSamplerFunc(func(float32) bool { return true })
	return client, nil
}

// takes a float64 representing a number of seconds and returns a time.Duration
// representing it
func durationFromFloat64(f float64) time.Duration {
	// in English:
	// 1. the float represents seconds, so convert to nanoseconds by multiplying by 10^9
	// 2. truncate off any precision beyond nanoseconds by converting to int64
	// 3. int64 can be directly cast to time.Duration, and it assumes
	//    the unit is nanoseconds, so do that...
	// 4. return the result, which will be a time.Duration representation of the original float64
	return time.Duration(int64(f * math.Pow10(9)))
}

func createCommonSender(processIdentifier *identifier.ProcessIdentifier, endpoint string, logger logging.Logger) (*CommonSender, error) {
	// create a statsd.Statter to the default endpoint with no prefix
	// TODO: would be cool if this had a lock around it, and periodically re-create it
	// so that the UDP.Dial would be called... else, some long-running apps never re-resolve
	// the DNS address and this can lead to staleness issues where apps send to dead IPs
	s, err := statsd.NewBufferedClient(endpoint, "", 1*time.Second, 1432)
	if err != nil {
		return nil, err
	}
	return &CommonSender{
		endpoint:          endpoint,
		statter:           s,
		processIdentifier: processIdentifier,
		logger:            logger,
	}, nil
}
