package scanproto

import (
	"log"
	"sync"
	"time"

	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/events"
	"code.justin.tv/release/trace/pbmsg"
	"github.com/golang/protobuf/proto"
)

// decodeEvents converts a channel of byte-encoded EventSets into a
// channel of TimestampedEvents, using nWorkers goroutines to do
// so. Blocks until the input channel closes and all workers have
// successfully passed all their messages into the output channel.
func decodeEvents(out chan<- *TimestampedEvent, in <-chan []byte, nWorkers int) {
	var wg sync.WaitGroup
	for i := 0; i < nWorkers; i++ {
		wg.Add(1)
		go func() {
			for b := range in {
				decoded, err := decodeSingleEventSet(b)
				if err != nil {
					log.Fatalf("proto unmarshal err=%q", err)
				}
				for _, ev := range decoded {
					out <- ev
				}
			}
			wg.Done()
		}()
	}
	wg.Wait()
}

// decodeSingleEventSet decodes an encoded pbmsg.EventSet into a slice
// of TimestampedEvents
func decodeSingleEventSet(b []byte) ([]*TimestampedEvent, error) {
	var es events.EventSet
	err := proto.Unmarshal(b, &es)
	if err != nil {
		return nil, err
	}
	events, err := events.UnpackEventSet(&es)
	if err != nil {
		return nil, err
	}
	out := make([]*TimestampedEvent, len(events))
	for i, ev := range events {
		out[i] = &TimestampedEvent{
			Time:  time.Unix(0, ev.Time),
			Event: ev,
		}
	}
	return out, nil
}

// decodeTransaction parses a byte slice as a proto-encoded
// Transaction and returns the Transaction so decoded.
func decodeTransaction(b []byte) (*pbmsg.Transaction, error) {
	var tx pbmsg.Transaction
	if err := proto.Unmarshal(b, &tx); err != nil {
		return nil, err
	}
	return &tx, nil
}

// decodeTransactionSet parses a byte slice as a proto-encoded
// TransactionSet and returns old-style, deprecated pbmsg.Transactions
// as a slice.
func decodeTransactionSet_deprecated(b []byte) ([]*pbmsg.Transaction, error) {
	var ts api.TransactionSet
	if err := proto.Unmarshal(b, &ts); err != nil {
		return nil, err
	}
	return pbmsg.DowngradeTransactionSet(&ts), nil
}

// Try to decode bytes as a TransactionSet. Fall back on decoding as a
// single Transaction. Write output Transaction(s) into the given
// channel. Return err if both fail.
func decodeIntoTxChannel(b []byte, ch chan *pbmsg.Transaction) error {
	txs, err := decodeTransactionSet_deprecated(b)
	if err == nil {
		for _, tx := range txs {
			ch <- tx
		}
		return nil
	}

	tx, err := decodeTransaction(b)
	if err != nil {
		return err
	}

	ch <- tx
	return nil
}
