@@ -13,7 +13,6 @@
 	"alin.ovh/x/log"
 	"github.com/Southclaws/fault"
 	"github.com/Southclaws/fault/fmsg"
-	"github.com/blevesearch/bleve/v2"
 
 	"alin.ovh/searchix/internal/config"
 	"alin.ovh/searchix/internal/fetcher"
@@ -226,7 +225,7 @@ if float64(res.Total) > (0.9 * float64(maxCount)) {
 		return fault.Newf("too many entities to prune: %d/%d (threshold: 90%%)", res.Total, maxCount)
 	}
 
-	err = write.WithBatch(func(batch *bleve.Batch) {
+	err = write.WithBatch(func(batch *index.Batcher) {
 		for _, dm := range res.Hits {
 			batch.Delete(dm.ID)
 		}
@@ -398,18 +397,42 @@ pdb,
 			)
 		}
 
-		objects := processor.Process(ctx)
+		hadWarnings := false
+		write := imp.options.WriteIndex
+		err = write.WithAutoBatch(func(batch *index.Batcher) {
+			for object := range processor.Process(ctx) {
+				select {
+				case <-ctx.Done():
+					imp.options.Logger.Warn("context canceled")
 
-		err = imp.options.WriteIndex.Import(ctx, objects)
+					return
+				default:
+					if err := processor.Err(); err != nil {
+						hadWarnings = true
+
+						imp.options.Logger.Warn("error processing object", "error", err)
+					}
+				}
+
+				doc, err := write.MapDocument(object)
+				if err != nil {
+					imp.options.Logger.Warn("error mapping document", "error", err)
+
+					return
+				}
+
+				err = batch.IndexAdvanced(doc)
+				if err != nil {
+					imp.options.Logger.Warn("error indexing document", "error", err)
+
+					return
+				}
+			}
+		})
 		if err != nil {
-			return fault.Wrap(err, fmsg.With("error writing batch"))
+			return fault.Wrap(err, fmsg.With("error writing index"))
 		}
 
-		hadWarnings := false
-		if err := processor.Err(); err != nil {
-			hadWarnings = true
-			imp.options.Logger.Warn("error processing objects", "error", err)
-		}
 		imp.options.Logger.Debug("ingest completed")
 
 		sourceMeta.StoredAt = time.Now()

@@ -79,6 +79,9 @@ return func(yield func(index.Indexable) bool) {
 		defer i.infile.Close()
 
 		for mv := range i.dec.Stream() {
+			if ctx.Err() != nil {
+				break
+			}
 			if err := i.dec.Err(); err != nil {
 				i.errs = append(i.errs, fault.Wrap(err, fmsg.With("could not decode JSON")))
 
@@ -102,9 +105,10 @@ case string:
 					s := decl
 					link, err := MakeChannelLink(i.source.Repo, s)
 					if err != nil {
-						i.errs = append(i.errs, fault.Wrap(err, fmsg.Withf("could not make a channel link for channel %s, revision %s and subpath %s",
-							i.source.Channel, i.source.Repo.Revision, s,
-						)))
+						i.errs = append(i.errs,
+							fault.Wrap(err, fmsg.Withf("could not make a channel link for channel %s, revision %s and subpath %s",
+								i.source.Channel, i.source.Repo.Revision, s,
+							)))
 
 						continue
 					}
@@ -183,11 +187,15 @@ }
 }
 
 func (i *OptionIngester) Err() error {
-	if len(i.errs) == 0 {
+	errs := i.errs
+	if len(errs) == 0 {
 		return nil
 	}
-	if len(i.errs) == 1 {
+	if len(errs) == 1 {
 		return i.errs[0]
 	}
-	return fault.Newf("encountered %d errors during processing", len(i.errs))
+
+	i.errs = []error{}
+
+	return fault.Newf("encountered %d errors during processing", len(errs))
 }

@@ -359,12 +359,15 @@ }
 }
 
 func (i *PackageIngester) Err() error {
-	if len(i.errs) == 0 {
+	errs := i.errs
+	if len(errs) == 0 {
 		return nil
 	}
-	if len(i.errs) == 1 {
+	if len(errs) == 1 {
 		return i.errs[0]
 	}
 
-	return fault.Newf("encountered %d errors during processing", len(i.errs))
+	i.errs = []error{}
+
+	return fault.Newf("encountered %d errors during processing", len(errs))
 }

@@ -0,0 +1,82 @@
+package index
+
+import (
+	"github.com/Southclaws/fault"
+	"github.com/Southclaws/fault/fmsg"
+	"github.com/blevesearch/bleve/v2"
+	"github.com/blevesearch/bleve/v2/document"
+)
+
+type Batcher struct {
+	batch *bleve.Batch
+	write *WriteIndex
+	err   error
+	n     int
+	max   int
+}
+
+func (index *WriteIndex) NewBatcher() *Batcher {
+	return &Batcher{
+		batch: index.index.NewBatch(),
+		write: index,
+		max:   index.batchSize,
+	}
+}
+
+func (b *Batcher) Size() int {
+	return b.max
+}
+
+func (b *Batcher) Flush() error {
+	b.flush()
+
+	return b.err
+}
+
+func (b *Batcher) flush() {
+	if b.n == 0 {
+		return
+	}
+
+	b.write.log.Debug("flushing batch", "size", b.n)
+
+	err := b.write.index.Batch(b.batch)
+	if err != nil {
+		b.err = &BatchError{fault.Wrap(err, fmsg.Withf("could not flush batch"))}
+
+		return
+	}
+
+	b.Reset()
+}
+
+func (b *Batcher) Reset() {
+	b.err = nil
+	b.n = 0
+	b.batch.Reset()
+}
+
+func (b *Batcher) countAndFlush() {
+	if b.n++; b.n >= b.max {
+		b.flush()
+	}
+}
+
+func (b *Batcher) IndexAdvanced(doc *document.Document) error {
+	err := b.batch.IndexAdvanced(doc)
+	if err != nil {
+		return fault.Wrap(err, fmsg.Withf("could not index document"))
+	}
+
+	b.countAndFlush()
+
+	if b.err != nil {
+		return b.err
+	}
+
+	return nil
+}
+
+func (b *Batcher) Delete(id string) {
+	b.batch.Delete(id)
+}

@@ -2,9 +2,7 @@ package index
 
 import (
 	"bytes"
-	"context"
 	"encoding/gob"
-	"iter"
 	"math"
 
 	"alin.ovh/searchix/internal/config"
@@ -326,52 +324,36 @@
 	return nil
 }
 
-func (i *WriteIndex) Import(
-	ctx context.Context,
-	objects iter.Seq[Indexable],
-) error {
-	indexMapping := i.index.Mapping()
-
-	return i.WithBatchObjects(ctx, objects, func(batch *bleve.Batch, obj Indexable) {
-		doc := document.NewDocument(GetKey(obj))
-		if err := indexMapping.MapDocument(doc, obj); err != nil {
-			i.log.Warn("could not map document for object", "name", obj.GetName())
-
-			return
-		}
-
-		var data bytes.Buffer
-		enc := gob.NewEncoder(&data)
-		if err := enc.Encode(&obj); err != nil {
-			i.log.Error("could not store object in search index", "name", obj.GetName())
-
-			return
-		}
-		field := document.NewTextFieldWithIndexingOptions("_data", nil, data.Bytes(), index.StoreField)
-		doc.AddField(field)
-		idField := document.NewTextFieldCustom(
-			"_id", nil, []byte(doc.ID()),
-			index.IndexField|index.StoreField|index.IncludeTermVectors,
-			idAnalyzer,
+func (i *WriteIndex) MapDocument(obj Indexable) (*document.Document, error) {
+	doc := document.NewDocument(GetKey(obj))
+	if err := i.index.Mapping().MapDocument(doc, obj); err != nil {
+		return nil, fault.Wrap(
+			err,
+			fmsg.Withf("could not map document for object %s", obj.GetName()),
 		)
-		doc.AddField(idField)
+	}
 
-		// log.Debug("adding object to index", "name", opt.Name)
-		if err := batch.IndexAdvanced(doc); err != nil {
-			i.log.Error("could not index object", "name", obj.GetName())
-		}
-	})
-}
+	var data bytes.Buffer
+	enc := gob.NewEncoder(&data)
+	if err := enc.Encode(&obj); err != nil {
+		return nil, fault.Wrap(err, fmsg.Withf("could not encode object %s", obj.GetName()))
+	}
+	field := document.NewTextFieldWithIndexingOptions("_data", nil, data.Bytes(), index.StoreField)
+	doc.AddField(field)
+	idField := document.NewTextFieldCustom(
+		"_id", nil, []byte(doc.ID()),
+		index.IndexField|index.StoreField|index.IncludeTermVectors,
+		idAnalyzer,
+	)
+	doc.AddField(idField)
 
-func (i *WriteIndex) GetBatchSize() int {
-	return i.batchSize
+	return doc, nil
 }
 
-func (i *WriteIndex) WithBatch(fn func(batch *bleve.Batch)) error {
-	batch := i.index.NewBatch()
+func withBatch(batch *Batcher, fn func(batch *Batcher)) error {
 	fn(batch)
 
-	err := i.Flush(batch)
+	err := batch.Flush()
 	if err != nil {
 		return fault.Wrap(err, fmsg.With("could not flush batch"))
 	}
@@ -379,60 +361,15 @@
 	return nil
 }
 
-func (i *WriteIndex) WithBatchObjects(
-	ctx context.Context,
-	objects iter.Seq[Indexable],
-	processor func(batch *bleve.Batch, obj Indexable),
-) error {
-	k := 0
-	batch := i.index.NewBatch()
+func (i *WriteIndex) WithBatch(fn func(batch *Batcher)) error {
+	batch := i.NewBatcher()
+	batch.max = math.MaxInt
 
-	for obj := range objects {
-		select {
-		case <-ctx.Done():
-			i.log.Warn("batch process aborted")
-
-			return ctx.Err()
-		default:
-		}
-
-		processor(batch, obj)
-
-		if k++; k%i.batchSize == 0 {
-			err := i.Flush(batch)
-			if err != nil {
-				return err
-			}
-		}
-	}
-
-	err := i.Flush(batch)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return withBatch(batch, fn)
 }
 
-func (i *WriteIndex) Flush(batch *bleve.Batch) error {
-	size := batch.Size()
-	if size == 0 {
-		return &BatchError{
-			fault.New("no documents to flush"),
-		}
-	}
-	i.log.Debug("flushing batch", "size", size)
-
-	err := i.index.Batch(batch)
-	if err != nil {
-		return &BatchError{
-			fault.Wrap(err, fmsg.Withf("could not flush batch")),
-		}
-	}
-
-	batch.Reset()
-
-	return nil
+func (i *WriteIndex) WithAutoBatch(fn func(batch *Batcher)) error {
+	return withBatch(i.NewBatcher(), fn)
 }
 
 func (i *WriteIndex) Close() (err error) {
@@ -459,18 +396,11 @@ if err != nil {
 		return fault.Wrap(err, fmsg.Withf("failed to query documents of retired index %s", source))
 	}
 
-	batch := i.index.NewBatch()
-	var k int
+	batch := i.NewBatcher()
 	for _, hit := range results.Hits {
 		batch.Delete(hit.ID)
-		if k++; k%i.batchSize == 0 {
-			err := i.Flush(batch)
-			if err != nil {
-				return err
-			}
-		}
 	}
-	err = i.Flush(batch)
+	err = batch.Flush()
 	if err != nil {
 		return fault.Wrap(err)
 	}