package main

import (
	"log"
	"sort"
	"time"

	"github.com/golang/protobuf/proto"

	"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/pbmsg"
	"code.justin.tv/release/trace/pbmsg/iana"
	"code.justin.tv/release/trace/pbmsg/memcached"
	"code.justin.tv/release/trace/scanproto"
)

const flushInterval time.Duration = 1 * time.Second

type TimestampedTransaction struct {
	expires     time.Time
	transaction *pbmsg.Transaction
}

func processEventStream(events <-chan *scanproto.TimestampedEvent, txs chan<- *pbmsg.Transaction, txTimeout time.Duration) {
	txMap := make(map[tx.TransactionID]*TimestampedTransaction)
	var (
		curTime time.Time
		evTime  time.Time
		txid    tx.TransactionID
	)
	for e := range events {
		ev := e.Event

		// Convert ev.TransactionId into an array for the map
		txid = *tx.IDForEvent(ev)

		evTime = e.Time

		// Get the transaction for this event
		var tx *pbmsg.Transaction
		transaction, ok := txMap[txid]
		if ok {
			tx = transaction.transaction
		} else {
			tx = &pbmsg.Transaction{}
			tx.TransactionId = ev.TransactionId
			tx.Root = &pbmsg.Call{}
			txMap[txid] = &TimestampedTransaction{evTime.Add(txTimeout), tx}
		}

		// Process event
		addEvent(tx, ev)

		// Check for expired transactions
		if evTime.Sub(curTime) >= flushInterval {
			curTime = evTime

			// Clean out expired transactions
			for k, t := range txMap {
				if t.expires.Before(curTime) {
					txs <- t.transaction
					delete(txMap, k)
				}
			}
		}
	}

	log.Println("flushing all remaining transactions")
	// Flush all remaining transactions
	for _, t := range txMap {
		txs <- t.transaction
	}

	close(txs)
}

// Add the data from a single event onto a transaction. No checks are
// on the Transaction ID - it's the caller's responsibility to route
// events properly.
func addEvent(tx *pbmsg.Transaction, ev *events.Event) {
	// Find the subcall that corresponds to this event's path
	call := findSubcallInTx(tx, ev.Path)

	fromServer := common.IsServerKind(ev.Kind)
	isRequest := common.IsRequestKind(ev.Kind)
	// Add service information if server-sent event
	if call.Svc == nil && fromServer {
		call.Svc = &pbmsg.Service{
			Name: proto.String(ev.Svcname),
			Host: proto.String(ev.Hostname),
			Pid:  proto.Int32(ev.Pid),
		}
		if extra := ev.GetExtra(); extra != nil {
			call.Svc.Peer = proto.String(extra.Peer)
		}
	}

	if !fromServer {
		if call == tx.Root {
			// This is data on the root client
			tx.Client = &pbmsg.Service{
				Name: proto.String(ev.Svcname),
				Host: proto.String(ev.Hostname),
				Pid:  proto.Int32(ev.Pid),
			}
		}
		if extra := ev.GetExtra(); extra != nil && extra.Peer != "" {
			call.RequestSentTo = proto.String(extra.Peer)
		}
	}

	if http := ev.GetExtra().GetHttp(); http != nil {
		addHTTPParams(call, http, fromServer)
	}
	if sql := ev.GetExtra().GetSql(); sql != nil {
		addSQLParams(call, sql, fromServer)
	}
	if grpc := ev.GetExtra().GetGrpc(); grpc != nil {
		addGRPCParams(call, grpc, fromServer)
	}
	if mc := ev.GetExtra().GetMemcached(); mc != nil {
		addMemcachedParams(call, mc, fromServer, isRequest)
	}

	// Add timestamps
	ts := &pbmsg.Timestamp{
		Time: proto.Int64(ev.Time),
		Kind: pbmsg.Event_Kind(ev.Kind).Enum(),
	}
	if fromServer {
		if call.ServiceTimestamps == nil {
			call.ServiceTimestamps = make([]*pbmsg.Timestamp, 0)
		}
		call.ServiceTimestamps = sortedInsert(call.ServiceTimestamps, ts)
	} else {
		if call.ClientTimestamps == nil {
			call.ClientTimestamps = make([]*pbmsg.Timestamp, 0)
		}
		call.ClientTimestamps = sortedInsert(call.ClientTimestamps, ts)
	}
}

// findSubcallInTx returns the appropriate pbmsg.Call for the given
// path within the given Transaction. If any call along the path does
// not exist, a new empty one is created and inserted into the
// transaction. The sorted order of subcalls is maintained through
// this creation.
func findSubcallInTx(tx *pbmsg.Transaction, path []uint32) *pbmsg.Call {
	depth := 0
	call := tx.Root
	// while not an exact match, dig deeper
	for paths.PathCompare(call.Path, path) != 0 {
		n := len(call.Subcalls)
		// Find the first subcall whose path, at this depth, is not
		// less than the input path
		idx := sort.Search(n, func(i int) bool {
			return call.Subcalls[i].Path[depth] >= path[depth]
		})
		// If the result of our search is len(call.Subcalls), or if it
		// points to a subcall whose path at this depth doesn't match
		// exactly, then no node exists which matches - we have to
		// fill in the subtree from here
		if idx == n || call.Subcalls[idx].Path[depth] != path[depth] {
			// insert the new call at the proper spot. This is bit of
			// cleverness is lifted from
			// [SliceTricks](https://code.google.com/p/go-wiki/wiki/SliceTricks)
			newCall := &pbmsg.Call{Path: path[:depth+1]}
			call.Subcalls = append(call.Subcalls, nil)
			copy(call.Subcalls[idx+1:], call.Subcalls[idx:])
			call.Subcalls[idx] = newCall
			// Short-circuit the outer loop logic for the remaining
			// depths of the path: we already know that it's not going
			// to find anything, after all.
			call = newCall
			for depth+1 < len(path) {
				depth += 1
				newCall = &pbmsg.Call{Path: path[:depth+1]}
				call.Subcalls = []*pbmsg.Call{newCall}
				call = newCall
			}
			return call

		} else {
			// Otherwise, our search was successful - we found a subcall
			// which has the right path component at this depth.
			call = call.Subcalls[idx]
			// we must go deeper!
			depth += 1
		}
	}
	return call
}

