feat(BRIDGE-150): Observability service modification; user distinction utility & heartbeat; various observbility metrics & relevant integration tests

internal/services/imapservice/observabilitymetrics/evtloopmsgevents/metrics.go

@@ -0,0 +1,67 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package evtloopmsgevents
+
+import (
+	"time"
+
+	"github.com/ProtonMail/go-proton-api"
+)
+
+const (
+	messageEventErrorCaseSchemaName    = "bridge_event_loop_message_event_failures_total"
+	messageEventErrorCaseSchemaVersion = 1
+)
+
+func generateMessageEventFailureObservabilityMetric(eventType string) proton.ObservabilityMetric {
+	return proton.ObservabilityMetric{
+		Name:      messageEventErrorCaseSchemaName,
+		Version:   messageEventErrorCaseSchemaVersion,
+		Timestamp: time.Now().Unix(),
+		Data: map[string]interface{}{
+			"Value": 1,
+			"Labels": map[string]string{
+				"eventType": eventType,
+			},
+		},
+	}
+}
+
+func GenerateMessageEventFailureCreateMessageMetric() proton.ObservabilityMetric {
+	return generateMessageEventFailureObservabilityMetric("createMessageEvent")
+}
+
+func GenerateMessageEventFailureDeleteMessageMetric() proton.ObservabilityMetric {
+	return generateMessageEventFailureObservabilityMetric("deleteMessageEvent")
+}
+
+func GenerateMessageEventFailureUpdateMetric() proton.ObservabilityMetric {
+	return generateMessageEventFailureObservabilityMetric("updateEvent")
+}
+
+func GenerateMessageEventFailedToBuildMessage() proton.ObservabilityMetric {
+	return generateMessageEventFailureObservabilityMetric("failedToBuildMessage")
+}
+
+func GenerateMessageEventFailedToBuildDraft() proton.ObservabilityMetric {
+	return generateMessageEventFailureObservabilityMetric("failedToBuildDraft")
+}
+
+func GenerateMessageEventUpdateChannelDoesNotExist() proton.ObservabilityMetric {
+	return generateMessageEventFailureObservabilityMetric("messageUpdateChannelDoesNotExist")
+}

internal/services/imapservice/observabilitymetrics/syncmsgevents/metrics.go

@@ -0,0 +1,51 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package syncmsgevents
+
+import (
+	"time"
+
+	"github.com/ProtonMail/go-proton-api"
+)
+
+const (
+	syncEventErrorCaseSchemaName    = "bridge_sync_message_event_failures_total"
+	syncEventErrorCaseSchemaVersion = 1
+)
+
+func generateSyncEventFailureObservabilityMetric(eventType string) proton.ObservabilityMetric {
+	return proton.ObservabilityMetric{
+		Name:      syncEventErrorCaseSchemaName,
+		Version:   syncEventErrorCaseSchemaVersion,
+		Timestamp: time.Now().Unix(),
+		Data: map[string]interface{}{
+			"Value": 1,
+			"Labels": map[string]string{
+				"eventType": eventType,
+			},
+		},
+	}
+}
+
+func GenerateSyncFailureCreateMessageEventMetric() proton.ObservabilityMetric {
+	return generateSyncEventFailureObservabilityMetric("createMessageEvent")
+}
+
+func GenerateSyncFailureDeleteMessageEventMetric() proton.ObservabilityMetric {
+	return generateSyncEventFailureObservabilityMetric("deleteMessageEvent")
+}

internal/services/imapservice/service.go

