package stream

import (
	"reflect"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/multierr"
)

type Stream struct {
	ite Iterator
}

func OfMap(o any) *Stream {
	v := reflect.ValueOf(o)
	if v.Kind() != reflect.Map {
		panic("not map")
	}

	keys := v.MapKeys()
	return &Stream{
		ite: &baseIterator{
			keys:  keys,
			index: 0,
			len:   len(keys),
			src:   v,
		},
	}
}

func (s *Stream) CollectEmptySlice(dst any) {
	fv := reflect.ValueOf(dst).Elem()

	results := reflect.MakeSlice(fv.Type(), 0, 0)
	for val, ok := s.ite.Next(); ok; val, ok = s.ite.Next() {
		results = reflect.Append(results, reflect.ValueOf(val))
	}

	// arr := reflect.Append(fv, results...)
	fv.Set(results)
}

func (s *Stream) Collect(dst any) {
	fv := reflect.ValueOf(dst).Elem()
	len, ok := s.ite.Len()
	if !ok {
		log.Error("unsupport dynamic slice")
		return
	}
	newSlice := reflect.MakeSlice(fv.Type(), len, len)

	count := 0
	for val, ok := s.ite.Next(); ok; val, ok = s.ite.Next() {
		newSlice.Index(count).Set(reflect.ValueOf(val))
		count++
	}
	fv.Set(newSlice)
}

func (s *Stream) Filter(f any) *Stream {
	t := reflect.TypeOf(f)
	if t.Kind() != reflect.Func {
		panic("no func")
	}

	if t.NumIn() != 1 {
		panic("parameter count must be 1")
	}

	if t.Out(0).Kind() != reflect.Bool {
		panic("no bool")
	}

	return &Stream{
		ite: &filterIterator{
			ite: s.ite,
			fv:  reflect.ValueOf(f),
		},
	}
}

type filterIterator struct {
	ite Iterator
	fv  reflect.Value
}

func (ite *filterIterator) Len() (int, bool) {
	return 0, false
}

func (ite *filterIterator) Next() (any, bool) {
	for {
		v, existed := ite.ite.Next()
		if !existed {
			return v, false
		}

		ok := ite.fv.Call([]reflect.Value{reflect.ValueOf(v)})[0].Interface().(bool)
		if ok {
			return v, true
		}
	}
}

func (s *Stream) Map(f any) *Stream {
	ft := reflect.TypeOf(f)
	if ft.Kind() != reflect.Func {
		panic("no func")
	}

	ite := &mapIterator{
		ite: s.ite,
		fv:  reflect.ValueOf(f),
	}

	return &Stream{
		ite: ite,
	}
}

func OfSlice(o any) *Stream {
	rt := reflect.ValueOf(o)
	if rt.Kind() != reflect.Slice {
		panic("no slice")
	}

	return &Stream{
		ite: &sliceIterator{
			index: 0,
			len:   rt.Len(),
			src:   rt,
		},
	}
}

func OfIterator(ite Iterator) *Stream {
	return &Stream{
		ite: ite,
	}
}

type mapIterator struct {
	ite Iterator
	fv  reflect.Value
}

func (ite *mapIterator) Next() (any, bool) {
	val, ok := ite.ite.Next()
	if !ok {
		return nil, false
	}

	results := ite.fv.Call([]reflect.Value{reflect.ValueOf(val)})
	if len(results) != 1 {
		panic("map return count must be 1")
	}

	return results[0].Interface(), ok
}

func (ite *mapIterator) Len() (int, bool) {
	return ite.ite.Len()
}

func (s *Stream) Each(f any) {
	t := reflect.TypeOf(f)
	if t.Kind() != reflect.Func {
		panic("no func")
	}

	if t.NumIn() != 1 {
		panic("parameter count must be 1")
	}

	fv := reflect.ValueOf(f)

	for val, ok := s.ite.Next(); ok; val, ok = s.ite.Next() {
		fv.Call([]reflect.Value{reflect.ValueOf(val)})
	}
}

