package main

import (
	"math"
	"sync"

	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/pbmsg"
)

func filterQueries(ch <-chan *api.Transaction, manager *queryManager, workers int) {
	var wg sync.WaitGroup

	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for tx := range ch {
				cleanAPITransaction(tx)
				pbTx := pbmsg.DowngradeTransaction(tx)
				manager.Each(func(q *query) {
					if matchRequest(tx, q.req) {
						// If we can't write immediately, drop it and move on
						select {
						case q.ch <- tx:
						default:
						}
						// If we can't write immediately, drop it and move on
						select {
						case q.pbCh <- pbTx:
						default:
						}
					}
				})
			}
		}()
	}

	wg.Wait()

	var reqs []*api.FirehoseRequest
	manager.Each(func(q *query) {
		reqs = append(reqs, q.req)
	})
	for _, req := range reqs {
		manager.Remove(req)
	}
}

func matchRequest(tx *api.Transaction, req *api.FirehoseRequest) bool {
	if req.Sampling > 0 && (len(tx.TransactionId) < 1 ||
		float64(tx.TransactionId[0])/math.MaxUint64 > req.Sampling) {
		return false
	}
	return matchQuery(tx.GetRoot(), req.GetQuery())
}

// Performs a breadth-first search of the call tree
func matchQuery(call *api.Call, query *api.Query) bool {
	calls := make([]*api.Call, 0)
	calls = append(calls, call)

CallMatching:
	for len(calls) > 0 {
		call, calls = calls[0], calls[1:]
		calls = append(calls, call.GetSubcalls()...)
		for _, comparison := range query.GetComparisons() {
			if !matchComparison(call, comparison) {
				continue CallMatching
			}
		}
		for _, subquery := range query.GetSubqueries() {
			if !matchQuery(call, subquery) {
				continue CallMatching
			}
		}
		return true
	}

	return false
}

func matchComparison(call *api.Call, comparison *api.Comparison) bool {
	// bool != bool is go's version of XOR
	svc := call.GetSvc()
	if svc == nil {
		svc = new(api.Service)
	}
	http := call.GetParams().GetHttp()
	if http == nil {
		http = new(api.HTTPParams)
	}

	return comparison.Not != (matchStringComparison(svc.Name, comparison.GetServiceName()) &&
		matchStringComparison(svc.Host, comparison.GetServiceHost()) &&
		matchIntegerComparison(int(svc.Pid), comparison.GetServicePid()) &&
		matchStringComparison(svc.Peer, comparison.GetServicePeer()) &&

		matchIntegerComparison(int(api.ClientDuration(call)), comparison.GetClientDuration()) &&
		matchIntegerComparison(int(api.ServiceDuration(call)), comparison.GetServerDuration()) &&

		matchStringComparison(http.Method.String(), comparison.GetHttpMethod()) &&
		matchStringComparison(http.Route, comparison.GetHttpRoute()) &&
		matchIntegerComparison(int(http.Status), comparison.GetHttpStatusCode()))
}

func matchStringComparison(value string, comparison *api.StringComparison) bool {
	if comparison == nil {
		return true
	}
	return value == comparison.Value
}

func matchIntegerComparison(value int, comparison *api.IntegerComparison) bool {
	if comparison == nil {
		return true
	}
	if comparison.Type == api.IntegerComparison_EQUALS {
		return value == int(comparison.Value)
	} else if comparison.Type == api.IntegerComparison_GREATER {
		return value > int(comparison.Value)
	} else if comparison.Type == api.IntegerComparison_LESSER {
		return value < int(comparison.Value)
	} else {
		return false
	}
}