@@ -30,6 +30,7 @@ import (
 	"github.com/ProtonMail/gluon/watcher"
 	"github.com/ProtonMail/go-proton-api"
 	"github.com/ProtonMail/proton-bridge/v3/internal/events"
+	"github.com/ProtonMail/proton-bridge/v3/internal/services/observability"
 	"github.com/ProtonMail/proton-bridge/v3/internal/services/orderedtasks"
 	"github.com/ProtonMail/proton-bridge/v3/internal/services/sendrecorder"
 	"github.com/ProtonMail/proton-bridge/v3/internal/services/syncservice"
@@ -96,6 +97,8 @@ type Service struct {
 	syncConfigPath     string
 	lastHandledEventID string
 	isSyncing          atomic.Bool
+
+	observabilitySender observability.Sender
 }
 
 func NewService(
@@ -116,6 +119,7 @@ func NewService(
 	syncConfigDir string,
 	maxSyncMemory uint64,
 	showAllMail bool,
+	observabilitySender observability.Sender,
 ) *Service {
 	subscriberName := fmt.Sprintf("imap-%v", identityState.User.ID)
 
@@ -160,6 +164,8 @@ func NewService(
 		syncMessageBuilder: syncMessageBuilder,
 		syncReporter:       syncReporter,
 		syncConfigPath:     GetSyncConfigPath(syncConfigDir, identityState.User.ID),
+
+		observabilitySender: observabilitySender,
 	}
 }
 

internal/services/imapservice/service_message_events.go

@@ -26,11 +26,11 @@ import (
 
 	"github.com/ProtonMail/gluon"
 	"github.com/ProtonMail/gluon/imap"
-	"github.com/ProtonMail/gluon/reporter"
 	"github.com/ProtonMail/go-proton-api"
 	"github.com/ProtonMail/gopenpgp/v2/crypto"
-	"github.com/ProtonMail/proton-bridge/v3/internal"
 	"github.com/ProtonMail/proton-bridge/v3/internal/logging"
+	obsMetrics "github.com/ProtonMail/proton-bridge/v3/internal/services/imapservice/observabilitymetrics/evtloopmsgevents"
+	"github.com/ProtonMail/proton-bridge/v3/internal/services/observability"
 	"github.com/ProtonMail/proton-bridge/v3/internal/usertypes"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/exp/maps"
@@ -46,7 +46,7 @@ func (s *Service) HandleMessageEvents(ctx context.Context, events []proton.Messa
 		case proton.EventCreate:
 			updates, err := onMessageCreated(logging.WithLogrusField(ctx, "action", "create message"), s, event.Message, false)
 			if err != nil {
-				reportError(s.reporter, s.log, "Failed to apply create message event", err)
+				s.observabilitySender.AddDistinctMetrics(observability.EventLoopError, obsMetrics.GenerateMessageEventFailureCreateMessageMetric())
 				return fmt.Errorf("failed to handle create message event: %w", err)
 			}
 
@@ -64,7 +64,7 @@ func (s *Service) HandleMessageEvents(ctx context.Context, events []proton.Messa
 					event,
 				)
 				if err != nil {
-					reportError(s.reporter, s.log, "Failed to apply update draft message event", err)
+					s.observabilitySender.AddDistinctMetrics(observability.EventLoopError, obsMetrics.GenerateMessageEventFailureUpdateMetric())
 					return fmt.Errorf("failed to handle update draft event: %w", err)
 				}
 
@@ -85,7 +85,7 @@ func (s *Service) HandleMessageEvents(ctx context.Context, events []proton.Messa
 				event.Message,
 			)
 			if err != nil {
-				reportError(s.reporter, s.log, "Failed to apply update message event", err)
+				s.observabilitySender.AddDistinctMetrics(observability.EventLoopError, obsMetrics.GenerateMessageEventFailureUpdateMetric())
 				return fmt.Errorf("failed to handle update message event: %w", err)
 			}
 
@@ -113,6 +113,7 @@ func (s *Service) HandleMessageEvents(ctx context.Context, events []proton.Messa
 			)
 
 			if err := waitOnIMAPUpdates(ctx, updates); err != nil {
+				s.observabilitySender.AddDistinctMetrics(observability.EventLoopError, obsMetrics.GenerateMessageEventFailureDeleteMessageMetric())
 				return fmt.Errorf("failed to handle delete message event in gluon: %w", err)
 			}
 		}
@@ -158,8 +159,7 @@ func onMessageCreated(
 				s.log.WithError(err).Error("Failed to add failed message ID to vault")
 			}
 
-			reportErrorAndMessageID(s.reporter, s.log, "Failed to build message (event create)", res.err, res.messageID)
-
+			s.observabilitySender.AddDistinctMetrics(observability.EventLoopError, obsMetrics.GenerateMessageEventFailedToBuildMessage())
 			return nil
 		}
 
@@ -221,8 +221,7 @@ func onMessageUpdateDraftOrSent(ctx context.Context, s *Service, event proton.Me
 				s.log.WithError(err).Error("Failed to add failed message ID to vault")
 			}
 
-			reportErrorAndMessageID(s.reporter, s.log, "Failed to build draft message (event update)", res.err, res.messageID)
-
+			s.observabilitySender.AddDistinctMetrics(observability.EventLoopError, obsMetrics.GenerateMessageEventFailedToBuildDraft())
 			return nil
 		}
 
@@ -299,24 +298,6 @@ func onMessageDeleted(ctx context.Context, s *Service, event proton.MessageEvent
 	return updates
 }
 
-func reportError(r reporter.Reporter, entry *logrus.Entry, title string, err error) {
-	reportErrorNoContextCancel(r, entry, title, err, reporter.Context{})
-}
-
-func reportErrorAndMessageID(r reporter.Reporter, entry *logrus.Entry, title string, err error, messgeID string) {
-	reportErrorNoContextCancel(r, entry, title, err, reporter.Context{"messageID": messgeID})
-}
-
-func reportErrorNoContextCancel(r reporter.Reporter, entry *logrus.Entry, title string, err error, reportContext reporter.Context) {
-	if !errors.Is(err, context.Canceled) {
-		reportContext["error"] = err
-		reportContext["error_type"] = internal.ErrCauseType(err)
-		if rerr := r.ReportMessageWithContext(title, reportContext); rerr != nil {
-			entry.WithError(err).WithField("title", title).Error("Failed to report message")
-		}
-	}
-}
-
 // safePublishMessageUpdate handles the rare case where the address' update channel may have been deleted in the same
 // event. This rare case can take place if in the same event fetch request there is an update for delete address and
 // create/update message.
@@ -341,7 +322,7 @@ func safePublishMessageUpdate(ctx context.Context, s *Service, addressID string,
 		}
 
 		logrus.Warnf("Update channel not found for address %v, it may have been already deleted", addressID)
-		_ = s.reporter.ReportMessage("Message Update channel does not exist")
+		s.observabilitySender.AddDistinctMetrics(observability.EventLoopError, obsMetrics.GenerateMessageEventUpdateChannelDoesNotExist())
 
 		return false, nil
 	}

internal/services/imapservice/service_sync_events.go

@@ -23,6 +23,8 @@ import (
 
 	"github.com/ProtonMail/go-proton-api"
 	"github.com/ProtonMail/proton-bridge/v3/internal/logging"
+	obsMetrics "github.com/ProtonMail/proton-bridge/v3/internal/services/imapservice/observabilitymetrics/syncmsgevents"
+	"github.com/ProtonMail/proton-bridge/v3/internal/services/observability"
 	"github.com/ProtonMail/proton-bridge/v3/internal/services/userevents"
 )
 
@@ -55,7 +57,7 @@ func (s syncMessageEventHandler) HandleMessageEvents(ctx context.Context, events
 				true,
 			)
 			if err != nil {
-				reportError(s.service.reporter, s.service.log, "Failed to apply create message event", err)
+				s.service.observabilitySender.AddDistinctMetrics(observability.SyncError, obsMetrics.GenerateSyncFailureCreateMessageEventMetric())
 				return fmt.Errorf("failed to handle create message event: %w", err)
 			}
 
@@ -71,6 +73,7 @@ func (s syncMessageEventHandler) HandleMessageEvents(ctx context.Context, events
 			)
 
 			if err := waitOnIMAPUpdates(ctx, updates); err != nil {
+				s.service.observabilitySender.AddDistinctMetrics(observability.SyncError, obsMetrics.GenerateSyncFailureDeleteMessageEventMetric())
 				return fmt.Errorf("failed to handle delete message event in gluon: %w", err)
 			}
 		default:

internal/services/notifications/service.go

