package aggregator

import (
	"github.com/olivere/elastic/v7"
	log "github.com/sirupsen/logrus"
	"github.com/spf13/viper"
	"iwarma.ru/console/correlator/config"
	"iwarma.ru/console/correlator/events"
	"time"
)

func Aggregator(store *events.AggregatedEventStore, window time.Duration, stop chan interface{}) {
	cl := log.WithField("part", "aggregator")
	ticker1 := time.NewTicker(window)
	ticker2 := time.NewTicker(viper.GetDuration(config.AggregatorIterationDuration))

	// We have one query to all iterations
	// Maybe, we should use MatchAll query
	query := elastic.NewMatchAllQuery()

	for {
		select {
		case <-stop:
			cl.Info("Stopping aggregator")
			ticker1.Stop()
			ticker2.Stop()
			store.SendUpdateBulk()
			return

		case <-ticker1.C:
			store.UpdateWindow(events.TimeWindow{
				Begin: time.Now().UTC(),
				End:   time.Now().UTC().Add(window),
			})
			cl.Debugf("Aggrerator windows change: %v", store.GetWindow())

		case <-ticker2.C:
			cl.Trace("Start aggregator iteration")
			raw, errs1 := store.GetClient().Query(events.GetNormalizedIndexName(), query)
			eventsChan, errs2 := events.ParseEvents(raw, errs1, viper.GetInt(config.Threads))

			store.AddEvents(eventsChan)

			// Log errors if we have some
			go func() {
				for err := range errs2 {
					if err != nil {
						cl.Errorf("%+v", err)
					}
				}
			}()
			store.SendUpdateBulk()
			cl.Trace("Finish aggregator iteration")
		}
	}
}
