Merge pull request #1317 from aduffeck/improve-batches

Do not intertwine different batch operations

services/search/.mockery.yaml

@@ -6,13 +6,12 @@ pkgname: mocks
 
 template: testify
 packages:
-  github.com/opencloud-eu/opencloud/services/search/pkg/engine:
-    interfaces:
-      Engine: {}
   github.com/opencloud-eu/opencloud/services/search/pkg/content:
     interfaces:
       Extractor: {}
       Retriever: {}
   github.com/opencloud-eu/opencloud/services/search/pkg/search:
     interfaces:
-      Searcher: {}
+      Searcher: {}
+      Engine: { }
+      BatchOperator: { }

services/search/pkg/bleve/backend.go

@@ -0,0 +1,229 @@
+package bleve
+
+import (
+	"context"
+	"math"
+	"strings"
+	"time"
+
+	"github.com/blevesearch/bleve/v2"
+	"github.com/blevesearch/bleve/v2/search/query"
+	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/opencloud-eu/reva/v2/pkg/errtypes"
+	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
+	"github.com/opencloud-eu/reva/v2/pkg/utils"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+
+	searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
+	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	searchQuery "github.com/opencloud-eu/opencloud/services/search/pkg/query"
+)
+
+const defaultBatchSize = 50
+
+var _ search.Engine = &Backend{} // ensure Backend implements Engine
+
+type Backend struct {
+	index        bleve.Index
+	queryCreator searchQuery.Creator[query.Query]
+	log          log.Logger
+}
+
+func NewBackend(index bleve.Index, queryCreator searchQuery.Creator[query.Query], log log.Logger) *Backend {
+	return &Backend{
+		index:        index,
+		queryCreator: queryCreator,
+		log:          log,
+	}
+}
+
+// Search executes a search request operation within the index.
+// Returns a SearchIndexResponse object or an error.
+func (b *Backend) Search(_ context.Context, sir *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error) {
+	createdQuery, err := b.queryCreator.Create(sir.Query)
+	if err != nil {
+		if searchQuery.IsValidationError(err) {
+			return nil, errtypes.BadRequest(err.Error())
+		}
+		return nil, err
+	}
+
+	q := bleve.NewConjunctionQuery(
+		// Skip documents that have been marked as deleted
+		&query.BoolFieldQuery{
+			Bool:     false,
+			FieldVal: "Deleted",
+		},
+		createdQuery,
+	)
+
+	if sir.Ref != nil {
+		q.Conjuncts = append(
+			q.Conjuncts,
+			&query.TermQuery{
+				FieldVal: "RootID",
+				Term: storagespace.FormatResourceID(
+					&storageProvider.ResourceId{
+						StorageId: sir.Ref.GetResourceId().GetStorageId(),
+						SpaceId:   sir.Ref.GetResourceId().GetSpaceId(),
+						OpaqueId:  sir.Ref.GetResourceId().GetOpaqueId(),
+					},
+				),
+			},
+		)
+	}
+
+	bleveReq := bleve.NewSearchRequest(q)
+	bleveReq.Highlight = bleve.NewHighlight()
+
+	switch {
+	case sir.PageSize == -1:
+		bleveReq.Size = math.MaxInt
+	case sir.PageSize == 0:
+		bleveReq.Size = 200
+	default:
+		bleveReq.Size = int(sir.PageSize)
+	}
+
+	bleveReq.Fields = []string{"*"}
+	res, err := b.index.Search(bleveReq)
+	if err != nil {
+		return nil, err
+	}
+
+	matches := make([]*searchMessage.Match, 0, len(res.Hits))
+	totalMatches := res.Total
+	for _, hit := range res.Hits {
+		if sir.Ref != nil {
+			hitPath := strings.TrimSuffix(getFieldValue[string](hit.Fields, "Path"), "/")
+			requestedPath := utils.MakeRelativePath(sir.Ref.Path)
+			isRoot := hitPath == requestedPath
+
+			if !isRoot && requestedPath != "." && !strings.HasPrefix(hitPath, requestedPath+"/") {
+				totalMatches--
+				continue
+			}
+		}
+
+		rootID, err := storagespace.ParseID(getFieldValue[string](hit.Fields, "RootID"))
+		if err != nil {
+			return nil, err
+		}
+
+		rID, err := storagespace.ParseID(getFieldValue[string](hit.Fields, "ID"))
+		if err != nil {
+			return nil, err
+		}
+
+		pID, _ := storagespace.ParseID(getFieldValue[string](hit.Fields, "ParentID"))
+		match := &searchMessage.Match{
+			Score: float32(hit.Score),
+			Entity: &searchMessage.Entity{
+				Ref: &searchMessage.Reference{
+					ResourceId: resourceIDtoSearchID(rootID),
+					Path:       getFieldValue[string](hit.Fields, "Path"),
+				},
+				Id:         resourceIDtoSearchID(rID),
+				Name:       getFieldValue[string](hit.Fields, "Name"),
+				ParentId:   resourceIDtoSearchID(pID),
+				Size:       uint64(getFieldValue[float64](hit.Fields, "Size")),
+				Type:       uint64(getFieldValue[float64](hit.Fields, "Type")),
+				MimeType:   getFieldValue[string](hit.Fields, "MimeType"),
+				Deleted:    getFieldValue[bool](hit.Fields, "Deleted"),
+				Tags:       getFieldSliceValue[string](hit.Fields, "Tags"),
+				Highlights: getFragmentValue(hit.Fragments, "Content", 0),
+				Audio:      getAudioValue[searchMessage.Audio](hit.Fields),
+				Image:      getImageValue[searchMessage.Image](hit.Fields),
+				Location:   getLocationValue[searchMessage.GeoCoordinates](hit.Fields),
+				Photo:      getPhotoValue[searchMessage.Photo](hit.Fields),
+			},
+		}
+
+		if mtime, err := time.Parse(time.RFC3339, getFieldValue[string](hit.Fields, "Mtime")); err == nil {
+			match.Entity.LastModifiedTime = &timestamppb.Timestamp{Seconds: mtime.Unix(), Nanos: int32(mtime.Nanosecond())}
+		}
+
+		matches = append(matches, match)
+	}
+
+	return &searchService.SearchIndexResponse{
+		Matches:      matches,
+		TotalMatches: int32(totalMatches),
+	}, nil
+}
+
+func (b *Backend) DocCount() (uint64, error) {
+	return b.index.DocCount()
+}
+
+func (b *Backend) Upsert(id string, r search.Resource) error {
+	batch, err := b.NewBatch(defaultBatchSize)
+	if err != nil {
+		return err
+	}
+
+	if err := batch.Upsert(id, r); err != nil {
+		return err
+	}
+
+	return batch.Push()
+}
+
+func (b *Backend) Move(rootID, parentID, location string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
+	if err != nil {
+		return err
+	}
+
+	if err := batch.Move(rootID, parentID, location); err != nil {
+		return err
+	}
+
+	return batch.Push()
+}
+
+func (b *Backend) Delete(id string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
+	if err != nil {
+		return err
+	}
+
+	if err := batch.Delete(id); err != nil {
+		return err
+	}
+
+	return batch.Push()
+}
+
+func (b *Backend) Restore(id string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
+	if err != nil {
+		return err
+	}
+
+	if err := batch.Restore(id); err != nil {
+		return err
+	}
+
+	return batch.Push()
+}
+
+func (b *Backend) Purge(id string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
+	if err != nil {
+		return err
+	}
+
+	if err := batch.Purge(id); err != nil {
+		return err
+	}
+
+	return batch.Push()
+}
+
+func (b *Backend) NewBatch(size int) (search.BatchOperator, error) {
+	return NewBatch(b.index, size)
+}

services/search/pkg/bleve/backend_test.go

