test(GODT-2442): test bad event feedback and clean-up.

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/Masterminds/semver/v3 v3.1.1
 	github.com/ProtonMail/gluon v0.14.2-0.20230307115918-2c30af37e4dd
 	github.com/ProtonMail/go-autostart v0.0.0-20210130080809-00ed301c8e9a
-	github.com/ProtonMail/go-proton-api v0.3.1-0.20230209110241-fe7894c4931a
+	github.com/ProtonMail/go-proton-api v0.3.1-0.20230306124657-5801649f875e
 	github.com/ProtonMail/go-rfc5322 v0.11.0
 	github.com/ProtonMail/gopenpgp/v2 v2.4.10
 	github.com/PuerkitoBio/goquery v1.8.0
@@ -119,7 +119,6 @@ require (
 )
 
 replace (
-	github.com/ProtonMail/go-proton-api => /home/dev/gopath18/src/go-proton-api
 	github.com/docker/docker-credential-helpers => github.com/ProtonMail/docker-credential-helpers v1.1.0
 	github.com/emersion/go-message => github.com/ProtonMail/go-message v0.0.0-20210611055058-fabeff2ec753
 	github.com/keybase/go-keychain => github.com/cuthix/go-keychain v0.0.0-20220405075754-31e7cee908fe

go.sum

@@ -41,8 +41,8 @@ github.com/ProtonMail/go-message v0.0.0-20210611055058-fabeff2ec753/go.mod h1:NB
 github.com/ProtonMail/go-mime v0.0.0-20220302105931-303f85f7fe0f/go.mod h1:NYt+V3/4rEeDuaev/zw1zCq8uqVEuPHzDPo3OZrlGJ4=
 github.com/ProtonMail/go-mime v0.0.0-20220429130430-2192574d760f h1:4IWzKjHzZxdrW9k4zl/qCwenOVHDbVDADPPHFLjs0Oc=
 github.com/ProtonMail/go-mime v0.0.0-20220429130430-2192574d760f/go.mod h1:qRZgbeASl2a9OwmsV85aWwRqic0NHPh+9ewGAzb4cgM=
-github.com/ProtonMail/go-proton-api v0.3.1-0.20230209110241-fe7894c4931a h1:h9KLPt0HTCJjILYHREWCYnZv+1xaYmOVx/rxiT/1dIg=
+github.com/ProtonMail/go-proton-api v0.3.1-0.20230306124657-5801649f875e h1:hx3jf6I2VRGLG1Kd/kvwks9rtByHiQKhD7mCd9LVcJ0=
-github.com/ProtonMail/go-proton-api v0.3.1-0.20230209110241-fe7894c4931a/go.mod h1:JUo5IQG0hNuPRuDpOUsCOvtee6UjTEHHF1QN2i8RSos=
+github.com/ProtonMail/go-proton-api v0.3.1-0.20230306124657-5801649f875e/go.mod h1:JUo5IQG0hNuPRuDpOUsCOvtee6UjTEHHF1QN2i8RSos=
 github.com/ProtonMail/go-rfc5322 v0.11.0 h1:o5Obrm4DpmQEffvgsVqG6S4BKwC1Wat+hYwjIp2YcCY=
 github.com/ProtonMail/go-rfc5322 v0.11.0/go.mod h1:6oOKr0jXvpoE6pwTx/HukigQpX2J9WUf6h0auplrFTw=
 github.com/ProtonMail/go-srp v0.0.5 h1:xhUioxZgDbCnpo9JehyFhwwsn9JLWkUGfB0oiKXgiGg=

internal/bridge/user.go

@@ -298,10 +298,9 @@ func (bridge *Bridge) SetAddressMode(ctx context.Context, userID string, mode va
 	}, bridge.usersLock)
 }
 
