package jsonl

import (
	"encoding/json"
	"io"

	"github.com/appbaseio/abc/importer/client"
	"github.com/appbaseio/abc/importer/message"
	"github.com/appbaseio/abc/importer/message/data"
	"github.com/appbaseio/abc/importer/message/ops"
	"github.com/appbaseio/abc/log"
)

var _ client.Reader = &Reader{}

// Reader implements the behavior defined by client.Reader for interfacing with the file.
type Reader struct {
	fileName string
	typeName string
}

func newReader(typeName string) client.Reader {
	return &Reader{typeName: typeName}
}

func (r *Reader) Read(_ map[string]client.MessageSet, filterFn client.NsFilterFunc) client.MessageChanFunc {
	return func(s client.Session, done chan struct{}) (chan client.MessageSet, error) {
		out := make(chan client.MessageSet)
		session := s.(*Session)
		ns := session.file.Name()

		go func() {
			defer close(out)
			results := r.decodeJSONL(session, done)
			for {
				select {
				case <-done:
					return
				case result, ok := <-results:
					if !ok {
						log.With("file", ns).Infoln("Read completed")
						return
					}
					if filterFn(ns) {
						out <- client.MessageSet{
							Msg: message.From(ops.Insert, ns, result),
						}
					}
				}
			}
		}()

		return out, nil
	}
}

func (r *Reader) decodeJSONL(s *Session, done chan struct{}) chan data.Data {
	out := make(chan data.Data)
	go func() {
		defer close(out)
		dec := json.NewDecoder(s.file)
		for {
			doc := make(data.Data)
			if err := dec.Decode(&doc); err == io.EOF {
				return
			} else if err != nil {
				log.With("file", s.file.Name()).Errorf("Can't unmarshal document (%v)", err)
				continue
			}
			out <- doc
		}
	}()
	return out
}
