Revert object sync status events

.mockery.yaml

@@ -182,6 +182,11 @@ packages:
   github.com/anyproto/anytype-heart/core/syncstatus/objectsyncstatus:
     interfaces:
       Updater:
+      UpdateReceiver:
+        config:
+          dir: "{{.InterfaceDir}}"
+          outpkg: "{{.PackageName}}"
+          inpackage: true
   github.com/anyproto/anytype-heart/space/spacecore/peermanager:
     interfaces:
       Updater:

core/syncstatus/filestatus.go

@@ -1,6 +1,7 @@
 package syncstatus
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -51,7 +52,10 @@ func (s *service) indexFileSyncStatus(fileObjectId string, status filesyncstatus
 	if err != nil {
 		return fmt.Errorf("get object: %w", err)
 	}
-
+	err = s.updateReceiver.UpdateTree(context.Background(), fileObjectId, status.ToSyncStatus())
+	if err != nil {
+		return fmt.Errorf("update tree: %w", err)
+	}
 	s.sendSpaceStatusUpdate(status, spaceId, bytesLeftPercentage)
 	return nil
 }

core/syncstatus/objectsyncstatus/syncstatus.go

@@ -31,7 +31,8 @@ const (
 var log = logger.NewNamed(syncstatus.CName)
 
 type UpdateReceiver interface {
-	UpdateTree(ctx context.Context, treeId string, status SyncStatus)
+	UpdateTree(ctx context.Context, treeId string, status SyncStatus) (err error)
+	UpdateNodeStatus()
 }
 
 type SyncStatus int
@@ -51,6 +52,7 @@ type StatusUpdater interface {
 type StatusWatcher interface {
 	Watch(treeId string) (err error)
 	Unwatch(treeId string)
+	SetUpdateReceiver(updater UpdateReceiver)
 }
 
 type StatusService interface {
@@ -77,9 +79,10 @@ type Updater interface {
 
 type syncStatusService struct {
 	sync.Mutex
-	configuration nodeconf.NodeConf
-	periodicSync  periodicsync.PeriodicSync
-	storage       spacestorage.SpaceStorage
+	configuration  nodeconf.NodeConf
+	periodicSync   periodicsync.PeriodicSync
+	updateReceiver UpdateReceiver
+	storage        spacestorage.SpaceStorage
 
 	spaceId      string
 	treeHeads    map[string]treeHeadsEntry
@@ -127,6 +130,13 @@ func (s *syncStatusService) Name() (name string) {
 	return syncstatus.CName
 }
 
+func (s *syncStatusService) SetUpdateReceiver(updater UpdateReceiver) {
+	s.Lock()
+	defer s.Unlock()
+
+	s.updateReceiver = updater
+}
+
 func (s *syncStatusService) Run(ctx context.Context) error {
 	s.periodicSync.Run()
 	return nil
@@ -152,6 +162,10 @@ func (s *syncStatusService) update(ctx context.Context) (err error) {
 	s.treeStatusBuf = s.treeStatusBuf[:0]
 
 	s.Lock()
+	if s.updateReceiver == nil {
+		s.Unlock()
+		return
+	}
 	for treeId := range s.watchers {
 		// that means that we haven't yet got the status update
 		treeHeads, exists := s.treeHeads[treeId]
@@ -163,7 +177,12 @@ func (s *syncStatusService) update(ctx context.Context) (err error) {
 		s.treeStatusBuf = append(s.treeStatusBuf, treeStatus{treeId, treeHeads.syncStatus})
 	}
 	s.Unlock()
+	s.updateReceiver.UpdateNodeStatus()
 	for _, entry := range s.treeStatusBuf {
+		err = s.updateReceiver.UpdateTree(ctx, entry.treeId, entry.status)
+		if err != nil {
+			return
+		}
 		s.updateDetails(entry.treeId, mapStatus(entry.status))
 	}
 	return

core/syncstatus/objectsyncstatus/syncstatus_test.go

@@ -228,6 +228,10 @@ func TestSyncStatusService_update(t *testing.T) {
 	t.Run("update: got updates on objects", func(t *testing.T) {
 		// given
 		s := newFixture(t)
+		updateReceiver := NewMockUpdateReceiver(t)
+		updateReceiver.EXPECT().UpdateNodeStatus().Return()
+		updateReceiver.EXPECT().UpdateTree(context.Background(), "id", StatusNotSynced).Return(nil)
+		s.SetUpdateReceiver(updateReceiver)
 
 		// when
 		s.detailsUpdater.EXPECT().UpdateDetails([]string{"id"}, domain.ObjectSyncing, domain.Null, "spaceId")
@@ -239,6 +243,7 @@ func TestSyncStatusService_update(t *testing.T) {
 
 		// then
 		assert.Nil(t, err)
+		updateReceiver.AssertCalled(t, "UpdateTree", context.Background(), "id", StatusNotSynced)
 	})
 }
 

core/syncstatus/service.go

@@ -6,9 +6,13 @@ import (
 
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/commonspace"
+	"github.com/anyproto/any-sync/nodeconf"
 
+	"github.com/anyproto/anytype-heart/core/anytype/config"
 	"github.com/anyproto/anytype-heart/core/block/cache"
+	"github.com/anyproto/anytype-heart/core/event"
 	"github.com/anyproto/anytype-heart/core/filestorage/filesync"
+	"github.com/anyproto/anytype-heart/core/syncstatus/nodestatus"
 	"github.com/anyproto/anytype-heart/core/syncstatus/objectsyncstatus"
 	"github.com/anyproto/anytype-heart/core/syncstatus/spacesyncstatus"
 	"github.com/anyproto/anytype-heart/pkg/lib/localstore/objectstore"
@@ -27,6 +31,7 @@ type Service interface {
 var _ Service = (*service)(nil)
 
 type service struct {
+	updateReceiver  *updateReceiver
 	fileSyncService filesync.FileSync
 
 	objectWatchersLock sync.Mutex
@@ -53,6 +58,12 @@ func (s *service) Init(a *app.App) (err error) {
 	s.fileSyncService.OnUploadStarted(s.onFileUploadStarted)
 	s.fileSyncService.OnLimited(s.onFileLimited)
 	s.fileSyncService.OnDelete(s.OnFileDelete)
+
+	nodeConfService := app.MustComponent[nodeconf.Service](a)
+	cfg := app.MustComponent[*config.Config](a)
+	eventSender := app.MustComponent[event.Sender](a)
+	nodeStatus := app.MustComponent[nodestatus.NodeStatus](a)
+	s.updateReceiver = newUpdateReceiver(nodeConfService, cfg, eventSender, s.objectStore, nodeStatus)
 	return nil
 }
 
@@ -68,7 +79,9 @@ func (s *service) RegisterSpace(space commonspace.Space, sw objectsyncstatus.Sta
 	s.objectWatchersLock.Lock()
 	defer s.objectWatchersLock.Unlock()
 
+	sw.SetUpdateReceiver(s.updateReceiver)
 	s.objectWatchers[space.Id()] = sw
+	s.updateReceiver.spaceId = space.Id()
 }
 
 func (s *service) UnregisterSpace(space commonspace.Space) {