package main

import (
	"code.justin.tv/ear/kinesis/combiner"
	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/common"
	"code.justin.tv/release/trace/pbmsg"
	"github.com/golang/protobuf/proto"
)

func combineTransactions(txCh <-chan *pbmsg.Transaction, blobber *combiner.RecordCombiner) {
	for tx := range txCh {
		blobber.Add((*transactionRecord)(convertTransaction(tx)))
	}
	blobber.Close()
}

type transactionRecord api.Transaction

func (r *transactionRecord) Key() int {
	return int(r.TransactionId[0])
}

func (r *transactionRecord) NewSet() combiner.RecordSet {
	return newTransactionRecordSet()
}

type transactionRecordSet struct {
	transactions *api.TransactionSet
	buf          *proto.Buffer
}

func newTransactionRecordSet() *transactionRecordSet {
	return &transactionRecordSet{
		transactions: &api.TransactionSet{
			Transaction: make([]*api.Transaction, 0),
		},
		buf: proto.NewBuffer(nil),
	}
}

func (rs *transactionRecordSet) Push(r combiner.Record) {
	tx := (*api.Transaction)(r.(*transactionRecord))
	rs.transactions.Transaction = append(rs.transactions.Transaction, tx)
	// Append the Transaction to the data, as well. Transactionsets can be simply
	// concatenated.
	rs.buf.Marshal(&api.TransactionSet{Transaction: []*api.Transaction{tx}})
}

func (rs *transactionRecordSet) Pop() combiner.Record {
	var r *api.Transaction
	r, rs.transactions.Transaction = rs.transactions.Transaction[len(rs.transactions.Transaction)-1], rs.transactions.Transaction[:len(rs.transactions.Transaction)-1]
	rs.buf.Reset()
	rs.buf.Marshal(rs.transactions)
	return (*transactionRecord)(r)
}

func (rs *transactionRecordSet) Size() int {
	return len(rs.buf.Bytes())
}

func (rs *transactionRecordSet) Encode() []byte {
	return rs.buf.Bytes()
}

func convertTransaction(old *pbmsg.Transaction) *api.Transaction {
	return &api.Transaction{
		TransactionId: old.GetTransactionId(),
		Root:          convertCall(old.Root),
		Client:        convertService(old.Client),
	}
}

func convertService(old *pbmsg.Service) *api.Service {
	return &api.Service{
		Name: old.GetName(),
		Host: old.GetHost(),
		Pid:  old.GetPid(),
		Peer: old.GetPeer(),
	}
}

func convertTimestamp(old *pbmsg.Timestamp) *api.Timestamp {
	return &api.Timestamp{
		Time: old.GetTime(),
		Kind: common.Kind(old.GetKind()),
	}
}

func convertCall(old *pbmsg.Call) *api.Call {
	c := &api.Call{
		Path:          old.GetPath(),
		Svc:           convertService(old.GetSvc()),
		RequestSentTo: old.GetRequestSentTo(),
	}

	for _, oc := range old.GetSubcalls() {
		c.Subcalls = append(c.Subcalls, convertCall(oc))
	}

	if old.Params != nil {
		c.Params = &api.RPCParams{}

		if old.Params.Http != nil {
			c.Params.Http = &api.HTTPParams{
				UriPath: old.Params.Http.GetUriPath(),
				Status:  old.Params.Http.GetStatus(),
				Method:  common.Method(old.Params.Http.GetMethod()),
				Route:   old.Params.Http.GetRoute(),
			}
		}

		if old.Params.Sql != nil {
			c.Params.Sql = &api.SQLParams{
				Dbname:        old.Params.Sql.GetDbname(),
				Dbuser:        old.Params.Sql.GetDbuser(),
				StrippedQuery: old.Params.Sql.GetStrippedQuery(),
				Tables:        old.Params.Sql.GetTables(),
			}
		}

		if old.Params.Grpc != nil {
			c.Params.Grpc = &api.GRPCParams{
				Method: old.Params.Grpc.GetMethod(),
			}
		}

		if old.Params.Memcached != nil {
			c.Params.Memcached = &api.MemcachedParams{
				Command:       common.Command(common.Command_value[old.Params.Memcached.GetCommand().String()]),
				NKeysRequest:  old.Params.Memcached.GetNKeysRequest(),
				NKeysResponse: old.Params.Memcached.GetNKeysResponse(),
			}
		}
	}

	for _, ts := range old.GetClientTimestamps() {
		c.ClientTimestamps = append(c.ClientTimestamps, convertTimestamp(ts))
	}

	for _, ts := range old.GetServiceTimestamps() {
		c.ServiceTimestamps = append(c.ServiceTimestamps, convertTimestamp(ts))
	}

	return c
}
