package main

import (
	"fmt"
	"io"
	"log"
	"net/http"
	"sort"
	"strings"
	"time"

	"github.com/patrickmn/go-cache"
	"goji.io/pat"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"

	"code.justin.tv/common/chitin"
	"code.justin.tv/common/config"
	"code.justin.tv/common/twitchhttp"
	"code.justin.tv/release/trace/api"
)

var firehoseSampleRate float64 = 0.01
var tableCache *cache.Cache
var uptime time.Time

func main() {
	config.Register(map[string]string{
		"FirehoseHostPort": "", // The host/port of the trace firehose we're going to read from
		"Service":          "", //The name of the service to watch for firehose events
		"Databases":        "", //Comma-separated list of databases to watch
	})
	uptime = time.Now()

	//Pull config
	err := config.Parse()
	if err != nil {
		log.Fatal(err)
	}

	firehoseAddr := config.MustResolve("FirehoseHostPort")
	databases := config.MustResolve("Databases")
	watchService := config.MustResolve("Service")

	var databaseList = map[string]string{}
	for _, dbName := range strings.Split(databases, ",") {
		databaseList[dbName] = dbName
	}

	//Auto-trace
	err = chitin.ExperimentalTraceProcessOptIn()
	if err != nil {
		log.Fatal(err)
	}

	tableCache = cache.New(72*time.Hour, 5*time.Minute)

	go func(tableCache *cache.Cache) {
		for {
			err = runTrace(tableCache, firehoseAddr, watchService, databaseList)

			time.Sleep(200 * time.Millisecond)

			if err == nil || grpc.Code(err) == codes.DeadlineExceeded || err == io.EOF {
				continue
			}

			if config.RollbarErrorLogger() != nil {
				config.RollbarErrorLogger().Error(err)
			}
			log.Println(err)
		}
	}(tableCache)

	server := twitchhttp.NewServer()
	server.HandleFuncC(pat.Get("/joins"), readJoins)
	server.HandleFuncC(pat.Get("/lastquery"), lastQuery)
	//Launch the healthcheck endpoint- this will return current health status on :8000/debug/running
	log.Fatal(twitchhttp.ListenAndServe(server))
}

func runTrace(tableCache *cache.Cache, firehoseAddr string, watchService string, databaseList map[string]string) error {
	//Open a GRPC connection to the trace URL
	log.Println("Dialing")
	traceConn, err := grpc.Dial(firehoseAddr, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("Could not reach firehose: %v", err)
	}
	defer attemptClose(traceConn)

	//Build a firehose siphon to pipe into stats
	firehoseClient := api.NewTraceClient(traceConn)
	stats := NewStatsWriter()
	siphon := NewFirehoseSiphon(firehoseClient, stats, tableCache, watchService, databaseList)

	//Run the siphon - if we have an error, keep rebuilding & trying again
	return siphon.PullTransactions(context.Background())
}

func attemptClose(grpcConn *grpc.ClientConn) {
	err := grpcConn.Close()

	if err != nil {
		log.Println("Error closing conn: ", err)
	}
}

func lastQuery(c context.Context, w http.ResponseWriter, r *http.Request) {
	query, present := tableCache.Get("LASTQUERY")
	if !present {
		query = "none"
	}

	queryStr, ok := query.(string)
	if !ok {
		queryStr = "none"
	}

	outputStr := fmt.Sprintf("<html><head><title>Frankenjoin Last Query</title></head><body>%s</body></html>", queryStr)
	_, err := w.Write([]byte(outputStr))
	if err != nil {
		log.Println(w.Write([]byte(err.Error())))
	}
}

func readJoins(c context.Context, w http.ResponseWriter, r *http.Request) {
	cachedJoins := []string{}

	for join, _ := range tableCache.Items() {
		if join == "LASTQUERY" {
			continue
		}
		cachedJoins = append(cachedJoins, join)
	}

	sort.Strings(cachedJoins)

	liveDuration := time.Now().Sub(uptime)
	liveSeconds := float32(liveDuration / time.Second)

	output := "<html><head><title>Frankenjoin Joins</title></head><body>"
	for _, join := range cachedJoins {
		item, ok := tableCache.Get(join)
		if !ok {
			continue
		}

		queryItem, ok := item.(*QueryData)
		if !ok {
			continue
		}

		rate := float32(queryItem.TotalCalls) / liveSeconds

		output += fmt.Sprintf("%s<br/>&nbsp;&nbsp;Last Seen: %s<br/>&nbsp;&nbsp;Rate: %f/sec<br/>&nbsp;&nbsp;%s<hr />", join, queryItem.LastSeen, rate, queryItem.QueryText)
	}
	output += "</body></html>"

	_, err := w.Write([]byte(output))
	if err != nil {
		log.Println(w.Write([]byte(err.Error())))
	}
}
