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

@@ -5,8 +5,10 @@ package main
 
 import (
 	"crypto/sha256"
+	"encoding/binary"
 	"encoding/hex"
 	"fmt"
+	"hash"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -21,8 +23,9 @@ import (
 )
 
 const (
-	notificationDedupTTL    = 30 * time.Second
+	webhookDedupTTL         = 30 * time.Second
 	notificationDedupKeyFmt = "notif_dedup_%s"
+	channelPostDedupKeyFmt  = "chan_dedup_%s"
 )
 
 const (
@@ -78,8 +81,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,17 +127,35 @@ 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 {
+		if claimed {
+			p.releaseDedupKey("PostToChannel", dedupKey)
+		}
 		return nil, http.StatusInternalServerError, err
 	}
 
@@ -224,7 +250,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.
@@ -237,9 +263,9 @@ func (wh *webhook) PostNotifications(p *Plugin, instanceID types.ID) ([]*model.P
 		post, err := p.CreateBotDMPost(instance.GetID(), mattermostUserID, notification.message, notification.postType)
 		if err != nil {
 			p.errorf("PostNotifications: failed to create notification post, err: %v", err)
-			// Keep the claim: CreateBotDMPost may have persisted the post despite
-			// returning an error, so releasing it would let a retry post a
-			// duplicate. Let the claim TTL expire on its own.
+			if claimed {
+				p.releaseDedupKey("PostNotifications", dedupKey)
+			}
 			continue
 		}
 		posts = append(posts, post)
@@ -255,15 +281,44 @@ func newWebhook(jwh *JiraWebhook, eventType string, format string, args ...inter
 	}
 }
 
+// releaseDedupKey drops a dedup claim after the post it guarded failed, so a
+// later delivery of the same event can retry instead of being skipped until the
+// claim expires. Only safe to call when this delivery won the claim.
+func (p *Plugin) releaseDedupKey(caller, dedupKey string) {
+	if err := p.client.KV.Delete(dedupKey); err != nil {
+		p.client.Log.Warn(caller+": failed to release dedup key after a failed post; duplicates of this event will be skipped until the claim expires",
+			"key", dedupKey, "error", err.Error())
+	}
+}
+
+// writeDedupToken writes s into h prefixed with its length, so concatenated
+// tokens can't collide due to separator characters appearing inside values.
+func writeDedupToken(h hash.Hash, s string) {
+	var length [8]byte
+	binary.BigEndian.PutUint64(length[:], uint64(len(s)))
+	h.Write(length[:]) //nolint:errcheck // hash.Hash.Write never returns an error
+	h.Write([]byte(s)) //nolint:errcheck // hash.Hash.Write never returns an error
+}
+
 func notificationDedupKey(instanceID types.ID, wh *webhook, recipientID types.ID, message string) string {
-	raw := fmt.Sprintf("%s_%s_%s_%s",
-		string(instanceID),
-		wh.Issue.Key,
-		string(recipientID),
-		message,
-	)
-	hash := sha256.Sum256([]byte(raw))
-	return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(hash[:]))
+	h := sha256.New()
+	for _, token := range []string{string(instanceID), wh.Issue.Key, string(recipientID), message} {
+		writeDedupToken(h, token)
+	}
+	return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(h.Sum(nil)))
+}
+
+func channelPostDedupKey(instanceID types.ID, wh *webhook, channelID string) string {
+	h := sha256.New()
+	for _, token := range []string{string(instanceID), wh.Issue.Key, channelID, wh.headline, wh.text} {
+		writeDedupToken(h, token)
+	}
+	for _, f := range wh.fields {
+		writeDedupToken(h, f.Title)
+		writeDedupToken(h, fmt.Sprintf("%v", f.Value))
+		writeDedupToken(h, fmt.Sprintf("%t", bool(f.Short)))
+	}
+	return fmt.Sprintf(channelPostDedupKeyFmt, hex.EncodeToString(h.Sum(nil)))
 }
 
 func (p *Plugin) GetWebhookURL(jiraURL string, teamID, channelID string) (subURL, legacyURL string, err error) {

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"
 
@@ -336,4 +337,113 @@ func TestNotificationDedupKey(t *testing.T) {
 			notificationDedupKey(instanceID, wh, "user-abc", "Actor **assigned** you to PROJ-1"),
 			notificationDedupKey(instanceID, wh, "user-abc", "Actor **commented** on PROJ-1"))
 	})
+
+	t.Run("values containing separator-like characters do not collide", func(t *testing.T) {
+		// A naive "_"-joined encoding would make ("user-abc", "extra_hello")
+		// collide with ("user-abc_extra", "hello"). The length-prefixed encoding
+		// must keep these distinct.
+		wh := makeWebhook("PROJ-1")
+		assert.NotEqual(t,
+			notificationDedupKey(instanceID, wh, "user-abc", "extra_hello"),
+			notificationDedupKey(instanceID, wh, "user-abc_extra", "hello"))
+
+		// Same idea across the issue key/recipient boundary.
+		assert.NotEqual(t,
+			notificationDedupKey(instanceID, makeWebhook("PROJ-1_user"), "abc", "hello"),
+			notificationDedupKey(instanceID, makeWebhook("PROJ-1"), "user_abc", "hello"))
+	})
+}
+
+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"))
+	})
+
+	t.Run("different field Short flags produce different keys", func(t *testing.T) {
+		wh1 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "High", Short: true}})
+		wh2 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "High", Short: false}})
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, wh1, "channel-abc"),
+			channelPostDedupKey(instanceID, wh2, "channel-abc"))
+	})
+
+	t.Run("values containing separator-like characters do not collide", func(t *testing.T) {
+		// A naive "_"-joined encoding would make ("A_B", "C") collide with ("A", "B_C").
+		// The length-prefixed encoding must keep these distinct.
+		wh1 := makeWebhook("PROJ-1", "A_B", "C", nil)
+		wh2 := makeWebhook("PROJ-1", "A", "B_C", nil)
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, wh1, "channel-abc"),
+			channelPostDedupKey(instanceID, wh2, "channel-abc"))
+
+		// Same idea across the channelID/headline boundary.
+		wh3 := makeWebhook("PROJ-1", "B", "text", nil)
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, wh3, "channel-abc_extra"),
+			channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "extra_B", "text", nil), "channel-abc"))
+
+		// And across a field's title/value boundary.
+		wh4 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "A=B", Value: "C"}})
+		wh5 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "A", Value: "B=C"}})
+		assert.NotEqual(t,
+			channelPostDedupKey(instanceID, wh4, "channel-abc"),
+			channelPostDedupKey(instanceID, wh5, "channel-abc"))
+	})
 }