Update Cloud Slack to always respond in a thread (#1372)

* Refactor e2e tests to handle thread messages assertion

* Add OnChannel assertion

.github/actions/setup-go-mod-private/action.yaml

@@ -12,6 +12,11 @@ inputs:
 runs:
   using: "composite"
   steps:
+    - name: Setup Go
+      uses: actions/setup-go@v4
+      with:
+        go-version-file: './test/go.mod'
+        cache: true
     - name: Download Go modules with private repository
       shell: bash
       run: |

.github/workflows/branch-build.yml

@@ -71,7 +71,6 @@ jobs:
           GORELEASER_CURRENT_TAG: ${{ matrix.image-version }}
           IMAGE_TAG: ${{ matrix.image-version }}
 
-
   integration-tests:
     if: github.event_name != 'repository_dispatch' # skip if triggered by repository_dispatch
     name: Integration tests
@@ -97,12 +96,6 @@ jobs:
         with:
           persist-credentials: false
 
-      - name: Setup Go
-        uses: actions/setup-go@v4
-        with:
-          go-version-file: 'go.mod'
-          cache: true
-
       - name: Setup Go modules
         uses: ./.github/actions/setup-go-mod-private
         with:
@@ -298,14 +291,14 @@ jobs:
         uses: ./.github/actions/cloud-slack-e2e
         with:
           access_token: ${{ secrets.E2E_TEST_GH_DEV_ACCOUNT_PAT }}
-          
+
           slack_workspace_name: ${{ secrets.E2E_DEV_SLACK_WORKSPACE_NAME }}
           slack_email: ${{ secrets.E2E_DEV_SLACK_EMAIL }}
           slack_password: ${{ secrets.E2E_DEV_SLACK_USER_PASSWORD }}
           slack_bot_display_name: "BotkubeDev"
           slack_tester_bot_token: ${{ secrets.E2E_DEV_SLACK_TESTER_BOT_TOKEN }}
           slack_tester_bot_name: "botkubedev"
-            
+
           botkube_cloud_api_base_url: "https://api-dev.botkube.io"
           botkube_cloud_email: ${{ secrets.E2E_DEV_BOTKUBE_CLOUD_EMAIL }}
           botkube_cloud_password: ${{ secrets.E2E_DEV_BOTKUBE_CLOUD_PASSWORD }}

.github/workflows/prod-e2e-test.yml

@@ -8,9 +8,6 @@ on:
 env:
   HELM_VERSION: v3.9.0
   K3D_VERSION: v5.4.6
-  IMAGE_REGISTRY: "ghcr.io"
-  IMAGE_REPOSITORY: "kubeshop/botkube"
-  IMAGE_TAG: v9.99.9-dev # TODO: Use commit hash tag to make the predictable builds for each commit on branch
   GIT_USER: botkube-dev
 
 jobs:

pkg/bot/slack_cloud.go

@@ -550,7 +550,7 @@ func (b *CloudSlack) send(ctx context.Context, event slackMessage, resp interact
 	var file *slack.File
 	var err error
 	if len(markdown) >= slackMaxMessageSize {
-		file, err = uploadFileToSlack(ctx, event.Channel, resp, b.client, event.ThreadTimeStamp)
+		file, err = b.uploadFileToSlack(ctx, event, resp)
 		if err != nil {
 			return err
 		}
@@ -578,10 +578,6 @@ func (b *CloudSlack) send(ctx context.Context, event slackMessage, resp interact
 		b.renderer.RenderInteractiveMessage(resp),
 	}
 
-	if ts := b.getThreadOptionIfNeeded(event, file); ts != nil {
-		options = append(options, ts)
-	}
-
 	if resp.ReplaceOriginal && event.ResponseURL != "" {
 		options = append(options, slack.MsgOptionReplaceOriginal(event.ResponseURL))
 	}
@@ -591,6 +587,10 @@ func (b *CloudSlack) send(ctx context.Context, event slackMessage, resp interact
 			return fmt.Errorf("while posting Slack message visible only to user: %w", err)
 		}
 	} else {
+		if ts := b.getThreadOptionIfNeeded(event, file); ts != nil {
+			options = append(options, ts)
+		}
+
 		if _, _, err := b.client.PostMessageContext(ctx, event.Channel, options...); err != nil {
 			return fmt.Errorf("while posting Slack message: %w", err)
 		}
@@ -600,6 +600,24 @@ func (b *CloudSlack) send(ctx context.Context, event slackMessage, resp interact
 	return nil
 }
 
+func (b *CloudSlack) uploadFileToSlack(ctx context.Context, event slackMessage, resp interactive.CoreMessage) (*slack.File, error) {
+	params := slack.FileUploadParameters{
+		Filename:        "Response.txt",
+		Title:           "Response.txt",
+		InitialComment:  resp.Description,
+		Content:         interactive.MessageToPlaintext(resp, interactive.NewlineFormatter),
+		Channels:        []string{event.Channel},
+		ThreadTimestamp: event.GetTimestamp(),
+	}
+
+	file, err := b.client.UploadFileContext(ctx, params)
+	if err != nil {
+		return nil, fmt.Errorf("while uploading file: %w", err)
+	}
+
+	return file, nil
+}
+
 func (b *CloudSlack) findAndTrimBotMention(msg string) (string, bool) {
 	if !b.botMentionRegex.MatchString(msg) {
 		return "", false
@@ -619,20 +637,17 @@ func (b *CloudSlack) BotName() string {
 }
 
 func (b *CloudSlack) getThreadOptionIfNeeded(event slackMessage, file *slack.File) slack.MsgOption {
-	//if the message is from thread then add an option to return the response to the thread
-	if event.ThreadTimeStamp != "" {
-		return slack.MsgOptionTS(event.ThreadTimeStamp)
-	}
-
-	if file == nil {
-		return nil
+	if file != nil {
+		// If the message was already as a file attachment, reply it a given thread
+		for _, share := range file.Shares.Public {
+			if len(share) >= 1 && share[0].Ts != "" {
+				return slack.MsgOptionTS(share[0].Ts)
+			}
+		}
 	}
 
-	// If the message was already as a file attachment, reply it a given thread
-	for _, share := range file.Shares.Public {
-		if len(share) >= 1 && share[0].Ts != "" {
-			return slack.MsgOptionTS(share[0].Ts)
-		}
+	if ts := event.GetTimestamp(); ts != "" {
+		return slack.MsgOptionTS(ts)
 	}
 
 	return nil

pkg/bot/slack_shared.go

@@ -69,3 +69,13 @@ type slackMessage struct {
 	EventTimeStamp       string
 	RootMessageTimeStamp string
 }
+
+// GetTimestamp returns the timestamp for the response message.
+func (s *slackMessage) GetTimestamp() string {
+	// If the event is coming from the thread, then we simply respond in that thread
+	if s.ThreadTimeStamp != "" {
+		return s.ThreadTimeStamp
+	}
+	// otherwise, we use the event timestamp to respond in the thread to the message that triggered our response
+	return s.EventTimeStamp
+}

test/cloud-slack-dev-e2e/cloud_slack_dev_e2e_test.go

@@ -374,34 +374,46 @@ func TestCloudSlackE2E(t *testing.T) {
 			t.Log("Testing ping with --cluster-name")
 			command := fmt.Sprintf("ping --cluster-name %s", deployment.Name)
 			expectedMessage := fmt.Sprintf("`%s` on `%s`\n```\npong", command, deployment.Name)
-			tester.PostMessageToBot(t, channel.ID(), command)
+			tester.PostMessageToBot(t, channel.Identifier(), command)
 			err = tester.WaitForLastMessageContains(tester.BotUserID(), channel.ID(), expectedMessage)
 			require.NoError(t, err)
 
 			t.Log("Testing ping for not connected deployment #2")
 			command = "ping"
 			expectedBlockMessage := notConnectedMessage(deployment2.Name, deployment2.ID)
-			tester.PostMessageToBot(t, channel.ID(), fmt.Sprintf("%s --cluster-name %s", command, deployment2.Name))
+			tester.PostMessageToBot(t, channel.Identifier(), fmt.Sprintf("%s --cluster-name %s", command, deployment2.Name))
 
 			renderedMsg := interactive.RenderMessage(tester.MDFormatter(), expectedBlockMessage)
 			renderedMsg = strings.Replace(renderedMsg, "\n", " ", -1)
 			renderedMsg = strings.TrimSuffix(renderedMsg, " ")
 			err = tester.WaitForLastInteractiveMessagePostedEqualWithCustomRender(tester.BotUserID(), channel.ID(), renderedMsg)
+			if err != nil { // the new cloud backend not release yet
+				t.Logf("Fallback to the old behavior with message sent at the channel level...")
+				err = tester.OnChannel().WaitForLastInteractiveMessagePostedEqualWithCustomRender(tester.BotUserID(), channel.ID(), renderedMsg)
+			}
 			require.NoError(t, err)
 
 			t.Log("Testing ping for not existing deployment")
 			command = "ping"
 			deployName := "non-existing-deployment"
 			expectedMessage = fmt.Sprintf("*Instance not found* The cluster %q does not exist.", deployName)
-			tester.PostMessageToBot(t, channel.ID(), fmt.Sprintf("%s --cluster-name %s", command, deployName))
+			tester.PostMessageToBot(t, channel.Identifier(), fmt.Sprintf("%s --cluster-name %s", command, deployName))
 			err = tester.WaitForLastMessageContains(tester.BotUserID(), channel.ID(), expectedMessage)
+			if err != nil { // the new cloud backend not release yet
+				t.Logf("Fallback to the old behavior with message sent at the channel level...")
+				err = tester.OnChannel().WaitForLastMessageContains(tester.BotUserID(), channel.ID(), expectedMessage)
+			}
 			require.NoError(t, err)
 
 			t.Log("Setting cluster as default")
-			tester.PostMessageToBot(t, channel.ID(), fmt.Sprintf("cloud set default-instance %s", deployment.ID))
+			tester.PostMessageToBot(t, channel.Identifier(), fmt.Sprintf("cloud set default-instance %s", deployment.ID))
 			t.Log("Waiting for confirmation message...")
 			expectedClusterDefaultMsg := fmt.Sprintf(":white_check_mark: Instance %s was successfully selected as the default cluster for this channel.", deployment.Name)
 			err = tester.WaitForLastMessageEqual(tester.BotUserID(), channel.ID(), expectedClusterDefaultMsg)
+			if err != nil { // the new cloud backend not release yet
+				t.Logf("Fallback to the old behavior with message sent at the channel level...")
+				err = tester.OnChannel().WaitForLastMessageEqual(tester.BotUserID(), channel.ID(), expectedClusterDefaultMsg)
+			}
 			require.NoError(t, err)
 
 			t.Log("Testing getting all deployments")
@@ -411,7 +423,7 @@ func TestCloudSlackE2E(t *testing.T) {
 					strings.Contains(msg, "coredns") &&
 					strings.Contains(msg, "botkube"), 0, ""
 			}
-			tester.PostMessageToBot(t, channel.ID(), command)
+			tester.PostMessageToBot(t, channel.Identifier(), command)
 			err = tester.WaitForMessagePosted(tester.BotUserID(), channel.ID(), 1, assertionFn)
 			require.NoError(t, err)
 		})
@@ -458,22 +470,23 @@ func TestCloudSlackE2E(t *testing.T) {
 				}
 				return result, 0, ""
 			}
-			err = tester.WaitForMessagePosted(tester.BotUserID(), channel.ID(), 1, assertionFn)
+			err = tester.OnChannel().WaitForMessagePosted(tester.BotUserID(), channel.ID(), 1, assertionFn)
 			require.NoError(t, err)
 		})
 
 		t.Run("Botkube Deployment -> Cloud sync", func(t *testing.T) {
 			t.Log("Disabling notification...")
-			tester.PostMessageToBot(t, channel.ID(), "disable notifications")
+			tester.PostMessageToBot(t, channel.Identifier(), "disable notifications")
 			t.Log("Waiting for config reload message...")
 			expectedReloadMsg := fmt.Sprintf(":arrows_counterclockwise: Configuration reload requested for cluster '%s'. Hold on a sec...", deployment.Name)
-			err = tester.WaitForMessagePostedRecentlyEqual(tester.BotUserID(), channel.ID(), expectedReloadMsg)
+
+			err = tester.OnChannel().WaitForMessagePostedRecentlyEqual(tester.BotUserID(), channel.ID(), expectedReloadMsg)
 			require.NoError(t, err)
 
 			t.Log("Waiting for watch begin message...")
 			expectedWatchBeginMsg := fmt.Sprintf("My watch begins for cluster '%s'! :crossed_swords:", deployment.Name)
 			recentMessages := 2 // take into the account the optional "upgrade checker message"
-			err = tester.WaitForMessagePosted(tester.BotUserID(), channel.ID(), recentMessages, func(msg string) (bool, int, string) {
+			err = tester.OnChannel().WaitForMessagePosted(tester.BotUserID(), channel.ID(), recentMessages, func(msg string) (bool, int, string) {
 				if !strings.EqualFold(expectedWatchBeginMsg, msg) {
 					count := diff.CountMatchBlock(expectedWatchBeginMsg, msg)
 					msgDiff := diff.Diff(expectedWatchBeginMsg, msg)
@@ -490,7 +503,7 @@ func TestCloudSlackE2E(t *testing.T) {
 			command := "status notifications"
 			expectedBody := formatx.CodeBlock(fmt.Sprintf("Notifications from cluster '%s' are disabled here.", deployment.Name))
 			expectedMessage := fmt.Sprintf("%s\n%s", cmdHeader(command), expectedBody)
-			tester.PostMessageToBot(t, channel.ID(), "status notifications")
+			tester.PostMessageToBot(t, channel.Identifier(), "status notifications")
 			err = tester.WaitForLastMessageEqual(tester.BotUserID(), channel.ID(), expectedMessage)
 			require.NoError(t, err)
 		})
@@ -502,13 +515,13 @@ func TestCloudSlackE2E(t *testing.T) {
 
 			t.Log("Waiting for config reload message...")
 			expectedReloadMsg := fmt.Sprintf(":arrows_counterclockwise: Configuration reload requested for cluster '%s'. Hold on a sec...", deployment.Name)
-			err = tester.WaitForMessagePostedRecentlyEqual(tester.BotUserID(), channel.ID(), expectedReloadMsg)
+			err = tester.OnChannel().WaitForMessagePostedRecentlyEqual(tester.BotUserID(), channel.ID(), expectedReloadMsg)
 			require.NoError(t, err)
 
 			t.Log("Waiting for watch begin message...")
 			expectedWatchBeginMsg := fmt.Sprintf("My watch begins for cluster '%s'! :crossed_swords:", deployment.Name)
 			recentMessages := 2 // take into the account the  optional "upgrade checker message"
-			err = tester.WaitForMessagePosted(tester.BotUserID(), channel.ID(), recentMessages, func(msg string) (bool, int, string) {
+			err = tester.OnChannel().WaitForMessagePosted(tester.BotUserID(), channel.ID(), recentMessages, func(msg string) (bool, int, string) {
 				if !strings.EqualFold(expectedWatchBeginMsg, msg) {
 					count := diff.CountMatchBlock(expectedWatchBeginMsg, msg)
 					msgDiff := diff.Diff(expectedWatchBeginMsg, msg)
@@ -517,13 +530,13 @@ func TestCloudSlackE2E(t *testing.T) {
 				return true, 0, ""
 			})
 			require.NoError(t, err)
-			tester.PostMessageToBot(t, channel.ID(), "list sources")
+			tester.PostMessageToBot(t, channel.Identifier(), "list sources")
 
 			t.Log("Waiting for empty source list...")
 			expectedSourceListMsg := fmt.Sprintf("%s\n```\nSOURCE ENABLED RESTARTS STATUS LAST_RESTART\n```", cmdHeader("list sources"))
 			err = tester.WaitForLastMessageEqual(tester.BotUserID(), channel.ID(), expectedSourceListMsg)
 			require.NoError(t, err)
-			tester.PostMessageToBot(t, channel.ID(), "list actions")
+			tester.PostMessageToBot(t, channel.Identifier(), "list actions")
 			t.Log("Waiting for actions list...")
 			expectedActionsListMsg := fmt.Sprintf("%s\n```\nACTION       ENABLED  DISPLAY NAME\naction_xxx22 true     Action Name\n```", cmdHeader("list actions"))
 			err = tester.WaitForLastMessageEqual(tester.BotUserID(), channel.ID(), expectedActionsListMsg)

test/commplatform/discord_tester.go

@@ -446,3 +446,8 @@ func (d *DiscordTester) findUserID(t *testing.T, name string) string {
 func (d *DiscordTester) ReplaceBotNamePlaceholder(msg *interactive.CoreMessage, clusterName string) {
 	msg.ReplaceBotNamePlaceholder(d.BotName())
 }
+
+// OnChannel assertion is the default mode for Discord, no action needed.
+func (d *DiscordTester) OnChannel() BotDriver {
+	return d
+}

test/commplatform/generic.go

@@ -51,6 +51,11 @@ type BotDriver interface {
 	Timeout() time.Duration
 	ReplaceBotNamePlaceholder(msg *interactive.CoreMessage, clusterName string)
 	AssertEquals(expectedMessage string) MessageAssertion
+	// OnChannel sets the expectation that the message should be posted in the channel. This is necessary when Bots
+	// by default expect a given message to be posted in the thread of the recently sent message.
+	// For example, in the context of source notification, we need to alter that default behavior
+	// and expect the message on the channel instead.
+	OnChannel() BotDriver
 }
 
 type MessageAssertion func(content string) (bool, int, string)