fix: use replicated events instead of progress events

Signed-off-by: gfanton <8671905+gfanton@users.noreply.github.com>

go.mod

@@ -6,7 +6,7 @@ require (
 	bazil.org/fuse v0.0.0-20200524192727-fb710f7dfd05 // indirect
 	berty.tech/go-ipfs-log v1.7.0
 	berty.tech/go-ipfs-repo-encrypted v1.1.3
-	berty.tech/go-orbit-db v1.16.0
+	berty.tech/go-orbit-db v1.16.1
 	berty.tech/ipfs-webui-packed v1.0.0-v2.11.4-1
 	fyne.io/fyne/v2 v2.1.1
 	github.com/Masterminds/goutils v1.1.0 // indirect

go/pkg/bertyprotocol/store_message.go

@@ -428,8 +428,8 @@ func constructorFactoryGroupMessage(s *BertyOrbitDB, logger *zap.Logger) iface.S
 
 		chSub, err := store.EventBus().Subscribe([]interface{}{
 			new(stores.EventWrite),
-			new(stores.EventReplicateProgress),
-		})
+			new(stores.EventReplicated),
+		}, eventbus.BufSize(128))
 		if err != nil {
 			return nil, fmt.Errorf("unable to subscribe to store events")
 		}
@@ -444,26 +444,27 @@ func constructorFactoryGroupMessage(s *BertyOrbitDB, logger *zap.Logger) iface.S
 					return
 				}
 
-				var entry ipfslog.Entry
+				var entries []ipfslog.Entry
 
 				switch evt := e.(type) {
 				case stores.EventWrite:
-					entry = evt.Entry
+					entries = []ipfslog.Entry{evt.Entry}
 
-				case stores.EventReplicateProgress:
-					entry = evt.Entry
+				case stores.EventReplicated:
+					entries = evt.Entries
 				}
 
-				ctx = tyber.ContextWithConstantTraceID(ctx, "msgrcvd-"+entry.GetHash().String())
-				store.logger.Debug("Received message store event", tyber.FormatTraceLogFields(ctx)...)
-
-				store.logger.Debug(
-					"Message store event",
-					tyber.FormatStepLogFields(ctx, []tyber.Detail{{Name: "RawEvent", Description: fmt.Sprint(e)}})...,
-				)
-
-				if err := store.addToMessageQueue(ctx, entry); err != nil {
-					logger.Error("unable to add message to queue", zap.Error(err))
+				for _, entry := range entries {
+					ctx = tyber.ContextWithConstantTraceID(ctx, "msgrcvd-"+entry.GetHash().String())
+					store.logger.Debug("Received message store event", tyber.FormatTraceLogFields(ctx)...)
+					store.logger.Debug(
+						"Message store event",
+						tyber.FormatStepLogFields(ctx, []tyber.Detail{{Name: "RawEvent", Description: fmt.Sprint(e)}})...,
+					)
+
+					if err := store.addToMessageQueue(ctx, entry); err != nil {
+						logger.Error("unable to add message to queue", zap.Error(err))
+					}
 				}
 			}
 		}()

go/pkg/bertyprotocol/store_metadata.go

@@ -1029,8 +1029,8 @@ func constructorFactoryGroupMetadata(s *BertyOrbitDB, logger *zap.Logger) iface.
 
 		chSub, err := store.eventBus.Subscribe([]interface{}{
 			new(stores.EventWrite),
-			new(stores.EventReplicateProgress),
-		})
+			new(stores.EventReplicated),
+		}, eventbus.BufSize(128))
 		if err != nil {
 			return nil, fmt.Errorf("unable to subscribe to store events")
 		}
@@ -1049,48 +1049,46 @@ func constructorFactoryGroupMetadata(s *BertyOrbitDB, logger *zap.Logger) iface.
 					return
 				}
 
-				var entry ipfslog.Entry
+				var entries []ipfslog.Entry
 
 				switch evt := e.(type) {
 				case stores.EventWrite:
-					entry = evt.Entry
+					entries = []ipfslog.Entry{evt.Entry}
 
-				case stores.EventReplicateProgress:
-					entry = evt.Entry
+				case stores.EventReplicated:
+					entries = evt.Entries
 				}
 
-				if entry == nil {
-					continue
-				}
-
-				ctx = tyber.ContextWithConstantTraceID(ctx, "msgrcvd-"+entry.GetHash().String())
-				tyber.LogTraceStart(ctx, store.logger, fmt.Sprintf("Received metadata from %s group %s", shortGroupType, b64GroupPK))
-
-				metaEvent, event, err := openMetadataEntry(store.OpLog(), entry, g, store.devKS)
-				if err != nil {
-					_ = tyber.LogFatalError(ctx, store.logger, "Unable to open metadata event", err, tyber.WithDetail("RawEvent", fmt.Sprint(e)), tyber.ForceReopen)
-					continue
-				}
-
-				tyber.LogStep(ctx, store.logger, "Opened metadata store event",
-					tyber.ForceReopen,
-					tyber.EndTrace,
-					tyber.WithJSONDetail("MetaEvent", metaEvent),
-					tyber.WithJSONDetail("Event", event),
-					tyber.UpdateTraceName(fmt.Sprintf("Received %s from %s group %s", strings.TrimPrefix(metaEvent.GetMetadata().GetEventType().String(), "EventType"), shortGroupType, b64GroupPK)),
-				)
-
-				recvEvent := EventMetadataReceived{
-					MetaEvent: metaEvent,
-					Event:     event,
-				}
-
-				if err := store.emitters.metadataReceived.Emit(recvEvent); err != nil {
-					store.logger.Warn("unable to emit recv event", zap.Error(err))
-				}
-
-				if err := store.emitters.groupMetadata.Emit(*metaEvent); err != nil {
-					store.logger.Warn("unable to emit group metadata event", zap.Error(err))
+				for _, entry := range entries {
+					ctx = tyber.ContextWithConstantTraceID(ctx, "msgrcvd-"+entry.GetHash().String())
+					tyber.LogTraceStart(ctx, store.logger, fmt.Sprintf("Received metadata from %s group %s", shortGroupType, b64GroupPK))
+
+					metaEvent, event, err := openMetadataEntry(store.OpLog(), entry, g, store.devKS)
+					if err != nil {
+						_ = tyber.LogFatalError(ctx, store.logger, "Unable to open metadata event", err, tyber.WithDetail("RawEvent", fmt.Sprint(e)), tyber.ForceReopen)
+						continue
+					}
+
+					tyber.LogStep(ctx, store.logger, "Opened metadata store event",
+						tyber.ForceReopen,
+						tyber.EndTrace,
+						tyber.WithJSONDetail("MetaEvent", metaEvent),
+						tyber.WithJSONDetail("Event", event),
+						tyber.UpdateTraceName(fmt.Sprintf("Received %s from %s group %s", strings.TrimPrefix(metaEvent.GetMetadata().GetEventType().String(), "EventType"), shortGroupType, b64GroupPK)),
+					)
+
+					recvEvent := EventMetadataReceived{
+						MetaEvent: metaEvent,
+						Event:     event,
+					}
+
+					if err := store.emitters.metadataReceived.Emit(recvEvent); err != nil {
+						store.logger.Warn("unable to emit recv event", zap.Error(err))
+					}
+
+					if err := store.emitters.groupMetadata.Emit(*metaEvent); err != nil {
+						store.logger.Warn("unable to emit group metadata event", zap.Error(err))
+					}
 				}
 			}
 		}()