MM-67925 Added deduplication of DM notifications

server/plugin.go

@@ -610,11 +610,16 @@ func (p *Plugin) registerChimeraURL() {
 	p.chimeraURL = os.Getenv("MM_PLUGINSETTINGS_CHIMERAOAUTHPROXYURL")
 }
 
+// errDMChannelUnavailable marks a CreateBotDMPost failure that happened
+// before any post was attempted (i.e. resolving the bot's DM channel), so
+// callers can be sure nothing was persisted.
+var errDMChannelUnavailable = errors.New("bot DM channel unavailable")
+
 func (p *Plugin) CreateBotDMPost(userID, message, postType string) error {
 	channel, err := p.client.Channel.GetDirect(userID, p.BotUserID)
 	if err != nil {
 		p.client.Log.Warn("Couldn't get bot's DM channel", "user_id", userID)
-		return err
+		return fmt.Errorf("%w: %w", errDMChannelUnavailable, err)
 	}
 
 	post := &model.Post{

server/webhook.go

@@ -5,6 +5,9 @@ package main
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -18,11 +21,18 @@ import (
 	"github.com/mattermost/mattermost-plugin-gitlab/server/webhook"
 
 	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/pluginapi"
 )
 
 const (
 	webhookTimeout            = 10 * time.Second
 	eventSourceParentPipeline = "parent_pipeline"
+
+	// notificationDedupTTL is the window during which a duplicate DM for the
+	// same recipient and message is suppressed (e.g. duplicate webhook
+	// deliveries from overlapping group/project hooks or GitLab retries).
+	notificationDedupTTL    = 30 * time.Second
+	notificationDedupKeyFmt = "notif_dedup_%s"
 )
 
 type gitlabRetreiver struct {
@@ -184,9 +194,7 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) {
 					continue
 				}
 				if info.Settings.Notifications {
-					if err := p.CreateBotDMPost(userTo, res.Message, "custom_git_review_request"); err != nil {
-						p.client.Log.Warn("can't send dm post", "err", err.Error())
-					}
+					p.sendDMNotification(userTo, res.Message)
 				}
 			}
 		}
