package fhevent

import (
	"bytes"
	"context"
	"errors"
	"sync"

	"code.justin.tv/sse/malachai/pkg/log"
	"github.com/cactus/go-statsd-client/statsd"
)

// RecordBuilder builds items that are up to size Item and writes them to
// ReadyChannel.
type RecordBuilder struct {
	Item *bytes.Buffer
	// How large records can be before they're shipped
	MaxItemSizeInBytes int
	// Channel we will ship complete records to
	ReadyChannel chan []byte

	Statter statsd.Statter
	Logger  log.S2SLogger

	itemLock sync.Mutex
	closed   bool
}

var (
	errItemLargerThanMaxItemSize = errors.New("item is too large to send to buffer")
	errClosed                    = errors.New("record builder is closed")
)

// Add ships the buffer to ReadyChannel if ready then adds item to the
// item buffer.
func (bb *RecordBuilder) Add(ctx context.Context, item []byte) error {
	if bb.closed {
		return errClosed
	}

	if len(item) > bb.MaxItemSizeInBytes {
		return errItemLargerThanMaxItemSize
	}

	bb.itemLock.Lock()
	// ship if the item is going to be bigger than MaxItemSizeInBytes
	if bb.Item.Len() > 0 {
		if bb.Item.Len()+len(item) > bb.MaxItemSizeInBytes {
			if err := bb.ship(ctx); err != nil {
				bb.itemLock.Unlock()
				return err
			}
		}
	}

	_, err := bb.Item.Write(item)
	bb.itemLock.Unlock()
	return err
}

// Ship the current buffer to ReadyChannel.
func (bb *RecordBuilder) Ship(ctx context.Context) error {
	bb.itemLock.Lock()
	err := bb.ship(ctx)
	bb.itemLock.Unlock()
	return err
}

// Close blocks Add calls
func (bb *RecordBuilder) Close() {
	bb.itemLock.Lock()
	bb.closed = true
	bb.itemLock.Unlock()
}

func (bb *RecordBuilder) ship(ctx context.Context) error {
	if bb.Item.Len() == 0 {
		return nil
	}

	bs := bb.Item.Bytes()
	bsToShip := make([]byte, len(bs))
	copy(bsToShip, bs)

	select {
	case bb.ReadyChannel <- bsToShip:
		bb.Item.Reset()
	case <-ctx.Done():
		return ctx.Err()
	default:
		// drops batch if ReadyChannel is blocked.
		bb.Item.Reset()

		bb.Logger.Warn("firehose events client buffer full, dropping log batch")
		statErr := bb.Statter.Inc("event_logger.batch.dropped", 1, 1.0)
		if statErr != nil {
			bb.Logger.Error("failed to increment stat event_logger.batch.dropped: " + statErr.Error())
		}
	}
	return nil
}