@@ -44,15 +44,16 @@ type Service struct {
 
 	store *Store
 
-	getFlagValueFn            unleash.GetFlagValueFn
-	pushObservabilityMetricFn observability.PushObsMetricFn
+	getFlagValueFn unleash.GetFlagValueFn
+
+	observabilitySender observability.Sender
 }
 
 const bitfieldRegexPattern = `^\\\d+`
 const disableNotificationsKillSwitch = "InboxBridgeEventLoopNotificationDisabled"
 
 func NewService(userID string, service userevents.Subscribable, eventPublisher events.EventPublisher, store *Store,
-	getFlagFn unleash.GetFlagValueFn, pushMetricFn observability.PushObsMetricFn) *Service {
+	getFlagFn unleash.GetFlagValueFn, observabilitySender observability.Sender) *Service {
 	return &Service{
 		userID: userID,
 
@@ -68,8 +69,8 @@ func NewService(userID string, service userevents.Subscribable, eventPublisher e
 
 		store: store,
 
-		getFlagValueFn:            getFlagFn,
-		pushObservabilityMetricFn: pushMetricFn,
+		getFlagValueFn:      getFlagFn,
+		observabilitySender: observabilitySender,
 	}
 }
 
@@ -110,7 +111,7 @@ func (s *Service) HandleNotificationEvents(ctx context.Context, notificationEven
 	s.log.Debug("Handling notification events")
 
 	// Publish observability metrics that we've received notifications
-	s.pushObservabilityMetricFn(GenerateReceivedMetric(len(notificationEvents)))
+	s.observabilitySender.AddMetrics(GenerateReceivedMetric(len(notificationEvents)))
 
 	for _, event := range notificationEvents {
 		ctx = logging.WithLogrusField(ctx, "notificationID", event.ID)
@@ -133,7 +134,7 @@ func (s *Service) HandleNotificationEvents(ctx context.Context, notificationEven
 					Subtitle: event.Payload.Subtitle, Body: event.Payload.Body})
 
 				// Publish observability metric that we've successfully processed notifications
-				s.pushObservabilityMetricFn(GenerateProcessedMetric(1))
+				s.observabilitySender.AddMetrics(GenerateProcessedMetric(1))
 			}
 
 		default:

internal/services/observability/distinction_error_types.go

@@ -0,0 +1,47 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observability
+
+import "time"
+
+// DistinctionErrorTypeEnum - maps to the specific error schema for which we
+// want to send a user update.
+type DistinctionErrorTypeEnum int
+
+const (
+	SyncError DistinctionErrorTypeEnum = iota
+	EventLoopError
+)
+
+// errorSchemaMap - maps between the DistinctionErrorTypeEnum and the relevant schema name.
+var errorSchemaMap = map[DistinctionErrorTypeEnum]string{ //nolint:gochecknoglobals
+	SyncError:      "bridge_sync_errors_users_total",
+	EventLoopError: "bridge_event_loop_events_errors_users_total",
+}
+
+// createLastSentMap - needs to be updated whenever we make changes to the enum.
+func createLastSentMap() map[DistinctionErrorTypeEnum]time.Time {
+	registerTime := time.Now().Add(-updateInterval)
+	lastSentMap := make(map[DistinctionErrorTypeEnum]time.Time)
+
+	for errType := SyncError; errType <= EventLoopError; errType++ {
+		lastSentMap[errType] = registerTime
+	}
+
+	return lastSentMap
+}

internal/services/observability/distinction_utility.go

@@ -0,0 +1,170 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observability
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/ProtonMail/gluon/async"
+	"github.com/ProtonMail/go-proton-api"
+	"github.com/ProtonMail/proton-bridge/v3/internal/updater"
+)
+
+var updateInterval = time.Minute * 5 //nolint:gochecknoglobals
+
+type observabilitySender interface {
+	addMetricsIfClients(metric ...proton.ObservabilityMetric)
+}
+
+// distinctionUtility - used to discern whether X number of events stem from Y number of users.
+type distinctionUtility struct {
+	ctx context.Context
+
+	panicHandler async.PanicHandler
+
+	lastSentMap map[DistinctionErrorTypeEnum]time.Time // Ensures we don't step over the limit of one user update every 5 mins.
+
+	observabilitySender observabilitySender
+	settingsGetter      settingsGetter
+
+	userPlanUnsafe string
+	userPlanLock   sync.Mutex
+
+	heartbeatData     heartbeatData
+	heartbeatDataLock sync.Mutex
+	heartbeatTicker   *time.Ticker
+}
+
+func newDistinctionUtility(ctx context.Context, panicHandler async.PanicHandler, observabilitySender observabilitySender) *distinctionUtility {
+	distinctionUtility := &distinctionUtility{
+		ctx: ctx,
+
+		panicHandler: panicHandler,
+
+		lastSentMap: createLastSentMap(),
+
+		observabilitySender: observabilitySender,
+
+		userPlanUnsafe: planUnknown,
+
+		heartbeatData:   heartbeatData{},
+		heartbeatTicker: time.NewTicker(updateInterval),
+	}
+
+	return distinctionUtility
+}
+
+// sendMetricsWithGuard - schedules the metrics to be sent only if there are authenticated clients.
+func (d *distinctionUtility) sendMetricsWithGuard(metrics ...proton.ObservabilityMetric) {
+	if d.observabilitySender == nil {
+		return
+	}
+
+	d.observabilitySender.addMetricsIfClients(metrics...)
+}
+
+func (d *distinctionUtility) setSettingsGetter(getter settingsGetter) {
+	d.settingsGetter = getter
+}
+
+// checkAndUpdateLastSentMap - checks whether we have sent a relevant user update metric
+// within the last 5 minutes.
+func (d *distinctionUtility) checkAndUpdateLastSentMap(key DistinctionErrorTypeEnum) bool {
+	curTime := time.Now()
+	val, ok := d.lastSentMap[key]
+	if !ok {
+		d.lastSentMap[key] = curTime
+		return true
+	}
+
+	if val.Add(updateInterval).Before(curTime) {
+		d.lastSentMap[key] = curTime
+		return true
+	}
+
+	return false
+}
+
+// generateUserMetric creates the relevant user update metric based on its type
+// and the relevant settings. In the future this will need to be expanded to support multiple
+// versions of the metric if we ever decide to change them.
+func (d *distinctionUtility) generateUserMetric(
+	metricType DistinctionErrorTypeEnum,
+) proton.ObservabilityMetric {
+	schemaName, ok := errorSchemaMap[metricType]
+	if !ok {
+		return proton.ObservabilityMetric{}
+	}
+
+	return generateUserMetric(schemaName, d.getUserPlanSafe(),
+		d.getEmailClientUserAgent(),
+		getEnabled(d.getProxyAllowed()),
+		getEnabled(d.getBetaAccessEnabled()),
+	)
+}
+
+func generateUserMetric(schemaName, plan, mailClient, dohEnabled, betaAccess string) proton.ObservabilityMetric {
+	return proton.ObservabilityMetric{
+		Name:      schemaName,
+		Version:   1,
+		Timestamp: time.Now().Unix(),
+		Data: map[string]interface{}{
+			"Value": 1,
+			"Labels": map[string]string{
+				"plan":              plan,
+				"mailClient":        mailClient,
+				"dohEnabled":        dohEnabled,
+				"betaAccessEnabled": betaAccess,
+			},
+		},
+	}
+}
+
+func (d *distinctionUtility) generateDistinctMetrics(errType DistinctionErrorTypeEnum, metrics ...proton.ObservabilityMetric) []proton.ObservabilityMetric {
+	d.updateHeartbeatData(errType)
+
+	if d.checkAndUpdateLastSentMap(errType) {
+		metrics = append(metrics, d.generateUserMetric(errType))
+	}
+	return metrics
+}
+
+func (d *distinctionUtility) getEmailClientUserAgent() string {
+	ua := ""
+	if d.settingsGetter != nil {
+		ua = d.settingsGetter.GetCurrentUserAgent()
+	}
+
+	return matchUserAgent(ua)
+}
+
+func (d *distinctionUtility) getBetaAccessEnabled() bool {
+	if d.settingsGetter == nil {
+		return false
+	}
+	return d.settingsGetter.GetUpdateChannel() == updater.EarlyChannel
+}
+
+func (d *distinctionUtility) getProxyAllowed() bool {
+	if d.settingsGetter == nil {
+		return false
+	}
+	return d.settingsGetter.GetProxyAllowed()
+}