func addHTTPParams(call *pbmsg.Call, params *events.ExtraHTTP, fromServer bool) {
	// ensure that data structs aren't nil
	if call.Params == nil {
		call.Params = &pbmsg.RPCParams{
			Http: &pbmsg.HTTPParams{},
		}
	} else if call.Params.Http == nil {
		call.Params.Http = &pbmsg.HTTPParams{}
	}

	currentHttp := call.GetParams().GetHttp()

	currentStatusCode := currentHttp.GetStatus()
	if params.StatusCode != 0 && (currentStatusCode == 0 || fromServer) {
		call.Params.Http.Status = proto.Int32(params.StatusCode)
	}

	currentMethod := currentHttp.GetMethod()
	if params.Method != common.Method_UNKNOWN && (currentMethod == iana.HTTPMethod_UNKNOWN || fromServer) {
		httpMethod := iana.HTTPMethod(iana.HTTPMethod_value[params.Method.String()])
		call.Params.Http.Method = httpMethod.Enum()
	}

	currentUri := currentHttp.GetUriPath()
	if params.UriPath != "" && (currentUri == "" || fromServer) {
		call.Params.Http.UriPath = proto.String(params.UriPath)
	}

	svcname := call.GetSvc().GetName()
	if routeMap, ok := routes.Routes[svcname]; ok {
		call.Params.Http.Route = proto.String(routeMap.LookupTrimSlash(params.UriPath, params.Method.String()))
	}
}

func addSQLParams(call *pbmsg.Call, params *events.ExtraSQL, fromServer bool) {
	if call.Params == nil {
		call.Params = &pbmsg.RPCParams{
			Sql: &pbmsg.SQLParams{},
		}
	} else if call.Params.Sql == nil {
		call.Params.Sql = &pbmsg.SQLParams{}
	}

	currentSql := call.GetParams().GetSql()

	currentDb := currentSql.GetDbname()
	if params.DatabaseName != "" && (currentDb == "" || fromServer) {
		call.Params.Sql.Dbname = proto.String(params.DatabaseName)
	}

	currentUser := currentSql.GetDbuser()
	if params.DatabaseUser != "" && (currentUser == "" || fromServer) {
		call.Params.Sql.Dbuser = proto.String(params.DatabaseUser)
	}

	currentQuery := currentSql.GetStrippedQuery()
	if params.StrippedQuery != "" && (currentQuery == "" || fromServer) {
		call.Params.Sql.StrippedQuery = proto.String(params.StrippedQuery)
	}

	// TODO: call.params.Sql.Tables
}

func addGRPCParams(call *pbmsg.Call, params *events.ExtraGRPC, fromServer bool) {
	if call.Params == nil {
		call.Params = &pbmsg.RPCParams{
			Grpc: &pbmsg.GRPCParams{},
		}
	} else if call.Params.Grpc == nil {
		call.Params.Grpc = &pbmsg.GRPCParams{}
	}

	if params.Method != "" && (call.Params.Grpc.GetMethod() == "" || fromServer) {
		call.Params.Grpc.Method = proto.String(params.Method)
	}
}

func addMemcachedParams(dst *pbmsg.Call, src *events.ExtraMemcached, fromServer, isRequest bool) {
	if dst.Params == nil {
		dst.Params = &pbmsg.RPCParams{
			Memcached: &pbmsg.MemcachedParams{},
		}
	} else if dst.Params.Memcached == nil {
		dst.Params.Memcached = &pbmsg.MemcachedParams{}
	}

	if src.Command != common.Command_UNKNOWN_COMMAND && (dst.Params.Memcached.GetCommand() == memcached.Command_UNKNOWN_COMMAND || fromServer) {
		dst.Params.Memcached.Command = memcached.Command(memcached.Command_value[src.Command.String()]).Enum()
	}

	if isRequest {
		if src.NKeys != 0 && (dst.Params.Memcached.GetNKeysRequest() == 0 || fromServer) {
			dst.Params.Memcached.NKeysRequest = proto.Uint32(src.NKeys)
		}
	} else {
		if src.NKeys != 0 && (dst.Params.Memcached.GetNKeysResponse() == 0 || fromServer) {
			dst.Params.Memcached.NKeysResponse = proto.Uint32(src.NKeys)
		}
	}
}

// insert a new timestamp into the slice, maintaining sorted order
func sortedInsert(timestamps []*pbmsg.Timestamp, ts *pbmsg.Timestamp) []*pbmsg.Timestamp {
	idx := sort.Search(len(timestamps), func(i int) bool {
		return timestamps[i].GetTime() > ts.GetTime()
	})
	timestamps = append(timestamps, nil)
	copy(timestamps[idx+1:], timestamps[idx:])
	timestamps[idx] = ts
	return timestamps
}
