// Package scanproto provides tools for reading streams of proto-encoded Events.
package scanproto

import (
	"bufio"
	"log"
	"os"
	"sync"
	"time"

	"code.justin.tv/release/trace/events"
)

// readBuffer is the number of messages to hold in internal decoding
// buffers by default.
var readBuffer = 1024

// TimestampedEvent is a wrapper around an events.Event that allows an
// eventSource to override the internal event time with a more
// authoritative source while preserving the original event time
// inside the event
type TimestampedEvent struct {
	Time  time.Time
	Event *events.Event
}

// EventSource provides a channel of *event.Events. Its Run function
// starts writing data into the channel.
type EventSource interface {
	Run()
	Events() <-chan *TimestampedEvent
	Errors() <-chan error
	Stop() // a blocking call which closes inputs and its events channel
}

// fileEventSource implements the eventSource interface by reading
// data from a protobuf file.
type fileEventSource struct {
	filename string
	workers  int
	ch       chan *TimestampedEvent

	closed bool
	wg     *sync.WaitGroup
}

// NewFileEventSource creates a new fileEventSource, set up to consume
// from filename using workers goroutines, but does not actually open
// the file or start processing.
func NewFileEventSource(filename string, workers int) EventSource {
	return &fileEventSource{
		filename: filename,
		workers:  workers,
		ch:       make(chan *TimestampedEvent, readBuffer),
		closed:   false,
	}
}

func (es *fileEventSource) Stop() {
	es.closed = true
	es.wg.Wait()
}

// run starts the fileEventSource's processing: it opens the file and
// spins up fileEventSource.workers goroutines to process the data.
func (es *fileEventSource) Run() {
	handle, err := os.Open(es.filename)
	if err != nil {
		log.Fatal(err)
	}
	defer handle.Close()

	var wg sync.WaitGroup

	// reserve a channel for encoded data
	encoded := make(chan []byte, readBuffer)

	// start the workers to process the encoded bytes
	es.wg.Add(1)
	go func() {
		// decodeEvents blocks until the input encoded channel is closed
		decodeEvents(es.ch, encoded, es.workers)
		close(es.ch)
		es.wg.Done()
	}()

	// start populating the encoded data channel
	es.wg.Add(1)
	go func() {
		sc := NewEventSetScanner(bufio.NewReader(handle))
		for sc.Scan() && !es.closed {
			b := make([]byte, len(sc.Bytes()))
			copy(b, sc.Bytes())
			encoded <- b
		}
		if err = sc.Err(); err != nil {
			log.Fatalf("scanner err=%q", err)
		}
		close(encoded)
		es.wg.Done()
	}()

	wg.Wait()
}

func (es *fileEventSource) Events() <-chan *TimestampedEvent {
	return es.ch
}

func (es *fileEventSource) Errors() <-chan error {
	return nil
}
