package main

import (
	"fmt"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/cactus/go-statsd-client/statsd"

	"code.justin.tv/release/trace/analysis/paths"
	"code.justin.tv/release/trace/analysis/routes"
	"code.justin.tv/release/trace/analysis/tx"
	"code.justin.tv/release/trace/common"
	"code.justin.tv/release/trace/events"
	"code.justin.tv/release/trace/scanproto"
)

const flushInterval time.Duration = 1 * time.Second

// processBuffer is the number of EventSets to hold in internal decoding
// buffers by default.
var processBuffer = 1024

type Call struct {
	req     *events.Event
	resp    *events.Event
	time    time.Time
	expires time.Time
	emitted bool
}

type Metric struct {
	time        time.Time
	service     string
	environment string
	hostname    string
	route       string
	status      int
	duration    time.Duration
}

func processEventStream(metrics chan<- *Metric, events <-chan *scanproto.TimestampedEvent, eventTimeout time.Duration, workers int, debugStats statsd.Statter) {
	var wg sync.WaitGroup

	// We split up the input stream to ensure events are consistently sharded
	// by TransactionId so that each worker can have it's own callMap.
	inputs := make([]chan *scanproto.TimestampedEvent, workers)

	for i := 0; i < workers; i++ {
		inputs[i] = make(chan *scanproto.TimestampedEvent, processBuffer)
		wg.Add(1)
		go func(in <-chan *scanproto.TimestampedEvent) {
			defer wg.Done()
			callMap := make(map[string]*Call)
			curTime := time.Unix(0, 0)

			for e := range in {
				debugStats.Inc("consumed", 1, statsdRate)
				ev := e.Event

				// Abort quickly if it's a client request
				if !common.IsServerKind(ev.Kind) {
					continue
				}

				// Abort quickly if it's not HTTP
				if ev.GetExtra().GetHttp() == nil {
					continue
				}

				// Turn the event path into a string for the mapping
				key := fmt.Sprintf("%s.%s", tx.IDForEvent(ev), paths.PathToString(ev.Path))

				// Load the call for our given event, or make one
				call, ok := callMap[key]
				if !ok {
					call = &Call{}
					call.expires = e.Time.Add(eventTimeout)
					call.emitted = false
					callMap[key] = call
				}

				// Process event
				if common.IsRequestKind(ev.Kind) {
					if call.req == nil {
						call.req = ev
					} else {
						debugStats.Inc("duplicate.request", 1, statsdRate)
					}
				}
				if common.IsResponseKind(ev.Kind) {
					if call.resp == nil {
						call.time = e.Time
						call.resp = ev
					} else {
						debugStats.Inc("duplicate.response", 1, statsdRate)
					}
				}
				if !call.emitted && call.req != nil && call.resp != nil {
					metrics <- parseCall(call)
					call.emitted = true
				}

				// Clean out expired calls
				if e.Time.Sub(curTime) >= flushInterval {
					curTime = e.Time
					expireCalls(callMap, curTime, debugStats)
				}
			}
		}(inputs[i])
	}

	for e := range events {
		inputs[e.Event.TransactionId[0]%uint64(workers)] <- e
	}

	for i := 0; i < workers; i++ {
		close(inputs[i])
	}

	wg.Wait()
	close(metrics)
}

func parseCall(call *Call) *Metric {
	var (
		route  string
		status int32
	)

	// The request side dictates the route
	if reqHttp := call.req.GetExtra().GetHttp(); reqHttp != nil {
		if routeMap, ok := routes.Routes[call.req.Svcname]; ok {
			route = routeMap.LookupTrimSlash(reqHttp.UriPath, reqHttp.Method.String())
		}
	}
	if route == "" {
		route = "xxx"
	}

	// The response dictates the status code
	if respHttp := call.resp.GetExtra().GetHttp(); respHttp != nil {
		status = respHttp.StatusCode
	}

	return &Metric{
		time:        call.time,
		service:     clean(call.req.Svcname),
		environment: "xxx", // TODO
		hostname:    "xxx", // Waiting for systems to approve this: clean(call.req.GetHostname()),
		route:       clean(route),
		status:      int(status),
		duration:    time.Duration(call.resp.Time-call.req.Time) * time.Nanosecond,
	}
}

func expireCalls(callMap map[string]*Call, curTime time.Time, debugStats statsd.Statter) {
	start := time.Now()
	defer debugStats.TimingDuration("expire-duration", time.Since(start), statsdRate)

	for k, c := range callMap {
		if c.expires.Before(curTime) {
			delete(callMap, k)
			if !c.emitted {
				debugStats.Inc("dropped", 1, statsdRate)
			}
		}
	}
}

var cleanRegex = regexp.MustCompile(`\W+`) // 1+ non-alphanumeric characters
func clean(segment string) string {
	return strings.Trim(cleanRegex.ReplaceAllLiteralString(segment, "-"), "-")
}