-// SendBadEventUserFeedback sets the address mode for the given user.
+// SendBadEventUserFeedback passes the feedback to the given user.
 func (bridge *Bridge) SendBadEventUserFeedback(ctx context.Context, userID string, doResync bool) error {
-	l := logrus.WithField("userID", userID).WithField("doResycn", doResync)
+	logrus.WithField("userID", userID).WithField("doResycn", doResync).Info("Passing bad event feedback to user")
-	l.Info("Passing bad event feedback to user")
 
 	user, ok := bridge.users[userID]
 	if !ok {

internal/bridge/user_event_test.go

@@ -68,9 +68,13 @@ func TestBridge_User_RefreshEvent(t *testing.T) {
 		require.NoError(t, s.RefreshUser(userID, proton.RefreshMail))
 
 		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, mocks *bridge.Mocks) {
+			syncCh, closeCh := chToType[events.Event, events.SyncFinished](bridge.GetEvents(events.SyncFinished{}))
+
 			mocks.Reporter.EXPECT().ReportMessageWithContext(gomock.Any(), gomock.Any()).MinTimes(1)
 
-			time.Sleep(time.Second)
+			require.Equal(t, userID, (<-syncCh).UserID)
+			closeCh()
+
 			userContinueEventProcess(ctx, t, s, bridge)
 		})
 
@@ -85,64 +89,86 @@ func TestBridge_User_RefreshEvent(t *testing.T) {
 }
 
 func TestBridge_User_BadMessage_BadEvent(t *testing.T) {
-	t.Run("Logout-Login", badMessage_badEvent)
+	t.Run("Resync", test_badMessage_badEvent(func(t *testing.T, ctx context.Context, bridge *bridge.Bridge, badUserID string) {
+		// User feedback is resync
+		require.NoError(t, bridge.SendBadEventUserFeedback(ctx, badUserID, true))
+
+		// Wait for sync to finish
+		syncCh, closeCh := chToType[events.Event, events.SyncFinished](bridge.GetEvents(events.SyncFinished{}))
+		require.Equal(t, badUserID, (<-syncCh).UserID)
+		closeCh()
+	}))
+
+	t.Run("LogoutAndLogin", test_badMessage_badEvent(func(t *testing.T, ctx context.Context, bridge *bridge.Bridge, badUserID string) {
+		// User feedback is logout
+		require.NoError(t, bridge.SendBadEventUserFeedback(ctx, badUserID, false))
+
+		// The user will eventually be logged out due to the bad request errors.
+		require.Eventually(t, func() bool {
+			return len(bridge.GetUserIDs()) == 1 && len(getConnectedUserIDs(t, bridge)) == 0
+		}, 100*user.EventPeriod, user.EventPeriod)
+
+		// Login again
+		_, err := bridge.LoginFull(ctx, "user", password, nil, nil)
+		require.NoError(t, err)
+	}))
 }
 
-func badMessage_badEvent(t *testing.T) {
+func test_badMessage_badEvent(userFeedback func(t *testing.T, ctx context.Context, bridge *bridge.Bridge, badUserID string)) func(t *testing.T) {
-	withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
+	return func(t *testing.T) {
-		// Create a user.
+		withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
-		userID, addrID, err := s.CreateUser("user", password)
+			// Create a user.
-		require.NoError(t, err)
+			userID, addrID, err := s.CreateUser("user", password)
-
-		labelID, err := s.CreateLabel(userID, "folder", "", proton.LabelTypeFolder)
-		require.NoError(t, err)
-
-		// Create 10 messages for the user.
-		withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-			createNumMessages(ctx, t, c, addrID, labelID, 10)
-		})
-
-		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, mocks *bridge.Mocks) {
-			userLoginAndSync(ctx, t, bridge, "user", password)
-
-			var messageIDs []string
-
-			// Create 10 more messages for the user, generating events.
-			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-				messageIDs = createNumMessages(ctx, t, c, addrID, labelID, 10)
-			})
-
-			// If bridge attempts to sync the new messages, it should get a BadRequest error.
-			doBadRequest := true
-			s.AddStatusHook(func(req *http.Request) (int, bool) {
-				if !doBadRequest {
-					return 0, false
-				}
-
-				if xslices.Index(xslices.Map(messageIDs, func(messageID string) string {
-					return "/mail/v4/messages/" + messageID
-				}), req.URL.Path) < 0 {
-					return 0, false
-				}
-
-				return http.StatusBadRequest, true
-			})
-
-			userReceiveBadErrorAndLogout(t, bridge, mocks)
-
-			// Remove messages
-			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-				require.NoError(t, c.DeleteMessage(ctx, messageIDs...))
-			})
-			doBadRequest = false
-
-			// Login again
-			_, err := bridge.LoginFull(ctx, "user", password, nil, nil)
 			require.NoError(t, err)
 
