app.go

// app.go
package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"math"
	"net/http"
	"time"

	"go.uber.org/zap"
	"golang.org/x/time/rate"
)

type SlackMessenger interface {
	PostMessage(req SlackPostMessageRequest, url string, token string) error
}

type SlackClient struct {
	client *http.Client
}

var slackPermanentErrors = map[string]string{
	"as_user_not_supported":                    "The as_user parameter does not function with workspace apps.",
	"channel_not_found":                        "Value passed for channel was invalid.",
	"duplicate_channel_not_found":              "Channel associated with client_msg_id was invalid.",
	"duplicate_message_not_found":              "No duplicate message exists associated with client_msg_id.",
	"ekm_access_denied":                        "Administrators have suspended the ability to post a message.",
	"invalid_blocks":                           "Blocks submitted with this message are not valid",
	"invalid_blocks_format":                    "The blocks is not a valid JSON object or doesn't match the Block Kit syntax.",
	"invalid_metadata_format":                  "Invalid metadata format provided",
	"invalid_metadata_schema":                  "Invalid metadata schema provided",
	"is_archived":                              "Channel has been archived.",
	"messages_tab_disabled":                    "Messages tab for the app is disabled.",
	"metadata_must_be_sent_from_app":           "Message metadata can only be posted or updated using an app token",
	"metadata_too_large":                       "Metadata exceeds size limit",
	"msg_too_long":                             "Message text is too long",
	"no_text":                                  "No message text provided",
	"not_in_channel":                           "Cannot post user messages to a channel they are not in.",
	"restricted_action":                        "A workspace preference prevents the authenticated user from posting.",
	"restricted_action_non_threadable_channel": "Cannot post thread replies into a non_threadable channel.",
	"restricted_action_read_only_channel":      "Cannot post any message into a read-only channel.",
	"restricted_action_thread_locked":          "Cannot post replies to a thread that has been locked by admins.",
	"restricted_action_thread_only_channel":    "Cannot post top-level messages into a thread-only channel.",
	"slack_connect_canvas_sharing_blocked":     "Admin has disabled Canvas File sharing in all Slack Connect communications",
	"slack_connect_file_link_sharing_blocked":  "Admin has disabled Slack File sharing in all Slack Connect communications",
	"team_access_not_granted":                  "The token used is not granted the specific workspace access required to complete this request.",
	"too_many_attachments":                     "Too many attachments were provided with this message. A maximum of 100 attachments are allowed on a message.",
	"too_many_contact_cards":                   "Too many contact_cards were provided with this message. A maximum of 10 contact cards are allowed on a message.",
	"cannot_reply_to_message":                  "This message type cannot have thread replies.",
	"access_denied":                            "Access to a resource specified in the request is denied.",
	"account_inactive":                         "Authentication token is for a deleted user or workspace when using a bot token.",
	"deprecated_endpoint":                      "The endpoint has been deprecated.",
	"enterprise_is_restricted":                 "The method cannot be called from an Enterprise.",
	"invalid_auth":                             "Some aspect of authentication cannot be validated. Either the provided token is invalid or the request originates from an IP address disallowed from making the request.",
	"method_deprecated":                        "The method has been deprecated.",
	"missing_scope":                            "The token used is not granted the specific scope permissions required to complete this request.",
	"not_allowed_token_type":                   "The token type used in this request is not allowed.",
	"not_authed":                               "No authentication token provided.",
	"no_permission":                            "The workspace token used in this request does not have the permissions necessary to complete the request. Make sure your app is a member of the conversation it's attempting to post a message to.",
	"org_login_required":                       "The workspace is undergoing an enterprise migration and will not be available until migration is complete.",
	"token_expired":                            "Authentication token has expired",
	"token_revoked":                            "Authentication token is for a deleted user or workspace or the app has been removed when using a user token.",
	"two_factor_setup_required":                "Two factor setup is required.",
	"accesslimited":                            "Access to this method is limited on the current network",
	"fatal_error":                              "The server could not complete your operation(s) without encountering a catastrophic error. It's possible some aspect of the operation succeeded before the error was raised.",
	"internal_error":                           "The server could not complete your operation(s) without encountering an error, likely due to a transient issue on our end. It's possible some aspect of the operation succeeded before the error was raised.",
	"invalid_arg_name":                         "The method was passed an argument whose name falls outside the bounds of accepted or expected values. This includes very long names and names with non-alphanumeric characters other than _. If you get this error, it is typically an indication that you have made a very malformed API call.",
	"invalid_arguments":                        "The method was either called with invalid arguments or some detail about the arguments passed is invalid, which is more likely when using complex arguments like blocks or attachments.",
	"invalid_array_arg":                        "The method was passed an array as an argument. Please only input valid strings.",
	"invalid_charset":                          "The method was called via a POST request, but the charset specified in the Content-Type header was invalid. Valid charset names are: utf-8 iso-8859-1.",
	"invalid_form_data":                        "The method was called via a POST request with Content-Type application/x-www-form-urlencoded or multipart/form-data, but the form data was either missing or syntactically invalid.",
	"invalid_post_type":                        "The method was called via a POST request, but the specified Content-Type was invalid. Valid types are: application/json application/x-www-form-urlencoded multipart/form-data text/plain.",
	"missing_post_type":                        "The method was called via a POST request and included a data payload, but the request did not include a Content-Type header.",
	"ratelimited":                              "The request has been ratelimited. Refer to the Retry-After header for when to retry the request.",
	"service_unavailable":                      "The service is temporarily unavailable",
	"team_added_to_org":                        "The workspace associated with your request is currently undergoing migration to an Enterprise Organization. Web API and other platform operations will be intermittently unavailable until the transition is complete.",
}

