func assert(condition bool, msg string, v ...interface{}) {
	if !condition {
		panic(fmt.Sprintf("assert failed: "+msg, v...))
	}
}
assert will panic with a given formatted message if the given condition is false.
package ekanite
import (
	"expvar"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/ekanite/ekanite/input"
)
const (
	DefaultNumShards       = 16
	DefaultIndexDuration   = 24 * time.Hour
	DefaultRetentionPeriod = 24 * time.Hour
	RetentionCheckInterval = time.Hour
)
var (
	stats = expvar.NewMap("engine")
)
type EventIndexer interface {
	Index(events []*Event) error
}
type Batcher struct {
	indexer  EventIndexer
	size     int
	duration time.Duration
	c chan *input.Event
}
func NewBatcher(e EventIndexer, sz int, dur time.Duration, max int) *Batcher {
	return &Batcher{
		indexer:  e,
		size:     sz,
		duration: dur,
		c:        make(chan *input.Event, max),
	}
}
func (b *Batcher) Start(errChan chan<- error) error {
	go func() {
		batch := make([]*Event, 0, b.size)
		timer := time.NewTimer(b.duration)
		timer.Stop()  
		send := func() {
			err := b.indexer.Index(batch)
			if err != nil {
				stats.Add("batchIndexedError", 1)
				return
			}
			stats.Add("batchIndexed", 1)
			stats.Add("eventsIndexed", int64(len(batch)))
			if errChan != nil {
				errChan <- err
			}
			batch = make([]*Event, 0, b.size)
		}
		for {
			select {
			case event := <-b.c:
				idxEvent := &Event{
					event,
				}
				batch = append(batch, idxEvent)
				if len(batch) == 1 {
					timer.Reset(b.duration)
				}
				if len(batch) == b.size {
					timer.Stop()
					send()
				}
			case <-timer.C:
				stats.Add("batchTimeout", 1)
				send()
			}
		}
	}()
	return nil
}
func (b *Batcher) C() chan<- *input.Event {
	return b.c
}
type Engine struct {
	path            string         
	NumShards       int            
	IndexDuration   time.Duration  
	RetentionPeriod time.Duration  
	mu      sync.RWMutex
	indexes Indexes
	open bool
	done chan struct{}
	wg   sync.WaitGroup
	Logger *log.Logger
}
func NewEngine(path string) *Engine {
	return &Engine{
		path:            path,
		NumShards:       DefaultNumShards,
		IndexDuration:   DefaultIndexDuration,
		RetentionPeriod: DefaultRetentionPeriod,
		done:            make(chan struct{}),
		Logger:          log.New(os.Stderr, "[engine] ", log.LstdFlags),
	}
}
func (e *Engine) Open() error {
	if err := os.MkdirAll(e.path, 0755); err != nil {
		return err
	}
	d, err := os.Open(e.path)
	if err != nil {
		return fmt.Errorf("failed to open engine: %s", err.Error())
	}
	fis, err := d.Readdir(0)
	if err != nil {
		return err
	}
	for _, fi := range fis {
		if !fi.IsDir() || strings.HasPrefix(fi.Name(), ".") {
			continue
		}
		indexPath := filepath.Join(e.path, fi.Name())
		i, err := OpenIndex(indexPath)
		if err != nil {
			return fmt.Errorf("engine failed to open at index %s: %s", indexPath, err.Error())
		}
		log.Printf("engine opened index with %d shard(s) at %s", len(i.Shards), indexPath)
		e.indexes = append(e.indexes, i)
		sort.Sort(e.indexes)
	}
	e.wg.Add(1)
	go e.runRetentionEnforcement()
	e.open = true
	return nil
}
func (e *Engine) Close() error {
	if !e.open {
		return nil
	}
	for _, i := range e.indexes {
		if err := i.Close(); err != nil {
			return err
		}
	}
	close(e.done)
	e.wg.Wait()
	e.open = false
	return nil
}
func (e *Engine) Total() (uint64, error) {
	e.mu.RLock()
	defer e.mu.RUnlock()
	var total uint64
	for _, i := range e.indexes {
		t, err := i.Total()
		if err != nil {
			return 0, err
		}
		total += t
	}
	return total, nil
}
func (e *Engine) runRetentionEnforcement() {
	defer e.wg.Done()
	for {
		select {
		case <-e.done:
			return
		case <-time.After(RetentionCheckInterval):
			stats.Add("retentionEnforcementRun", 1)
			e.enforceRetention()
		}
	}
}
func (e *Engine) enforceRetention() {
	e.mu.Lock()
	defer e.mu.Unlock()
	filtered := e.indexes[:0]
	for _, i := range e.indexes {
		if i.Expired(time.Now().UTC(), e.RetentionPeriod) {
			if err := DeleteIndex(i); err != nil {
				e.Logger.Printf("retention enforcement failed to delete index %s: %s", i.path, err.Error())
			} else {
				e.Logger.Printf("retention enforcement deleted index %s", i.path)
				stats.Add("retentionEnforcementDeletions", 1)
			}
		} else {
			filtered = append(filtered, i)
		}
	}
	e.indexes = filtered
	return
}
func (e *Engine) indexForReferenceTime(t time.Time) *Index {
	for _, i := range e.indexes {
		if i.Contains(t) {
			return i
		}
	}
	return nil
}
func (e *Engine) createIndex(startTime, endTime time.Time) (*Index, error) {
	var idx *Index
	for _, i := range e.indexes {
		if i.startTime == startTime {
			idx = i
			break
		}
	}
	if idx != nil {
		startTime = idx.endTime  
		assert(!startTime.After(endTime), "new start time after end time")
	}
	i, err := NewIndex(e.path, startTime, endTime, e.NumShards)
	if err != nil {
		return nil, err
	}
	e.indexes = append(e.indexes, i)
	sort.Sort(e.indexes)
	e.Logger.Printf("index %s created with %d shards, start time: %s, end time: %s",
		i.Path(), e.NumShards, i.StartTime(), i.EndTime())
	return i, nil
}
func (e *Engine) createIndexForReferenceTime(rt time.Time) (*Index, error) {
	start := rt.Truncate(e.IndexDuration).UTC()
	end := start.Add(e.IndexDuration).UTC()
	return e.createIndex(start, end)
}
func (e *Engine) Index(events []*Event) error {
	e.mu.RLock()
	defer e.mu.RUnlock()
	var wg sync.WaitGroup
	subBatches := make(map[*Index][]Document, 0)
	for _, ev := range events {
		index := e.indexForReferenceTime(ev.ReferenceTime())
		if index == nil {
			func() {
				e.mu.RUnlock()
				defer e.mu.RLock()
				e.mu.Lock()
				defer e.mu.Unlock()
				index = e.indexForReferenceTime(ev.ReferenceTime())
				if index == nil {
					var err error
					index, err = e.createIndexForReferenceTime(ev.ReferenceTime())
					if err != nil || index == nil {
						panic(fmt.Sprintf("failed to create index for %s: %s", ev.ReferenceTime(), err))
					}
				}
			}()
		}
		if _, ok := subBatches[index]; !ok {
			subBatches[index] = make([]Document, 0)
		}
		subBatches[index] = append(subBatches[index], ev)
	}
	for index, subBatch := range subBatches {
		wg.Add(1)
		go func(i *Index, b []Document) {
			defer wg.Done()
			i.Index(b)
		}(index, subBatch)
	}
	wg.Wait()
	return nil
}
func (e *Engine) Search(query string) (<-chan string, error) {
	e.mu.RLock()
	defer e.mu.RUnlock()
	stats.Add("queriesRx", 1)
	c := make(chan string, 1)
	go func() {
		for i := len(e.indexes) - 1; i >= 0; i-- {
			e.Logger.Printf("searching index %s", e.indexes[i].Path())
			ids, err := e.indexes[i].Search(query)
			if err != nil {
				e.Logger.Println("error performing search:", err.Error())
				break
			}
			for _, id := range ids {
				b, err := e.indexes[i].Document(id)
				if err != nil {
					e.Logger.Println("error getting document:", err.Error())
					break
				}
				stats.Add("docsIDsRetrived", 1)
				c <- string(b)  
			}
		}
		close(c)
	}()
	return c, nil
}
func (e *Engine) Path() string {
	return e.path
}