func (sg *Sandwich) SendWebhook(ctx context.Context, _url string,
	message discord.WebhookMessage) (status int, err error) {
	var c *Client
	_url = strings.TrimSpace(_url)
	_, err = url.Parse(_url)
	if err != nil {
		return -1, xerrors.Errorf("failed to parse webhook URL: %w", err)
	}
	if sg.RestTunnelEnabled.IsSet() {
		c = NewClient("", sg.Configuration.RestTunnel.URL, sg.RestTunnelReverse.IsSet(), false)
	} else {
		c = NewClient("", "", false, false)
	}
	res, err := json.Marshal(message)
	if err != nil {
		return -1, xerrors.Errorf("failed to marshal webhook message: %w", err)
	}
	_, status, err = c.Fetch(ctx, "POST", _url, bytes.NewBuffer(res), map[string]string{
		"Content-Type": "application/json",
	})
	return status, err
}
SendWebhook executes a webhook request. This does not currently support sending.
files.
package gateway
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"os"
	"path"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	bucketstore "github.com/TheRockettek/Sandwich-Daemon/pkg/bucketstore"
	consolepump "github.com/TheRockettek/Sandwich-Daemon/pkg/consolepump"
	"github.com/TheRockettek/Sandwich-Daemon/pkg/limiter"
	methodrouter "github.com/TheRockettek/Sandwich-Daemon/pkg/methodrouter"
	"github.com/TheRockettek/Sandwich-Daemon/pkg/snowflake"
	gatewayServer "github.com/TheRockettek/Sandwich-Daemon/protobuf"
	"github.com/TheRockettek/Sandwich-Daemon/structs"
	discord "github.com/TheRockettek/Sandwich-Daemon/structs/discord"
	"github.com/gorilla/sessions"
	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
	"github.com/tevino/abool"
	"github.com/valyala/fasthttp"
	"golang.org/x/oauth2"
	"golang.org/x/xerrors"
	"google.golang.org/grpc"
	"gopkg.in/natefinch/lumberjack.v2"
	"gopkg.in/yaml.v2"
)
const VERSION = "0.9.2+050620212040"
const (
	ConfigurationPath = "sandwich.yaml"
	ErrOnConfigurationFailure = true
	Interval = time.Second * 15
	Samples = 720
	distCacheDuration = 720
	poolConcurrency = 512
	webRootPath = "web/dist"
)
type SandwichConfiguration struct {
	Logging struct {
		Level                 string `json:"level" yaml:"level"`
		ConsoleLoggingEnabled bool   `json:"console_logging" yaml:"console_logging"`
		FileLoggingEnabled    bool   `json:"file_logging" yaml:"file_logging"`
		EncodeAsJSON bool `json:"encode_as_json" yaml:"encode_as_json"`  
		Directory  string `json:"directory" yaml:"directory"`      
		Filename   string `json:"filename" yaml:"filename"`        
		MaxSize    int    `json:"max_size" yaml:"max_size"`        
		MaxBackups int    `json:"max_backups" yaml:"max_backups"`  
		MaxAge     int    `json:"max_age" yaml:"max_age"`          
		MinimalWebhooks bool `json:"minimal_webhooks" yaml:"minimal_webhooks"`
	} `json:"logging" yaml:"logging"`
	RestTunnel struct {
		Enabled bool   `json:"enabled" yaml:"enabled"`
		URL     string `json:"url" yaml:"url"`
	} `json:"resttunnel" yaml:"resttunnel"`
	Producer struct {
		Type          string                 `json:"type" yaml:"type"`
		Configuration map[string]interface{} `json:"configuration" yaml:"configuration"`
	} `json:"producer" yaml:"producer"`
	GRPC struct {
		Network string `json:"network" yaml:"network"`
		Host    string `json:"host" yaml:"host"`
	} `json:"grpc" yaml:"grpc"`
	HTTP struct {
		Host          string `json:"host" yaml:"host"`
		SessionSecret string `json:"secret" yaml:"secret"`
		Enabled       bool   `json:"enabled" yaml:"enabled"`
		Public        bool   `json:"public" yaml:"public"`
	} `json:"http" yaml:"http"`
	Webhooks      []string       `json:"webhooks" yaml:"webhooks"`
	ElevatedUsers []string       `json:"elevated_users" yaml:"elevated_users"`
	OAuth         *oauth2.Config `json:"oauth" yaml:"oauth"`
	Managers []*ManagerConfiguration `json:"managers" yaml:"managers"`
}
type Sandwich struct {
	sync.RWMutex  
	Logger zerolog.Logger `json:"-"`
	Start time.Time `json:"uptime"`
	ConfigurationMu sync.RWMutex           `json:"-"`
	Configuration   *SandwichConfiguration `json:"configuration"`
	RestTunnelReverse abool.AtomicBool `json:"-"`
	RestTunnelEnabled abool.AtomicBool `json:"-"`
	ManagersMu sync.RWMutex        `json:"-"`
	Managers   map[string]*Manager `json:"-"`
	TotalEvents *int64 `json:"-"`
	Buckets *bucketstore.BucketStore `json:"-"`
	ProducerClient *MQClient `json:"-"`
	State *SandwichState `json:"-"`
	Router *methodrouter.MethodRouter `json:"-"`
	Store  *sessions.CookieStore      `json:"-"`
	distHandler fasthttp.RequestHandler
	fs          *fasthttp.FS
	ConsolePump *consolepump.ConsolePump `json:"-"`
	Pool        *limiter.ConcurrencyLimiter `json:"-"`
	PoolWaiting *int64                      `json:"-"`
}
type SandwichState struct {
	GuildsMu sync.RWMutex                         `json:"-"`
	Guilds   map[snowflake.ID]*discord.StateGuild `json:"-"`
	GuildMembersMu sync.RWMutex                                `json:"-"`
	GuildMembers   map[snowflake.ID]*discord.StateGuildMembers `json:"-"`
	ChannelsMu sync.RWMutex                      `json:"-"`
	Channels   map[snowflake.ID]*discord.Channel `json:"-"`
	RolesMu sync.RWMutex                   `json:"-"`
	Roles   map[snowflake.ID]*discord.Role `json:"-"`
	EmojisMu sync.RWMutex                    `json:"-"`
	Emojis   map[snowflake.ID]*discord.Emoji `json:"-"`
	UsersMu sync.RWMutex                   `json:"-"`
	Users   map[snowflake.ID]*discord.User `json:"-"`
}
func NewSandwich(logger io.Writer) (sg *Sandwich, err error) {
	sg = &Sandwich{
		Logger:          zerolog.New(logger).With().Timestamp().Logger(),
		ConfigurationMu: sync.RWMutex{},
		Configuration:   &SandwichConfiguration{},
		ManagersMu:      sync.RWMutex{},
		Managers:        make(map[string]*Manager),
		TotalEvents:     new(int64),
		Buckets:         bucketstore.NewBucketStore(),
		State:           NewSandwichState(),
		Pool:            limiter.NewConcurrencyLimiter("eventPool", poolConcurrency),
		PoolWaiting:     new(int64),
	}
	sg.Lock()
	defer sg.Unlock()
	configuration, err := sg.LoadConfiguration(ConfigurationPath)
	if err != nil {
		return nil, xerrors.Errorf("new sandwich: %w", err)
	}
	sg.ConfigurationMu.Lock()
	defer sg.ConfigurationMu.Unlock()
	sg.Configuration = configuration
	var writers []io.Writer
	zlLevel, err := zerolog.ParseLevel(sg.Configuration.Logging.Level)
	if err != nil {
		sg.Logger.Warn().
			Str("lvl", sg.Configuration.Logging.Level).
			Msg("Current zerolog level provided is not valid")
	} else {
		sg.Logger.Info().
			Str("lvl", sg.Configuration.Logging.Level).
			Msg("Changed logging level")
		zerolog.SetGlobalLevel(zlLevel)
	}
	if sg.Configuration.Logging.ConsoleLoggingEnabled {
		writers = append(writers, logger)
	}
	if sg.Configuration.Logging.FileLoggingEnabled {
		if err := os.MkdirAll(sg.Configuration.Logging.Directory, 0o744); err != nil {
			log.Error().Err(err).Str("path", sg.Configuration.Logging.Directory).Msg("Unable to create log directory")
		} else {
			lumber := &lumberjack.Logger{
				Filename:   path.Join(sg.Configuration.Logging.Directory, sg.Configuration.Logging.Filename),
				MaxBackups: sg.Configuration.Logging.MaxBackups,
				MaxSize:    sg.Configuration.Logging.MaxSize,
				MaxAge:     sg.Configuration.Logging.MaxAge,
			}
			if sg.Configuration.Logging.EncodeAsJSON {
				writers = append(writers, lumber)
			} else {
				writers = append(writers, zerolog.ConsoleWriter{
					Out:        lumber,
					TimeFormat: time.Stamp,
					NoColor:    true,
				})
			}
		}
	}
	if sg.Configuration.HTTP.Enabled {
		sg.ConsolePump = consolepump.NewConsolePump()
		writers = append(writers, sg.ConsolePump)
	}
	mw := io.MultiWriter(writers...)
	sg.Logger = zerolog.New(mw).With().Timestamp().Logger()
	sg.Logger.Info().Msg("Logging configured")
	return sg, nil
}
func (sg *Sandwich) LoadConfiguration(path string) (configuration *SandwichConfiguration, err error) {
	sg.Logger.Debug().Msg("Loading configuration")
	defer func() {
		if err == nil {
			sg.Logger.Info().Msg("Configuration loaded")
		}
	}()
	file, err := ioutil.ReadFile(path)
	if err != nil {
		if ErrOnConfigurationFailure {
			return configuration, xerrors.Errorf("load configuration readfile: %w", err)
		}
		sg.Logger.Warn().Msg("Failed to read configuration but ErrOnConfigurationFailure is disabled")
	}
	configuration = &SandwichConfiguration{}
	err = yaml.Unmarshal(file, &configuration)
	if err != nil {
		if ErrOnConfigurationFailure {
			return configuration, xerrors.Errorf("load configuration unmarshal: %w", err)
		}
		sg.Logger.Warn().Msg("Failed to unmarshal configuration but ErrOnConfigurationFailure is disabled")
	}
	err = sg.NormalizeConfiguration(configuration)
	if err != nil {
		if ErrOnConfigurationFailure {
			return configuration, xerrors.Errorf("load configuration normalize: %w", err)
		}
		sg.Logger.Warn().Msg("Failed to normalize configuration but ErrOnConfigurationFailure is disabled")
	}
	return configuration, err
}
func (sg *Sandwich) SaveConfiguration(configuration *SandwichConfiguration, path string) (err error) {
	sg.Logger.Debug().Msg("Saving configuration")
	defer func() {
		if err == nil {
			sg.Logger.Info().Msg("Flushed configuration to disk")
		}
	}()
	var config *SandwichConfiguration
	oldmanagers := make(map[string]*ManagerConfiguration)
	for _, manager := range configuration.Managers {
		if !manager.Persist {
			config, err = sg.LoadConfiguration(path)
			if err != nil {
				return err
			}
			for _, mg := range config.Managers {
				oldmanagers[mg.Identifier] = mg
			}
		}
	}
	storedManagers := []*ManagerConfiguration{}
	for _, manager := range configuration.Managers {
		if !manager.Persist {
			oldmanager, ok := oldmanagers[manager.Identifier]
			if ok {
				manager = oldmanager
			} else {
				continue
			}
		}
		storedManagers = append(storedManagers, manager)
	}
	configuration.Managers = storedManagers
	data, err := yaml.Marshal(configuration)
	if err != nil {
		return xerrors.Errorf("save configuration marshal: %w", err)
	}
	err = ioutil.WriteFile(path, data, 0o600)
	if err != nil {
		return xerrors.Errorf("save configuration write: %w", err)
	}
	return nil
}
func (sg *Sandwich) NormalizeConfiguration(configuration *SandwichConfiguration) (err error) {
	if configuration.HTTP.Host == "" {
		return xerrors.Errorf("Configuration missing HTTP host. Try 127.0.0.1:5469")
	}
	if configuration.GRPC.Host == "" {
		return xerrors.Errorf("Configuration missing GRPC host. Try 127.0.0.1:10000")
	}
	return nil
}
func (sg *Sandwich) Open() (err error) {
	sg.ConfigurationMu.RLock()
	defer sg.ConfigurationMu.RUnlock()
	sg.Start = time.Now().UTC()
	sg.Logger.Info().Msgf("Starting sandwich\n\n         _-**--__\n"+
		"     _--*         *--__         Sandwich Daemon %s\n"+
		" _-**                  **-_\n"+
		"|_*--_                _-* _|    HTTP: %s\n"+
		"| *-_ *---_     _----* _-* |    Managers: %d\n"+
		" *-_ *--__ *****  __---* _*\n"+
		"     *--__ *-----** ___--*      %s\n"+
		"         **-____-**\n",
		VERSION, sg.Configuration.HTTP.Host, len(sg.Configuration.Managers), "┬─┬ ノ( ゜-゜ノ)")
	if sg.Configuration.HTTP.Enabled {
		if sg.Configuration.HTTP.Public {
			sg.Logger.Warn().Msg(
				"Public mode is enabled on the HTTP API. This can allow anyone to" +
					"get bot credentials if exposed publicly. It is recommended you disable" +
					"this and add trusted user ids in sandwich.yaml under \"elevated_users\"")
		}
		sg.Logger.Info().Msg("Starting up http server")
		sg.fs = &fasthttp.FS{
			Root:            webRootPath,
			Compress:        true,
			CompressBrotli:  true,
			AcceptByteRange: true,
			CacheDuration:   time.Hour * distCacheDuration,
			PathNotFound:    fasthttp.RequestHandler(func(ctx *fasthttp.RequestCtx) {}),
		}
		sg.distHandler = sg.fs.NewRequestHandler()
		sg.Store = sessions.NewCookieStore([]byte(sg.Configuration.HTTP.SessionSecret))
		sg.Logger.Info().Msg("Creating endpoints")
		sg.Router = createEndpoints(sg)
		go func() {
			sg.Logger.Info().Msgf("Serving dashboard on %s (Press CTRL+C to quit)\n", sg.Configuration.HTTP.Host)
			err = fasthttp.ListenAndServe(sg.Configuration.HTTP.Host, sg.HandleRequest)
			if err != nil {
				sg.Logger.Error().Str("host", sg.Configuration.HTTP.Host).Err(err).Msg("Failed to serve http server")
			}
		}()
	} else {
		sg.Logger.Info().Msg("The web interface will not start as HTTP is disabled in the configuration")
	}
	go func() {
		lis, err := net.Listen(sg.Configuration.GRPC.Network, sg.Configuration.GRPC.Host)
		if err != nil {
			sg.Logger.Error().Str("host", sg.Configuration.GRPC.Host).Err(err).Msg("Failed to bind address for gRPC server")
			return
		}
		var opts []grpc.ServerOption
		grpcServer := grpc.NewServer(opts...)
		gatewayServer.RegisterGatewayServer(grpcServer, sg.NewGatewayServer())
		sg.Logger.Info().Msgf("Serving gRPC on %s (Press CTRL+C to quit)\n", sg.Configuration.GRPC.Host)
		err = grpcServer.Serve(lis)
		if err != nil {
			sg.Logger.Error().Str("host", sg.Configuration.GRPC.Host).Err(err).Msg("Failed to serve gRPC server")
		}
	}()
	sg.Logger.Info().Msg("Configuring RestTunnel")
	if sg.Configuration.RestTunnel.Enabled {
		enabled, reverse, err := sg.VerifyRestTunnel(sg.Configuration.RestTunnel.URL)
		if err != nil {
			sg.Logger.Error().Err(err).Msg("Failed to verify RestTunnel")
		}
		sg.RestTunnelReverse.SetTo(reverse)
		sg.RestTunnelEnabled.SetTo(enabled)
	} else {
		sg.RestTunnelEnabled.UnSet()
	}
	go sg.PublishWebhook(context.Background(), discord.WebhookMessage{
		Embeds: []discord.Embed{
			{
				Title:       "Starting up Sandwich-Daemon",
				URL:         "https://github.com/TheRockettek/Sandwich-Daemon",
				Description: fmt.Sprintf("Version %s", VERSION),
				Color:       discord.EmbedSandwich,
				Timestamp:   WebhookTime(time.Now().UTC()),
			},
		},
	})
	sg.Logger.Info().Msg("Creating managers")
	sg.startManagers()
	go sg.gatherAnalytics()
	go sg.analyticsRunner()
	return nil
}
func (sg *Sandwich) startManagers() {
	sg.ManagersMu.Lock()
	for _, managerConfiguration := range sg.Configuration.Managers {
		manager, err := sg.NewManager(managerConfiguration)
		if err != nil {
			sg.Logger.Error().Err(err).Msg("Could not create manager")
			continue
		}
		if _, ok := sg.Managers[managerConfiguration.Identifier]; ok {
			sg.Logger.Warn().
				Str("identifier", managerConfiguration.Identifier).
				Msg("Found conflicting manager identifiers. Ignoring!")
			go sg.PublishWebhook(context.Background(), discord.WebhookMessage{
				Embeds: []discord.Embed{
					{
						Title:     "Found conflicting manager identifiers. Ignoring!",
						Color:     discord.EmbedWarning,
						Timestamp: WebhookTime(time.Now().UTC()),
						Footer: &discord.EmbedFooter{
							Text: fmt.Sprintf("Manager %s",
								manager.Configuration.DisplayName),
						},
					},
				},
			})
			continue
		}
		sg.Managers[managerConfiguration.Identifier] = manager
		err = manager.Open()
		if err != nil {
			manager.ErrorMu.Lock()
			manager.Error = err.Error()
			manager.ErrorMu.Unlock()
			sg.Logger.Error().Err(err).Msg("Failed to start up manager")
		} else if manager.Configuration.AutoStart {
			go func() {
				manager.Gateway, err = manager.GetGateway()
				if err != nil {
					manager.Logger.Error().Err(err).Msg("Failed to start up manager")
					return
				}
				manager.GatewayMu.RLock()
				manager.Logger.Info().
					Int("sessions", manager.Gateway.SessionStartLimit.Remaining).
					Msg("Retrieved gateway information")
				shardCount := manager.GatherShardCount()
				if shardCount >= manager.Gateway.SessionStartLimit.Remaining {
					manager.Logger.Error().Err(ErrSessionLimitExhausted).Msg("Failed to start up manager")
					manager.GatewayMu.RUnlock()
					go sg.PublishWebhook(context.Background(), discord.WebhookMessage{
						Embeds: []discord.Embed{
							{
								Title: fmt.Sprintf("Failed to start up manager as not enough sessions to start %d shard(s). %d remain",
									shardCount, manager.Gateway.SessionStartLimit.Remaining),
								Color:     discord.EmbedDanger,
								Timestamp: WebhookTime(time.Now().UTC()),
								Footer: &discord.EmbedFooter{
									Text: fmt.Sprintf("Manager %s",
										manager.Configuration.DisplayName),
								},
							},
						},
					})
					return
				}
				manager.GatewayMu.RUnlock()
				ready, err := manager.Scale(manager.GenerateShardIDs(shardCount), shardCount, true)
				if err != nil {
					manager.Logger.Error().Err(err).Msg("Failed to start up manager")
					go sg.PublishWebhook(context.Background(), discord.WebhookMessage{
						Embeds: []discord.Embed{
							{
								Title:       "Failed to start up manager",
								Description: err.Error(),
								Color:       discord.EmbedDanger,
								Timestamp:   WebhookTime(time.Now().UTC()),
								Footer: &discord.EmbedFooter{
									Text: fmt.Sprintf("Manager %s",
										manager.Configuration.DisplayName),
								},
							},
						},
					})
					return
				}
				<-ready
			}()
		}
	}
	sg.ManagersMu.Unlock()
}
func (sg *Sandwich) gatherAnalytics() {
	var events int64
	var managerEvents int64
	t := time.NewTicker(time.Second * 1)
	for {
		<-t.C
		events = 0
		sg.ManagersMu.RLock()
		for _, mg := range sg.Managers {
			managerEvents = 0
			mg.ShardGroupsMu.RLock()
			for _, sg := range mg.ShardGroups {
				sg.ShardsMu.RLock()
				for _, sh := range sg.Shards {
					managerEvents += atomic.SwapInt64(sh.events, 0)
				}
				sg.ShardsMu.RUnlock()
			}
			mg.ShardGroupsMu.RUnlock()
			mg.AnalyticsMu.RLock()
			if mg.Analytics != nil {
				mg.Analytics.IncrementBy(managerEvents)
			}
			mg.AnalyticsMu.RUnlock()
			events += managerEvents
		}
		sg.ManagersMu.RUnlock()
		atomic.AddInt64(sg.TotalEvents, events)
		now := time.Now().UTC()
		uptime := now.Sub(sg.Start).Round(time.Second)
		sg.Logger.Debug().Str("Elapsed", uptime.String()).Int64("q", atomic.LoadInt64(sg.PoolWaiting)).Msgf("%d/s", events)
	}
}
func (sg *Sandwich) analyticsRunner() {
	e := time.NewTicker(Interval)
	for {
		<-e.C
		now := time.Now().UTC().Round(time.Second)
		sg.ManagersMu.RLock()
		for _, mg := range sg.Managers {
			mg.AnalyticsMu.RLock()
			if mg.Analytics != nil {
				go mg.Analytics.RunOnce(now)
			}
			mg.AnalyticsMu.RUnlock()
		}
		sg.ManagersMu.RUnlock()
	}
}
func (sg *Sandwich) VerifyRestTunnel(restTunnelURL string) (enabled bool, reverse bool, err error) {
	if restTunnelURL == "" {
		return
	}
	_url, err := url.Parse(restTunnelURL)
	if err != nil {
		sg.Logger.Error().Err(err).Msgf("Failed to parse RestTunnel URL %s", restTunnelURL)
		return
	}
	resp, err := http.Get(_url.Scheme + "://" + _url.Host + "/resttunnel")  
	if err != nil {
		sg.Logger.Error().Err(err).Msgf("Failed to connect to RestTunnel")
		return
	}
	baseResponse := structs.RestTunnelAliveResponse{}
	defer resp.Body.Close()
	err = json.NewDecoder(resp.Body).Decode(&baseResponse)
	if err != nil {
		sg.Logger.Error().Err(err).Msg("Failed to parse RestTunnel alive response")
	}
	aliveResponse := baseResponse.Data
	sg.Logger.Info().Msgf(
		"\\o/ Detected RestTunnel version %s. Running in reverse mode: %t",
		aliveResponse.Version, aliveResponse.Reverse)
	return true, aliveResponse.Reverse, nil
}
func (sg *Sandwich) Close() (err error) {
	sg.Logger.Info().Msg("Closing sandwich")
	go sg.PublishWebhook(context.Background(), discord.WebhookMessage{
		Embeds: []discord.Embed{
			{
				Title:     "Shutting down sandwich",
				Color:     discord.EmbedSandwich,
				Timestamp: WebhookTime(time.Now().UTC()),
			},
		},
	})
	sg.ManagersMu.RLock()
	for _, manager := range sg.Managers {
		manager.Close()
	}
	sg.ManagersMu.RUnlock()
	return
}
func (sg *Sandwich) PublishWebhook(ctx context.Context, message discord.WebhookMessage) {
	for _, webhook := range sg.Configuration.Webhooks {
		_, err := sg.SendWebhook(ctx, webhook, message)
		if err != nil && !xerrors.Is(err, context.Canceled) {
			sg.Logger.Warn().Err(err).Str("url", webhook).Msg("Failed to send webhook")
		}
	}
}