package poller

import (
	"fmt"
	"time"

	logging "code.justin.tv/amzn/TwitchLogging"
	telemetry "code.justin.tv/amzn/TwitchTelemetry"
	"log"
	"sync"
)

// PollingCollector is a harness that seats a job that will be run periodically to generate and observe
// samples.  The PollingCollector is given a job to run, an interval on which to run it, a sample builder,
// and an observer to flush samples to
//
// When the PollingJob is run and samples are returned, the PollingCollector will mangle every sample, doing the
// following:
// - The PollingCollector will build a sample with its SampleBuilder using the polled sample's name, value, and units
// - Any dimensions in the returned sample will be written into the built sample. In case of collisions,
//   the polling job's dimension value will override the sample builder's value
// - The PollingCollector will unconditionally assign a timestamp to every sample
//
// If the PollingCollector does not have both non-nil SampleBuilder and SampleObserver, then calls to
// Start() will simply return without starting the PollingCollector. A call to IsRunning() can verify
// that the PollingCollector was started successfully.
type PollingCollector struct {
	pollingJob    PollingJob
	sampleBuilder *telemetry.SampleBuilder
	observer      telemetry.SampleObserver
	interval      time.Duration
	logger        logging.Logger

	// internals to control starting and stopping
	stopWait   sync.WaitGroup
	stopSignal chan bool
	isRunning  bool
}

// NewPollingCollector returns a PollingCollector using the given configuration
func NewPollingCollector(job PollingJob, interval time.Duration, sb *telemetry.SampleBuilder, obs telemetry.SampleObserver, logger logging.Logger) *PollingCollector {
	return &PollingCollector{
		pollingJob:    job,
		observer:      obs,
		interval:      interval,
		stopSignal:    make(chan bool),
		isRunning:     false,
		sampleBuilder: sb,
		logger:        logger,
	}
}

// WithSampleBuilder will set the base rollup dimensions for the samples generated by this
// PollingCollector
func (pc *PollingCollector) WithSampleBuilder(sb *telemetry.SampleBuilder) *PollingCollector {
	pc.sampleBuilder = sb
	return pc
}

// WithSampleObserver sets the observer for the PollingCollector
func (pc *PollingCollector) WithSampleObserver(o telemetry.SampleObserver) *PollingCollector {
	pc.observer = o
	return pc
}

// IsRunning returns a boolean indicating if the PollingCollector is currently running
func (pc *PollingCollector) IsRunning() bool {
	return pc.isRunning
}

// Start initiates the poller, causing it to start collecting data from the
// PollingJob every `interval`
func (pc *PollingCollector) Start() {
	// Don't spin another go routine if already running... just return!
	if pc.isRunning {
		pc.log("Cannot start poller, already running")
		return
	}
	// If there is no job in the harness, return
	if pc.pollingJob == nil || pc.sampleBuilder == nil || pc.observer == nil {
		pc.log("Poller must have job, sample builder, and observer to start")
		return
	}
	pc.isRunning = true
	// Add a stopWait, this is solely used to ensure we properly shut down
	pc.stopWait.Add(1)
	go func() {
		pc.log("Starting poller job")
		// Sleep until aligned on the interval to collect on
		now := time.Now()
		startTime := now.Truncate(pc.interval).Add(pc.interval)
		if now.Before(startTime) {
			time.Sleep(startTime.Sub(now))
		}
		// Create a ticker to collect every interval
		ticker := time.NewTicker(pc.interval)
		defer ticker.Stop()
		defer pc.stopWait.Done()
		// Start the collection loop
		for {
			select {
			case <-pc.stopSignal:
				// told to stop, shut down the loop
				// ... mark as not running
				pc.log("Stopping poller job")
				// Emit metrics one final time
				pc.fetchAndSubmitSamples(time.Now())
				pc.isRunning = false
				return
			case t := <-ticker.C:
				pc.fetchAndSubmitSamples(t)
			}
		}
	}()
}

// Get samples and submit them
func (pc *PollingCollector) fetchAndSubmitSamples(t time.Time) {
	// The interval has elapsed, collect more data
	samples, err := pc.fetchSamples(t)
	if err == nil {
		pc.submitSamples(samples)
	} else {
		pc.logError("Error fetching samples from polling job", err)
	}
}

// Stop turns off the PollingCollector, causing it to stop collecting data
// Put another way: Stop trying to make Fetch happen
func (pc *PollingCollector) Stop() {
	pc.stopSignal <- true
	pc.stopWait.Wait()
}

func (pc *PollingCollector) fetchSamples(t time.Time) ([]*telemetry.Sample, error) {

	// 1. get data, do _something_ if errors
	var err error
	rawSamples, err := pc.pollingJob.Fetch()
	builtSamples := make([]*telemetry.Sample, 0)
	var builtSample *telemetry.Sample

	if err != nil {
		return nil, err
	}
	// 2. fill in samples

	for _, rawSample := range rawSamples {
		builtSample, err = pc.buildFullSample(rawSample, t)
		if err != nil {
			pc.logError("Error constructing full sample from polling job", err)
			continue
		}
		builtSamples = append(builtSamples, builtSample)
	}
	return builtSamples, nil
}

func (pc *PollingCollector) submitSamples(samples []*telemetry.Sample) {
	for _, s := range samples {
		pc.observer.ObserveSample(s)
	}
}

// uses the sample builder to build a sample, then copies in all of the information
// provided by the polling job (value, extra dimensions, etc)
func (pc *PollingCollector) buildFullSample(s *telemetry.Sample, timestamp time.Time) (*telemetry.Sample, error) {
	if s == nil || s.MetricID.Name == "" {
		return nil, fmt.Errorf("trying to construct empty sample")
	}
	// Build the base sample
	builtSample, err := pc.sampleBuilder.Build(s.MetricID.Name, s.Value, s.Unit)
	if err != nil {
		return nil, err
	}
	builtSample.Timestamp = timestamp

	// whatever dimensions the sample returned will override the sample builder's dimensions
	for name, value := range s.MetricID.Dimensions {
		builtSample.MetricID.AddDimension(name, value)
	}

	return builtSample, nil
}

func (pc *PollingCollector) logError(errMsg string, err error) {
	if pc.logger != nil {
		pc.logger.Log(errMsg, "err", err.Error())
	} else {
		log.Printf("%v: %v", errMsg, err)
	}
}

func (pc *PollingCollector) log(msg string) {
	if pc.logger != nil {
		pc.logger.Log(msg)
	} else {
		log.Printf(msg)
	}
}
