Implement purging items from the index

services/search/pkg/search/events.go

@@ -18,6 +18,7 @@ import (
 func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, m *metrics.Metrics, logger log.Logger) error {
 	evts := []events.Unmarshaller{
 		events.ItemTrashed{},
+		events.ItemPurged{},
 		events.ItemRestored{},
 		events.ItemMoved{},
 		events.ContainerCreated{},
@@ -76,6 +77,9 @@ func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, m *metrics.
 					case events.ItemTrashed:
 						s.TrashItem(ev.ID)
 						indexSpaceDebouncer.Debounce(getSpaceID(ev.Ref), e.Ack)
+					case events.ItemPurged:
+						s.PurgeItem(ev.Ref)
+						e.Ack()
 					case events.ItemMoved:
 						s.MoveItem(ev.Ref)
 						indexSpaceDebouncer.Debounce(getSpaceID(ev.Ref), e.Ack)

services/search/pkg/search/service.go

@@ -47,6 +47,7 @@ type Searcher interface {
 	Search(ctx context.Context, req *searchsvc.SearchRequest) (*searchsvc.SearchResponse, error)
 	IndexSpace(rID *provider.StorageSpaceId) error
 	TrashItem(rID *provider.ResourceId)
+	PurgeItem(rID *provider.Reference)
 	UpsertItem(ref *provider.Reference)
 	RestoreItem(ref *provider.Reference)
 	MoveItem(ref *provider.Reference)
@@ -522,6 +523,27 @@ func (s *Service) TrashItem(rID *provider.ResourceId) {
 	}
 }
 
+func (s *Service) PurgeItem(ref *provider.Reference) {
+	if ref.Path != "" && ref.Path != "." {
+		s.logger.Warn().Str("path", ref.Path).Msg("purging an item with a path is not supported")
+		return
+	}
+
+	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)
+	}()
+	err := s.engine.Purge(storagespace.FormatResourceID(ref.ResourceId))
+	if err != nil {
+		s.logger.Error().Err(err).Interface("Id", ref.ResourceId).Msg("failed to purge item from index")
+		return
+	}
+	s.logger.Info().Interface("Id", ref.ResourceId).Msg("purged item from index")
+}
+
 // UpsertItem indexes or stores Resource data fields.
 func (s *Service) UpsertItem(ref *provider.Reference) {
 	s.doUpsertItem(ref, nil)