var errType = reflect.TypeOf((*error)(nil)).Elem()

func (s *Stream) Consume(f any) error {
	ft := reflect.TypeOf(f)
	if ft.Kind() != reflect.Func {
		return errors.New("f is not a func kind")
	}

	if ft.NumIn() != 1 {
		return errors.New("f only has 1 input parameter")
	}

	rv := reflect.ValueOf(f)

	var err error
	for val, ok := s.ite.Next(); ok; val, ok = s.ite.Next() {
		results := rv.Call([]reflect.Value{reflect.ValueOf(val)})

		for _, r := range results {
			if !r.Type().Implements(errType) {
				continue
			}

			if !r.IsZero() {
				err = multierr.Append(err, r.Interface().(error))
			}
		}
	}

	return err
}

func (s *Stream) BatchSink(f any, batchSize int) error {
	ft := reflect.TypeOf(f)
	if ft.Kind() != reflect.Func {
		return errors.New("f is not a func kind")
	}
	if ft.NumIn() != 1 {
		return errors.New("f only has 1 input parameter")
	}

	arr := reflect.MakeSlice(ft.In(0), batchSize, batchSize)

	var err error
	count := 0

	rv := reflect.ValueOf(f)
	for val, ok := s.ite.Next(); ok; val, ok = s.ite.Next() {
		arr.Index(count).Set(reflect.ValueOf(val))
		count++

		if count >= batchSize {
			err = multierr.Append(err, processErr(rv.Call([]reflect.Value{arr})))
			count = 0
			arr = reflect.MakeSlice(ft.In(0), batchSize, batchSize)
		}
	}

	if count > 0 {
		newarr := reflect.MakeSlice(ft.In(0), count, count)
		for i := 0; i < count; i++ {
			newarr.Index(i).Set(arr.Index(i))
		}
		err = multierr.Append(err, processErr(rv.Call([]reflect.Value{newarr})))
	}

	return err
}

func processErr(results []reflect.Value) error {
	var err error
	for _, r := range results {
		if !r.Type().Implements(errType) {
			continue
		}

		if !r.IsZero() {
			err = multierr.Append(err, r.Interface().(error))
		}
	}

	return err
}

func (s *Stream) Sink(f any) error {
	ft := reflect.TypeOf(f)
	if ft.Kind() != reflect.Func {
		return errors.New("f is not a func kind")
	}
	if ft.NumIn() != 1 {
		return errors.New("f only has 1 input parameter")
	}

	size, ok := s.ite.Len()
	if !ok {
		log.Error("unsupport dynamic slice")
		return errors.New("unsupport dynamic slice")
	}
	slice := reflect.MakeSlice(ft.In(0), size, size)

	count := 0
	for val, ok := s.ite.Next(); ok; val, ok = s.ite.Next() {
		slice.Index(count).Set(reflect.ValueOf(val))
		count++
	}

	results := reflect.ValueOf(f).Call([]reflect.Value{slice})
	errType := reflect.TypeOf((*error)(nil)).Elem()
	for _, r := range results {
		if !r.Type().Implements(errType) {
			continue
		}

		if !r.IsZero() {
			return r.Interface().(error)
		}
	}
	return nil
}

type Iterator interface {
	Next() (any, bool)
	Len() (int, bool)
}

type sliceIterator struct {
	index int
	len   int
	src   reflect.Value
}

func (ite *sliceIterator) Len() (int, bool) {
	return ite.len, true
}

func (ite *sliceIterator) Next() (any, bool) {
	if ite.index > ite.len-1 {
		return nil, false
	}

	v := ite.src.Index(ite.index).Interface()
	ite.index++

	return v, true
}

type baseIterator struct {
	keys  []reflect.Value
	index int
	len   int
	src   reflect.Value
}

func (ite *baseIterator) Len() (int, bool) {
	return ite.len, true
}

func (ite *baseIterator) Next() (any, bool) {
	if ite.index > ite.len-1 {
		return nil, false
	}

	k := ite.keys[ite.index]
	val := ite.src.MapIndex(k)

	ite.index++

	return MapEntry{k, val}, true
}

