MM-70280 Applying dedup on channel subscription posts

server/plugin_test.go

@@ -127,6 +127,7 @@ func TestPlugin(t *testing.T) {
 			api.On("LogWarn", mockAnythingOfTypeBatch("string", 13)...).Return(nil)
 
 			api.On("KVGet", mock.AnythingOfType("string")).Return(make([]byte, 0), (*model.AppError)(nil))
+			api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(true, (*model.AppError)(nil))
 			api.On("GetDirectChannel", mockAnythingOfTypeBatch("string", 2)...).Return(
 				&model.Channel{}, (*model.AppError)(nil))
 			api.On("GetUserByUsername", "theuser").Return(&model.User{

server/webhook.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/pkg/errors"
@@ -21,8 +22,9 @@ import (
 )
 
 const (
-	notificationDedupTTL    = 30 * time.Second
+	webhookDedupTTL         = 30 * time.Second
 	notificationDedupKeyFmt = "notif_dedup_%s"
+	channelPostDedupKeyFmt  = "chan_dedup_%s"
 )
 
 const (
@@ -78,8 +80,13 @@ func (wh webhook) PostToChannel(p *Plugin, instanceID types.ID, channelID, fromU
 
 	if wh.headline == "" {
 		return nil, http.StatusBadRequest, errors.Errorf("unsupported webhook")
-	} else if pluginConfig.DisplaySubscriptionNameInNotifications && subscriptionName != "" {
-		wh.headline = fmt.Sprintf("%s\nSubscription: **%s**", wh.headline, subscriptionName)
+	}
+
+	// Keep the dedup identity independent of the subscription name so overlapping
+	// subscriptions on the same channel collapse into one post.
+	headline := wh.headline
+	if pluginConfig.DisplaySubscriptionNameInNotifications && subscriptionName != "" {
+		headline = fmt.Sprintf("%s\nSubscription: **%s**", headline, subscriptionName)
 	}
 
 	post := &model.Post{
@@ -119,14 +126,29 @@ func (wh webhook) PostToChannel(p *Plugin, instanceID types.ID, channelID, fromU
 			{
 				// TODO is this supposed to be themed?
 				Color:    "#95b7d0",
-				Fallback: wh.headline,
-				Pretext:  wh.headline,
+				Fallback: headline,
+				Pretext:  headline,
 				Text:     text,
 				Fields:   wh.fields,
 			},
 		})
 	} else {
-		post.Message = wh.headline
+		post.Message = headline
+	}
+
+	// Atomically claim the dedup key before posting so concurrent webhook
+	// deliveries can't both pass the check and post duplicates. SetAtomic(nil)
+	// only writes when the key does not already exist.
+	dedupKey := channelPostDedupKey(instanceID, &wh, channelID)
+	claimed, kvErr := p.client.KV.Set(dedupKey, true, pluginapi.SetExpiry(webhookDedupTTL), pluginapi.SetAtomic(nil))
+	switch {
+	case kvErr != nil:
+		// Fail open: post rather than dropping the event.
+		p.client.Log.Warn("PostToChannel: failed to claim dedup key, posting anyway", "key", dedupKey, "error", kvErr.Error())
+	case !claimed:
+		// Another delivery already claimed this post.
+		p.client.Log.Debug("PostToChannel: another delivery already claimed this post, skipping", "key", dedupKey)
+		return nil, http.StatusOK, nil
 	}
 
 	if err := p.client.Post.CreatePost(post); err != nil {
@@ -224,7 +246,7 @@ func (wh *webhook) PostNotifications(p *Plugin, instanceID types.ID) ([]*model.P
 		// deliveries can't both pass the check and send duplicates. SetAtomic(nil)
 		// only writes when the key does not already exist.
 		dedupKey := notificationDedupKey(instance.GetID(), wh, mattermostUserID, notification.message)
-		claimed, kvErr := p.client.KV.Set(dedupKey, true, pluginapi.SetExpiry(notificationDedupTTL), pluginapi.SetAtomic(nil))
+		claimed, kvErr := p.client.KV.Set(dedupKey, true, pluginapi.SetExpiry(webhookDedupTTL), pluginapi.SetAtomic(nil))
 		switch {
 		case kvErr != nil:
 			// Fail open: send the notification rather than dropping it.
@@ -266,6 +288,17 @@ func notificationDedupKey(instanceID types.ID, wh *webhook, recipientID types.ID
 	return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(hash[:]))
 }
 
+func channelPostDedupKey(instanceID types.ID, wh *webhook, channelID string) string {
+	var sb strings.Builder
+	fmt.Fprintf(&sb, "%s_%s_%s_%s_%s",
+		string(instanceID), wh.Issue.Key, channelID, wh.headline, wh.text)
+	for _, f := range wh.fields {
+		fmt.Fprintf(&sb, "_%s=%s", f.Title, f.Value)
+	}
+	hash := sha256.Sum256([]byte(sb.String()))
+	return fmt.Sprintf(channelPostDedupKeyFmt, hex.EncodeToString(hash[:]))
+}
+
 func (p *Plugin) GetWebhookURL(jiraURL string, teamID, channelID string) (subURL, legacyURL string, err error) {
 	cf := p.getConfig()
 

server/webhook_http_test.go

@@ -645,6 +645,7 @@ func TestWebhookHTTP(t *testing.T) {
 			} else {
 				api.On("KVGet", mock.AnythingOfType("string")).Return(nil, (*model.AppError)(nil))
 			}
+			api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(true, nil)
 
 			api.On("LogDebug", mockAnythingOfTypeBatch("string", 11)...).Return(nil)
 			api.On("LogWarn", mockAnythingOfTypeBatch("string", 10)...).Return(nil)

server/webhook_parser_misc_test.go

@@ -11,6 +11,7 @@ import (
 	"testing"
 
 	jira "github.com/andygrunwald/go-jira"
+	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -337,3 +338,66 @@ func TestNotificationDedupKey(t *testing.T) {
 			notificationDedupKey(instanceID, wh, "user-abc", "Actor **commented** on PROJ-1"))
 	})
 }
+
+func TestChannelPostDedupKey(t *testing.T) {
+	makeWebhook := func(issueKey, headline, text string, fields []*model.SlackAttachmentField) *webhook {
+		return &webhook{
+			JiraWebhook: &JiraWebhook{
+				Issue: jira.Issue{Key: issueKey},
+			},
+			headline: headline,
+			text:     text,
+			fields:   fields,
+		}
+	}
+
+	const instanceID = types.ID("https://jira.example.com")
+
+	t.Run("same instance, issue, channel and content produce the same key", func(t *testing.T) {
+		wh1 := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil)
+		wh2 := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil)
+		assert.Equal(t,
+			channelPostDedupKey(instanceID, wh1, "channel-abc"),
+			channelPostDedupKey(instanceID, wh2, "channel-abc"))
+	})
+
+	t.Run("different channels produce different keys", func(t *testing.T) {
+		wh := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil)
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, wh, "channel-abc"),
+			channelPostDedupKey(instanceID, wh, "channel-xyz"))
+	})
+
+	t.Run("different instances produce different keys", func(t *testing.T) {
+		wh := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil)
+		assert.NotEqual(t,
+			channelPostDedupKey(types.ID("https://jira-a.example.com"), wh, "channel-abc"),
+			channelPostDedupKey(types.ID("https://jira-b.example.com"), wh, "channel-abc"))
+	})
+
+	t.Run("different issues produce different keys", func(t *testing.T) {
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "headline", "text", nil), "channel-abc"),
+			channelPostDedupKey(instanceID, makeWebhook("PROJ-2", "headline", "text", nil), "channel-abc"))
+	})
+
+	t.Run("different headlines produce different keys", func(t *testing.T) {
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "text", nil), "channel-abc"),
+			channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "Actor **updated** PROJ-1", "text", nil), "channel-abc"))
+	})
+
+	t.Run("different text produce different keys", func(t *testing.T) {
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "headline", "first comment", nil), "channel-abc"),
+			channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "headline", "second comment", nil), "channel-abc"))
+	})
+
+	t.Run("different fields produce different keys", func(t *testing.T) {
+		wh1 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "High"}})
+		wh2 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "Low"}})
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, wh1, "channel-abc"),
+			channelPostDedupKey(instanceID, wh2, "channel-abc"))
+	})
+}

