package txstore

import (
	"sync"

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

type byteSource struct {
	workers int

	txSet    bool
	nextFn   func() bool
	bytesFn  func() []byte
	endErrFn func() error

	wg   sync.WaitGroup
	ch   chan *api.Transaction
	errs chan error
	stop chan struct{}
}

func (src *byteSource) Stop() {
	close(src.stop)
	src.wg.Wait()
}

func (src *byteSource) Run() {
	defer close(src.errs)
	defer close(src.ch)

	encoded := make(chan []byte, 1<<10)

	for i := 0; i < src.workers; i++ {
		src.wg.Add(1)
		go func() {
			defer src.wg.Done()

			if src.txSet {
				for b := range encoded {
					var txs api.TransactionSet
					if err := proto.Unmarshal(b, &txs); err != nil {
						src.errs <- err
						continue
					}
					for _, tx := range txs.Transaction {
						CleanTx(tx)
						src.ch <- tx
					}
				}
			} else {
				for b := range encoded {
					var tx api.Transaction
					if err := proto.Unmarshal(b, &tx); err != nil {
						src.errs <- err
						continue
					}
					CleanTx(&tx)
					src.ch <- &tx
				}
			}
		}()
	}

	src.wg.Add(1)
	go func() {
		defer src.wg.Done()
		defer close(encoded)

		for src.next() {
			select {
			case encoded <- append([]byte(nil), src.bytes()...):
			case <-src.stop:
				return
			}
		}
		if err := src.endErr(); err != nil {
			src.errs <- err
		}
	}()

	src.wg.Wait()
}

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

func (src *byteSource) Errors() <-chan error {
	return src.errs
}

func (src *byteSource) next() bool {
	if fn := src.nextFn; fn != nil {
		return fn()
	}
	return false
}

func (src *byteSource) bytes() []byte {
	if fn := src.bytesFn; fn != nil {
		return fn()
	}
	return nil
}

func (src *byteSource) endErr() error {
	if fn := src.endErrFn; fn != nil {
		return fn()
	}
	return nil
}
