package scanproto

import (
	"io"
	"strings"
	"sync"

	"golang.org/x/net/context"
	"google.golang.org/grpc"

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

// simpleApiTransactionSource implements the TransactionSource interface by reading
// data from the trace grpc api.
type simpleApiTransactionSource struct {
	hostport string
	conn     *grpc.ClientConn
	client   api.TraceClient
	queries  []*api.Query

	ch     chan *api.Transaction
	errors chan error
	wg     sync.WaitGroup

	stopMu sync.Mutex
	stop   bool
	ctx    context.Context
	cancel context.CancelFunc
}

func NewSimpleApiTransactionSource(hostport string, services string, hosts string) TransactionSource {
	queries := []*api.Query{}
	for _, service := range strings.Split(services, ",") {
		queries = append(queries, &api.Query{Comparisons: []*api.Comparison{&api.Comparison{ServiceName: &api.StringComparison{Value: service}}}})
	}
	for _, service := range strings.Split(hosts, ",") {
		queries = append(queries, &api.Query{Comparisons: []*api.Comparison{&api.Comparison{ServiceHost: &api.StringComparison{Value: service}}}})
	}

	ctx, cancel := context.WithCancel(context.Background())

	return &simpleApiTransactionSource{
		hostport: hostport,
		queries:  queries,
		ch:       make(chan *api.Transaction, readBuffer),
		errors:   make(chan error, 1),
		ctx:      ctx,
		cancel:   cancel,
	}
}

func (src *simpleApiTransactionSource) Run() {
	defer func() {
		src.wg.Wait()
		close(src.ch)
		close(src.errors)
	}()

	src.stopMu.Lock()
	if src.stop {
		src.stopMu.Unlock()
		return
	}
	src.wg.Add(1)
	defer src.wg.Done()
	src.ctx, src.cancel = context.WithCancel(context.Background())
	src.stopMu.Unlock()

	var err error
	src.conn, err = grpc.Dial(src.hostport, grpc.WithInsecure())
	if err != nil {
		src.errors <- err
		return
	}

	defer func() {
		err := src.conn.Close()
		if err != nil {
			src.errors <- err
		}
	}()

	src.client = api.NewTraceClient(src.conn)

	firehose, err := src.client.Firehose(src.ctx, &api.FirehoseRequest{Query: &api.Query{Subqueries: src.queries}})
	if err != nil {
		src.errors <- err
		return
	}

	for {
		tx, err := firehose.Recv()
		if err == io.EOF {
			break
		}
		if err == nil {
			src.ch <- tx
		} else if src.ctx.Err() != nil {
			break
		} else {
			src.errors <- err
		}
	}
}

func (src *simpleApiTransactionSource) Transactions() <-chan *api.Transaction {
	return src.ch
}

func (src *simpleApiTransactionSource) Errors() <-chan error {
	return src.errors
}

func (src *simpleApiTransactionSource) Stop() {
	src.stopMu.Lock()
	src.stop = true
	src.cancel()
	src.stopMu.Unlock()

	src.wg.Wait()
}
