package main

import (
	"flag"
	"log"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"code.justin.tv/common/chitin"
	"code.justin.tv/common/gometrics"
	"github.com/cactus/go-statsd-client/statsd"

	_ "code.justin.tv/common/golibs/bininfo"
	"code.justin.tv/release/trace/scanproto"

	// Install variable export handler, per current good practice
	_ "expvar"
	// Install profiling handlers, per current good practice
	_ "net/http/pprof"
)

const (
	defaultCallTimeout = 65 * time.Second
	defaultStatsdRate  = 0.1
	defaultZookeeper   = "trace-zk.prod.us-west2.justin.tv"

	traceStatsdPrefix      = "trace.data"
	statsdsinkStatsdPrefix = "trace.statsdsink"
	statsdFlushInterval    = 100 * time.Millisecond
	statsdFlushSize        = 512 // bytes. As recommended by http://godoc.org/github.com/cactus/go-statsd-client/statsd#NewBufferedClient
)

var (
	inFile string

	inKCL bool

	outHostPort string

	readWorkers    int
	processWorkers int
	writeWorkers   int

	callTimeout  time.Duration
	statsdRate   float32
	statsdRate64 float64

	unused string
)

func main() {
	flag.StringVar(&inFile, "in-file", "events.pb", "input file of encoded EventSets")

	flag.BoolVar(&inKCL, "in-kinesis", false, "read input from a kinesis client library MultiLangDaemon")

	flag.StringVar(&unused, "zk", "", "Unused. Reserved for backwards compatibility.")
	flag.StringVar(&unused, "kafka-group", "", "Unused. Reserved for backwards compatibility.")
	flag.StringVar(&unused, "in-topic", "", "Unused. Reserved for backwards compatibility.")

	flag.StringVar(&outHostPort, "out", "localhost:8125", "hostport of a statsd-compatible collector")

	flag.IntVar(&readWorkers, "nread", 1, "number of read workers")
	flag.IntVar(&processWorkers, "nprocess", 1, "number of process workers")
	flag.IntVar(&writeWorkers, "nwrite", 1, "number of write workers")

	flag.DurationVar(&callTimeout, "tx-max-duration", defaultCallTimeout, "max allowed duration of a transaction")
	flag.Float64Var(&statsdRate64, "rate", defaultStatsdRate, "sampling rate of statsd events emitted, from 0.0 to 1.0")
	flag.Parse()

	statsdRate = float32(statsdRate64)

	var (
		source scanproto.EventSource
		sink   MetricSink
		err    error
	)

	err = chitin.ExperimentalTraceProcessOptIn()
	if err != nil {
		log.Fatalf("trace enable err=%q", err)
	}

	switch {
	case inKCL:
		source = scanproto.NewKinesisEventSource()
	case inFile != "":
		source = scanproto.NewFileEventSource(inFile, readWorkers)
	}

	sink, err = NewStatsdMetricSink(outHostPort, writeWorkers)
	if err != nil {
		log.Fatalf("sink open err=%q", err)
	}

	evCh := source.Events()
	mCh := make(chan *Metric, writeBuffer)
	debugStats, err := statsd.NewBufferedClient(outHostPort, statsdsinkStatsdPrefix, statsdFlushInterval, statsdFlushSize)
	if err != nil {
		log.Fatalf("debugStats open err=%q", err)
	}
	gometrics.Monitor(debugStats, time.Second*5)

	var wg sync.WaitGroup

	wg.Add(1)
	go func() {
		defer wg.Done()
		for e := range source.Errors() {
			log.Printf("Event source error! shutting down! err=%q", e)
			source.Stop()
			return
		}
	}()

	wg.Add(1)
	go func() {
		log.Println("source.run start")
		source.Run()
		wg.Done()
		log.Println("source.run end")
	}()

	log.Println("turning on signal handling")
	sigch := make(chan os.Signal, 1)
	signal.Notify(sigch, syscall.SIGINT)
	go func() {
		for _ = range sigch {
			log.Println("source.stop start")
			wg.Add(1)
			source.Stop()
			wg.Done()
			log.Println("source.stop end")
		}
	}()

	wg.Add(1)
	go func() {
		log.Println("processEventStream start")
		processEventStream(mCh, evCh, callTimeout, processWorkers, debugStats)
		wg.Done()
		log.Println("processEventStream end")
	}()

	wg.Add(1)
	go func() {
		log.Println("sink.consume start")
		sink.Consume(mCh)
		wg.Done()
		log.Println("sink.consume end")
	}()

	// Start up an HTTP server for pprof, then ignore it
	go func() {
		log.Println(http.ListenAndServe(":8080", nil))
	}()

	wg.Wait()
}