type MapEntry struct {
	Key   reflect.Value
	Value reflect.Value
}

type SmartDataAccessor interface {
	Accessor() func(query PaginationFilter) (any, int64, error)
}

type PaginationFilter struct {
	Page  int
	Limit int

	Query string
}

// JoinDataAccessorToSlice return specify data from accessor1 and accessor2
//
// There are 3 cases:
// 1. accessor1 contains all wanted records
// 2. accessor1 contains part of the data. Need to fetch data from accessor2
// 3. accessor2 contains all wanted records, the results need to skip the data in case2
func JoinDataAccessorToSlice(accessor1, accessor2 SmartDataAccessor, query PaginationFilter, dst any) (int64, error) {
	if reflect.TypeOf(dst).Kind() != reflect.Ptr {
		return 0, errors.New("the type must be prt type")
	}

	dstRV := reflect.ValueOf(dst).Elem()

	v1, total1, err := accessor1.Accessor()(query)
	if err != nil {
		return 0, err
	}
	rv1 := reflect.ValueOf(v1)
	v1Len := 0

	if v1 != nil && !rv1.IsZero() {
		r1 := reflect.TypeOf(v1)
		if r1.Kind() != dstRV.Kind() {
			return 0, errors.New("accessor1 must be slice return type")
		}

		v1Len = rv1.Len()
	}

	results := reflect.MakeSlice(dstRV.Type(), 0, 0)
	for i := 0; i < v1Len; i++ {
		results = reflect.Append(results, rv1.Index(i))
	}
	defer func() {
		dstRV.Set(results)
	}()
	// case1: accessor1 include all wanted data
	if v1Len >= query.Limit {
		_, total2, err := accessor2.Accessor()(query)
		if err != nil {
			return 0, err
		}
		return total1 + total2, nil
	}

	result2 := reflect.MakeSlice(dstRV.Type(), 0, 0)
	queryPages := calculatePages(int(total1), query)

	var (
		total2   int64
		v2       any
		queryErr error
	)

	for _, page := range queryPages {
		query2 := PaginationFilter{
			Page:  page,
			Limit: query.Limit,
			Query: query.Query,
		}

		v2, total2, queryErr = accessor2.Accessor()(query2)
		if queryErr != nil {
			return 0, queryErr
		}

		if v2 == nil {
			break
		}

		r2 := reflect.TypeOf(v2)
		if r2.Kind() != dstRV.Kind() {
			return 0, errors.New("accessor2 must be slice return type")
		}

		// accessor2 has been drained
		rv2 := reflect.ValueOf(v2)
		if rv2.IsZero() {
			break
		}

		v2Len := rv2.Len()
		for i := 0; i < v2Len; i++ {
			result2 = reflect.Append(result2, rv2.Index(i))
		}

		if v2Len < query.Limit {
			// accessor2 has been drained
			break
		}
	}

	skip := 0
	if result2.Len() > 0 {
		skip = calculateSkip(int(total1), queryPages, query)
	}

	for i, j := skip, 0; i < min(result2.Len(), skip+query.Limit) && j < query.Limit-v1Len; i, j = i+1, j+1 {
		results = reflect.Append(results, result2.Index(i))
	}

	return total1 + total2, nil
}

func min(v1, v2 int) int {
	if v1 > v2 {
		return v2
	}

	return v1
}

func calculatePages(total1 int, pq PaginationFilter) []int {
	if total1 >= pq.Page*pq.Limit {
		return nil
	}
	if total1%pq.Limit == 0 {
		return []int{pq.Page - total1/pq.Limit}
	}

	page := pq.Page - total1/pq.Limit
	if page <= 1 {
		return []int{page}
	}

	return []int{page - 1, page}
}

func calculateSkip(total1 int, pages []int, pq PaginationFilter) int {
	if len(pages) <= 1 {
		return 0
	}

	return pq.Limit - total1%pq.Limit
}