feat: implement threaded message strategy for Slack notifier

Signed-off-by: Santiago Fernández Núñez <santiago.nunez@nubank.com.br>

Author: santiagofn

notify/slack/internal/apiurl/apiurl.go

@@ -39,6 +39,9 @@ func NewResolver(apiURL *amcommoncfg.SecretURL, apiURLFile string) *Resolver {
 	return &Resolver{apiURL: apiURL, apiURLFile: apiURLFile}
 }
 
+// URLForMethod returns the Slack Web API URL for the given method name
+// (e.g. "chat.update", "reactions.add"). An empty method returns the base
+// api_url as-is, which is suitable for the default chat.postMessage call.
 func (r *Resolver) URLForMethod(method string) (string, error) {
 	if method == "" {
 		if r.apiURL != nil {
@@ -74,6 +77,7 @@ func (r *Resolver) URLForMethod(method string) (string, error) {
 	return webAPIMethodURL(baseURL, method)
 }
 
+// getURLFromFile reads and parses the URL from the api_url_file path on disk.
 func (r *Resolver) getURLFromFile() (*amcommoncfg.URL, error) {
 	content, err := os.ReadFile(r.apiURLFile)
 	if err != nil {

notify/slack/slack.go

@@ -36,7 +36,14 @@ import (
 // https://api.slack.com/reference/messaging/attachments#legacy_fields - 1024, no units given, assuming runes or characters.
 const maxTitleLenRunes = 1024
 
-// New returns a new Slack notification handler.
+// nflog store keys for persisting Slack-specific state across notifications.
+const (
+	storeKeyThreadTs    = "threadTs"
+	storeKeyChannelId   = "channelId"
+	storeKeyTransitions = "transitions"
+)
+
+// New builds a Slack Notifier with tracing enabled on the configured HTTP client.
 func New(c *config.SlackConfig, t *template.Template, l *slog.Logger, httpOpts ...commoncfg.HTTPClientOption) (*Notifier, error) {
 	client, err := notify.NewClientWithTracing(*c.HTTPConfig, "slack", httpOpts...)
 	if err != nil {
@@ -54,7 +61,10 @@ func New(c *config.SlackConfig, t *template.Template, l *slog.Logger, httpOpts .
 	}, nil
 }
 
-// Notify implements the Notifier interface.
+// Notify implements the Notifier interface. It expands templates, builds the Slack
+// payload, and sends it (or updates an existing message / thread) based on
+// message_strategy and nflog state. The returned bool is true when the delivery
+// should be retried (e.g. transport or retryable HTTP errors).
 func (n *Notifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error) {
 	var err error
 	key, err := notify.ExtractGroupKey(ctx)
@@ -143,6 +153,21 @@ func (n *Notifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error)
 		att.Actions = actions
 	}
 
+	req := &request{
+		Channel:     tmplText(n.conf.Channel),
+		Username:    tmplText(n.conf.Username),
+		IconEmoji:   tmplText(n.conf.IconEmoji),
+		IconURL:     tmplText(n.conf.IconURL),
+		LinkNames:   n.conf.LinkNames,
+		Text:        tmplText(n.conf.MessageText),
+		Attachments: []attachment{*att},
+	}
+	// tmplText is notify.TmplText(..., &err): every field execution appends template errors
+	// into the same err. Check here so we never call Slack after a failed template render.
+	if err != nil {
+		return false, err
+	}
+
 	u, err := n.urlResolver.URLForMethod("")
 	if err != nil {
 		return false, err
@@ -154,44 +179,57 @@ func (n *Notifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error)
 		ctx = postCtx
 	}
 
-	req := &request{
-		Channel:     tmplText(n.conf.Channel),
-		Username:    tmplText(n.conf.Username),
-		IconEmoji:   tmplText(n.conf.IconEmoji),
-		IconURL:     tmplText(n.conf.IconURL),
-		LinkNames:   n.conf.LinkNames,
-		Text:        tmplText(n.conf.MessageText),
-		Attachments: []attachment{*att},
-	}
-
-	// If a notification for this alert group has already been sent and `update_message` config is set
-	// edit API endpoint and payload to update notification instead of sending a new one.
 	var store *nflog.Store
 
-	if n.conf.UpdateMessage {
+	if n.conf.HasStrategyThatUpdatesParent() {
 		var ok bool
 		store, ok = notify.NflogStore(ctx)
 		if !ok {
-			logger.Warn("cannot create NflogStore, updatable messages will be disabled.")
+			logger.Warn("cannot create NflogStore, updatable/threaded messages will be disabled.")
+		} else if store == nil {
+			logger.Warn("NflogStore is nil, updatable/threaded messages will be disabled.")
 		} else {
-			threadTs, _ := store.GetStr("threadTs")
-			channelId, _ := store.GetStr("channelId")
-			logger.Debug("attempt recovering threadTs and channelId to update an existing message", "threadTs", threadTs, "channelId", channelId)
-			if threadTs != "" && channelId != "" {
-				updateURL, err := n.urlResolver.URLForMethod("chat.update")
-				if err != nil {
-					return false, err
+
+			// If message_strategy is "update", edit the API endpoint and payload to update
+			// the existing notification instead of sending a new one.
+			if n.conf.HasUpdateStrategy() {
+				threadTs, _ := store.GetStr(storeKeyThreadTs)
+				channelId, _ := store.GetStr(storeKeyChannelId)
+				logger.Debug("attempt recovering threadTs and channelId to update an existing message", storeKeyThreadTs, threadTs, storeKeyChannelId, channelId)
+				if threadTs != "" && channelId != "" {
+					updateURL, err := n.urlResolver.URLForMethod("chat.update")
+					if err != nil {
+						return false, err
+					}
+					u = updateURL
+					req.Timestamp = threadTs
+					req.Channel = channelId
+					logger.Debug("updating previously sent message", storeKeyThreadTs, threadTs, storeKeyChannelId, channelId)
+				}
+			} else if n.conf.HasThreadStrategy() {
+				// If message_strategy is "thread", there are two modes controlled by the flag use_summary_header.
+				if n.conf.UseSummaryHeaderInThread() {
+					return n.handleThreadedSummaryHeaderMode(ctx, data, tmplText, &err, store, u, req, logger)
 				}
-				u = updateURL
-				req.Timestamp = threadTs
-				req.Channel = channelId
-				logger.Debug("updating previously sent message", "threadTs", threadTs, "channelId", channelId)
+				return n.handleThreadedDirectMode(ctx, store, req, u, logger)
 			}
 		}
 	}
+
+	// Default path: post the message directly (for "new" and "update" strategies, or when thread strategy falls
+	// through due to missing nflog store).
+	return n.postAndHandle(ctx, u, req.Channel, req, store, slackResponseOpts{})
+}
+
+// postAndHandle JSON-encodes payload, POSTs it to u, applies HTTP retry classification,
+// then parses the Slack body. channel is only used in error messages. When store is
+// non-nil and the response is successful JSON with ts/channel, persistResponseState may
+// persist nflog keys for update/thread strategies. opts.IgnoreAPIErrors lists Slack
+// JSON error codes treated as success (e.g. already_reacted for reactions.add).
+func (n *Notifier) postAndHandle(ctx context.Context, u, channel string, payload any, store *nflog.Store, opts slackResponseOpts) (bool, error) {
 	var buf bytes.Buffer
-	if err := json.NewEncoder(&buf).Encode(req); err != nil {
-		return false, err
+	if err := json.NewEncoder(&buf).Encode(payload); err != nil {
+		return false, fmt.Errorf("encode slack request: %w", err)
 	}
 
 	resp, err := n.postJSONFunc(ctx, n.client, u, &buf)
@@ -207,51 +245,77 @@ func (n *Notifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error)
 	// classify them as retriable or not.
 	retry, err := n.retrier.Check(resp.StatusCode, resp.Body)
 	if err != nil {
-		err = fmt.Errorf("channel %q: %w", req.Channel, err)
+		err = fmt.Errorf("channel %q: %w", channel, err)
 		return retry, notify.NewErrorWithReason(notify.GetFailureReasonFromStatusCode(resp.StatusCode), err)
 	}
 
-	retry, err = n.slackResponseHandler(resp, store)
+	data, retry, err := readAndParseSlackResponse(resp, opts)
 	if err != nil {
-		err = fmt.Errorf("channel %q: %w", req.Channel, err)
+		err = fmt.Errorf("channel %q: %w", channel, err)
 		return retry, notify.NewErrorWithReason(notify.ClientErrorReason, err)
 	}
-	return retry, nil
+
+	n.persistResponseState(store, data)
+
+	return false, nil
 }
 
-// slackResponseHandler parses the response body of the request, handles retryable errors
-// and saves the response timestamp and channelId to nflog.
-func (n *Notifier) slackResponseHandler(resp *http.Response, store *nflog.Store) (bool, error) {
-	body, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return true, fmt.Errorf("could not read response body: %w", err)
-	}
-	if !strings.HasPrefix(resp.Header.Get("Content-Type"), "application/json") {
-		return checkTextResponseError(body)
-	}
-	var data slackResponse
-	if err := json.Unmarshal(body, &data); err != nil {
-		return true, fmt.Errorf("could not unmarshal JSON response %q: %w", string(body), err)
+// persistResponseState persists the threadTs and channelId of a message in nflog. For message_strategy "thread", only
+// the first message is saved, so later replies do not replace the thread root.
+func (n *Notifier) persistResponseState(store *nflog.Store, data slackResponse) {
+	if store == nil || data.Timestamp == "" || data.Channel == "" {
+		return
 	}
-	if !data.OK {
-		return false, fmt.Errorf("error response from Slack: %s", data.Error)
-	}
-	// If store, TS and Channel are set, store the threadTS and channelId
-	if store != nil && data.Timestamp != "" && data.Channel != "" {
-		store.SetStr("threadTs", data.Timestamp)
-		store.SetStr("channelId", data.Channel)
-		n.logger.Debug("stored threadTs and channelId", "threadTs", data.Timestamp, "channelId", data.Channel)
+	if n.conf.HasThreadStrategy() {
+		parentThreadTs, parentChannelId, parentFound := getStoredParent(store)
+		if !parentFound {
+			store.SetStr(storeKeyThreadTs, data.Timestamp)
+			store.SetStr(storeKeyChannelId, data.Channel)
+			n.logger.Debug("stored threadTs and channelId for thread parent", storeKeyThreadTs, data.Timestamp, storeKeyChannelId, data.Channel)
+		} else {
+			n.logger.Debug("skipping storing reply as thread parent is already stored", storeKeyThreadTs, parentThreadTs, storeKeyChannelId, parentChannelId)
+		}
+	} else {
+		store.SetStr(storeKeyThreadTs, data.Timestamp)
+		store.SetStr(storeKeyChannelId, data.Channel)
+		n.logger.Debug("stored threadTs and channelId", storeKeyThreadTs, data.Timestamp, storeKeyChannelId, data.Channel)
 	}
-	return false, nil
 }
 
-// checkTextResponseError classifies plaintext responses from Slack.
-// A plaintext (non-JSON) response is successful if it's a string "ok".
-// This is typically a response for an Incoming Webhook
-// (https://api.slack.com/messaging/webhooks#handling_errors)
+// checkTextResponseError classifies incoming-webhook plaintext responses.
+// Success requires body exactly "ok". The bool is the retry hint (always false here).
+// See https://api.slack.com/messaging/webhooks#handling_errors
 func checkTextResponseError(body []byte) (bool, error) {
-	if !bytes.Equal(body, []byte("ok")) {
+	if !bytes.Equal(bytes.TrimSpace(body), []byte("ok")) {
 		return false, fmt.Errorf("received an error response from Slack: %s", string(body))
 	}
 	return false, nil
 }
+
+// readAndParseSlackResponse reads the response body. For Content-Type application/json
+// it unmarshals slackResponse; ok=false is an error unless data.Error is listed in
+// opts.IgnoreAPIErrors. Non-JSON bodies use incoming-webhook plaintext rules (body "ok").
+// retry is true for read/unmarshal failures that may be transient; false for definitive
+// Slack API errors (ok=false without ignore) or successful plaintext.
+func readAndParseSlackResponse(resp *http.Response, opts slackResponseOpts) (slackResponse, bool, error) {
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return slackResponse{}, true, fmt.Errorf("could not read response body: %w", err)
+	}
+	contentType := strings.TrimSpace(strings.ToLower(resp.Header.Get("Content-Type")))
+	if !strings.HasPrefix(contentType, "application/json") {
+		retry, err := checkTextResponseError(body)
+		return slackResponse{}, retry, err
+	}
+	var data slackResponse
+	if err = json.Unmarshal(body, &data); err != nil {
+		return slackResponse{}, true, fmt.Errorf("could not unmarshal JSON response %q: %w", string(body), err)
+	}
+	if !data.OK {
+		if opts.treatsSlackErrorAsSuccess(data.Error) {
+			return data, false, nil
+		}
+		return slackResponse{}, false, fmt.Errorf("error response from Slack: %s", data.Error)
+	}
+	return data, false, nil
+}