-			userContinueEventProcess(ctx, t, s, bridge)
+			labelID, err := s.CreateLabel(userID, "folder", "", proton.LabelTypeFolder)
+			require.NoError(t, err)
+
+			// Create 10 messages for the user.
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				createNumMessages(ctx, t, c, addrID, labelID, 10)
+			})
+
+			withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, mocks *bridge.Mocks) {
+				userLoginAndSync(ctx, t, bridge, "user", password)
+
+				var messageIDs []string
+
+				// Create 10 more messages for the user, generating events.
+				withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+					messageIDs = createNumMessages(ctx, t, c, addrID, labelID, 10)
+				})
+
+				// If bridge attempts to sync the new messages, it should get a BadRequest error.
+				doBadRequest := true
+				s.AddStatusHook(func(req *http.Request) (int, bool) {
+					if !doBadRequest {
+						return 0, false
+					}
+
+					if xslices.Index(xslices.Map(messageIDs[0:5], func(messageID string) string {
+						return "/mail/v4/messages/" + messageID
+					}), req.URL.Path) < 0 {
+						return 0, false
+					}
+
+					return http.StatusBadRequest, true
+				})
+
+				badUserID := userReceivesBadError(t, bridge, mocks)
+
+				// Remove messages, make response OK again
+				withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+					require.NoError(t, c.DeleteMessage(ctx, messageIDs[0:5]...))
+				})
+				doBadRequest = false
+
+				userFeedback(t, ctx, bridge, badUserID)
+
+				userContinueEventProcess(ctx, t, s, bridge)
+			})
 		})
-	})
+	}
 }
 
 func TestBridge_User_BadMessage_NoBadEvent(t *testing.T) {
@@ -359,21 +385,24 @@ func userLoginAndSync(
 	require.Equal(t, userID, (<-syncCh).UserID)
 }
 
-func userReceiveBadErrorAndLogout(
+func userReceivesBadError(
 	t *testing.T,
 	bridge *bridge.Bridge,
 	mocks *bridge.Mocks,
-) {
+) (userID string) {
+	badEventCh, closeCh := bridge.GetEvents(events.UserBadEvent{})
+
 	// The user will continue to process events and will receive bad request errors.
 	mocks.Reporter.EXPECT().ReportMessageWithContext(gomock.Any(), gomock.Any()).MinTimes(1)
 
-	// The user will eventually be logged out due to the bad request errors.
+	badEvent, ok := (<-badEventCh).(events.UserBadEvent)
-	require.Eventually(t, func() bool {
+	require.True(t, ok)
-		return len(bridge.GetUserIDs()) == 1 && len(getConnectedUserIDs(t, bridge)) == 0
+
-	}, 100*user.EventPeriod, user.EventPeriod)
+	closeCh()
+
+	return badEvent.UserID
 }
 
-// userContinueEventProcess checks that user will continue to process events and will not receive any bad request errors.
 func userContinueEventProcess(
 	ctx context.Context,
 	t *testing.T,

internal/bridge/user_events.go

@@ -141,7 +141,7 @@ func (bridge *Bridge) handleUserDeauth(ctx context.Context, user *user.User) {
 }
 
 func (bridge *Bridge) handleUserBadEvent(ctx context.Context, user *user.User, event events.UserBadEvent) {
-	safe.Lock(func() {
+	go safe.Lock(func() {
 		reportContext := reporter.Context{
 			"user_id":      user.ID(),
 			"old_event_id": event.OldEventID,
@@ -185,6 +185,9 @@ func (bridge *Bridge) getBadEventUserFeedback(userID string) (doResyc bool, err
 		return false, ErrNoSuchUser
 	}
 
+	user.LockEvents()
+	defer user.UnlockEvents()
+
 	return user.GetBadEventFeedback(), nil
 }
 

internal/user/events.go

@@ -90,18 +90,12 @@ func (user *User) handleRefreshEvent(ctx context.Context, refresh proton.Refresh
 }
 
 func (user *User) SyncEvent(ctx context.Context) error {
-	// Cancel the event stream
+	// Abort the event stream
-	user.pollAbort.Abort() // ??? There was a defer here. But I think it's best to do this immediately, there is no reason to continue with polls while having re-sync.
+	defer user.pollAbort.Abort()
 
 	// Re-sync messages after the user, address and label refresh.
 	defer user.goSync()
 
-	// stop IMAP and SMTP
-
-	if err := user.vault.SetEventID(""); err != nil {
-		return fmt.Errorf("failed to clean latest event ID: %w", err)
-	}
-
 	return safe.LockRet(func() error {
 		// Fetch latest user info.
 		apiUser, err := user.client.GetUser(ctx)

internal/user/user.go

@@ -525,6 +525,14 @@ func (user *User) clearSyncStatus() error {
 	return nil
 }
 
+func (user *User) LockEvents() {
+	user.eventLock.Lock()
+}
+
+func (user *User) UnlockEvents() {
+	user.eventLock.Unlock()
+}
+
 // Logout logs the user out from the API.
 func (user *User) Logout(ctx context.Context, withAPI bool) error {
 	user.log.WithField("withAPI", withAPI).Info("Logging out user")