var slackRetryErrors = map[string]string{
	"message_limit_exceeded": "Members on this team are sending too many messages. For more details, see https://slack.com/help/articles/115002422943-Usage-limits-for-free-workspaces",
	"rate_limited":           "Application has posted too many messages, read the Rate Limit documentation for more information",
	"fatal_error":            "The server could not complete your operation(s) without encountering a catastrophic error. It's possible some aspect of the operation succeeded before the error was raised.",
	"internal_error":         "The server could not complete your operation(s) without encountering an error, likely due to a transient issue on our end. It's possible some aspect of the operation succeeded before the error was raised.",
	"ratelimited":            "The request has been ratelimited. Refer to the Retry-After header for when to retry the request.",
	"request_timeout":        "The method was called via a POST request, but the POST data was either missing or truncated.",
}

var doNotProcessChannels = map[string]time.Time{}

func CheckError(err string, channel string) (retryable bool, pause bool, description string) {

	// Special case for channel_not_found, we don't want to retry this one right away.
	// We are making it a 'soft failure' so that we don't keep retrying it for a period of time for any message that is sent to a channel that doesn't exist.
	// We keep track of said channel in a map, and we will retry it after a period of time.
	if err == "channel_not_found" {
		doNotProcessChannels[channel] = time.Now()
		return true, true, "Channel not found"
	}

	description, exists := slackRetryErrors[err]
	if exists {
		return true, false, description
	}

	description, exists = slackPermanentErrors[err]
	if exists {
		return false, false, description
	}

	// This should not happen, but if it does, we just try to retry it
	return true, false, "Unknown error"

}

func (s *SlackClient) PostMessage(request SlackPostMessageRequest, url string, token string) error {
	jsonValue, err := json.Marshal(request)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonValue))
	if err != nil {
		return err
	}

	// Charset is required to remove warnings from Slack. Maybe it's nice to have it configurable. /shrug
	req.Header.Set("Content-Type", "application/json; charset=utf-8")
	// Documentation says that you are allowed the POST the token instead, however that does simply not work. Hence why we are using the Authorization header.
	req.Header.Set("Authorization", "Bearer "+token)

	resp, err := s.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return err
	}

	var slackResp SlackResponse
	if err = json.Unmarshal(body, &slackResp); err != nil {
		return err
	}

	if !slackResp.Ok {
		return fmt.Errorf(slackResp.Error)
	}

	return nil
}

func NewApp(queueSize int, httpClient *http.Client, metrics *Metrics) *App {

	logger, err := zap.NewProduction()
	if err != nil {
		panic("failed to initialize logger: " + err.Error())
	}

	return &App{
		slackQueue: make(chan SlackPostMessageRequest, queueSize),
		messenger:  &SlackClient{client: httpClient},
		logger:     logger,
		metrics:    metrics,
	}
}