internal/services/observability/heartbeat.go

@@ -0,0 +1,119 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observability
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/ProtonMail/gluon/async"
+	"github.com/ProtonMail/go-proton-api"
+)
+
+const genericHeartbeatSchemaName = "bridge_generic_user_heartbeat_total"
+
+type heartbeatData struct {
+	receivedSyncError      bool
+	receivedEventLoopError bool
+	receivedOtherError     bool
+}
+
+func (d *distinctionUtility) resetHeartbeatData() {
+	d.heartbeatData.receivedSyncError = false
+	d.heartbeatData.receivedOtherError = false
+	d.heartbeatData.receivedEventLoopError = false
+}
+
+func (d *distinctionUtility) updateHeartbeatData(errType DistinctionErrorTypeEnum) {
+	d.withUpdateHeartbeatDataLock(func() {
+		switch errType {
+		case SyncError:
+			d.heartbeatData.receivedSyncError = true
+		case EventLoopError:
+			d.heartbeatData.receivedEventLoopError = true
+		}
+	})
+}
+
+func (d *distinctionUtility) runHeartbeat() {
+	go func() {
+		defer async.HandlePanic(d.panicHandler)
+		defer d.heartbeatTicker.Stop()
+
+		for {
+			select {
+			case <-d.ctx.Done():
+				return
+			case <-d.heartbeatTicker.C:
+				d.sendHeartbeat()
+			}
+		}
+	}()
+}
+
+func (d *distinctionUtility) withUpdateHeartbeatDataLock(fn func()) {
+	d.heartbeatDataLock.Lock()
+	defer d.heartbeatDataLock.Unlock()
+	fn()
+}
+
+// sendHeartbeat - will only send a heartbeat if there is an authenticated client
+// otherwise we might end up polluting the cache and therefore our metrics.
+func (d *distinctionUtility) sendHeartbeat() {
+	d.withUpdateHeartbeatDataLock(func() {
+		d.sendMetricsWithGuard(d.generateHeartbeatUserMetric())
+		d.resetHeartbeatData()
+	})
+}
+
+func formatBool(value bool) string {
+	return fmt.Sprintf("%t", value)
+}
+
+// generateHeartbeatUserMetric creates the heartbeat user metric and includes the relevant data.
+func (d *distinctionUtility) generateHeartbeatUserMetric() proton.ObservabilityMetric {
+	return generateHeartbeatMetric(
+		d.getUserPlanSafe(),
+		d.getEmailClientUserAgent(),
+		getEnabled(d.settingsGetter.GetProxyAllowed()),
+		getEnabled(d.getBetaAccessEnabled()),
+		formatBool(d.heartbeatData.receivedOtherError),
+		formatBool(d.heartbeatData.receivedSyncError),
+		formatBool(d.heartbeatData.receivedEventLoopError),
+	)
+}
+
+func generateHeartbeatMetric(plan, mailClient, dohEnabled, betaAccess, otherError, syncError, eventLoopError string) proton.ObservabilityMetric {
+	return proton.ObservabilityMetric{
+		Name:      genericHeartbeatSchemaName,
+		Version:   1,
+		Timestamp: time.Now().Unix(),
+		Data: map[string]interface{}{
+			"Value": 1,
+			"Labels": map[string]string{
+				"plan":                   plan,
+				"mailClient":             mailClient,
+				"dohEnabled":             dohEnabled,
+				"betaAccessEnabled":      betaAccess,
+				"receivedOtherError":     otherError,
+				"receivedSyncError":      syncError,
+				"receivedEventLoopError": eventLoopError,
+			},
+		},
+	}
+}

internal/services/observability/plan_utils.go

@@ -0,0 +1,121 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observability
+
+import (
+	"context"
+	"strings"
+
+	"github.com/ProtonMail/gluon/async"
+	"github.com/ProtonMail/go-proton-api"
+)
+
+const (
+	planUnknown    = "unknown"
+	planOther      = "other"
+	planBusiness   = "business"
+	planIndividual = "individual"
+	planGroup      = "group"
+)
+
+var planHierarchy = map[string]int{ //nolint:gochecknoglobals
+	planBusiness:   4,
+	planGroup:      3,
+	planIndividual: 2,
+	planOther:      1,
+	planUnknown:    0,
+}
+
+type planGetter interface {
+	GetOrganizationData(ctx context.Context) (proton.OrganizationResponse, error)
+}
+
+func isHigherPriority(currentPlan, newPlan string) bool {
+	newRank, ok := planHierarchy[newPlan]
+	if !ok {
+		return false
+	}
+
+	currentRank, ok2 := planHierarchy[currentPlan]
+	if !ok2 {
+		return true // we don't have a valid plan, might as well replace it
+	}
+
+	return newRank > currentRank
+}
+
+func mapUserPlan(planName string) string {
+	if planName == "" {
+		return planUnknown
+	}
+	switch strings.TrimSpace(strings.ToLower(planName)) {
+	case "mail2022":
+		return planIndividual
+	case "bundle2022":
+		return planIndividual
+	case "family2022":
+		return planGroup
+	case "visionary2022":
+		return planGroup
+	case "mailpro2022":
+		return planBusiness
+	case "planbiz2024":
+		return planBusiness
+	case "bundlepro2022":
+		return planBusiness
+	case "bundlepro2024":
+		return planBusiness
+	case "duo2024":
+		return planGroup
+
+	default:
+		return planOther
+	}
+}
+
+func (d *distinctionUtility) setUserPlan(planName string) {
+	if planName == "" {
+		return
+	}
+
+	d.userPlanLock.Lock()
+	defer d.userPlanLock.Unlock()
+
+	userPlanMapped := mapUserPlan(planName)
+	if isHigherPriority(d.userPlanUnsafe, userPlanMapped) {
+		d.userPlanUnsafe = userPlanMapped
+	}
+}
+
+func (d *distinctionUtility) registerUserPlan(ctx context.Context, getter planGetter, panicHandler async.PanicHandler) {
+	go func() {
+		defer async.HandlePanic(panicHandler)
+
+		orgRes, err := getter.GetOrganizationData(ctx)
+		if err != nil {
+			return
+		}
+		d.setUserPlan(orgRes.Organization.PlanName)
+	}()
+}
+
+func (d *distinctionUtility) getUserPlanSafe() string {
+	d.userPlanLock.Lock()
+	defer d.userPlanLock.Unlock()
+	return d.userPlanUnsafe
+}

