<commit_before>package es

import (
	"log"
	"os"
	"time"
)

var timer = time.NewTimer(1 * time.Second)

type IndexerStats struct {
	Runs        int64
	IndexedDocs int64
	TotalTime   time.Duration
}

func (stats *IndexerStats) Add(count int, dur time.Duration) {
	stats.Runs++
	stats.IndexedDocs += int64(count)
	stats.TotalTime += dur
}

type Indexer struct {
	Index *Index

	IndexEvery time.Duration // triggers a new index run after that duration, will be reset when BatchSize reached
	BatchSize  int           // triggers a new index run when the batch reaches that size

	docsBatch   []*Doc
	docsChannel chan *Doc
	timer       *time.Timer
	Stats       IndexerStats
}

func (indexer *Indexer) Finish() error {
	return indexer.indexBatch()
}

func (indexer *Indexer) resetBatch() {
	indexer.docsBatch = make([]*Doc, 0, indexer.BatchSize)
}

func (indexer *Indexer) resetTimer() bool {
	return indexer.timer.Reset(indexer.IndexEvery)
}

var logger = log.New(os.Stderr, "", 0)

func (indexer *Indexer) Start() chan *Doc {
	if indexer.BatchSize == 0 {
		indexer.BatchSize = 100
	}
	if indexer.IndexEvery == 0 {
		indexer.IndexEvery = 1 * time.Hour
	}
	indexer.timer = time.NewTimer(indexer.IndexEvery)
	indexer.docsChannel = make(chan *Doc, indexer.BatchSize)
	indexer.resetBatch()
	go func(indexer *Indexer) {
		for {
			select {
			case <-indexer.timer.C:
				e := indexer.indexBatch()
				if e != nil {
					logger.Printf("ERROR=%q", e)
				}
				indexer.resetTimer()
			case doc, ok := <-indexer.docsChannel:
				if !ok {
					indexer.timer.Stop()
					e := indexer.indexBatch()<commit_msg>reminder to refactor indexer<commit_after>package es

import (
	"log"
	"os"
	"time"
)

var timer = time.NewTimer(1 * time.Second)

type IndexerStats struct {
	Runs        int64
	IndexedDocs int64
	TotalTime   time.Duration
}

func (stats *IndexerStats) Add(count int, dur time.Duration) {
	stats.Runs++
	stats.IndexedDocs += int64(count)
	stats.TotalTime += dur
}

type Indexer struct {
	Index *Index

	IndexEvery time.Duration // triggers a new index run after that duration, will be reset when BatchSize reached
	BatchSize  int           // triggers a new index run when the batch reaches that size

	docsBatch   []*Doc
	docsChannel chan *Doc
	timer       *time.Timer
	Stats       IndexerStats
}

func (indexer *Indexer) Finish() error {
	return indexer.indexBatch()
}

func (indexer *Indexer) resetBatch() {
	indexer.docsBatch = make([]*Doc, 0, indexer.BatchSize)
}

func (indexer *Indexer) resetTimer() bool {
	return indexer.timer.Reset(indexer.IndexEvery)
}

var logger = log.New(os.Stderr, "", 0)

// TODO change handling to e.g. Add command with implizit opening of channel
// to not expose the channel to the public
func (indexer *Indexer) Start() chan *Doc {
	if indexer.BatchSize == 0 {
		indexer.BatchSize = 100
	}
	if indexer.IndexEvery == 0 {
		indexer.IndexEvery = 1 * time.Hour
	}
	indexer.timer = time.NewTimer(indexer.IndexEvery)
	indexer.docsChannel = make(chan *Doc, indexer.BatchSize)
	indexer.resetBatch()
	go func(indexer *Indexer) {
		for {
			select {
			case <-indexer.timer.C:
				e := indexer.indexBatch()
				if e != nil {
					logger.Printf("ERROR=%q", e)
				}
				indexer.resetTimer()
			case doc, ok := <-indexer.docsChannel:
				if !ok {
					indexer.timer.Stop()
					e := indexer.indexBatch()
{'fraction_non_alphanumeric': 0.11683227943390545, 'fraction_numerical': 0.009033423667570008, 'mean_word_length': 3.4411764705882355, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2073632', 'n_tokens_mistral': 1258, 'n_tokens_neox': 1150, 'n_words': 318}