@@ -1,4 +1,4 @@
-package engine_test
+package bleve_test
 
 import (
 	"context"
@@ -14,14 +14,15 @@ import (
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	searchmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
 	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/query/bleve"
+	bleveQuery "github.com/opencloud-eu/opencloud/services/search/pkg/query/bleve"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 )
 
 var _ = Describe("Bleve", func() {
 	var (
-		eng *engine.Bleve
+		eng *bleve.Backend
 		idx bleveSearch.Index
 
 		doSearch = func(id string, query, path string) (*searchsvc.SearchIndexResponse, error) {
@@ -51,30 +52,30 @@ var _ = Describe("Bleve", func() {
 			return res.Matches
 		}
 
-		rootResource   engine.Resource
-		parentResource engine.Resource
-		childResource  engine.Resource
-		childResource2 engine.Resource
+		rootResource   search.Resource
+		parentResource search.Resource
+		childResource  search.Resource
+		childResource2 search.Resource
 	)
 
 	BeforeEach(func() {
-		mapping, err := engine.BuildBleveMapping()
+		mapping, err := bleve.NewMapping()
 		Expect(err).ToNot(HaveOccurred())
 
 		idx, err = bleveSearch.NewMemOnly(mapping)
 		Expect(err).ToNot(HaveOccurred())
 
-		eng = engine.NewBleveEngine(idx, bleve.DefaultCreator, log.Logger{})
+		eng = bleve.NewBackend(idx, bleveQuery.DefaultCreator, log.Logger{})
 		Expect(err).ToNot(HaveOccurred())
 
-		rootResource = engine.Resource{
+		rootResource = search.Resource{
 			ID:       "1$2!2",
 			RootID:   "1$2!2",
 			Path:     ".",
 			Document: content.Document{},
 		}
 
-		parentResource = engine.Resource{
+		parentResource = search.Resource{
 			ID:       "1$2!3",
 			ParentID: rootResource.ID,
 			RootID:   rootResource.ID,
@@ -83,7 +84,7 @@ var _ = Describe("Bleve", func() {
 			Document: content.Document{Name: "parent d!r"},
 		}
 
-		childResource = engine.Resource{
+		childResource = search.Resource{
 			ID:       "1$2!4",
 			ParentID: parentResource.ID,
 			RootID:   rootResource.ID,
@@ -92,7 +93,7 @@ var _ = Describe("Bleve", func() {
 			Document: content.Document{Name: "child.pdf"},
 		}
 
-		childResource2 = engine.Resource{
+		childResource2 = search.Resource{
 			ID:       "1$2!5",
 			ParentID: parentResource.ID,
 			RootID:   rootResource.ID,
@@ -104,7 +105,7 @@ var _ = Describe("Bleve", func() {
 
 	Describe("New", func() {
 		It("returns a new index instance", func() {
-			b := engine.NewBleveEngine(idx, bleve.DefaultCreator, log.Logger{})
+			b := bleve.NewBackend(idx, bleveQuery.DefaultCreator, log.Logger{})
 			Expect(b).ToNot(BeNil())
 		})
 	})
@@ -286,7 +287,7 @@ var _ = Describe("Bleve", func() {
 
 		Context("with a file in the root of the space and folder with a file. all of them have the same name", func() {
 			BeforeEach(func() {
-				parentResource := engine.Resource{
+				parentResource := search.Resource{
 					ID:       "1$2!3",
 					ParentID: rootResource.ID,
 					RootID:   rootResource.ID,
@@ -295,7 +296,7 @@ var _ = Describe("Bleve", func() {
 					Document: content.Document{Name: "doc"},
 				}
 
-				childResource := engine.Resource{
+				childResource := search.Resource{
 					ID:       "1$2!4",
 					ParentID: parentResource.ID,
 					RootID:   rootResource.ID,
@@ -304,7 +305,7 @@ var _ = Describe("Bleve", func() {
 					Document: content.Document{Name: "doc.pdf"},
 				}
 
-				childResource2 := engine.Resource{
+				childResource2 := search.Resource{
 					ID:       "1$2!7",
 					ParentID: parentResource.ID,
 					RootID:   rootResource.ID,
@@ -313,7 +314,7 @@ var _ = Describe("Bleve", func() {
 					Document: content.Document{Name: "file.pdf"},
 				}
 
-				rootChildResource := engine.Resource{
+				rootChildResource := search.Resource{
 					ID:       "1$2!5",
 					ParentID: rootResource.ID,
 					RootID:   rootResource.ID,
@@ -322,7 +323,7 @@ var _ = Describe("Bleve", func() {
 					Document: content.Document{Name: "doc.pdf"},
 				}
 
-				rootChildResource2 := engine.Resource{
+				rootChildResource2 := search.Resource{
 					ID:       "1$2!6",
 					ParentID: rootResource.ID,
 					RootID:   rootResource.ID,
@@ -499,17 +500,17 @@ var _ = Describe("Bleve", func() {
 
 	Describe("StartBatch", func() {
 		It("starts a new batch", func() {
-			err := eng.StartBatch(100)
+			b, err := eng.NewBatch(100)
 			Expect(err).ToNot(HaveOccurred())
 
-			err = eng.Upsert(childResource.ID, childResource)
+			err = b.Upsert(childResource.ID, childResource)
 			Expect(err).ToNot(HaveOccurred())
 
 			count, err := idx.DocCount()
 			Expect(err).ToNot(HaveOccurred())
 			Expect(count).To(Equal(uint64(0)))
 
-			err = eng.EndBatch()
+			err = b.Push()
 			Expect(err).ToNot(HaveOccurred())
 
 			count, err = idx.DocCount()
@@ -522,24 +523,29 @@ var _ = Describe("Bleve", func() {
 			Expect(res.Hits.Len()).To(Equal(1))
 		})
 
-		It("doesn't overwrite batches that are already in progress", func() {
-			err := eng.StartBatch(100)
+		It("doesn't intertwine different batches", func() {
+			b, err := eng.NewBatch(100)
 			Expect(err).ToNot(HaveOccurred())
 
-			err = eng.Upsert(childResource.ID, childResource)
+			err = b.Upsert(childResource.ID, childResource)
 			Expect(err).ToNot(HaveOccurred())
 
 			count, err := idx.DocCount()
 			Expect(err).ToNot(HaveOccurred())
 			Expect(count).To(Equal(uint64(0)))
 
-			err = eng.StartBatch(100)
+			b2, err := eng.NewBatch(100)
 			Expect(err).ToNot(HaveOccurred())
 
-			err = eng.Upsert(childResource2.ID, childResource2)
+			err = b2.Upsert(childResource2.ID, childResource2)
 			Expect(err).ToNot(HaveOccurred())
 
-			Expect(eng.EndBatch()).To(Succeed())
+			Expect(b.Push()).To(Succeed())
+			count, err = idx.DocCount()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(count).To(Equal(uint64(1)))
+
+			Expect(b2.Push()).To(Succeed())
 			count, err = idx.DocCount()
 			Expect(err).ToNot(HaveOccurred())
 			Expect(count).To(Equal(uint64(2)))
@@ -550,7 +556,7 @@ var _ = Describe("Bleve", func() {
 
 		Context("with audio metadata", func() {
 			BeforeEach(func() {
-				resource := engine.Resource{
+				resource := search.Resource{
 					ID:       "1$2!7",
 					ParentID: rootResource.ID,
 					RootID:   rootResource.ID,
@@ -610,7 +616,7 @@ var _ = Describe("Bleve", func() {
 
 		Context("with location metadata", func() {
 			BeforeEach(func() {
-				resource := engine.Resource{
+				resource := search.Resource{
 					ID:       "1$2!7",
 					ParentID: rootResource.ID,
 					RootID:   rootResource.ID,

services/search/pkg/bleve/batch.go

@@ -0,0 +1,121 @@
+package bleve
+
+import (
+	"errors"
+
+	"github.com/blevesearch/bleve/v2"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+var _ search.BatchOperator = &Batch{} // ensure Batch implements BatchOperator
+
+type Batch struct {
+	batch *bleve.Batch
+	index bleve.Index
+	size  int
+	log   log.Logger
+}
+
+func NewBatch(index bleve.Index, size int) (*Batch, error) {
+	if size <= 0 {
+		return nil, errors.New("batch size must be greater than 0")
+	}
+
+	return &Batch{
+		batch: index.NewBatch(),
+		index: index,
+		size:  size,
+	}, nil
+}
+
+func (b *Batch) Upsert(id string, r search.Resource) error {
+	return b.withSizeLimit(func() error {
+		return b.batch.Index(id, r)
+	})
+}
+
+func (b *Batch) Move(id, parentID, location string) error {
+	return b.withSizeLimit(func() error {
+		affectedResources, err := searchAndUpdateResourcesLocation(id, parentID, location, b.index)
+		if err != nil {
+			return err
+		}
+
+		for _, resource := range affectedResources {
+			if err := b.batch.Index(resource.ID, resource); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+func (b *Batch) Delete(id string) error {
+	return b.withSizeLimit(func() error {
+		affectedResources, err := searchAndUpdateResourcesDeletionState(id, true, b.index)
+		if err != nil {
+			return err
+		}
+
+		for _, resource := range affectedResources {
+			if err := b.batch.Index(resource.ID, resource); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+func (b *Batch) Restore(id string) error {
+	return b.withSizeLimit(func() error {
+		affectedResources, err := searchAndUpdateResourcesDeletionState(id, false, b.index)
+		if err != nil {
+			return err
+		}
+
+		for _, resource := range affectedResources {
+			if err := b.batch.Index(resource.ID, resource); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+func (b *Batch) Purge(id string) error {
+	return b.withSizeLimit(func() error {
+		b.batch.Delete(id)
+		return nil
+	})
+}
+
+func (b *Batch) Push() error {
+	if b.batch.Size() == 0 {
+		return nil
+	}
+
+	if err := b.index.Batch(b.batch); err != nil {
+		return err
+	}
+
+	b.batch.Reset()
+
+	return nil
+}
+
+func (b *Batch) withSizeLimit(f func() error) error {
+	if err := f(); err != nil {
+		return err
+	}
+
+	if b.batch.Size() >= b.size {
+		return b.Push()
+	}
+
+	return nil
+}

services/search/pkg/bleve/bleve.go

@@ -0,0 +1,206 @@
+package bleve
+
+import (
+	"reflect"
+	"regexp"
+	"strings"
+	"time"
+
+	bleveSearch "github.com/blevesearch/bleve/v2/search"
+	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	libregraph "github.com/opencloud-eu/libre-graph-api-go"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+var queryEscape = regexp.MustCompile(`([` + regexp.QuoteMeta(`+=&|><!(){}[]^\"~*?:\/`) + `\-\s])`)
+
+func getFieldValue[T any](m map[string]interface{}, key string) (out T) {
+	val, ok := m[key]
+	if !ok {
+		return
+	}
+
+	out, _ = val.(T)
+
+	return
+}
+
+func resourceIDtoSearchID(id storageProvider.ResourceId) *searchMessage.ResourceID {
+	return &searchMessage.ResourceID{
+		StorageId: id.GetStorageId(),
+		SpaceId:   id.GetSpaceId(),
+		OpaqueId:  id.GetOpaqueId()}
+}
+
+func getFieldSliceValue[T any](m map[string]interface{}, key string) (out []T) {
+	iv := getFieldValue[interface{}](m, key)
+	add := func(v interface{}) {
+		cv, ok := v.(T)
+		if !ok {
+			return
+		}
+
+		out = append(out, cv)
+	}
+
+	// bleve tend to convert []string{"foo"} to type string if slice contains only one value
+	// bleve: []string{"foo"} -> "foo"
+	// bleve: []string{"foo", "bar"} -> []string{"foo", "bar"}
+	switch v := iv.(type) {
+	case T:
+		add(v)
+	case []interface{}:
+		for _, rv := range v {
+			add(rv)
+		}
+	}
+
+	return
+}
+
+func getFragmentValue(m bleveSearch.FieldFragmentMap, key string, idx int) string {
+	val, ok := m[key]
+	if !ok {
+		return ""
+	}
+
+	if len(val) <= idx {
+		return ""
+	}
+
+	return val[idx]
+}
+
+func getAudioValue[T any](fields map[string]interface{}) *T {
+	if !strings.HasPrefix(getFieldValue[string](fields, "MimeType"), "audio/") {
+		return nil
+	}
+
+	var audio = newPointerOfType[T]()
+	if ok := unmarshalInterfaceMap(audio, fields, "audio."); ok {
+		return audio
+	}
+
+	return nil
+}
+
+func getImageValue[T any](fields map[string]interface{}) *T {
+	var image = newPointerOfType[T]()
+	if ok := unmarshalInterfaceMap(image, fields, "image."); ok {
+		return image
+	}
+
+	return nil
+}
+
+func getLocationValue[T any](fields map[string]interface{}) *T {
+	var location = newPointerOfType[T]()
+	if ok := unmarshalInterfaceMap(location, fields, "location."); ok {
+		return location
+	}
+
+	return nil
+}
+
+func getPhotoValue[T any](fields map[string]interface{}) *T {
+	var photo = newPointerOfType[T]()
+	if ok := unmarshalInterfaceMap(photo, fields, "photo."); ok {
+		return photo
+	}
+
+	return nil
+}
+
+func newPointerOfType[T any]() *T {
+	t := reflect.TypeOf((*T)(nil)).Elem()
+	ptr := reflect.New(t).Interface()
+	return ptr.(*T)
+}
+
+func unmarshalInterfaceMap(out any, flatMap map[string]interface{}, prefix string) bool {
+	nonEmpty := false
+	obj := reflect.ValueOf(out).Elem()
+	for i := 0; i < obj.NumField(); i++ {
+		field := obj.Field(i)
+		structField := obj.Type().Field(i)
+		mapKey := prefix + getFieldName(structField)
+
+		if value, ok := flatMap[mapKey]; ok {
+			if field.Kind() == reflect.Ptr {
+				alloc := reflect.New(field.Type().Elem())
+				elemType := field.Type().Elem()
+
+				// convert time strings from index for search requests
+				if elemType == reflect.TypeOf(timestamppb.Timestamp{}) {
+					if strValue, ok := value.(string); ok {
+						if parsedTime, err := time.Parse(time.RFC3339, strValue); err == nil {
+							alloc.Elem().Set(reflect.ValueOf(*timestamppb.New(parsedTime)))
+							field.Set(alloc)
+							nonEmpty = true
+						}
+					}
+					continue
+				}
+
+				// convert time strings from index for libregraph structs when updating resources
+				if elemType == reflect.TypeOf(time.Time{}) {
+					if strValue, ok := value.(string); ok {
+						if parsedTime, err := time.Parse(time.RFC3339, strValue); err == nil {
+							alloc.Elem().Set(reflect.ValueOf(parsedTime))
+							field.Set(alloc)
+							nonEmpty = true
+						}
+					}
+					continue
+				}
+
+				alloc.Elem().Set(reflect.ValueOf(value).Convert(elemType))
+				field.Set(alloc)
+				nonEmpty = true
+			}
+		}
+	}
+
+	return nonEmpty
+}
+
+func getFieldName(structField reflect.StructField) string {
+	tag := structField.Tag.Get("json")
+	if tag == "" {
+		return structField.Name
+	}
+
+	return strings.Split(tag, ",")[0]
+}
+
+func matchToResource(match *bleveSearch.DocumentMatch) *search.Resource {
+	return &search.Resource{
+		ID:       getFieldValue[string](match.Fields, "ID"),
+		RootID:   getFieldValue[string](match.Fields, "RootID"),
+		Path:     getFieldValue[string](match.Fields, "Path"),
+		ParentID: getFieldValue[string](match.Fields, "ParentID"),
+		Type:     uint64(getFieldValue[float64](match.Fields, "Type")),
+		Deleted:  getFieldValue[bool](match.Fields, "Deleted"),
+		Document: content.Document{
+			Name:     getFieldValue[string](match.Fields, "Name"),
+			Title:    getFieldValue[string](match.Fields, "Title"),
+			Size:     uint64(getFieldValue[float64](match.Fields, "Size")),
+			Mtime:    getFieldValue[string](match.Fields, "Mtime"),
+			MimeType: getFieldValue[string](match.Fields, "MimeType"),
+			Content:  getFieldValue[string](match.Fields, "Content"),
+			Tags:     getFieldSliceValue[string](match.Fields, "Tags"),
+			Audio:    getAudioValue[libregraph.Audio](match.Fields),
+			Image:    getImageValue[libregraph.Image](match.Fields),
+			Location: getLocationValue[libregraph.GeoCoordinates](match.Fields),
+			Photo:    getPhotoValue[libregraph.Photo](match.Fields),
+		},
+	}
+}
+
+func escapeQuery(s string) string {
+	return queryEscape.ReplaceAllString(s, "\\$1")
+}

services/search/pkg/bleve/engine_suite_test.go

@@ -1,4 +1,4 @@
-package engine_test
+package bleve_test
 
 import (
 	"testing"
@@ -9,5 +9,5 @@ import (
 
 func TestEngine(t *testing.T) {
 	RegisterFailHandler(Fail)
-	RunSpecs(t, "Engine Suite")
+	RunSpecs(t, "Bleve Suite")
 }

services/search/pkg/bleve/index.go

@@ -0,0 +1,179 @@
+package bleve
+
+import (
+	"errors"
+	"math"
+	"path"
+	"path/filepath"
+	"strings"
+
+	"github.com/blevesearch/bleve/v2"
+	"github.com/blevesearch/bleve/v2/analysis/analyzer/custom"
+	"github.com/blevesearch/bleve/v2/analysis/analyzer/keyword"
+	"github.com/blevesearch/bleve/v2/analysis/token/lowercase"
+	"github.com/blevesearch/bleve/v2/analysis/token/porter"
+	"github.com/blevesearch/bleve/v2/analysis/tokenizer/single"
+	"github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode"
+	"github.com/blevesearch/bleve/v2/mapping"
+	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/opencloud-eu/reva/v2/pkg/utils"
+
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+func NewIndex(root string) (bleve.Index, error) {
+	destination := filepath.Join(root, "bleve")
+	index, err := bleve.Open(destination)
+	if errors.Is(bleve.ErrorIndexPathDoesNotExist, err) {
+		indexMapping, err := NewMapping()
+		if err != nil {
+			return nil, err
+		}
+		index, err = bleve.New(destination, indexMapping)
+		if err != nil {
+			return nil, err
+		}
+
+		return index, nil
+	}
+
+	return index, err
+}
+
+func NewMapping() (mapping.IndexMapping, error) {
+	nameMapping := bleve.NewTextFieldMapping()
+	nameMapping.Analyzer = "lowercaseKeyword"
+
+	lowercaseMapping := bleve.NewTextFieldMapping()
+	lowercaseMapping.IncludeInAll = false
+	lowercaseMapping.Analyzer = "lowercaseKeyword"
+
+	fulltextFieldMapping := bleve.NewTextFieldMapping()
+	fulltextFieldMapping.Analyzer = "fulltext"
+	fulltextFieldMapping.IncludeInAll = false
+
+	docMapping := bleve.NewDocumentMapping()
+	docMapping.AddFieldMappingsAt("Name", nameMapping)
+	docMapping.AddFieldMappingsAt("Tags", lowercaseMapping)
+	docMapping.AddFieldMappingsAt("Content", fulltextFieldMapping)
+
+	indexMapping := bleve.NewIndexMapping()
+	indexMapping.DefaultAnalyzer = keyword.Name
+	indexMapping.DefaultMapping = docMapping
+	err := indexMapping.AddCustomAnalyzer("lowercaseKeyword",
+		map[string]interface{}{
+			"type":      custom.Name,
+			"tokenizer": single.Name,
+			"token_filters": []string{
+				lowercase.Name,
+			},
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	err = indexMapping.AddCustomAnalyzer("fulltext",
+		map[string]interface{}{
+			"type":      custom.Name,
+			"tokenizer": unicode.Name,
+			"token_filters": []string{
+				lowercase.Name,
+				porter.Name,
+			},
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return indexMapping, nil
+}
+
+func searchResourceByID(id string, index bleve.Index) (*search.Resource, error) {
+	req := bleve.NewSearchRequest(bleve.NewDocIDQuery([]string{id}))
+	req.Fields = []string{"*"}
+	res, err := index.Search(req)
+	if err != nil {
+		return nil, err
+	}
+	if res.Hits.Len() == 0 {
+		return nil, errors.New("entity not found")
+	}
+
+	return matchToResource(res.Hits[0]), nil
+}
+
+func searchResourcesByPath(rootId, lookupPath string, index bleve.Index) ([]*search.Resource, error) {
+	q := bleve.NewConjunctionQuery(
+		bleve.NewQueryStringQuery("RootID:"+rootId),
+		bleve.NewQueryStringQuery("Path:"+escapeQuery(lookupPath+"/*")),
+	)
+	bleveReq := bleve.NewSearchRequest(q)
+	bleveReq.Size = math.MaxInt
+	bleveReq.Fields = []string{"*"}
+	res, err := index.Search(bleveReq)
+	if err != nil {
+		return nil, err
+	}
+
+	resources := make([]*search.Resource, 0, res.Hits.Len())
+	for _, match := range res.Hits {
+		resources = append(resources, matchToResource(match))
+	}
+
+	return resources, nil
+}
+
+func searchAndUpdateResourcesLocation(rootID, parentID, location string, index bleve.Index) ([]*search.Resource, error) {
+	rootResource, err := searchResourceByID(rootID, index)
+	if err != nil {
+		return nil, err
+	}
+	currentPath := rootResource.Path
+	nextPath := utils.MakeRelativePath(location)
+
+	rootResource.Path = nextPath
+	rootResource.Name = path.Base(nextPath)
+	rootResource.ParentID = parentID
+
+	resources := []*search.Resource{rootResource}
+
+	if rootResource.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER) {
+		descendantResources, err := searchResourcesByPath(rootResource.RootID, currentPath, index)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, descendantResource := range descendantResources {
+			descendantResource.Path = strings.Replace(descendantResource.Path, currentPath, nextPath, 1)
+			resources = append(resources, descendantResource)
+		}
+	}
+
+	return resources, nil
+}
+
+func searchAndUpdateResourcesDeletionState(id string, state bool, index bleve.Index) ([]*search.Resource, error) {
+	rootResource, err := searchResourceByID(id, index)
+	if err != nil {
+		return nil, err
+	}
+	rootResource.Deleted = state
+
+	resources := []*search.Resource{rootResource}
+
+	if rootResource.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER) {
+		descendantResources, err := searchResourcesByPath(rootResource.RootID, rootResource.Path, index)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, descendantResource := range descendantResources {
+			descendantResource.Deleted = state
+			resources = append(resources, descendantResource)
+		}
+	}
+
+	return resources, nil
+}

services/search/pkg/engine/bleve.go

@@ -1,568 +0,0 @@
-package engine
-
-import (
-	"context"
-	"errors"
-	"math"
-	"path"
-	"path/filepath"
-	"reflect"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/blevesearch/bleve/v2"
-	"github.com/blevesearch/bleve/v2/analysis/analyzer/custom"
-	"github.com/blevesearch/bleve/v2/analysis/analyzer/keyword"
-	"github.com/blevesearch/bleve/v2/analysis/token/lowercase"
-	"github.com/blevesearch/bleve/v2/analysis/token/porter"
-	"github.com/blevesearch/bleve/v2/analysis/tokenizer/single"
-	"github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode"
-	"github.com/blevesearch/bleve/v2/mapping"
-	"github.com/blevesearch/bleve/v2/search/query"
-	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
-	libregraph "github.com/opencloud-eu/libre-graph-api-go"
-	"github.com/opencloud-eu/opencloud/pkg/log"
-	"github.com/opencloud-eu/reva/v2/pkg/errtypes"
-	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
-	"github.com/opencloud-eu/reva/v2/pkg/utils"
-	"google.golang.org/protobuf/types/known/timestamppb"
-
-	searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
-	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
-	searchQuery "github.com/opencloud-eu/opencloud/services/search/pkg/query"
-)
-
-const _batchSize = 500
-
-// Bleve represents a search engine which utilizes bleve to search and store resources.
-type Bleve struct {
-	index        bleve.Index
-	queryCreator searchQuery.Creator[query.Query]
-	batch        *bleve.Batch
-	batchSize    int
-	m            sync.Mutex // batch operations in bleve are not thread-safe
-	log          log.Logger
-}
-
-// NewBleveIndex returns a new bleve index
-// given path must exist.
-func NewBleveIndex(root string) (bleve.Index, error) {
-	destination := filepath.Join(root, "bleve")
-	index, err := bleve.Open(destination)
-	if errors.Is(bleve.ErrorIndexPathDoesNotExist, err) {
-		m, err := BuildBleveMapping()
-		if err != nil {
-			return nil, err
-		}
-		index, err = bleve.New(destination, m)
-		if err != nil {
-			return nil, err
-		}
-
-		return index, nil
-	}
-
-	return index, err
-}
-
-// NewBleveEngine creates a new Bleve instance
-func NewBleveEngine(index bleve.Index, queryCreator searchQuery.Creator[query.Query], log log.Logger) *Bleve {
-	return &Bleve{
-		index:        index,
-		queryCreator: queryCreator,
-		log:          log,
-	}
-}
-
-// BuildBleveMapping builds a bleve index mapping which can be used for indexing
-func BuildBleveMapping() (mapping.IndexMapping, error) {
-	nameMapping := bleve.NewTextFieldMapping()
-	nameMapping.Analyzer = "lowercaseKeyword"
-
-	lowercaseMapping := bleve.NewTextFieldMapping()
-	lowercaseMapping.IncludeInAll = false
-	lowercaseMapping.Analyzer = "lowercaseKeyword"
-
-	fulltextFieldMapping := bleve.NewTextFieldMapping()
-	fulltextFieldMapping.Analyzer = "fulltext"
-	fulltextFieldMapping.IncludeInAll = false
-
-	docMapping := bleve.NewDocumentMapping()
-	docMapping.AddFieldMappingsAt("Name", nameMapping)
-	docMapping.AddFieldMappingsAt("Tags", lowercaseMapping)
-	docMapping.AddFieldMappingsAt("Content", fulltextFieldMapping)
-
-	indexMapping := bleve.NewIndexMapping()
-	indexMapping.DefaultAnalyzer = keyword.Name
-	indexMapping.DefaultMapping = docMapping
-	err := indexMapping.AddCustomAnalyzer("lowercaseKeyword",
-		map[string]interface{}{
-			"type":      custom.Name,
-			"tokenizer": single.Name,
-			"token_filters": []string{
-				lowercase.Name,
-			},
-		},
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	err = indexMapping.AddCustomAnalyzer("fulltext",
-		map[string]interface{}{
-			"type":      custom.Name,
-			"tokenizer": unicode.Name,
-			"token_filters": []string{
-				lowercase.Name,
-				porter.Name,
-			},
-		},
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	return indexMapping, nil
-}
-
-// Search executes a search request operation within the index.
-// Returns a SearchIndexResponse object or an error.
-func (b *Bleve) Search(ctx context.Context, sir *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error) {
-	createdQuery, err := b.queryCreator.Create(sir.Query)
-	if err != nil {
-		if searchQuery.IsValidationError(err) {
-			return nil, errtypes.BadRequest(err.Error())
-		}
-		return nil, err
-	}
-
-	q := bleve.NewConjunctionQuery(
-		// Skip documents that have been marked as deleted
-		&query.BoolFieldQuery{
-			Bool:     false,
-			FieldVal: "Deleted",
-		},
-		createdQuery,
-	)
-
-	if sir.Ref != nil {
-		q.Conjuncts = append(
-			q.Conjuncts,
-			&query.TermQuery{
-				FieldVal: "RootID",
-				Term: storagespace.FormatResourceID(
-					&storageProvider.ResourceId{
-						StorageId: sir.Ref.GetResourceId().GetStorageId(),
-						SpaceId:   sir.Ref.GetResourceId().GetSpaceId(),
-						OpaqueId:  sir.Ref.GetResourceId().GetOpaqueId(),
-					},
-				),
-			},
-		)
-	}
-
-	bleveReq := bleve.NewSearchRequest(q)
-	bleveReq.Highlight = bleve.NewHighlight()
-
-	switch {
-	case sir.PageSize == -1:
-		bleveReq.Size = math.MaxInt
-	case sir.PageSize == 0:
-		bleveReq.Size = 200
-	default:
-		bleveReq.Size = int(sir.PageSize)
-	}
-
-	bleveReq.Fields = []string{"*"}
-	res, err := b.index.Search(bleveReq)
-	if err != nil {
-		return nil, err
-	}
-
-	matches := make([]*searchMessage.Match, 0, len(res.Hits))
-	totalMatches := res.Total
-	for _, hit := range res.Hits {
-		if sir.Ref != nil {
-			hitPath := strings.TrimSuffix(getFieldValue[string](hit.Fields, "Path"), "/")
-			requestedPath := utils.MakeRelativePath(sir.Ref.Path)
-			isRoot := hitPath == requestedPath
-
-			if !isRoot && requestedPath != "." && !strings.HasPrefix(hitPath, requestedPath+"/") {
-				totalMatches--
-				continue
-			}
-		}
-
-		rootID, err := storagespace.ParseID(getFieldValue[string](hit.Fields, "RootID"))
-		if err != nil {
-			return nil, err
-		}
-
-		rID, err := storagespace.ParseID(getFieldValue[string](hit.Fields, "ID"))
-		if err != nil {
-			return nil, err
-		}
-
-		pID, _ := storagespace.ParseID(getFieldValue[string](hit.Fields, "ParentID"))
-		match := &searchMessage.Match{
-			Score: float32(hit.Score),
-			Entity: &searchMessage.Entity{
-				Ref: &searchMessage.Reference{
-					ResourceId: resourceIDtoSearchID(rootID),
-					Path:       getFieldValue[string](hit.Fields, "Path"),
-				},
-				Id:         resourceIDtoSearchID(rID),
-				Name:       getFieldValue[string](hit.Fields, "Name"),
-				ParentId:   resourceIDtoSearchID(pID),
-				Size:       uint64(getFieldValue[float64](hit.Fields, "Size")),
-				Type:       uint64(getFieldValue[float64](hit.Fields, "Type")),
-				MimeType:   getFieldValue[string](hit.Fields, "MimeType"),
-				Deleted:    getFieldValue[bool](hit.Fields, "Deleted"),
-				Tags:       getFieldSliceValue[string](hit.Fields, "Tags"),
-				Highlights: getFragmentValue(hit.Fragments, "Content", 0),
-				Audio:      getAudioValue[searchMessage.Audio](hit.Fields),
-				Image:      getImageValue[searchMessage.Image](hit.Fields),
-				Location:   getLocationValue[searchMessage.GeoCoordinates](hit.Fields),
-				Photo:      getPhotoValue[searchMessage.Photo](hit.Fields),
-			},
-		}
-
-		if mtime, err := time.Parse(time.RFC3339, getFieldValue[string](hit.Fields, "Mtime")); err == nil {
-			match.Entity.LastModifiedTime = &timestamppb.Timestamp{Seconds: mtime.Unix(), Nanos: int32(mtime.Nanosecond())}
-		}
-
-		matches = append(matches, match)
-	}
-
-	return &searchService.SearchIndexResponse{
-		Matches:      matches,
-		TotalMatches: int32(totalMatches),
-	}, nil
-}
-
-func (b *Bleve) StartBatch(batchSize int) error {
-	b.m.Lock()
-	defer b.m.Unlock()
-
-	if batchSize <= 0 {
-		return errors.New("batch size must be greater than 0")
-	}
-
-	if b.batch != nil {
-		b.log.Debug().Msg("reusing another batch that has already been started")
-		return nil
-	}
-
-	b.log.Debug().Msg("Starting new batch")
-	b.batch = b.index.NewBatch()
-	b.batchSize = batchSize
-	return nil
-}
-
-func (b *Bleve) EndBatch() error {
-	b.m.Lock()
-	defer b.m.Unlock()
-
-	if b.batch == nil {
-		return errors.New("no batch started")
-	}
-
-	b.log.Debug().Int("size", b.batch.Size()).Msg("Ending batch")
-	if err := b.index.Batch(b.batch); err != nil {
-		return err
-	}
-
-	b.batch = nil
-	return nil
-}
-
-// Upsert indexes or stores Resource data fields.
-func (b *Bleve) Upsert(id string, r Resource) error {
-	b.m.Lock()
-	defer b.m.Unlock()
-
-	if b.batch != nil {
-		if err := b.batch.Index(id, r); err != nil {
-			return err
-		}
-		if b.batch.Size() >= b.batchSize {
-			b.log.Debug().Int("size", b.batch.Size()).Msg("Committing batch")
-			if err := b.index.Batch(b.batch); err != nil {
-				return err
-			}
-			b.batch = b.index.NewBatch()
-		}
-		return nil
-	}
-	return b.index.Index(id, r)
-}
-
-// Move updates the resource location and all of its necessary fields.
-func (b *Bleve) Move(id string, parentid string, target string) error {
-	r, err := b.getResource(id)
-	if err != nil {
-		return err
-	}
-	currentPath := r.Path
-	nextPath := utils.MakeRelativePath(target)
-
-	r, err = b.updateEntity(id, func(r *Resource) {
-		r.Path = nextPath
-		r.Name = path.Base(nextPath)
-		r.ParentID = parentid
-	})
-	if err != nil {
-		return err
-	}
-
-	if r.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER) {
-		q := bleve.NewConjunctionQuery(
-			bleve.NewQueryStringQuery("RootID:"+r.RootID),
-			bleve.NewQueryStringQuery("Path:"+escapeQuery(currentPath+"/*")),
-		)
-		bleveReq := bleve.NewSearchRequest(q)
-		bleveReq.Size = math.MaxInt
-		bleveReq.Fields = []string{"*"}
-		res, err := b.index.Search(bleveReq)
-		if err != nil {
-			return err
-		}
-
-		for _, h := range res.Hits {
-			_, err := b.updateEntity(h.ID, func(r *Resource) {
-				r.Path = strings.Replace(r.Path, currentPath, nextPath, 1)
-			})
-			if err != nil {
-				return err
-			}
-		}
-	}
-
-	return nil
-}
-
-// Delete marks the resource as deleted.
-// The resource object will stay in the bleve index,
-// instead of removing the resource it just marks it as deleted!
-// can be undone
-func (b *Bleve) Delete(id string) error {
-	return b.setDeleted(id, true)
-}
-
-// Restore is the counterpart to Delete.
-// It restores the resource which makes it available again.
-func (b *Bleve) Restore(id string) error {
-	return b.setDeleted(id, false)
-}
-
-// Purge removes a resource from the index, irreversible operation.
-func (b *Bleve) Purge(id string) error {
-	b.m.Lock()
-	defer b.m.Unlock()
-
-	if b.batch != nil {
-		b.batch.Delete(id)
-		if b.batch.Size() >= b.batchSize {
-			if err := b.index.Batch(b.batch); err != nil {
-				return err
-			}
-			b.batch = b.index.NewBatch()
-		}
-		return nil
-	}
-	return b.index.Delete(id)
-}
-
-// DocCount returns the number of resources in the index.
-func (b *Bleve) DocCount() (uint64, error) {
-	return b.index.DocCount()
-}
-
-func (b *Bleve) getResource(id string) (*Resource, error) {
-	req := bleve.NewSearchRequest(bleve.NewDocIDQuery([]string{id}))
-	req.Fields = []string{"*"}
-	res, err := b.index.Search(req)
-	if err != nil {
-		return nil, err
-	}
-	if res.Hits.Len() == 0 {
-		return nil, errors.New("entity not found")
-	}
-
-	fields := res.Hits[0].Fields
-
-	return &Resource{
-		ID:       getFieldValue[string](fields, "ID"),
-		RootID:   getFieldValue[string](fields, "RootID"),
-		Path:     getFieldValue[string](fields, "Path"),
-		ParentID: getFieldValue[string](fields, "ParentID"),
-		Type:     uint64(getFieldValue[float64](fields, "Type")),
-		Deleted:  getFieldValue[bool](fields, "Deleted"),
-		Document: content.Document{
-			Name:     getFieldValue[string](fields, "Name"),
-			Title:    getFieldValue[string](fields, "Title"),
-			Size:     uint64(getFieldValue[float64](fields, "Size")),
-			Mtime:    getFieldValue[string](fields, "Mtime"),
-			MimeType: getFieldValue[string](fields, "MimeType"),
-			Content:  getFieldValue[string](fields, "Content"),
-			Tags:     getFieldSliceValue[string](fields, "Tags"),
-			Audio:    getAudioValue[libregraph.Audio](fields),
-			Image:    getImageValue[libregraph.Image](fields),
-			Location: getLocationValue[libregraph.GeoCoordinates](fields),
-			Photo:    getPhotoValue[libregraph.Photo](fields),
-		},
-	}, nil
-}
-
-func newPointerOfType[T any]() *T {
-	t := reflect.TypeOf((*T)(nil)).Elem()
-	ptr := reflect.New(t).Interface()
-	return ptr.(*T)
-}
-
-func getFieldName(structField reflect.StructField) string {
-	tag := structField.Tag.Get("json")
-	if tag == "" {
-		return structField.Name
-	}
-
-	return strings.Split(tag, ",")[0]
-}
-
-func unmarshalInterfaceMap(out any, flatMap map[string]interface{}, prefix string) bool {
-	nonEmpty := false
-	obj := reflect.ValueOf(out).Elem()
-	for i := 0; i < obj.NumField(); i++ {
-		field := obj.Field(i)
-		structField := obj.Type().Field(i)
-		mapKey := prefix + getFieldName(structField)
-
-		if value, ok := flatMap[mapKey]; ok {
-			if field.Kind() == reflect.Ptr {
-				alloc := reflect.New(field.Type().Elem())
-				elemType := field.Type().Elem()
-
-				// convert time strings from index for search requests
-				if elemType == reflect.TypeOf(timestamppb.Timestamp{}) {
-					if strValue, ok := value.(string); ok {
-						if parsedTime, err := time.Parse(time.RFC3339, strValue); err == nil {
-							alloc.Elem().Set(reflect.ValueOf(*timestamppb.New(parsedTime)))
-							field.Set(alloc)
-							nonEmpty = true
-						}
-					}
-					continue
-				}
-
-				// convert time strings from index for libregraph structs when updating resources
-				if elemType == reflect.TypeOf(time.Time{}) {
-					if strValue, ok := value.(string); ok {
-						if parsedTime, err := time.Parse(time.RFC3339, strValue); err == nil {
-							alloc.Elem().Set(reflect.ValueOf(parsedTime))
-							field.Set(alloc)
-							nonEmpty = true
-						}
-					}
-					continue
-				}
-
-				alloc.Elem().Set(reflect.ValueOf(value).Convert(elemType))
-				field.Set(alloc)
-				nonEmpty = true
-			}
-		}
-	}
-
-	return nonEmpty
-}
-
-func getAudioValue[T any](fields map[string]interface{}) *T {
-	if !strings.HasPrefix(getFieldValue[string](fields, "MimeType"), "audio/") {
-		return nil
-	}
-
-	var audio = newPointerOfType[T]()
-	if ok := unmarshalInterfaceMap(audio, fields, "audio."); ok {
-		return audio
-	}
-
-	return nil
-}
-
-func getImageValue[T any](fields map[string]interface{}) *T {
-	var image = newPointerOfType[T]()
-	if ok := unmarshalInterfaceMap(image, fields, "image."); ok {
-		return image
-	}
-
-	return nil
-}
-
-func getLocationValue[T any](fields map[string]interface{}) *T {
-	var location = newPointerOfType[T]()
-	if ok := unmarshalInterfaceMap(location, fields, "location."); ok {
-		return location
-	}
-
-	return nil
-}
-
-func getPhotoValue[T any](fields map[string]interface{}) *T {
-	var photo = newPointerOfType[T]()
-	if ok := unmarshalInterfaceMap(photo, fields, "photo."); ok {
-		return photo
-	}
-
-	return nil
-}
-
-func (b *Bleve) updateEntity(id string, mutateFunc func(r *Resource)) (*Resource, error) {
-	it, err := b.getResource(id)
-	if err != nil {
-		return nil, err
-	}
-
-	mutateFunc(it)
-
-	return it, b.Upsert(id, *it)
-}
-
-func (b *Bleve) setDeleted(id string, deleted bool) error {
-	it, err := b.updateEntity(id, func(r *Resource) {
-		r.Deleted = deleted
-	})
-	if err != nil {
-		return err
-	}
-
-	if it.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER) {
-		q := bleve.NewConjunctionQuery(
-			bleve.NewQueryStringQuery("RootID:"+it.RootID),
-			bleve.NewQueryStringQuery("Path:"+escapeQuery(it.Path+"/*")),
-		)
-
-		bleveReq := bleve.NewSearchRequest(q)
-		bleveReq.Size = math.MaxInt
-		bleveReq.Fields = []string{"*"}
-		res, err := b.index.Search(bleveReq)
-		if err != nil {
-			return err
-		}
-
-		b.StartBatch(_batchSize)
-		defer b.EndBatch()
-		for _, h := range res.Hits {
-			_, err := b.updateEntity(h.ID, func(r *Resource) {
-				r.Deleted = deleted
-			})
-			if err != nil {
-				return err
-			}
-		}
-		b.EndBatch()
-	}
-
-	return nil
-}

services/search/pkg/engine/engine.go

@@ -1,103 +0,0 @@
-package engine
-
-import (
-	"context"
-	"regexp"
-
-	"github.com/blevesearch/bleve/v2/search"
-	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
-
-	searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
-	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
-)
-
-var queryEscape = regexp.MustCompile(`([` + regexp.QuoteMeta(`+=&|><!(){}[]^\"~*?:\/`) + `\-\s])`)
-
-// Engine is the interface to the search engine
-type Engine interface {
-	Search(ctx context.Context, req *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error)
-	Upsert(id string, r Resource) error
-	Move(id string, parentid string, target string) error
-	Delete(id string) error
-	Restore(id string) error
-	Purge(id string) error
-	DocCount() (uint64, error)
-
-	StartBatch(batchSize int) error
-	EndBatch() error
-}
-
-// Resource is the entity that is stored in the index.
-type Resource struct {
-	content.Document
-
-	ID       string
-	RootID   string
-	Path     string
-	ParentID string
-	Type     uint64
-	Deleted  bool
-	Hidden   bool
-}
-
-func resourceIDtoSearchID(id storageProvider.ResourceId) *searchMessage.ResourceID {
-	return &searchMessage.ResourceID{
-		StorageId: id.GetStorageId(),
-		SpaceId:   id.GetSpaceId(),
-		OpaqueId:  id.GetOpaqueId()}
-}
-
-func escapeQuery(s string) string {
-	return queryEscape.ReplaceAllString(s, "\\$1")
-}
-
-func getFragmentValue(m search.FieldFragmentMap, key string, idx int) string {
-	val, ok := m[key]
-	if !ok {
-		return ""
-	}
-
-	if len(val) <= idx {
-		return ""
-	}
-
-	return val[idx]
-}
-
-func getFieldValue[T any](m map[string]interface{}, key string) (out T) {
-	val, ok := m[key]
-	if !ok {
-		return
-	}
-
-	out, _ = val.(T)
-
-	return
-}
-
-func getFieldSliceValue[T any](m map[string]interface{}, key string) (out []T) {
-	iv := getFieldValue[interface{}](m, key)
-	add := func(v interface{}) {
-		cv, ok := v.(T)
-		if !ok {
-			return
-		}
-
-		out = append(out, cv)
-	}
-
-	// bleve tend to convert []string{"foo"} to type string if slice contains only one value
-	// bleve: []string{"foo"} -> "foo"
-	// bleve: []string{"foo", "bar"} -> []string{"foo", "bar"}
-	switch v := iv.(type) {
-	case T:
-		add(v)
-	case []interface{}:
-		for _, rv := range v {
-			add(rv)
-		}
-	}
-
-	return
-}

services/search/pkg/opensearch/backend.go

@@ -1,27 +1,27 @@
 package opensearch
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
 	"fmt"
-	"path"
 	"strings"
 	"time"
 
 	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+
 	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
 	"github.com/opencloud-eu/reva/v2/pkg/utils"
-	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
 
 	"github.com/opencloud-eu/opencloud/pkg/conversions"
 	searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
 	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/convert"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/osu"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 )
 
+const defaultBatchSize = 50
+
 var (
 	ErrUnhealthyCluster = fmt.Errorf("cluster is not healthy")
 )
@@ -66,7 +66,7 @@ func NewBackend(index string, client *opensearchgoAPI.Client) (*Backend, error)
 	return &Backend{index: index, client: client}, nil
 }
 
-func (be *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error) {
+func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error) {
 	boolQuery, err := convert.KQLToOpenSearchBoolQuery(sir.Query)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert KQL query to OpenSearch bool query: %w", err)
@@ -104,7 +104,7 @@ func (be *Backend) Search(ctx context.Context, sir *searchService.SearchIndexReq
 	}
 
 	req, err := osu.BuildSearchReq(&opensearchgoAPI.SearchReq{
-		Indices: []string{be.index},
+		Indices: []string{b.index},
 		Params:  searchParams,
 	},
 		boolQuery,
@@ -122,7 +122,7 @@ func (be *Backend) Search(ctx context.Context, sir *searchService.SearchIndexReq
 		return nil, fmt.Errorf("failed to build search request: %w", err)
 	}
 
-	resp, err := be.client.Search(ctx, req)
+	resp, err := b.client.Search(ctx, req)
 	if err != nil {
 		return nil, fmt.Errorf("failed to search: %w", err)
 	}
@@ -155,101 +155,10 @@ func (be *Backend) Search(ctx context.Context, sir *searchService.SearchIndexReq
 	}, nil
 }
 
-func (be *Backend) Upsert(id string, r engine.Resource) error {
-	body, err := json.Marshal(r)
-	if err != nil {
-		return fmt.Errorf("failed to marshal resource: %w", err)
-	}
-
-	_, err = be.client.Index(context.TODO(), opensearchgoAPI.IndexReq{
-		Index:      be.index,
-		DocumentID: id,
-		Body:       bytes.NewReader(body),
-	})
-	if err != nil {
-		return fmt.Errorf("failed to index document: %w", err)
-	}
-
-	return nil
-}
-
-func (be *Backend) Move(id string, parentID string, target string) error {
-	return be.updateSelfAndDescendants(id, func(rootResource engine.Resource) *osu.BodyParamScript {
-		return &osu.BodyParamScript{
-			Source: `
-					if (ctx._source.ID == params.id ) { ctx._source.Name = params.newName; ctx._source.ParentID = params.parentID; }
-					ctx._source.Path = ctx._source.Path.replace(params.oldPath, params.newPath)
-				`,
-			Lang: "painless",
-			Params: map[string]any{
-				"id":       id,
-				"parentID": parentID,
-				"oldPath":  rootResource.Path,
-				"newPath":  utils.MakeRelativePath(target),
-				"newName":  path.Base(utils.MakeRelativePath(target)),
-			},
-		}
-	})
-}
-
-func (be *Backend) Delete(id string) error {
-	return be.updateSelfAndDescendants(id, func(_ engine.Resource) *osu.BodyParamScript {
-		return &osu.BodyParamScript{
-			Source: "ctx._source.Deleted = params.deleted",
-			Lang:   "painless",
-			Params: map[string]any{
-				"deleted": true,
-			},
-		}
-	})
-}
-
-func (be *Backend) Restore(id string) error {
-	return be.updateSelfAndDescendants(id, func(_ engine.Resource) *osu.BodyParamScript {
-		return &osu.BodyParamScript{
-			Source: "ctx._source.Deleted = params.deleted",
-			Lang:   "painless",
-			Params: map[string]any{
-				"deleted": false,
-			},
-		}
-	})
-}
-
-func (be *Backend) Purge(id string) error {
-	resource, err := be.getResource(id)
-	if err != nil {
-		return fmt.Errorf("failed to get resource: %w", err)
-	}
-
-	req, err := osu.BuildDocumentDeleteByQueryReq(
-		opensearchgoAPI.DocumentDeleteByQueryReq{
-			Indices: []string{be.index},
-			Params: opensearchgoAPI.DocumentDeleteByQueryParams{
-				WaitForCompletion: conversions.ToPointer(true),
-			},
-		},
-		osu.NewTermQuery[string]("Path").Value(resource.Path),
-	)
-	if err != nil {
-		return fmt.Errorf("failed to build delete by query request: %w", err)
-	}
-
-	resp, err := be.client.Document.DeleteByQuery(context.TODO(), req)
-	switch {
-	case err != nil:
-		return fmt.Errorf("failed to delete by query: %w", err)
-	case len(resp.Failures) != 0:
-		return fmt.Errorf("failed to delete by query, failures: %v", resp.Failures)
-	}
-
-	return nil
-}
-
-func (be *Backend) DocCount() (uint64, error) {
+func (b *Backend) DocCount() (uint64, error) {
 	req, err := osu.BuildIndicesCountReq(
 		&opensearchgoAPI.IndicesCountReq{
-			Indices: []string{be.index},
+			Indices: []string{b.index},
 		},
 		osu.NewTermQuery[bool]("Deleted").Value(false),
 	)
@@ -257,7 +166,7 @@ func (be *Backend) DocCount() (uint64, error) {
 		return 0, fmt.Errorf("failed to build count request: %w", err)
 	}
 
-	resp, err := be.client.Indices.Count(context.TODO(), req)
+	resp, err := b.client.Indices.Count(context.TODO(), req)
 	if err != nil {
 		return 0, fmt.Errorf("failed to count documents: %w", err)
 	}
@@ -265,74 +174,71 @@ func (be *Backend) DocCount() (uint64, error) {
 	return uint64(resp.Count), nil
 }
 
-func (be *Backend) updateSelfAndDescendants(id string, scriptProvider func(engine.Resource) *osu.BodyParamScript) error {
-	if scriptProvider == nil {
-		return fmt.Errorf("script cannot be nil")
-	}
-
-	resource, err := be.getResource(id)
+func (b *Backend) Upsert(id string, r search.Resource) error {
+	batch, err := b.NewBatch(defaultBatchSize)
 	if err != nil {
-		return fmt.Errorf("failed to get resource: %w", err)
+		return err
 	}
 
-	req, err := osu.BuildUpdateByQueryReq(
-		opensearchgoAPI.UpdateByQueryReq{
-			Indices: []string{be.index},
-			Params: opensearchgoAPI.UpdateByQueryParams{
-				WaitForCompletion: conversions.ToPointer(true),
-			},
-		},
-		osu.NewTermQuery[string]("Path").Value(resource.Path),
-		osu.UpdateByQueryBodyParams{
-			Script: scriptProvider(resource),
-		},
-	)
-	if err != nil {
-		return fmt.Errorf("failed to build update by query request: %w", err)
+	if err := batch.Upsert(id, r); err != nil {
+		return err
 	}
 
-	resp, err := be.client.UpdateByQuery(context.TODO(), req)
-	switch {
-	case err != nil:
-		return fmt.Errorf("failed to update by query: %w", err)
-	case len(resp.Failures) != 0:
-		return fmt.Errorf("failed to update by query, failures: %v", resp.Failures)
-	}
-
-	return nil
+	return batch.Push()
 }
 
-func (be *Backend) getResource(id string) (engine.Resource, error) {
-	req, err := osu.BuildSearchReq(
-		&opensearchgoAPI.SearchReq{
-			Indices: []string{be.index},
-		},
-		osu.NewIDsQuery(id),
-	)
+func (b *Backend) Move(id string, parentID string, target string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
 	if err != nil {
-		return engine.Resource{}, fmt.Errorf("failed to build search request: %w", err)
+		return err
 	}
 
-	resp, err := be.client.Search(context.TODO(), req)
-	switch {
-	case err != nil:
-		return engine.Resource{}, fmt.Errorf("failed to search for resource: %w", err)
-	case resp.Hits.Total.Value == 0 || len(resp.Hits.Hits) == 0:
-		return engine.Resource{}, fmt.Errorf("document with id %s not found", id)
+	if err := batch.Move(id, parentID, target); err != nil {
+		return err
 	}
 
-	resource, err := conversions.To[engine.Resource](resp.Hits.Hits[0].Source)
+	return batch.Push()
+}
+
+func (b *Backend) Delete(id string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
 	if err != nil {
-		return engine.Resource{}, fmt.Errorf("failed to convert hit source: %w", err)
+		return err
 	}
 
-	return resource, nil
+	if err := batch.Delete(id); err != nil {
+		return err
+	}
+
+	return batch.Push()
 }
 
-func (be *Backend) StartBatch(_ int) error {
-	return nil // todo: implement batch processing
+func (b *Backend) Restore(id string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
+	if err != nil {
+		return err
+	}
+
+	if err := batch.Restore(id); err != nil {
+		return err
+	}
+
+	return batch.Push()
 }
 
-func (be *Backend) EndBatch() error {
-	return nil // todo: implement batch processing
+func (b *Backend) Purge(id string) error {
+	batch, err := b.NewBatch(defaultBatchSize)
+	if err != nil {
+		return err
+	}
+
+	if err := batch.Purge(id); err != nil {
+		return err
+	}
+
+	return batch.Push()
+}
+
+func (b *Backend) NewBatch(size int) (search.BatchOperator, error) {
+	return NewBatch(b.client, b.index, size)
 }

services/search/pkg/opensearch/backend_test.go

@@ -10,9 +10,9 @@ import (
 	"github.com/stretchr/testify/require"
 
 	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 )
 
 func TestNewBackend(t *testing.T) {
@@ -116,14 +116,14 @@ func TestEngine_Move(t *testing.T) {
 			},
 		})
 
-		resources := opensearchtest.SearchHitsMustBeConverted[engine.Resource](t, tc.Require.Search(indexName, strings.NewReader(body)).Hits)
+		resources := opensearchtest.SearchHitsMustBeConverted[search.Resource](t, tc.Require.Search(indexName, strings.NewReader(body)).Hits)
 		require.Len(t, resources, 1)
 		require.Equal(t, document.Path, resources[0].Path)
 
 		document.Path = "./new/path/to/resource"
 		require.NoError(t, backend.Move(document.ID, document.ParentID, document.Path))
 
-		resources = opensearchtest.SearchHitsMustBeConverted[engine.Resource](t, tc.Require.Search(indexName, strings.NewReader(body)).Hits)
+		resources = opensearchtest.SearchHitsMustBeConverted[search.Resource](t, tc.Require.Search(indexName, strings.NewReader(body)).Hits)
 		require.Len(t, resources, 1)
 		require.Equal(t, document.Path, resources[0].Path)
 	})

services/search/pkg/opensearch/batch.go

@@ -0,0 +1,243 @@
+package opensearch
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"path"
+	"strings"
+	"sync"
+
+	"github.com/opencloud-eu/reva/v2/pkg/utils"
+	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+
+	"github.com/opencloud-eu/opencloud/pkg/conversions"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/osu"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+var _ search.BatchOperator = &Batch{} // ensure Batch implements BatchOperator
+
+type Batch struct {
+	client     *opensearchgoAPI.Client
+	index      string
+	size       int
+	log        log.Logger
+	operations []any
+	mu         sync.Mutex
+}
+
+func NewBatch(client *opensearchgoAPI.Client, index string, size int) (*Batch, error) {
+	if size <= 0 {
+		return nil, errors.New("batch size must be greater than 0")
+	}
+
+	return &Batch{
+		client: client,
+		size:   size,
+		index:  index,
+	}, nil
+}
+
+func (b *Batch) Upsert(id string, r search.Resource) error {
+	return b.withSizeLimit(func() error {
+		body, err := conversions.To[map[string]any](r)
+		if err != nil {
+			return fmt.Errorf("failed to marshal resource: %w", err)
+		}
+
+		op := func() []map[string]any {
+			return []map[string]any{
+				{"index": map[string]any{"_index": b.index, "_id": id}},
+				body,
+			}
+		}
+
+		b.mu.Lock()
+		b.operations = append(b.operations, op)
+		b.mu.Unlock()
+
+		return nil
+	})
+}
+
+func (b *Batch) Move(id, parentID, location string) error {
+	return b.withSizeLimit(func() error {
+		op := func() error {
+			return updateSelfAndDescendants(context.Background(), b.client, b.index, id, func(rootResource search.Resource) *osu.BodyParamScript {
+				return &osu.BodyParamScript{
+					Source: `
+					if (ctx._source.ID == params.id ) { ctx._source.Name = params.newName; ctx._source.ParentID = params.parentID; }
+					ctx._source.Path = ctx._source.Path.replace(params.oldPath, params.newPath)
+				`,
+					Lang: "painless",
+					Params: map[string]any{
+						"id":       id,
+						"parentID": parentID,
+						"oldPath":  rootResource.Path,
+						"newPath":  utils.MakeRelativePath(location),
+						"newName":  path.Base(utils.MakeRelativePath(location)),
+					},
+				}
+			})
+		}
+
+		b.mu.Lock()
+		b.operations = append(b.operations, op)
+		b.mu.Unlock()
+
+		return nil
+	})
+}
+
+func (b *Batch) Delete(id string) error {
+	return b.withSizeLimit(func() error {
+		op := func() error {
+			return updateSelfAndDescendants(context.Background(), b.client, b.index, id, func(_ search.Resource) *osu.BodyParamScript {
+				return &osu.BodyParamScript{
+					Source: "ctx._source.Deleted = params.deleted",
+					Lang:   "painless",
+					Params: map[string]any{
+						"deleted": true,
+					},
+				}
+			})
+		}
+
+		b.mu.Lock()
+		b.operations = append(b.operations, op)
+		b.mu.Unlock()
+
+		return nil
+	})
+}
+
+func (b *Batch) Restore(id string) error {
+	return b.withSizeLimit(func() error {
+		op := func() error {
+			return updateSelfAndDescendants(context.Background(), b.client, b.index, id, func(_ search.Resource) *osu.BodyParamScript {
+				return &osu.BodyParamScript{
+					Source: "ctx._source.Deleted = params.deleted",
+					Lang:   "painless",
+					Params: map[string]any{
+						"deleted": false,
+					},
+				}
+			})
+		}
+
+		b.mu.Lock()
+		b.operations = append(b.operations, op)
+		b.mu.Unlock()
+
+		return nil
+	})
+}
+
+func (b *Batch) Purge(id string) error {
+	return b.withSizeLimit(func() error {
+		resource, err := searchResourceByID(context.Background(), b.client, b.index, id)
+		if err != nil {
+			return fmt.Errorf("failed to get resource: %w", err)
+		}
+
+		req, err := osu.BuildDocumentDeleteByQueryReq(
+			opensearchgoAPI.DocumentDeleteByQueryReq{
+				Indices: []string{b.index},
+				Params: opensearchgoAPI.DocumentDeleteByQueryParams{
+					WaitForCompletion: conversions.ToPointer(true),
+				},
+			},
+			osu.NewTermQuery[string]("Path").Value(resource.Path),
+		)
+		if err != nil {
+			return fmt.Errorf("failed to build delete by query request: %w", err)
+		}
+
+		op := func() error {
+			resp, err := b.client.Document.DeleteByQuery(context.TODO(), req)
+			switch {
+			case err != nil:
+				return fmt.Errorf("failed to delete by query: %w", err)
+			case len(resp.Failures) != 0:
+				return fmt.Errorf("failed to delete by query, failures: %v", resp.Failures)
+			}
+
+			return nil
+		}
+
+		b.mu.Lock()
+		b.operations = append(b.operations, op)
+		b.mu.Unlock()
+
+		return nil
+	})
+}
+
+func (b *Batch) Push() error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	defer func() { // cleanup
+		b.operations = nil
+	}()
+
+	var bulkOperations []map[string]any
+	pushBulkOperations := func() error {
+		if len(bulkOperations) == 0 {
+			return nil
+		}
+
+		var body strings.Builder
+		for _, operation := range bulkOperations {
+			part, err := json.Marshal(operation)
+			if err != nil {
+				return fmt.Errorf("failed to marshal bulk operation: %w", err)
+			}
+			body.Write(part)
+			body.WriteString("\n")
+		}
+
+		if _, err := b.client.Bulk(context.Background(), opensearchgoAPI.BulkReq{
+			Body: strings.NewReader(body.String()),
+		}); err != nil {
+			return fmt.Errorf("failed to execute bulk operations: %w", err)
+		}
+
+		bulkOperations = nil
+		return nil
+	}
+
+	// keep the order of operations in the batch intact,
+	//  unfortunately, operations like DeleteByQuery cannot be part of the bulk API,
+	//  so we need to push the previous bulk operations before executing such operations
+	//  this might lead to smaller bulks than the configured size, but ensures correct order
+	for _, operation := range b.operations {
+		switch op := operation.(type) {
+		case func() []map[string]any:
+			bulkOperations = append(bulkOperations, op()...)
+		case func() error:
+			if err := pushBulkOperations(); err != nil {
+				return fmt.Errorf("failed to push operations: %w", err)
+			}
+			if err := op(); err != nil {
+				return fmt.Errorf("failed to execute operation: %w", err)
+			}
+		}
+	}
+
+	return pushBulkOperations()
+}
+
+func (b *Batch) withSizeLimit(f func() error) error {
+	if err := f(); err != nil {
+		return err
+	}
+
+	if len(b.operations) >= b.size {
+		return b.Push()
+	}
+
+	return nil
+}

services/search/pkg/opensearch/internal/convert/opensearch.go

@@ -5,17 +5,18 @@ import (
 	"strings"
 	"time"
 
-	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
 	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
+
 	"github.com/opencloud-eu/opencloud/pkg/conversions"
 	searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 )
 
 func OpenSearchHitToMatch(hit opensearchgoAPI.SearchHit) (*searchMessage.Match, error) {
-	resource, err := conversions.To[engine.Resource](hit.Source)
+	resource, err := conversions.To[search.Resource](hit.Source)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert hit source: %w", err)
 	}

services/search/pkg/opensearch/internal/test/testdata.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"path"
 
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 )
 
 //go:embed testdata/*.json
@@ -16,12 +16,12 @@ var Testdata = struct {
 	Resources resourceTestdata
 }{
 	Resources: resourceTestdata{
-		File: fromTestData[engine.Resource]("resource_file.json"),
+		File: fromTestData[search.Resource]("resource_file.json"),
 	},
 }
 
 type resourceTestdata struct {
-	File engine.Resource
+	File search.Resource
 }
 
 func fromTestData[D any](name string) D {

services/search/pkg/opensearch/opensearch.go

@@ -0,0 +1,77 @@
+package opensearch
+
+import (
+	"context"
+	"fmt"
+
+	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+
+	"github.com/opencloud-eu/opencloud/pkg/conversions"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/osu"
+)
+
+func searchResourceByID(ctx context.Context, client *opensearchgoAPI.Client, index string, id string) (search.Resource, error) {
+	req, err := osu.BuildSearchReq(
+		&opensearchgoAPI.SearchReq{
+			Indices: []string{index},
+		},
+		osu.NewIDsQuery(id),
+	)
+	if err != nil {
+		return search.Resource{}, fmt.Errorf("failed to build search request: %w", err)
+	}
+
+	resp, err := client.Search(ctx, req)
+	switch {
+	case err != nil:
+		return search.Resource{}, fmt.Errorf("failed to search for resource: %w", err)
+	case resp.Hits.Total.Value == 0 || len(resp.Hits.Hits) == 0:
+		return search.Resource{}, fmt.Errorf("document with id %s not found", id)
+	}
+
+	resource, err := conversions.To[search.Resource](resp.Hits.Hits[0].Source)
+	if err != nil {
+		return search.Resource{}, fmt.Errorf("failed to convert hit source: %w", err)
+	}
+
+	return resource, nil
+}
+
+func updateSelfAndDescendants(ctx context.Context, client *opensearchgoAPI.Client, index string, id string, scriptProvider func(search.Resource) *osu.BodyParamScript) error {
+	if scriptProvider == nil {
+		return fmt.Errorf("script cannot be nil")
+	}
+
+	resource, err := searchResourceByID(context.Background(), client, index, id)
+	if err != nil {
+		return fmt.Errorf("failed to get resource: %w", err)
+	}
+
+	req, err := osu.BuildUpdateByQueryReq(
+		opensearchgoAPI.UpdateByQueryReq{
+			Indices: []string{index},
+			Params: opensearchgoAPI.UpdateByQueryParams{
+				WaitForCompletion: conversions.ToPointer(true),
+			},
+		},
+		osu.NewTermQuery[string]("Path").Value(resource.Path),
+		osu.UpdateByQueryBodyParams{
+			Script: scriptProvider(resource),
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("failed to build update by query request: %w", err)
+	}
+
+	resp, err := client.UpdateByQuery(ctx, req)
+	switch {
+	case err != nil:
+		return fmt.Errorf("failed to update by query: %w", err)
+	case len(resp.Failures) != 0:
+		return fmt.Errorf("failed to update by query, failures: %v", resp.Failures)
+	}
+
+	return nil
+}

services/search/pkg/search/mocks/batch_operator.go

@@ -0,0 +1,354 @@
+// Code generated by mockery; DO NOT EDIT.
+// github.com/vektra/mockery
+// template: testify
+
+package mocks
+
+import (
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+	mock "github.com/stretchr/testify/mock"
+)
+
+// NewBatchOperator creates a new instance of BatchOperator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewBatchOperator(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *BatchOperator {
+	mock := &BatchOperator{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
+
+// BatchOperator is an autogenerated mock type for the BatchOperator type
+type BatchOperator struct {
+	mock.Mock
+}
+
+type BatchOperator_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *BatchOperator) EXPECT() *BatchOperator_Expecter {
+	return &BatchOperator_Expecter{mock: &_m.Mock}
+}
+
+// Delete provides a mock function for the type BatchOperator
+func (_mock *BatchOperator) Delete(id string) error {
+	ret := _mock.Called(id)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Delete")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func(string) error); ok {
+		r0 = returnFunc(id)
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// BatchOperator_Delete_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Delete'
+type BatchOperator_Delete_Call struct {
+	*mock.Call
+}
+
+// Delete is a helper method to define mock.On call
+//   - id string
+func (_e *BatchOperator_Expecter) Delete(id interface{}) *BatchOperator_Delete_Call {
+	return &BatchOperator_Delete_Call{Call: _e.mock.On("Delete", id)}
+}
+
+func (_c *BatchOperator_Delete_Call) Run(run func(id string)) *BatchOperator_Delete_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 string
+		if args[0] != nil {
+			arg0 = args[0].(string)
+		}
+		run(
+			arg0,
+		)
+	})
+	return _c
+}
+
+func (_c *BatchOperator_Delete_Call) Return(err error) *BatchOperator_Delete_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *BatchOperator_Delete_Call) RunAndReturn(run func(id string) error) *BatchOperator_Delete_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// Move provides a mock function for the type BatchOperator
+func (_mock *BatchOperator) Move(rootID string, parentID string, location string) error {
+	ret := _mock.Called(rootID, parentID, location)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Move")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func(string, string, string) error); ok {
+		r0 = returnFunc(rootID, parentID, location)
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// BatchOperator_Move_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Move'
+type BatchOperator_Move_Call struct {
+	*mock.Call
+}
+
+// Move is a helper method to define mock.On call
+//   - rootID string
+//   - parentID string
+//   - location string
+func (_e *BatchOperator_Expecter) Move(rootID interface{}, parentID interface{}, location interface{}) *BatchOperator_Move_Call {
+	return &BatchOperator_Move_Call{Call: _e.mock.On("Move", rootID, parentID, location)}
+}
+
+func (_c *BatchOperator_Move_Call) Run(run func(rootID string, parentID string, location string)) *BatchOperator_Move_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 string
+		if args[0] != nil {
+			arg0 = args[0].(string)
+		}
+		var arg1 string
+		if args[1] != nil {
+			arg1 = args[1].(string)
+		}
+		var arg2 string
+		if args[2] != nil {
+			arg2 = args[2].(string)
+		}
+		run(
+			arg0,
+			arg1,
+			arg2,
+		)
+	})
+	return _c
+}
+
+func (_c *BatchOperator_Move_Call) Return(err error) *BatchOperator_Move_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *BatchOperator_Move_Call) RunAndReturn(run func(rootID string, parentID string, location string) error) *BatchOperator_Move_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// Purge provides a mock function for the type BatchOperator
+func (_mock *BatchOperator) Purge(id string) error {
+	ret := _mock.Called(id)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Purge")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func(string) error); ok {
+		r0 = returnFunc(id)
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// BatchOperator_Purge_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Purge'
+type BatchOperator_Purge_Call struct {
+	*mock.Call
+}
+
+// Purge is a helper method to define mock.On call
+//   - id string
+func (_e *BatchOperator_Expecter) Purge(id interface{}) *BatchOperator_Purge_Call {
+	return &BatchOperator_Purge_Call{Call: _e.mock.On("Purge", id)}
+}
+
+func (_c *BatchOperator_Purge_Call) Run(run func(id string)) *BatchOperator_Purge_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 string
+		if args[0] != nil {
+			arg0 = args[0].(string)
+		}
+		run(
+			arg0,
+		)
+	})
+	return _c
+}
+
+func (_c *BatchOperator_Purge_Call) Return(err error) *BatchOperator_Purge_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *BatchOperator_Purge_Call) RunAndReturn(run func(id string) error) *BatchOperator_Purge_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// Push provides a mock function for the type BatchOperator
+func (_mock *BatchOperator) Push() error {
+	ret := _mock.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for Push")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func() error); ok {
+		r0 = returnFunc()
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// BatchOperator_Push_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Push'
+type BatchOperator_Push_Call struct {
+	*mock.Call
+}
+
+// Push is a helper method to define mock.On call
+func (_e *BatchOperator_Expecter) Push() *BatchOperator_Push_Call {
+	return &BatchOperator_Push_Call{Call: _e.mock.On("Push")}
+}
+
+func (_c *BatchOperator_Push_Call) Run(run func()) *BatchOperator_Push_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run()
+	})
+	return _c
+}
+
+func (_c *BatchOperator_Push_Call) Return(err error) *BatchOperator_Push_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *BatchOperator_Push_Call) RunAndReturn(run func() error) *BatchOperator_Push_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// Restore provides a mock function for the type BatchOperator
+func (_mock *BatchOperator) Restore(id string) error {
+	ret := _mock.Called(id)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Restore")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func(string) error); ok {
+		r0 = returnFunc(id)
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// BatchOperator_Restore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Restore'
+type BatchOperator_Restore_Call struct {
+	*mock.Call
+}
+
+// Restore is a helper method to define mock.On call
+//   - id string
+func (_e *BatchOperator_Expecter) Restore(id interface{}) *BatchOperator_Restore_Call {
+	return &BatchOperator_Restore_Call{Call: _e.mock.On("Restore", id)}
+}
+
+func (_c *BatchOperator_Restore_Call) Run(run func(id string)) *BatchOperator_Restore_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 string
+		if args[0] != nil {
+			arg0 = args[0].(string)
+		}
+		run(
+			arg0,
+		)
+	})
+	return _c
+}
+
+func (_c *BatchOperator_Restore_Call) Return(err error) *BatchOperator_Restore_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *BatchOperator_Restore_Call) RunAndReturn(run func(id string) error) *BatchOperator_Restore_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// Upsert provides a mock function for the type BatchOperator
+func (_mock *BatchOperator) Upsert(id string, r search.Resource) error {
+	ret := _mock.Called(id, r)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Upsert")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func(string, search.Resource) error); ok {
+		r0 = returnFunc(id, r)
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// BatchOperator_Upsert_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Upsert'
+type BatchOperator_Upsert_Call struct {
+	*mock.Call
+}
+
+// Upsert is a helper method to define mock.On call
+//   - id string
+//   - r search.Resource
+func (_e *BatchOperator_Expecter) Upsert(id interface{}, r interface{}) *BatchOperator_Upsert_Call {
+	return &BatchOperator_Upsert_Call{Call: _e.mock.On("Upsert", id, r)}
+}
+
+func (_c *BatchOperator_Upsert_Call) Run(run func(id string, r search.Resource)) *BatchOperator_Upsert_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 string
+		if args[0] != nil {
+			arg0 = args[0].(string)
+		}
+		var arg1 search.Resource
+		if args[1] != nil {
+			arg1 = args[1].(search.Resource)
+		}
+		run(
+			arg0,
+			arg1,
+		)
+	})
+	return _c
+}
+
+func (_c *BatchOperator_Upsert_Call) Return(err error) *BatchOperator_Upsert_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *BatchOperator_Upsert_Call) RunAndReturn(run func(id string, r search.Resource) error) *BatchOperator_Upsert_Call {
+	_c.Call.Return(run)
+	return _c
+}

services/search/pkg/search/mocks/engine.go

@@ -8,7 +8,7 @@ import (
 	"context"
 
 	"github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 	mock "github.com/stretchr/testify/mock"
 )
 
@@ -143,50 +143,6 @@ func (_c *Engine_DocCount_Call) RunAndReturn(run func() (uint64, error)) *Engine
 	return _c
 }
 
-// EndBatch provides a mock function for the type Engine
-func (_mock *Engine) EndBatch() error {
-	ret := _mock.Called()
-
-	if len(ret) == 0 {
-		panic("no return value specified for EndBatch")
-	}
-
-	var r0 error
-	if returnFunc, ok := ret.Get(0).(func() error); ok {
-		r0 = returnFunc()
-	} else {
-		r0 = ret.Error(0)
-	}
-	return r0
-}
-
-// Engine_EndBatch_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'EndBatch'
-type Engine_EndBatch_Call struct {
-	*mock.Call
-}
-
-// EndBatch is a helper method to define mock.On call
-func (_e *Engine_Expecter) EndBatch() *Engine_EndBatch_Call {
-	return &Engine_EndBatch_Call{Call: _e.mock.On("EndBatch")}
-}
-
-func (_c *Engine_EndBatch_Call) Run(run func()) *Engine_EndBatch_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run()
-	})
-	return _c
-}
-
-func (_c *Engine_EndBatch_Call) Return(err error) *Engine_EndBatch_Call {
-	_c.Call.Return(err)
-	return _c
-}
-
-func (_c *Engine_EndBatch_Call) RunAndReturn(run func() error) *Engine_EndBatch_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
 // Move provides a mock function for the type Engine
 func (_mock *Engine) Move(id string, parentid string, target string) error {
 	ret := _mock.Called(id, parentid, target)
@@ -250,6 +206,68 @@ func (_c *Engine_Move_Call) RunAndReturn(run func(id string, parentid string, ta
 	return _c
 }
 
+// NewBatch provides a mock function for the type Engine
+func (_mock *Engine) NewBatch(batchSize int) (search.BatchOperator, error) {
+	ret := _mock.Called(batchSize)
+
+	if len(ret) == 0 {
+		panic("no return value specified for NewBatch")
+	}
+
+	var r0 search.BatchOperator
+	var r1 error
+	if returnFunc, ok := ret.Get(0).(func(int) (search.BatchOperator, error)); ok {
+		return returnFunc(batchSize)
+	}
+	if returnFunc, ok := ret.Get(0).(func(int) search.BatchOperator); ok {
+		r0 = returnFunc(batchSize)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(search.BatchOperator)
+		}
+	}
+	if returnFunc, ok := ret.Get(1).(func(int) error); ok {
+		r1 = returnFunc(batchSize)
+	} else {
+		r1 = ret.Error(1)
+	}
+	return r0, r1
+}
+
+// Engine_NewBatch_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'NewBatch'
+type Engine_NewBatch_Call struct {
+	*mock.Call
+}
+
+// NewBatch is a helper method to define mock.On call
+//   - batchSize int
+func (_e *Engine_Expecter) NewBatch(batchSize interface{}) *Engine_NewBatch_Call {
+	return &Engine_NewBatch_Call{Call: _e.mock.On("NewBatch", batchSize)}
+}
+
+func (_c *Engine_NewBatch_Call) Run(run func(batchSize int)) *Engine_NewBatch_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 int
+		if args[0] != nil {
+			arg0 = args[0].(int)
+		}
+		run(
+			arg0,
+		)
+	})
+	return _c
+}
+
+func (_c *Engine_NewBatch_Call) Return(batchOperator search.BatchOperator, err error) *Engine_NewBatch_Call {
+	_c.Call.Return(batchOperator, err)
+	return _c
+}
+
+func (_c *Engine_NewBatch_Call) RunAndReturn(run func(batchSize int) (search.BatchOperator, error)) *Engine_NewBatch_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Purge provides a mock function for the type Engine
 func (_mock *Engine) Purge(id string) error {
 	ret := _mock.Called(id)
@@ -420,59 +438,8 @@ func (_c *Engine_Search_Call) RunAndReturn(run func(ctx context.Context, req *v0
 	return _c
 }
 
-// StartBatch provides a mock function for the type Engine
-func (_mock *Engine) StartBatch(batchSize int) error {
-	ret := _mock.Called(batchSize)
-
-	if len(ret) == 0 {
-		panic("no return value specified for StartBatch")
-	}
-
-	var r0 error
-	if returnFunc, ok := ret.Get(0).(func(int) error); ok {
-		r0 = returnFunc(batchSize)
-	} else {
-		r0 = ret.Error(0)
-	}
-	return r0
-}
-
-// Engine_StartBatch_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'StartBatch'
-type Engine_StartBatch_Call struct {
-	*mock.Call
-}
-
-// StartBatch is a helper method to define mock.On call
-//   - batchSize int
-func (_e *Engine_Expecter) StartBatch(batchSize interface{}) *Engine_StartBatch_Call {
-	return &Engine_StartBatch_Call{Call: _e.mock.On("StartBatch", batchSize)}
-}
-
-func (_c *Engine_StartBatch_Call) Run(run func(batchSize int)) *Engine_StartBatch_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		var arg0 int
-		if args[0] != nil {
-			arg0 = args[0].(int)
-		}
-		run(
-			arg0,
-		)
-	})
-	return _c
-}
-
-func (_c *Engine_StartBatch_Call) Return(err error) *Engine_StartBatch_Call {
-	_c.Call.Return(err)
-	return _c
-}
-
-func (_c *Engine_StartBatch_Call) RunAndReturn(run func(batchSize int) error) *Engine_StartBatch_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
 // Upsert provides a mock function for the type Engine
-func (_mock *Engine) Upsert(id string, r engine.Resource) error {
+func (_mock *Engine) Upsert(id string, r search.Resource) error {
 	ret := _mock.Called(id, r)
 
 	if len(ret) == 0 {
@@ -480,7 +447,7 @@ func (_mock *Engine) Upsert(id string, r engine.Resource) error {
 	}
 
 	var r0 error
-	if returnFunc, ok := ret.Get(0).(func(string, engine.Resource) error); ok {
+	if returnFunc, ok := ret.Get(0).(func(string, search.Resource) error); ok {
 		r0 = returnFunc(id, r)
 	} else {
 		r0 = ret.Error(0)
@@ -495,20 +462,20 @@ type Engine_Upsert_Call struct {
 
 // Upsert is a helper method to define mock.On call
 //   - id string
-//   - r engine.Resource
+//   - r search.Resource
 func (_e *Engine_Expecter) Upsert(id interface{}, r interface{}) *Engine_Upsert_Call {
 	return &Engine_Upsert_Call{Call: _e.mock.On("Upsert", id, r)}
 }
 
-func (_c *Engine_Upsert_Call) Run(run func(id string, r engine.Resource)) *Engine_Upsert_Call {
+func (_c *Engine_Upsert_Call) Run(run func(id string, r search.Resource)) *Engine_Upsert_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		var arg0 string
 		if args[0] != nil {
 			arg0 = args[0].(string)
 		}
-		var arg1 engine.Resource
+		var arg1 search.Resource
 		if args[1] != nil {
-			arg1 = args[1].(engine.Resource)
+			arg1 = args[1].(search.Resource)
 		}
 		run(
 			arg0,
@@ -523,7 +490,7 @@ func (_c *Engine_Upsert_Call) Return(err error) *Engine_Upsert_Call {
 	return _c
 }
 
-func (_c *Engine_Upsert_Call) RunAndReturn(run func(id string, r engine.Resource) error) *Engine_Upsert_Call {
+func (_c *Engine_Upsert_Call) RunAndReturn(run func(id string, r search.Resource) error) *Engine_Upsert_Call {
 	_c.Call.Return(run)
 	return _c
 }

services/search/pkg/search/search.go

@@ -17,11 +17,49 @@ import (
 
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	searchmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
+	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
 )
 
 var scopeRegex = regexp.MustCompile(`scope:\s*([^" "\n\r]*)`)
 
+// Engine is the interface to the search engine
+type Engine interface {
+	Search(ctx context.Context, req *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error)
+	DocCount() (uint64, error)
+
+	Upsert(id string, r Resource) error
+	Move(id string, parentid string, target string) error
+	Delete(id string) error
+	Restore(id string) error
+	Purge(id string) error
+
+	NewBatch(batchSize int) (BatchOperator, error)
+}
+
+type BatchOperator interface {
+	Upsert(id string, r Resource) error
+	Move(rootID, parentID, location string) error
+	Delete(id string) error
+	Restore(id string) error
+	Purge(id string) error
+
+	Push() error
+}
+
+// Resource is the entity that is stored in the index.
+type Resource struct {
+	content.Document
+
+	ID       string
+	RootID   string
+	Path     string
+	ParentID string
+	Type     uint64
+	Deleted  bool
+	Hidden   bool
+}
+
 // ResolveReference makes sure the path is relative to the space root
 func ResolveReference(ctx context.Context, ref *provider.Reference, ri *provider.ResourceInfo, gatewaySelector pool.Selectable[gateway.GatewayAPIClient]) (*provider.Reference, error) {
 	if ref.GetResourceId().GetOpaqueId() == ref.GetResourceId().GetSpaceId() {
@@ -61,7 +99,7 @@ func (ma matchArray) Less(i, j int) bool {
 	return ma[i].GetScore() > ma[j].GetScore()
 }
 
-func logDocCount(engine engine.Engine, logger log.Logger) {
+func logDocCount(engine Engine, logger log.Logger) {
 	c, err := engine.DocCount()
 	if err != nil {
 		logger.Error().Err(err).Msg("error getting document count from the index")

services/search/pkg/search/service.go

@@ -30,7 +30,6 @@ import (
 	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/metrics"
 )
 
@@ -58,7 +57,7 @@ type Searcher interface {
 type Service struct {
 	logger          log.Logger
 	gatewaySelector pool.Selectable[gateway.GatewayAPIClient]
-	engine          engine.Engine
+	engine          Engine
 	extractor       content.Extractor
 	metrics         *metrics.Metrics
 
@@ -71,7 +70,7 @@ type Service struct {
 var errSkipSpace error
 
 // NewService creates a new Provider instance.
-func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng engine.Engine, extractor content.Extractor, metrics *metrics.Metrics, logger log.Logger, cfg *config.Config) *Service {
+func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng Engine, extractor content.Extractor, metrics *metrics.Metrics, logger log.Logger, cfg *config.Config) *Service {
 	var s = &Service{
 		gatewaySelector: gatewaySelector,
 		engine:          eng,
@@ -463,9 +462,12 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId) error {
 	}()
 
 	w := walker.NewWalker(s.gatewaySelector)
-	s.engine.StartBatch(s.batchSize)
+	batch, err := s.engine.NewBatch(s.batchSize)
+	if err != nil {
+		return err
+	}
 	defer func() {
-		if err := s.engine.EndBatch(); err != nil {
+		if err := batch.Push(); err != nil {
 			s.logger.Error().Err(err).Msg("failed to end batch")
 		}
 	}()
@@ -498,7 +500,7 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId) error {
 			return nil
 		}
 
-		s.UpsertItem(ref)
+		s.doUpsertItem(ref, batch)
 
 		return nil
 	})
@@ -515,14 +517,18 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId) error {
 
 // TrashItem marks the item as deleted.
 func (s *Service) TrashItem(rID *provider.ResourceId) {
-	err := s.engine.Delete(storagespace.FormatResourceID(rID))
-	if err != nil {
+	if err := s.engine.Delete(storagespace.FormatResourceID(rID)); err != nil {
 		s.logger.Error().Err(err).Interface("Id", rID).Msg("failed to remove item from index")
 	}
 }
 
 // UpsertItem indexes or stores Resource data fields.
 func (s *Service) UpsertItem(ref *provider.Reference) {
+	s.doUpsertItem(ref, nil)
+}
+
+// doUpsertItem indexes or stores Resource data fields.
+func (s *Service) doUpsertItem(ref *provider.Reference, batch BatchOperator) {
 	ctx, stat, path := s.resInfo(ref)
 	if ctx == nil || stat == nil || path == "" {
 		return
@@ -534,7 +540,7 @@ func (s *Service) UpsertItem(ref *provider.Reference) {
 		return
 	}
 
-	r := engine.Resource{
+	r := Resource{
 		ID: storagespace.FormatResourceID(stat.Info.Id),
 		RootID: storagespace.FormatResourceID(&provider.ResourceId{
 			StorageId: stat.Info.Id.StorageId,
@@ -551,7 +557,12 @@ func (s *Service) UpsertItem(ref *provider.Reference) {
 		r.ParentID = storagespace.FormatResourceID(parentID)
 	}
 
-	if err = s.engine.Upsert(r.ID, r); err != nil {
+	if batch != nil {
+		err = batch.Upsert(r.ID, r)
+	} else {
+		err = s.engine.Upsert(r.ID, r)
+	}
+	if err != nil {
 		s.logger.Error().Err(err).Msg("error adding updating the resource in the index")
 	} else {
 		logDocCount(s.engine, s.logger)

services/search/pkg/search/service_test.go

@@ -10,20 +10,21 @@ import (
 	typesv1beta1 "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"github.com/opencloud-eu/opencloud/pkg/log"
-	searchmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
-	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
-	contentMocks "github.com/opencloud-eu/opencloud/services/search/pkg/content/mocks"
-	engineMocks "github.com/opencloud-eu/opencloud/services/search/pkg/engine/mocks"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 	revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/status"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 	cs3mocks "github.com/opencloud-eu/reva/v2/tests/cs3mocks/mocks"
 	"github.com/stretchr/testify/mock"
 	"google.golang.org/grpc"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	searchmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
+	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
+	contentMocks "github.com/opencloud-eu/opencloud/services/search/pkg/content/mocks"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+	engineMocks "github.com/opencloud-eu/opencloud/services/search/pkg/search/mocks"
 )
 
 var _ = Describe("Searchprovider", func() {
@@ -117,19 +118,21 @@ var _ = Describe("Searchprovider", func() {
 
 	Describe("IndexSpace", func() {
 		It("walks the space and indexes all files", func() {
+			batch := &engineMocks.BatchOperator{}
+			batch.EXPECT().Push().Return(nil)
 			gatewayClient.On("GetUserByClaim", mock.Anything, mock.Anything).Return(&userv1beta1.GetUserByClaimResponse{
 				Status: status.NewOK(context.Background()),
 				User:   user,
 			}, nil)
 			extractor.On("Extract", mock.Anything, mock.Anything, mock.Anything).Return(content.Document{}, nil)
-			indexClient.On("StartBatch", mock.Anything, mock.Anything).Return(nil)
-			indexClient.On("EndBatch", mock.Anything, mock.Anything).Return(nil)
-			indexClient.On("Upsert", mock.Anything, mock.Anything).Return(nil)
+			indexClient.On("NewBatch", mock.Anything).Return(batch, nil)
+			batch.On("Upsert", mock.Anything, mock.Anything).Return(nil)
 			indexClient.On("Search", mock.Anything, mock.Anything).Return(&searchsvc.SearchIndexResponse{}, nil)
 			gatewayClient.On("Stat", mock.Anything, mock.Anything).Return(&sprovider.StatResponse{
 				Status: status.NewOK(context.Background()),
 				Info:   ri,
 			}, nil)
+
 			err := s.IndexSpace(&sprovider.StorageSpaceId{OpaqueId: "storageid$spaceid!spaceid"})
 			Expect(err).ShouldNot(HaveOccurred())
 		})

services/search/pkg/service/grpc/v0/service.go

@@ -28,11 +28,11 @@ import (
 	"github.com/opencloud-eu/opencloud/pkg/registry"
 	v0 "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
 	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
-	"github.com/opencloud-eu/opencloud/services/search/pkg/query/bleve"
+	bleveQuery "github.com/opencloud-eu/opencloud/services/search/pkg/query/bleve"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 )
 
@@ -44,10 +44,10 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 	cfg := options.Config
 
 	// initialize search engine
-	var eng engine.Engine
+	var eng search.Engine
 	switch cfg.Engine.Type {
 	case "bleve":
-		idx, err := engine.NewBleveIndex(cfg.Engine.Bleve.Datapath)
+		idx, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath)
 		if err != nil {
 			return nil, teardown, err
 		}
@@ -56,7 +56,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 			_ = idx.Close()
 		}
 
-		eng = engine.NewBleveEngine(idx, bleve.DefaultCreator, logger)
+		eng = bleve.NewBackend(idx, bleveQuery.DefaultCreator, logger)
 	case "open-search":
 		client, err := opensearchgoAPI.NewClient(opensearchgoAPI.Config{
 			Client: opensearchgo.Config{