Remove items from the index when they are purged from the trashbin

Also purge all deleted items when the whole trashbin is purged.

services/search/pkg/bleve/backend_test.go

@@ -460,7 +460,7 @@ var _ = Describe("Bleve", func() {
 			Expect(err).ToNot(HaveOccurred())
 			assertDocCount(rootResource.ID, "Name:child.pdf", 1)
 
-			err = eng.Purge(childResource.ID)
+			err = eng.Purge(childResource.ID, false)
 
 			Expect(err).ToNot(HaveOccurred())
 			assertDocCount(rootResource.ID, "Name:child.pdf", 0)
@@ -474,7 +474,7 @@ var _ = Describe("Bleve", func() {
 			assertDocCount(rootResource.ID, `"`+parentResource.Document.Name+`"`, 1)
 			assertDocCount(rootResource.ID, `"`+childResource.Document.Name+`"`, 1)
 
-			err = eng.Purge(parentResource.ID)
+			err = eng.Purge(parentResource.ID, false)
 			Expect(err).ToNot(HaveOccurred())
 
 			assertDocCount(rootResource.ID, `"`+parentResource.Document.Name+`"`, 0)

services/search/pkg/search/events.go

@@ -21,6 +21,7 @@ func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, m *metrics.
 		events.ItemPurged{},
 		events.ItemRestored{},
 		events.ItemMoved{},
+		events.TrashbinPurged{},
 		events.ContainerCreated{},
 		events.FileTouched{},
 		events.FileVersionRestored{},
@@ -80,6 +81,9 @@ func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, m *metrics.
 					case events.ItemPurged:
 						s.PurgeItem(ev.Ref)
 						e.Ack()
+					case events.TrashbinPurged:
+						s.PurgeDeleted(getSpaceID(ev.Ref))
+						e.Ack()
 					case events.ItemMoved:
 						s.MoveItem(ev.Ref)
 						indexSpaceDebouncer.Debounce(getSpaceID(ev.Ref), e.Ack)

services/search/pkg/search/service.go

@@ -45,7 +45,10 @@ const (
 // Searcher is the interface to the SearchService
 type Searcher interface {
 	Search(ctx context.Context, req *searchsvc.SearchRequest) (*searchsvc.SearchResponse, error)
+
 	IndexSpace(rID *provider.StorageSpaceId) error
+	PurgeDeleted(spaceID *provider.StorageSpaceId) error
+
 	TrashItem(rID *provider.ResourceId)
 	PurgeItem(rID *provider.Reference)
 	UpsertItem(ref *provider.Reference)
@@ -535,7 +538,7 @@ func (s *Service) PurgeItem(ref *provider.Reference) {
 		}
 		logDocCount(s.engine, s.logger)
 	}()
-	err := s.engine.Purge(storagespace.FormatResourceID(ref.ResourceId))
+	err := s.engine.Purge(storagespace.FormatResourceID(ref.ResourceId), false)
 	if err != nil {
 		s.logger.Error().Err(err).Interface("Id", ref.ResourceId).Msg("failed to purge item from index")
 		return
@@ -543,6 +546,32 @@ func (s *Service) PurgeItem(ref *provider.Reference) {
 	s.logger.Info().Interface("Id", ref.ResourceId).Msg("purged item from index")
 }
 
+func (s *Service) PurgeDeleted(spaceID *provider.StorageSpaceId) error {
+	if spaceID == nil {
+		return fmt.Errorf("spaceID must not be nil")
+	}
+
+	rootID, err := storagespace.ParseID(spaceID.OpaqueId)
+	if err != nil {
+		s.logger.Error().Err(err).Msg("invalid space id")
+		return err
+	}
+	if rootID.StorageId == "" || rootID.SpaceId == "" {
+		s.logger.Error().Err(err).Msg("invalid space id")
+		return fmt.Errorf("invalid space id")
+	}
+	rootID.OpaqueId = rootID.SpaceId
+
+	s.engine.StartBatch(s.batchSize)
+	defer func() {
+		if err := s.engine.EndBatch(); err != nil {
+			s.logger.Error().Err(err).Msg("failed to end batch")
+		}
+		logDocCount(s.engine, s.logger)
+	}()
+	return s.engine.Purge(storagespace.FormatResourceID(&rootID), true)
+}
+
 // UpsertItem indexes or stores Resource data fields.
 func (s *Service) UpsertItem(ref *provider.Reference) {
 	s.doUpsertItem(ref, nil)