server/webhook_test.go

@@ -0,0 +1,106 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package main
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	jira "github.com/andygrunwald/go-jira"
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/plugin/plugintest"
+	"github.com/mattermost/mattermost/server/public/plugin/plugintest/mock"
+	"github.com/mattermost/mattermost/server/public/pluginapi"
+	"github.com/stretchr/testify/require"
+)
+
+func newTestChannelWebhook() *webhook {
+	return &webhook{
+		JiraWebhook: &JiraWebhook{
+			Issue: jira.Issue{ID: "10001", Key: "PROJ-1"},
+		},
+		headline: "Actor **commented** on PROJ-1",
+	}
+}
+
+func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) {
+	t.Run("concurrent deliveries for the same channel post only once", func(t *testing.T) {
+		api := &plugintest.API{}
+
+		var kvWinners atomic.Int32
+		api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) {
+			if kvWinners.Add(1) == 1 {
+				return true, nil
+			}
+			return false, nil
+		})
+		api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once()
+
+		p := &Plugin{}
+		p.SetAPI(api)
+		p.client = pluginapi.NewClient(api, p.Driver)
+
+		const callers = 25
+		var wg sync.WaitGroup
+		wg.Add(callers)
+		start := make(chan struct{})
+		for i := 0; i < callers; i++ {
+			go func() {
+				defer wg.Done()
+				<-start
+				_, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "")
+				require.NoError(t, err)
+			}()
+		}
+		close(start)
+		wg.Wait()
+
+		api.AssertExpectations(t)
+	})
+
+	t.Run("overlapping subscriptions on the same channel still post only once", func(t *testing.T) {
+		// Two subscriptions on the same channel produce identical webhook content
+		// but different subscription names; the dedup key must ignore the name.
+		api := &plugintest.API{}
+		var kvWinners atomic.Int32
+		api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) {
+			if kvWinners.Add(1) == 1 {
+				return true, nil
+			}
+			return false, nil
+		}).Twice()
+		api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once()
+
+		p := &Plugin{}
+		p.SetAPI(api)
+		p.client = pluginapi.NewClient(api, p.Driver)
+
+		_, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "subscription-a")
+		require.NoError(t, err)
+
+		post, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "subscription-b")
+		require.NoError(t, err)
+		require.Nil(t, post)
+
+		api.AssertExpectations(t)
+	})
+
+	t.Run("different channels each get their own post", func(t *testing.T) {
+		api := &plugintest.API{}
+		api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(true, (*model.AppError)(nil)).Twice()
+		api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Twice()
+
+		p := &Plugin{}
+		p.SetAPI(api)
+		p.client = pluginapi.NewClient(api, p.Driver)
+
+		_, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "")
+		require.NoError(t, err)
+		_, _, err = newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-2", "bot-user-id", "")
+		require.NoError(t, err)
+
+		api.AssertExpectations(t)
+	})
+}