func (app *App) Shutdown() {
	close(app.slackQueue)
	// Very important to wait, so that we process all the messages in the queue before exiting!
	app.wg.Wait()
}

func (app *App) processQueue(ctx context.Context, MaxRetries int, InitialBackoffMs int, SlackPostMessageURL string, tokenFlag string, burst int) {
	// This is the rate limiter, which will block until it is allowed to continue on r.Wait(ctx).
	// I kept the rate at 1 per second, as doing more than that will cause Slack to reject the messages anyways. We can burst however.
	// Do note that this is best effort, in case of failures, we will exponentially backoff and retry, which will cause the rate to be lower than 1 per second due to obvious reasons.
	r := rate.NewLimiter(rate.Every(1*time.Second), burst)

	for {
		r.Wait(ctx)

		select {
		case msg, ok := <-app.slackQueue:
			// We do check if the channel is closed, but its important to note is that the channel will be closed when the queue is empty and the Shutdown() is called.
			// Simply calling close(app.slackQueue) will not close the channel, it will only prevent any more messages from being sent to the channel.
			// Only once the channel is empty, will it be closed.
			if !ok {
				return
			}
			app.logger.Debug("Got message from queue", zap.Any("message", msg))

			// Update the queue size metric after any change on the queue size
			app.metrics.QueueSize.With(nil).Set(float64(len(app.slackQueue)))

			retryCount := 0
			for {

				// Check if the channel is in the doNotProcessChannels map, if it is, check if it's been more than 15 minutes since we last tried to send a message to it.
				if (doNotProcessChannels[msg.Channel] != time.Time{}) {
					if time.Since(doNotProcessChannels[msg.Channel]) >= 15*time.Minute {
						// Remove the channel from the map, so that we can process it again. If the channel isn't created in the meantime, we will just add it again.
						delete(doNotProcessChannels, msg.Channel)
					} else {
						app.logger.Info("Channel is on the doNotProcess list, not trying to post this message", zap.String("channel", msg.Channel))
						app.metrics.RequestsNotProcessed.WithLabelValues(msg.Channel).Inc()
						break
					}
				}

				err := app.messenger.PostMessage(msg, SlackPostMessageURL, tokenFlag)
				if err != nil {

					retryable, pause, description := CheckError(err.Error(), msg.Channel)

					if pause {
						app.logger.Info("Channel not found, pausing for 15 minutes", zap.String("channel", msg.Channel))
						app.metrics.RequestsNotProcessed.WithLabelValues(msg.Channel).Inc()
						break
					}

					if !retryable {
						app.metrics.RequestsFailedTotal.WithLabelValues(msg.Channel).Inc()
						app.logger.Error("Permanent error, message will not be retried", zap.Error(err), zap.String("description", description), zap.String("channel", msg.Channel), zap.Any("message", msg))
						break
					}

					if description == "Unknown error" {
						app.logger.Error("Unknown error, since we can't infer what type of error it is, we will retry it. However, please create a ticket/issue for this project for this error", zap.Error(err))
					}
					app.logger.Warn("Temporary error, message will be retried", zap.Error(err), zap.String("description", description), zap.String("channel", msg.Channel), zap.Any("message", msg))

					app.metrics.RequestsRetriedTotal.WithLabelValues(msg.Channel).Inc()

					if retryCount < MaxRetries {
						retryCount++
						backoffDuration := time.Duration(InitialBackoffMs*int(math.Pow(2, float64(retryCount-1)))) * time.Millisecond
						time.Sleep(backoffDuration)
					} else {
						app.logger.Error("Message failed after retries", zap.Error(err), zap.Int("retryCount", retryCount))
						app.metrics.RequestsFailedTotal.WithLabelValues(msg.Channel).Inc()
						break
					}
				} else {
					app.logger.Debug("Message sent successfully")
					app.metrics.RequestsSucceededTotal.WithLabelValues(msg.Channel).Inc()
					break
				}
			}

			// Need to call this to clean up the wg, which is vital for the shutdown to work (so that we process all the messages in the queue before exiting cleanly)
			app.wg.Done()

		case <-ctx.Done():
			return
		}
	}
}