GO-3861 Merge branch 'main' of ssh://github.com/anyproto/anytype-hear…

…t into go-3861-prioritise-fulltext-indexing-for-spaces

core/indexer/indexer.go

@@ -81,13 +81,13 @@ type indexer struct {
 	btHash  Hasher
 	forceFt chan struct{}
 
-	mb                     *mb.MB[clientspace.Space]
-	lastSpacesSubscription *syncsubscriptions.ObjectSubscription[*types.Struct]
-	lock                   sync.Mutex
-	reindexLogFields       []zap.Field
-	techSpaceId            techSpaceIdGetter
-	spacesPriority         []string
-	spaceReindexQueue      *taskmanager.TasksManager
+	mb                         *mb.MB[clientspace.Space]
+	spacesPrioritySubscription *syncsubscriptions.ObjectSubscription[*types.Struct]
+	lock                       sync.Mutex
+	reindexLogFields           []zap.Field
+	techSpaceId                techSpaceIdGetter
+	spacesPriority             []string
+	spaceReindexQueue          *taskmanager.TasksManager
 }
 
 func (i *indexer) Init(a *app.App) (err error) {
@@ -132,6 +132,7 @@ func (i *indexer) StartFullTextIndex() (err error) {
 
 func (i *indexer) Close(ctx context.Context) (err error) {
 	i.componentCancel()
+	i.spacesPrioritySubscription.Close()
 	i.spaceReindexQueue.WaitAndClose()
 	// we need to wait for the ftQueue processing to be finished gracefully. Because we may be in the middle of badger transaction
 	<-i.ftQueueFinished
@@ -299,13 +300,13 @@ func (i *indexer) subscribeToSpaces() error {
 			},
 		},
 	}
-	updatesChan := make(chan []*types.Struct)
-	go i.spacesSubscriptionChanWatcher(updatesChan)
-	i.lastSpacesSubscription = syncsubscriptions.NewSubscription(i.subscriptionService, objectReq)
-	return i.lastSpacesSubscription.Run(updatesChan)
+	spacePriorityUpdateChan := make(chan []*types.Struct)
+	go i.spacesPrioritySubscriptionWatcher(spacePriorityUpdateChan)
+	i.spacesPrioritySubscription = syncsubscriptions.NewSubscription(i.subscriptionService, objectReq)
+	return i.spacesPrioritySubscription.Run(spacePriorityUpdateChan)
 }
 
-func (i *indexer) spacesSubscriptionSetPriority(priority []string) {
+func (i *indexer) spacesPriorityUpdate(priority []string) {
 	techSpaceId := i.techSpaceId.TechSpaceId()
 	i.lock.Lock()
 	i.spacesPriority = append([]string{techSpaceId}, slices.DeleteFunc(priority, func(s string) bool {
@@ -317,13 +318,15 @@ func (i *indexer) spacesSubscriptionSetPriority(priority []string) {
 	i.spaceReindexQueue.RefreshPriority()
 }
 
-func (i *indexer) spacesSubscriptionChanWatcher(ch chan []*types.Struct) {
+func (i *indexer) spacesPrioritySubscriptionWatcher(ch chan []*types.Struct) {
 	for {
 		select {
-		case <-i.componentCtx.Done():
-			return
+		// subscription and chan will be closed on indexer close
 		case records := <-ch:
-			i.spacesSubscriptionSetPriority(pbtypes.ExtractString(records, bundle.RelationKeyTargetSpaceId.String(), true))
+			if records == nil {
+				return
+			}
+			i.spacesPriorityUpdate(pbtypes.ExtractString(records, bundle.RelationKeyTargetSpaceId.String(), true))
 		}
 	}
 }