internal/services/observability/service.go

@@ -36,13 +36,18 @@ const (
 	maxBatchSize   = 1000
 )
 
-type PushObsMetricFn func(metric proton.ObservabilityMetric)
-
 type client struct {
 	isTelemetryEnabled func(context.Context) bool
 	sendMetrics        func(context.Context, proton.ObservabilityBatch) error
 }
 
+// Sender - interface maps to the observability service methods,
+// so we can easily pass them down to relevant components.
+type Sender interface {
+	AddMetrics(metrics ...proton.ObservabilityMetric)
+	AddDistinctMetrics(errType DistinctionErrorTypeEnum, metrics ...proton.ObservabilityMetric)
+}
+
 type Service struct {
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -62,6 +67,8 @@ type Service struct {
 
 	userClientStore     map[string]*client
 	userClientStoreLock sync.Mutex
+
+	distinctionUtility *distinctionUtility
 }
 
 func NewService(ctx context.Context, panicHandler async.PanicHandler) *Service {
@@ -85,11 +92,19 @@ func NewService(ctx context.Context, panicHandler async.PanicHandler) *Service {
 		userClientStore: make(map[string]*client),
 	}
 
+	service.distinctionUtility = newDistinctionUtility(ctx, panicHandler, service)
+
 	return service
 }
 
-func (s *Service) Run() {
+// Run starts the observability service goroutine.
+// The function also sets some utility functions to a helper struct aimed at differentiating the amount of users sending metric updates.
+func (s *Service) Run(settingsGetter settingsGetter) {
 	s.log.Info("Starting service")
+
+	s.distinctionUtility.setSettingsGetter(settingsGetter)
+	s.distinctionUtility.runHeartbeat()
+
 	go func() {
 		s.start()
 	}()
@@ -200,7 +215,7 @@ func (s *Service) scheduleDispatch() {
 	}()
 }
 
-func (s *Service) AddMetric(metric proton.ObservabilityMetric) {
+func (s *Service) addMetrics(metric ...proton.ObservabilityMetric) {
 	s.withMetricStoreLock(func() {
 		metricStoreLength := len(s.metricStore)
 		if metricStoreLength >= maxStorageSize {
@@ -209,12 +224,32 @@ func (s *Service) AddMetric(metric proton.ObservabilityMetric) {
 			dropCount := metricStoreLength - maxStorageSize + 1
 			s.metricStore = s.metricStore[dropCount:]
 		}
-		s.metricStore = append(s.metricStore, metric)
+		s.metricStore = append(s.metricStore, metric...)
 	})
 
+	// If the context has been cancelled i.e. the service has been stopped then we should be free to exit.
+	if s.ctx.Err() != nil {
+		return
+	}
+
 	s.sendSignal(s.signalDataArrived)
 }
 
+// addMetricsIfClients - will append a metric only if there are authenticated clients
+// via which we can reach the endpoint.
+func (s *Service) addMetricsIfClients(metric ...proton.ObservabilityMetric) {
+	hasClients := false
+	s.withUserClientStoreLock(func() {
+		hasClients = len(s.userClientStore) > 0
+	})
+
+	if !hasClients {
+		return
+	}
+
+	s.addMetrics(metric...)
+}
+
 func (s *Service) RegisterUserClient(userID string, protonClient *proton.Client, telemetryService *telemetry.Service) {
 	s.log.Info("Registering user client, ID:", userID)
 
@@ -225,6 +260,8 @@ func (s *Service) RegisterUserClient(userID string, protonClient *proton.Client,
 		}
 	})
 
+	s.distinctionUtility.registerUserPlan(s.ctx, protonClient, s.panicHandler)
+
 	// There may be a case where we already have metric updates stored, so try to flush;
 	s.sendSignal(s.signalDataArrived)
 }
@@ -279,3 +316,25 @@ func (s *Service) sendSignal(channel chan struct{}) {
 func ModifyThrottlePeriod(duration time.Duration) {
 	throttleDuration = duration
 }
+
+func (s *Service) AddMetrics(metrics ...proton.ObservabilityMetric) {
+	s.addMetrics(metrics...)
+}
+
+// AddDistinctMetrics - sends an additional metric related to the user, so we can determine
+// what number of events come from what number of users.
+// As the binning interval is what allows us to do this we
+// should not send these if there are no logged-in users at that moment.
+func (s *Service) AddDistinctMetrics(errType DistinctionErrorTypeEnum, metrics ...proton.ObservabilityMetric) {
+	metrics = s.distinctionUtility.generateDistinctMetrics(errType, metrics...)
+	s.addMetricsIfClients(metrics...)
+}
+
+// ModifyHeartbeatInterval - should only be used for testing. Resets the heartbeat ticker.
+func (s *Service) ModifyHeartbeatInterval(duration time.Duration) {
+	s.distinctionUtility.heartbeatTicker.Reset(duration)
+}
+
+func ModifyUserMetricInterval(duration time.Duration) {
+	updateInterval = duration
+}

internal/services/observability/test_utils.go

@@ -0,0 +1,106 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observability
+
+import (
+	"github.com/ProtonMail/go-proton-api"
+)
+
+func GenerateAllUsedDistinctionMetricPermutations() []proton.ObservabilityMetric {
+	planValues := []string{
+		planUnknown,
+		planOther,
+		planBusiness,
+		planIndividual,
+		planGroup}
+	mailClientValues := []string{
+		emailAgentAppleMail,
+		emailAgentOutlook,
+		emailAgentThunderbird,
+		emailAgentOther,
+		emailAgentUnknown,
+	}
+	enabledValues := []string{
+		getEnabled(true), getEnabled(false),
+	}
+
+	var metrics []proton.ObservabilityMetric
+
+	for _, schemaName := range errorSchemaMap {
+		for _, plan := range planValues {
+			for _, mailClient := range mailClientValues {
+				for _, dohEnabled := range enabledValues {
+					for _, betaAccess := range enabledValues {
+						metrics = append(metrics, generateUserMetric(schemaName, plan, mailClient, dohEnabled, betaAccess))
+					}
+				}
+			}
+		}
+	}
+	return metrics
+}
+
+func GenerateAllHeartbeatMetricPermutations() []proton.ObservabilityMetric {
+	planValues := []string{
+		planUnknown,
+		planOther,
+		planBusiness,
+		planIndividual,
+		planGroup}
+	mailClientValues := []string{
+		emailAgentAppleMail,
+		emailAgentOutlook,
+		emailAgentThunderbird,
+		emailAgentOther,
+		emailAgentUnknown,
+	}
+	enabledValues := []string{
+		getEnabled(true), getEnabled(false),
+	}
+
+	trueFalseValues := []string{
+		"true", "false",
+	}
+
+	var metrics []proton.ObservabilityMetric
+	for _, plan := range planValues {
+		for _, mailClient := range mailClientValues {
+			for _, dohEnabled := range enabledValues {
+				for _, betaAccess := range enabledValues {
+					for _, receivedOtherError := range trueFalseValues {
+						for _, receivedSyncError := range trueFalseValues {
+							for _, receivedEventLoopError := range trueFalseValues {
+								metrics = append(metrics,
+									generateHeartbeatMetric(plan,
+										mailClient,
+										dohEnabled,
+										betaAccess,
+										receivedOtherError,
+										receivedSyncError,
+										receivedEventLoopError,
+									),
+								)
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+	return metrics
+}

internal/services/observability/utils.go

@@ -0,0 +1,68 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observability
+
+import (
+	"strings"
+
+	"github.com/ProtonMail/proton-bridge/v3/internal/updater"
+)
+
+// settingsGetter - interface that maps to bridge object methods such that we
+// can pass the whole object instead of individual function callbacks.
+type settingsGetter interface {
+	GetCurrentUserAgent() string
+	GetProxyAllowed() bool
+	GetUpdateChannel() updater.Channel
+}
+
+// User agent mapping.
+const (
+	emailAgentAppleMail   = "apple_mail"
+	emailAgentOutlook     = "outlook"
+	emailAgentThunderbird = "thunderbird"
+	emailAgentOther       = "other"
+	emailAgentUnknown     = "unknown"
+)
+
+func matchUserAgent(userAgent string) string {
+	if userAgent == "" {
+		return emailAgentUnknown
+	}
+
+	userAgent = strings.ToLower(userAgent)
+	switch {
+	case strings.Contains(userAgent, "outlook"):
+		return emailAgentOutlook
+	case strings.Contains(userAgent, "thunderbird"):
+		return emailAgentThunderbird
+	case strings.Contains(userAgent, "mac") && strings.Contains(userAgent, "mail"):
+		return emailAgentAppleMail
+	case strings.Contains(userAgent, "mac") && strings.Contains(userAgent, "notes"):
+		return emailAgentUnknown
+	default:
+		return emailAgentOther
+	}
+}
+
+func getEnabled(value bool) string {
+	if !value {
+		return "disabled"
+	}
+	return "enabled"
+}

internal/services/observability/utils_test.go

@@ -0,0 +1,111 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observability
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMatchUserAgent(t *testing.T) {
+	type agentParseResult struct {
+		agent  string
+		result string
+	}
+
+	testCases := []agentParseResult{
+		{
+			agent:  "Microsoft Outlook/16.0.17928.20114 (Windows 11 Version 23H2)",
+			result: emailAgentOutlook,
+		},
+		{
+			agent:  "Mailbird/3.0.18.0 (Windows 11 Version 23H2)",
+			result: emailAgentOther,
+		},
+		{
+			agent:  "Microsoft Outlook/16.0.17830.20166 (Windows 11 Version 23H2)",
+			result: emailAgentOutlook,
+		},
+		{
+			agent:  "Mac OS X Mail/16.0-3776.700.51 (macOS 14.6)",
+			result: emailAgentAppleMail,
+		},
+		{
+			agent:  "/ (Windows 11 Version 23H2)",
+			result: emailAgentOther,
+		},
+		{
+			agent:  "Microsoft Outlook for Mac/16.88.0-BUILDDAY (macOS 14.6)",
+			result: emailAgentOutlook,
+		},
+		{
+			agent:  "/ (macOS 14.5)",
+			result: emailAgentOther,
+		},
+		{
+			agent:  "/ (Freedesktop SDK 23.08 (Flatpak runtime))",
+			result: emailAgentOther,
+		},
+		{
+			agent:  "Mac OS X Mail/16.0-3774.600.62 (macOS 14.5)",
+			result: emailAgentAppleMail,
+		},
+		{
+			agent:  "Mac OS X Notes/4.11-2817 (macOS 14.6)",
+			result: emailAgentUnknown,
+		},
+		{
+			agent:  "NoClient/0.0.1 (macOS 14.6)",
+			result: emailAgentOther,
+		},
+		{
+			agent:  "Thunderbird/115.15.0 (Ubuntu 20.04.6 LTS)",
+			result: emailAgentThunderbird,
+		},
+		{
+			agent:  "Thunderbird/115.14.0 (macOS 14.6)",
+			result: emailAgentThunderbird,
+		},
+		{
+			agent:  "Thunderbird/115.10.2 (Windows 11 Version 23H2)",
+			result: emailAgentThunderbird,
+		},
+		{
+			agent:  "Mac OS X Notes/4.9-1965 (macOS Monterey (12.0))",
+			result: emailAgentUnknown,
+		},
+		{
+			agent:  " Thunderbird/115.14.0 (macOS 14.6) ",
+			result: emailAgentThunderbird,
+		},
+		{
+			agent:  "",
+			result: emailAgentUnknown,
+		},
+	}
+
+	for _, testCase := range testCases {
+		require.Equal(t, testCase.result, matchUserAgent(testCase.agent))
+	}
+}
+
+func TestFormatBool(t *testing.T) {
+	require.Equal(t, "false", formatBool(false))
+	require.Equal(t, "true", formatBool(true))
+}

internal/services/syncservice/job.go

@@ -160,10 +160,6 @@ func (s *childJob) onError(err error) {
 	s.job.onError(err)
 }
 
-func (s *childJob) userID() string {
-	return s.job.userID
-}
-
 func (s *childJob) chunkDivide(chunks [][]proton.FullMessage) []childJob {
 	numChunks := len(chunks)
 

internal/services/syncservice/observabilitymetrics/metrics.go

@@ -0,0 +1,67 @@
+// Copyright (c) 2024 Proton AG
+//
+// This file is part of Proton Mail Bridge.
+//
+// Proton Mail Bridge is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Proton Mail Bridge is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with Proton Mail Bridge.  If not, see <https://www.gnu.org/licenses/>.
+
+package observabilitymetrics
+
+import (
+	"time"
+
+	"github.com/ProtonMail/go-proton-api"
+)
+
+const (
+	errorCaseSchemaName      = "bridge_sync_message_build_errors_total"
+	errorCaseSchemaVersion   = 1
+	successCaseSchemaName    = "bridge_sync_message_build_success_total"
+	successCaseSchemaVersion = 1
+)
+
+func generateStageBuildFailureObservabilityMetric(errorType string) proton.ObservabilityMetric {
+	return proton.ObservabilityMetric{
+		Name:      errorCaseSchemaName,
+		Version:   errorCaseSchemaVersion,
+		Timestamp: time.Now().Unix(),
+		Data: map[string]interface{}{
+			"Value": 1,
+			"Labels": map[string]string{
+				"errorType": errorType,
+			},
+		},
+	}
+}
+
+func GenerateNoUnlockedKeyringMetric() proton.ObservabilityMetric {
+	return generateStageBuildFailureObservabilityMetric("noUnlockedKeyring")
+}
+
+func GenerateFailedToBuildMetric() proton.ObservabilityMetric {
+	return generateStageBuildFailureObservabilityMetric("failedToBuild")
+}
+
+// GenerateMessageBuiltSuccessMetric - Maybe this is incorrect, I'm not sure how metrics with no labels
+// should be dealt with. The integration tests will tell us.
+func GenerateMessageBuiltSuccessMetric() proton.ObservabilityMetric {
+	return proton.ObservabilityMetric{
+		Name:      successCaseSchemaName,
+		Version:   successCaseSchemaVersion,
+		Timestamp: time.Now().Unix(),
+		Data: map[string]interface{}{
+			"Value":  1,
+			"Labels": map[string]string{},
+		},
+	}
+}

internal/services/syncservice/service.go

@@ -21,7 +21,7 @@ import (
 	"context"
 
 	"github.com/ProtonMail/gluon/async"
-	"github.com/ProtonMail/gluon/reporter"
+	"github.com/ProtonMail/proton-bridge/v3/internal/services/observability"
 )
 
 // Service which mediates IMAP syncing in Bridge.
@@ -36,8 +36,9 @@ type Service struct {
 	group         *async.Group
 }
 
-func NewService(reporter reporter.Reporter,
+func NewService(
 	panicHandler async.PanicHandler,
+	observabilitySender observability.Sender,
 ) *Service {
 	limits := newSyncLimits(2 * Gigabyte)
 
@@ -50,7 +51,7 @@ func NewService(reporter reporter.Reporter,
 		limits:        limits,
 		metadataStage: NewMetadataStage(metaCh, downloadCh, limits.DownloadRequestMem, panicHandler),
 		downloadStage: NewDownloadStage(downloadCh, buildCh, limits.MaxParallelDownloads, panicHandler),
-		buildStage:    NewBuildStage(buildCh, applyCh, limits.MessageBuildMem, panicHandler, reporter),
+		buildStage:    NewBuildStage(buildCh, applyCh, limits.MessageBuildMem, panicHandler, observabilitySender),
 		applyStage:    NewApplyStage(applyCh),
 		metaCh:        metaCh,
 		group:         async.NewGroup(context.Background(), panicHandler),

internal/services/syncservice/stage_build.go

@@ -26,9 +26,10 @@ import (
 
 	"github.com/ProtonMail/gluon/async"
 	"github.com/ProtonMail/gluon/logging"
-	"github.com/ProtonMail/gluon/reporter"
 	"github.com/ProtonMail/go-proton-api"
 	"github.com/ProtonMail/gopenpgp/v2/crypto"
+	"github.com/ProtonMail/proton-bridge/v3/internal/services/observability"
+	obsMetrics "github.com/ProtonMail/proton-bridge/v3/internal/services/syncservice/observabilitymetrics"
 	"github.com/bradenaw/juniper/parallel"
 	"github.com/bradenaw/juniper/xslices"
 	"github.com/sirupsen/logrus"
@@ -50,8 +51,10 @@ type BuildStage struct {
 	maxBuildMem uint64
 
 	panicHandler async.PanicHandler
-	reporter     reporter.Reporter
 	log          *logrus.Entry
+
+	// Observability
+	observabilitySender observability.Sender
 }
 
 func NewBuildStage(
@@ -59,15 +62,15 @@ func NewBuildStage(
 	output BuildStageOutput,
 	maxBuildMem uint64,
 	panicHandler async.PanicHandler,
-	reporter reporter.Reporter,
+	observabilitySender observability.Sender,
 ) *BuildStage {
 	return &BuildStage{
-		input:        input,
-		output:       output,
-		maxBuildMem:  maxBuildMem,
-		log:          logrus.WithField("sync-stage", "build"),
-		panicHandler: panicHandler,
-		reporter:     reporter,
+		input:               input,
+		output:              output,
+		maxBuildMem:         maxBuildMem,
+		log:                 logrus.WithField("sync-stage", "build"),
+		panicHandler:        panicHandler,
+		observabilitySender: observabilitySender,
 	}
 }
 
@@ -147,35 +150,24 @@ func (b *BuildStage) run(ctx context.Context) {
 							req.job.log.WithError(err).Error("Failed to add failed message ID")
 						}
 
-						if err := b.reporter.ReportMessageWithContext("Failed to build message - no unlocked keyring (sync)", reporter.Context{
-							"messageID": msg.ID,
-							"userID":    req.userID(),
-						}); err != nil {
-							req.job.log.WithError(err).Error("Failed to report message build error")
-						}
+						b.observabilitySender.AddDistinctMetrics(observability.SyncError, obsMetrics.GenerateNoUnlockedKeyringMetric())
 						return BuildResult{}, nil
 					}
 
 					res, err := req.job.messageBuilder.BuildMessage(req.job.labels, msg, kr, new(bytes.Buffer))
 					if err != nil {
-						req.job.log.WithError(err).WithField("msgID", msg.ID).Error("Failed to build message (syn)")
+						req.job.log.WithError(err).WithField("msgID", msg.ID).Error("Failed to build message (sync)")
 
 						if err := req.job.state.AddFailedMessageID(req.getContext(), msg.ID); err != nil {
 							req.job.log.WithError(err).Error("Failed to add failed message ID")
 						}
 
-						if err := b.reporter.ReportMessageWithContext("Failed to build message (sync)", reporter.Context{
-							"messageID": msg.ID,
-							"error":     err,
-							"userID":    req.userID(),
-						}); err != nil {
-							req.job.log.WithError(err).Error("Failed to report message build error")
-						}
-
+						b.observabilitySender.AddDistinctMetrics(observability.SyncError, obsMetrics.GenerateFailedToBuildMetric())
 						// We could sync a placeholder message here, but for now we skip it entirely.
 						return BuildResult{}, nil
 					}
 
+					b.observabilitySender.AddMetrics(obsMetrics.GenerateMessageBuiltSuccessMetric())
 					return res, nil
 				})
 				if err != nil {

internal/services/syncservice/stage_build_test.go

@@ -24,10 +24,11 @@ import (
 
 	"github.com/ProtonMail/gluon/async"
 	"github.com/ProtonMail/gluon/imap"
-	"github.com/ProtonMail/gluon/reporter"
 	"github.com/ProtonMail/go-proton-api"
 	"github.com/ProtonMail/gopenpgp/v2/crypto"
 	"github.com/ProtonMail/proton-bridge/v3/internal/bridge/mocks"
+	"github.com/ProtonMail/proton-bridge/v3/internal/services/observability"
+	obsMetrics "github.com/ProtonMail/proton-bridge/v3/internal/services/syncservice/observabilitymetrics"
 	"github.com/bradenaw/juniper/xslices"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
@@ -67,7 +68,6 @@ func TestBuildStage_SuccessRemovesFailedMessage(t *testing.T) {
 
 	input := NewChannelConsumerProducer[BuildRequest]()
 	output := NewChannelConsumerProducer[ApplyRequest]()
-	reporter := mocks.NewMockReporter(mockCtrl)
 
 	labels := getTestLabels()
 
@@ -105,7 +105,10 @@ func TestBuildStage_SuccessRemovesFailedMessage(t *testing.T) {
 	tj.messageBuilder.EXPECT().BuildMessage(gomock.Eq(labels), gomock.Eq(msg), gomock.Any(), gomock.Any()).Return(buildResult, nil)
 	tj.state.EXPECT().RemFailedMessageID(gomock.Any(), gomock.Eq("MSG"))
 
-	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, reporter)
+	observabilityService := mocks.NewMockObservabilitySender(mockCtrl)
+	observabilityService.EXPECT().AddMetrics(obsMetrics.GenerateMessageBuiltSuccessMetric())
+
+	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, observabilityService)
 
 	go func() {
 		stage.run(ctx)
@@ -125,7 +128,7 @@ func TestBuildStage_BuildFailureIsReportedButDoesNotCancelJob(t *testing.T) {
 
 	input := NewChannelConsumerProducer[BuildRequest]()
 	output := NewChannelConsumerProducer[ApplyRequest]()
-	mockReporter := mocks.NewMockReporter(mockCtrl)
+	mockObservabilityService := mocks.NewMockObservabilitySender(mockCtrl)
 
 	labels := getTestLabels()
 
@@ -156,15 +159,12 @@ func TestBuildStage_BuildFailureIsReportedButDoesNotCancelJob(t *testing.T) {
 
 	tj.messageBuilder.EXPECT().BuildMessage(gomock.Eq(labels), gomock.Eq(msg), gomock.Any(), gomock.Any()).Return(BuildResult{}, buildError)
 	tj.state.EXPECT().AddFailedMessageID(gomock.Any(), gomock.Eq([]string{"MSG"}))
-	mockReporter.EXPECT().ReportMessageWithContext(gomock.Any(), gomock.Eq(reporter.Context{
-		"userID":    "u",
-		"messageID": "MSG",
-		"error":     buildError,
-	})).Return(nil)
 
 	tj.syncReporter.EXPECT().OnProgress(gomock.Any(), gomock.Eq(int64(10)))
 
-	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mockReporter)
+	mockObservabilityService.EXPECT().AddDistinctMetrics(observability.SyncError, obsMetrics.GenerateNoUnlockedKeyringMetric())
+
+	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mockObservabilityService)
 
 	go func() {
 		stage.run(ctx)
@@ -183,7 +183,6 @@ func TestBuildStage_FailedToLocateKeyRingIsReportedButDoesNotFailBuild(t *testin
 
 	input := NewChannelConsumerProducer[BuildRequest]()
 	output := NewChannelConsumerProducer[ApplyRequest]()
-	mockReporter := mocks.NewMockReporter(mockCtrl)
 
 	labels := getTestLabels()
 
@@ -209,14 +208,13 @@ func TestBuildStage_FailedToLocateKeyRingIsReportedButDoesNotFailBuild(t *testin
 	tj.job.end()
 
 	tj.state.EXPECT().AddFailedMessageID(gomock.Any(), gomock.Eq([]string{"MSG"}))
-	mockReporter.EXPECT().ReportMessageWithContext(gomock.Any(), gomock.Eq(reporter.Context{
-		"userID":    "u",
-		"messageID": "MSG",
-	})).Return(nil)
 
 	tj.syncReporter.EXPECT().OnProgress(gomock.Any(), gomock.Eq(int64(10)))
 
-	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mockReporter)
+	observabilitySender := mocks.NewMockObservabilitySender(mockCtrl)
+	observabilitySender.EXPECT().AddDistinctMetrics(observability.SyncError)
+
+	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, observabilitySender)
 
 	go func() {
 		stage.run(ctx)
@@ -235,7 +233,6 @@ func TestBuildStage_OtherErrorsFailJob(t *testing.T) {
 
 	input := NewChannelConsumerProducer[BuildRequest]()
 	output := NewChannelConsumerProducer[ApplyRequest]()
-	mockReporter := mocks.NewMockReporter(mockCtrl)
 
 	labels := getTestLabels()
 
@@ -261,7 +258,7 @@ func TestBuildStage_OtherErrorsFailJob(t *testing.T) {
 	childJob := tj.job.newChildJob("f", 10)
 	tj.job.end()
 
-	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mockReporter)
+	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mocks.NewMockObservabilitySender(mockCtrl))
 
 	go func() {
 		stage.run(ctx)
@@ -283,7 +280,6 @@ func TestBuildStage_CancelledJobIsDiscarded(t *testing.T) {
 
 	input := NewChannelConsumerProducer[BuildRequest]()
 	output := NewChannelConsumerProducer[ApplyRequest]()
-	mockReporter := mocks.NewMockReporter(mockCtrl)
 
 	msg := proton.FullMessage{
 		Message: proton.Message{
@@ -294,7 +290,7 @@ func TestBuildStage_CancelledJobIsDiscarded(t *testing.T) {
 		},
 	}
 
-	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mockReporter)
+	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mocks.NewMockObservabilitySender(mockCtrl))
 
 	ctx, cancel := context.WithCancel(context.Background())
 
@@ -327,7 +323,6 @@ func TestTask_EmptyInputDoesNotCrash(t *testing.T) {
 
 	input := NewChannelConsumerProducer[BuildRequest]()
 	output := NewChannelConsumerProducer[ApplyRequest]()
-	reporter := mocks.NewMockReporter(mockCtrl)
 
 	labels := getTestLabels()
 
@@ -340,7 +335,7 @@ func TestTask_EmptyInputDoesNotCrash(t *testing.T) {
 	childJob := tj.job.newChildJob("f", 10)
 	tj.job.end()
 
-	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, reporter)
+	stage := NewBuildStage(input, output, 1024, &async.NoopPanicHandler{}, mocks.NewMockObservabilitySender(mockCtrl))
 
 	go func() {
 		stage.run(ctx)