@@ -206,6 +214,45 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// notificationDedupKey returns a KV key that uniquely identifies a DM
+// notification by its recipient and rendered message, so that repeated
+// deliveries of the same notification collapse onto the same key.
+func notificationDedupKey(recipientID, message string) string {
+	hash := sha256.Sum256([]byte(recipientID + "_" + message))
+	return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(hash[:]))
+}
+
+// sendDMNotification sends a bot DM to userID, deduplicating against
+// duplicate webhook deliveries (e.g. overlapping group/project hooks, GitLab
+// retries, or concurrent delivery across cluster nodes). It atomically claims
+// a short-lived KV key before posting, so only the first delivery to claim
+// the key actually sends the DM.
+func (p *Plugin) sendDMNotification(userID, message string) {
+	dedupKey := notificationDedupKey(userID, message)
+	claimed, kvErr := p.client.KV.Set(dedupKey, true,
+		pluginapi.SetExpiry(notificationDedupTTL),
+		pluginapi.SetAtomic(nil),
+	)
+	if kvErr != nil {
+		// Fail open: send the notification rather than dropping it.
+		p.client.Log.Warn("failed to claim notification dedup key, sending DM anyway", "err", kvErr.Error())
+	} else if !claimed {
+		// Another delivery already claimed this notification.
+		p.client.Log.Debug("notification already claimed, skipping", "dedup_key", dedupKey)
+		return
+	}
+
+	if err := p.CreateBotDMPost(userID, message, "custom_git_review_request"); err != nil {
+		p.client.Log.Warn("can't send dm post", "err", err.Error())
+		if errors.Is(err, errDMChannelUnavailable) {
+			if delErr := p.client.KV.Delete(dedupKey); delErr != nil {
+				p.client.Log.Warn("failed to release notification dedup key", "err", delErr.Error())
+			}
+			return
+		}
+	}
+}
+
 func (p *Plugin) sendRefreshIfNotAlreadySent(alreadySentRefresh map[string]bool, gitlabUsername string) string {
 	if len(gitlabUsername) == 0 || alreadySentRefresh[gitlabUsername] {
 		return ""

server/webhook_test.go

@@ -6,16 +6,23 @@ package main
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/plugin/plugintest"
 	"github.com/mattermost/mattermost/server/public/pluginapi"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 	gitlabLib "github.com/xanzy/go-gitlab"
 
+	"github.com/mattermost/mattermost-plugin-gitlab/server/gitlab"
 	"github.com/mattermost/mattermost-plugin-gitlab/server/webhook"
 )
 
@@ -38,7 +45,11 @@ func (fakeWebhookHandler) HandleMergeRequest(_ context.Context, _ *gitlabLib.Mer
 }
 
 func (fakeWebhookHandler) HandleIssueComment(_ context.Context, _ *gitlabLib.IssueCommentEvent) ([]*webhook.HandleWebhook, []string, error) {
-	return nil, []string{}, nil
+	return []*webhook.HandleWebhook{{
+		Message: "hello",
+		From:    "test",
+		ToUsers: []string{"known"},
+	}}, []string{}, nil
 }
 
 func (fakeWebhookHandler) HandleMergeRequestComment(_ context.Context, _ *gitlabLib.MergeCommentEvent) ([]*webhook.HandleWebhook, []string, error) {
@@ -191,3 +202,191 @@ func TestHandleWebhookForChildPipelineNotficationEnabled(t *testing.T) {
 	mock.AssertCalled(t, "PublishWebSocketEvent", WsEventRefresh, map[string]any(nil), &model.WebsocketBroadcast{UserId: "1"})
 	mock.AssertNumberOfCalls(t, "PublishWebSocketEvent", 1)
 }
+
+func TestNotificationDedupKey(t *testing.T) {
+	t.Run("same recipient and message produce the same key", func(t *testing.T) {
+		assert.Equal(t,
+			notificationDedupKey("user-1", "hello"),
+			notificationDedupKey("user-1", "hello"))
+	})
+
+	t.Run("different recipients produce different keys", func(t *testing.T) {
+		assert.NotEqual(t,
+			notificationDedupKey("user-1", "hello"),
+			notificationDedupKey("user-2", "hello"))
+	})
+
+	t.Run("different messages produce different keys", func(t *testing.T) {
+		assert.NotEqual(t,
+			notificationDedupKey("user-1", "hello"),
+			notificationDedupKey("user-1", "goodbye"))
+	})
+
+	t.Run("key has the expected prefix and length", func(t *testing.T) {
+		key := notificationDedupKey("user-1", "hello")
+		assert.True(t, strings.HasPrefix(key, "notif_dedup_"))
+		assert.Len(t, key, len("notif_dedup_")+64) // sha256 hex digest is 64 chars
+	})
+}
+
+// noteIssueCommentBody is a minimal GitLab "Note Hook" payload that parses
+// into an *gitlabLib.IssueCommentEvent.
+const noteIssueCommentBody = `{"object_kind":"note","user":{"username":"test"},"object_attributes":{"noteable_type":"Issue"}}` //nolint:misspell // "noteable_type" is GitLab's actual webhook field name
+
+func newIssueCommentWebhookRequest() *http.Request {
+	req := httptest.NewRequest("POST", "/", bytes.NewBufferString(noteIssueCommentBody))
+	req.Header.Add("X-Gitlab-Token", "secret")
+	req.Header.Add("X-Gitlab-Event", string(gitlabLib.EventTypeNote))
+	return req
+}
+
+// setupDedupTestPlugin wires up a Plugin and mock API for a single known
+// recipient ("known" GitLab username, mapped to mattermostUserID) with
+// notifications enabled, ready to receive the "hello" DM from
+// fakeWebhookHandler.HandleIssueComment.
+func setupDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) {
+	t.Helper()
+
+	const mattermostUserID = "1"
+
+	p := &Plugin{configuration: &configuration{WebhookSecret: "secret"}, WebhookHandler: fakeWebhookHandler{}}
+
+	userInfo := &gitlab.UserInfo{
+		UserID:   mattermostUserID,
+		Settings: &gitlab.UserSettings{Notifications: true},
+	}
+	infoJSON, err := json.Marshal(userInfo)
+	require.NoError(t, err)
+
+	api := &plugintest.API{}
+	api.On("KVGet", "test_gitlabusername").Return(nil, nil)
+	api.On("KVGet", "known_gitlabusername").Return([]byte(mattermostUserID), nil)
+	api.On("KVGet", mattermostUserID+GitlabUserInfoKey).Return(infoJSON, nil)
+	api.On("PublishWebSocketEvent", WsEventRefresh, mock.Anything, mock.Anything).Return(nil)
+	api.On("LogInfo", "new msg", "message", "hello", "from", "test").Return(nil)
+	api.On("LogDebug", "notification already claimed, skipping", "dedup_key", mock.Anything).Return(nil)
+	api.On("GetDirectChannel", mattermostUserID, p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil)
+	api.On("CreatePost", mock.MatchedBy(func(post *model.Post) bool {
+		return post.ChannelId == "dm-channel" && post.Message == "hello"
+	})).Return(&model.Post{}, nil)
+
+	p.SetAPI(api)
+	p.client = pluginapi.NewClient(api, p.Driver)
+
+	return p, api
+}
+
+func TestHandleWebhookDeduplicatesDuplicateDelivery(t *testing.T) {
+	p, api := setupDedupTestPlugin(t)
+
+	var claims atomic.Int32
+	api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(
+		func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) {
+			return claims.Add(1) == 1, nil
+		},
+	)
+
+	// Simulate the same GitLab event being delivered twice, e.g. via
+	// overlapping group/project hooks or a GitLab retry.
+	for range 2 {
+		w := httptest.NewRecorder()
+		p.handleWebhook(w, newIssueCommentWebhookRequest())
+		assert.Equal(t, http.StatusOK, w.Result().StatusCode)
+	}
+
+	api.AssertNumberOfCalls(t, "CreatePost", 1)
+}
+
+func TestHandleWebhookDeduplicatesConcurrentDeliveries(t *testing.T) {
+	p, api := setupDedupTestPlugin(t)
+
+	var claims atomic.Int32
+	api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(
+		func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) {
+			return claims.Add(1) == 1, nil
+		},
+	)
+
+	const deliveries = 25
+	var wg sync.WaitGroup
+	wg.Add(deliveries)
+	start := make(chan struct{})
+	for range deliveries {
+		go func() {
+			defer wg.Done()
+			<-start
+			w := httptest.NewRecorder()
+			p.handleWebhook(w, newIssueCommentWebhookRequest())
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	api.AssertNumberOfCalls(t, "CreatePost", 1)
+}
+
+func TestSendDMNotificationFailsOpenOnKVError(t *testing.T) {
+	p := &Plugin{}
+
+	api := &plugintest.API{}
+	api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).
+		Return(false, model.NewAppError("KVSetWithOptions", "id", nil, "boom", http.StatusInternalServerError))
+	api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil)
+	api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil)
+	api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+	p.SetAPI(api)
+	p.client = pluginapi.NewClient(api, p.Driver)
+
+	p.sendDMNotification("user-1", "hello")
+
+	api.AssertNumberOfCalls(t, "CreatePost", 1)
+}
+
+// TestSendDMNotificationReleasesClaimOnDirectChannelFailure verifies that a
+// failure resolving the bot's DM channel (before any post is attempted)
+// releases the dedup claim, so a subsequent retry isn't suppressed for the
+// rest of the TTL.
+func TestSendDMNotificationReleasesClaimOnDirectChannelFailure(t *testing.T) {
+	p := &Plugin{}
+	dedupKey := notificationDedupKey("user-1", "hello")
+
+	api := &plugintest.API{}
+	api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything).Return(true, nil).Once()
+	api.On("GetDirectChannel", "user-1", p.BotUserID).
+		Return(nil, model.NewAppError("GetDirectChannel", "id", nil, "boom", http.StatusInternalServerError))
+	api.On("KVSetWithOptions", dedupKey, isNilBytes, mock.Anything).Return(true, nil).Once()
+	api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+	p.SetAPI(api)
+	p.client = pluginapi.NewClient(api, p.Driver)
+
+	p.sendDMNotification("user-1", "hello")
+
+	api.AssertNotCalled(t, "CreatePost", mock.Anything)
+	api.AssertCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything)
+}
+
+// TestSendDMNotificationKeepsClaimOnCreatePostFailure verifies that a
+// CreatePost failure (which may have persisted the post despite the error)
+// does not release the dedup claim, so a retry can't post a duplicate.
+func TestSendDMNotificationKeepsClaimOnCreatePostFailure(t *testing.T) {
+	p := &Plugin{}
+	dedupKey := notificationDedupKey("user-1", "hello")
+
+	api := &plugintest.API{}
+	api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything).Return(true, nil).Once()
+	api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil)
+	api.On("CreatePost", mock.Anything).
+		Return(nil, model.NewAppError("CreatePost", "id", nil, "boom", http.StatusInternalServerError))
+	api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	// CreateBotDMPost logs its own "CreatePost failed" warning with 3 key/value pairs.
+	api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+	p.SetAPI(api)
+	p.client = pluginapi.NewClient(api, p.Driver)
+
+	p.sendDMNotification("user-1", "hello")
+
+	api.AssertNotCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything)
+}