package scanproto

import (
	"io"
	"strings"

	"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
	ctx      context.Context
	cancel   context.CancelFunc
	queries  []*api.Query

	ch       chan *api.Transaction
	errors   chan error
	done     chan struct{}
	shutdown bool
}

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}}}})
	}

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

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

	src.client = api.NewTraceClient(src.conn)
	src.ctx, src.cancel = context.WithCancel(context.Background())

	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 {
			src.errors <- err
		}
		if src.shutdown {
			break
		}
	}

	close(src.done)
	src.Stop()
}

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

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

func (src *simpleApiTransactionSource) Stop() {
	if src.shutdown {
		return
	}
	src.shutdown = true
	<-src.done
	src.cancel()
	err := src.conn.Close()
	if err != nil {
		src.errors <- err
	}
	close(src.ch)
	close(src.errors)
}
