func (sh *Shard) Listen() (err error) {
	wsConn := sh.wsConn
	for {
		select {
		case <-sh.ctx.Done():
			return
		default:
		}
		msg, err := sh.readMessage()
		if err != nil {
			if xerrors.Is(err, context.Canceled) || xerrors.Is(err, context.DeadlineExceeded) {
				break
			}
			sh.Logger.Error().Err(err).Msg("Error reading from gateway")
			var closeError *websocket.CloseError
			if errors.As(err, &closeError) {
				switch closeError.Code {
				case discord.CloseNotAuthenticated,  
					discord.CloseInvalidShard,       
					discord.CloseShardingRequired,   
					discord.CloseInvalidAPIVersion,  
					discord.CloseInvalidIntents,     
					discord.CloseDisallowedIntents:  
					sh.Logger.Warn().Msgf(
						"Closing ShardGroup as cannot continue without valid token. Received code %d",
						closeError.Code,
					)
					go sh.PublishWebhook("ShardGroup is closing due to invalid token being passed", "", 16760839, false)
					sh.ShardGroup.ErrorMu.Lock()
					sh.ShardGroup.Error = err.Error()
					sh.ShardGroup.ErrorMu.Unlock()
					sh.ShardGroup.Close()
					if err := sh.ShardGroup.SetStatus(structs.ShardGroupError); err != nil {
						sh.ShardGroup.Logger.Error().Err(err).Msg("Encountered error setting shard group status")
					}
					return err
				default:
					sh.Logger.Warn().Msgf("Websocket was closed with code %d", closeError.Code)
				}
			}
			if wsConn == sh.wsConn {
				sh.Logger.Warn().Msg("We have encountered an error whilst in the same connection, reconnecting...")
				err = sh.Reconnect(websocket.StatusNormalClosure)
				if err != nil {
					return err
				}
				return nil
			}
			wsConn = sh.wsConn
		}
		sh.OnEvent(msg)
		if sh.wsConn != wsConn {
			sh.Logger.Debug().Msg("New wsConn was assigned to shard")
			wsConn = sh.wsConn
		}
	}
	return err
}
Listen to gateway and process accordingly.
package gateway
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/TheRockettek/Sandwich-Daemon/pkg/snowflake"
	"github.com/TheRockettek/Sandwich-Daemon/structs"
	discord "github.com/TheRockettek/Sandwich-Daemon/structs/discord"
	"github.com/TheRockettek/czlib"
	"github.com/andybalholm/brotli"
	"github.com/rs/zerolog"
	"github.com/savsgio/gotils"
	"github.com/tevino/abool"
	"golang.org/x/xerrors"
	"nhooyr.io/websocket"
)
const (
	timeoutDuration     = 2 * time.Second
	dispatchTimeout     = 30 * time.Second
	waitForReadyTimeout = 10 * time.Second
	identifyRatelimit   = (5 * time.Second) + (500 * time.Millisecond)
	websocketReadLimit    = 512 << 20
	reconnectCloseCode    = 4000
	maxReconnectWait      = 600
	gatewayConnectTimeout = 5
	messageChannelBuffer      = 64
	minPayloadCompressionSize = 1000000  
	initialMemberChunkTimeout = 10 * time.Second
	memberChunkTimeout = 1 * time.Second
	chunkStatePersistTimeout = 10 * time.Second
)
type Shard struct {
	sync.RWMutex  
	Status   structs.ShardStatus `json:"status"`
	StatusMu sync.RWMutex        `json:"-"`
	Logger zerolog.Logger `json:"-"`
	ShardID    int         `json:"shard_id"`
	ShardGroup *ShardGroup `json:"-"`
	Manager    *Manager    `json:"-"`
	User *discord.User `json:"user"`
	ctx    context.Context
	cancel func()
	HeartbeatActive   *abool.AtomicBool `json:"-"`
	LastHeartbeatMu   sync.RWMutex      `json:"-"`
	LastHeartbeatAck  time.Time         `json:"last_heartbeat_ack"`
	LastHeartbeatSent time.Time         `json:"last_heartbeat_sent"`
	Heartbeater          *time.Ticker  `json:"-"`
	HeartbeatInterval    time.Duration `json:"heartbeat_interval"`
	MaxHeartbeatFailures time.Duration `json:"max_heartbeat_failures"`
	UnavailableMu sync.RWMutex          `json:"-"`
	Unavailable   map[snowflake.ID]bool `json:"-"`
	Start   time.Time `json:"start"`
	Retries *int32    `json:"retries"`  
	FastCompressor    sync.Pool
	DefaultCompressor sync.Pool
	wsConn *websocket.Conn
	mp sync.Pool
	rp sync.Pool
	pp sync.Pool
	cp sync.Pool
	MessageCh chan discord.ReceivedPayload
	ErrorCh   chan error
	events *int64
	seq       *int64
	sessionID string
	ready chan void
	errs chan error
}
func (sg *ShardGroup) NewShard(shardID int) *Shard {
	logger := sg.Logger.With().Int("shard", shardID).Logger()
	sh := &Shard{
		Status:   structs.ShardIdle,
		StatusMu: sync.RWMutex{},
		Logger: logger,
		ShardID:    shardID,
		ShardGroup: sg,
		Manager:    sg.Manager,
		HeartbeatActive:   abool.New(),
		LastHeartbeatMu:   sync.RWMutex{},
		LastHeartbeatAck:  time.Now().UTC(),
		LastHeartbeatSent: time.Now().UTC(),
		UnavailableMu: sync.RWMutex{},
		Start:   time.Now().UTC(),
		Retries: new(int32),
		FastCompressor: sync.Pool{
			New: func() interface{} { return brotli.NewWriterLevel(nil, brotli.BestSpeed) },
		},
		DefaultCompressor: sync.Pool{
			New: func() interface{} { return brotli.NewWriterLevel(nil, brotli.DefaultCompression) },
		},
		mp: sync.Pool{
			New: func() interface{} { return new(discord.ReceivedPayload) },
		},
		rp: sync.Pool{
			New: func() interface{} { return new(discord.SentPayload) },
		},
		pp: sync.Pool{
			New: func() interface{} { return new(structs.SandwichPayload) },
		},
		cp: sync.Pool{
			New: func() interface{} { return new(bytes.Buffer) },
		},
		events: new(int64),
		seq:       new(int64),
		sessionID: "",
		ready: make(chan void, 1),
		errs: make(chan error),
	}
	if sh.ctx == nil || sh.cancel == nil {
		sh.ctx, sh.cancel = context.WithCancel(context.Background())
	}
	atomic.StoreInt32(sh.Retries, sg.Manager.Configuration.Bot.Retries)
	return sh
}
func (sh *Shard) Open() {
	sh.Logger.Debug().Msg("Opening Shard")
	for {
		sh.Logger.Debug().Msg("Started listening to shard")
		err := sh.Listen()
		if xerrors.Is(err, context.Canceled) {
			sh.Logger.Debug().Msg("Shard received context Canceled")
			return
		}
		select {
		case <-sh.ctx.Done():
			sh.Logger.Debug().Msg("Shard context has finished")
			return
		default:
		}
	}
}
func (sh *Shard) Connect() (err error) {
	sh.Logger.Debug().Msg("Starting shard")
	if err := sh.SetStatus(structs.ShardWaiting); err != nil {
		sh.Logger.Error().Err(err).Msg("Encountered error setting shard status")
	}
	sh.Manager.GatewayMu.RLock()
	concurrencyBucket := sh.ShardID % sh.Manager.Gateway.SessionStartLimit.MaxConcurrency
	sh.Logger.Trace().Msgf("Using concurrency bucket %d", concurrencyBucket)
	sh.Manager.GatewayMu.RUnlock()
	select {
	case <-sh.ctx.Done():
		sh.Logger.Trace().Msg("Creating new context")
		sh.ctx, sh.cancel = context.WithCancel(context.Background())
	default:
		sh.Logger.Trace().Msg("No need for new context")
	}
	sh.Logger.Trace().Msg("Creating buckets")
	sh.Manager.Buckets.CreateBucket(fmt.Sprintf("ws:%d:%d", sh.ShardID, sh.ShardGroup.ShardCount), 120, time.Minute)
	hash, err := QuickHash(sh.Manager.Configuration.Token)
	if err != nil {
		sh.Logger.Error().Err(err).Msg("Failed to generate token hash")
		return err
	}
	sh.Manager.Sandwich.Buckets.CreateBucket(fmt.Sprintf("gw:%s:%d", hash, concurrencyBucket), 1, identifyRatelimit)
	if sh.ready == nil {
		sh.ready = make(chan void, 1)
	}
	sh.Manager.GatewayMu.RLock()
	gatewayURL := sh.Manager.Gateway.URL
	sh.Manager.GatewayMu.RUnlock()
	defer func() {
		if err != nil && sh.wsConn != nil {
			if _err := sh.CloseWS(websocket.StatusNormalClosure); _err != nil {
				sh.Logger.Error().Err(_err).Msg("Failed to close websocket")
			}
		}
	}()
	sh.Logger.Trace().Msg("Starting connecting")
	if err := sh.SetStatus(structs.ShardConnecting); err != nil {
		sh.Logger.Error().Err(err).Msg("Encountered error setting shard status")
	}
	if sh.wsConn == nil {
		var errorCh chan error
		var messageCh chan discord.ReceivedPayload
		errorCh, messageCh, err = sh.FeedWebsocket(sh.ctx, gatewayURL, nil)
		if err != nil {
			sh.Logger.Error().Err(err).Msg("Failed to dial")
			go sh.PublishWebhook(fmt.Sprintf("Failed to dial `%s`", gatewayURL), err.Error(), 14431557, false)
			return
		}
		sh.Lock()
		sh.ErrorCh = errorCh
		sh.MessageCh = messageCh
		sh.Unlock()
	} else {
		sh.Logger.Info().Msg("Reusing websocket connection")
	}
	sh.Logger.Trace().Msg("Reading from WS")
	msg, err := sh.readMessage()
	if err != nil {
		sh.Logger.Error().Err(err).Msg("Failed to read message")
		return
	}
	hello := discord.Hello{}
	err = sh.decodeContent(msg, &hello)
	sh.LastHeartbeatMu.Lock()
	sh.LastHeartbeatAck = time.Now().UTC()
	sh.LastHeartbeatSent = time.Now().UTC()
	sh.LastHeartbeatMu.Unlock()
	sh.Lock()
	sh.HeartbeatInterval = hello.HeartbeatInterval * time.Millisecond
	sh.MaxHeartbeatFailures = sh.HeartbeatInterval * time.Duration(sh.Manager.Configuration.Bot.MaxHeartbeatFailures)
	sh.Heartbeater = time.NewTicker(sh.HeartbeatInterval)
	sh.Unlock()
	if sh.HeartbeatActive.IsNotSet() {
		go sh.Heartbeat()
	}
	seq := atomic.LoadInt64(sh.seq)
	sh.Logger.Debug().
		Dur("interval", sh.HeartbeatInterval).
		Int("maxfails", sh.Manager.Configuration.Bot.MaxHeartbeatFailures).
		Msg("Retrieved HELLO event from discord")
	sh.RLock()
	sessionID := sh.sessionID
	sh.RUnlock()
	if sessionID == "" || seq == 0 {
		err = sh.Identify()
		if err != nil {
			sh.Logger.Error().Err(err).Msg("Failed to identify")
			go sh.PublishWebhook("Gateway `IDENTIFY` failed", err.Error(), 14431557, false)
			return
		}
	} else {
		err = sh.Resume()
		if err != nil {
			sh.Logger.Error().Err(err).Msg("Failed to resume")
			go sh.PublishWebhook("Gateway `RESUME` failed", err.Error(), 14431557, false)
			return
		}
		if err := sh.SetStatus(structs.ShardReady); err != nil {
			sh.Logger.Error().Err(err).Msg("Encountered error setting shard status")
		}
	}
	sh.Manager.ConfigurationMu.RLock()
	hash, err = QuickHash(sh.Manager.Configuration.Token)
	if err != nil {
		sh.Manager.ConfigurationMu.RUnlock()
		sh.Logger.Error().Err(err).Msg("Failed to generate token hash")
		return
	}
	bucket := fmt.Sprintf("gw:%s:%d", hash, sh.ShardID%sh.Manager.Gateway.SessionStartLimit.MaxConcurrency)
	sh.Manager.Buckets.ResetBucket(bucket)
	sh.Manager.ConfigurationMu.RUnlock()
	t := time.NewTicker(time.Second * gatewayConnectTimeout)
	sh.Logger.Trace().Msg("Waiting for first event")
	sh.RLock()
	errorch := sh.ErrorCh
	messagech := sh.MessageCh
	sh.RUnlock()
	select {
	case err = <-errorch:
		sh.Logger.Error().Err(err).Msg("Encountered error whilst connecting")
		go sh.PublishWebhook("Encountered error during connection", err.Error(), 14431557, false)
		return xerrors.Errorf("encountered error whilst connecting: %w", err)
	case msg = <-messagech:
		sh.Logger.Debug().Msgf("Received first event. %d %s", msg.Op, msg.Type)
		messagech <- msg
	case <-t.C:
	}
	sh.Logger.Trace().Msg("Finished connecting")
	return err
}
func (sh *Shard) FeedWebsocket(ctx context.Context, u string,
	opts *websocket.DialOptions) (errorCh chan error, messageCh chan discord.ReceivedPayload, err error) {
	messageCh = make(chan discord.ReceivedPayload, messageChannelBuffer)
	errorCh = make(chan error, 1)
	conn, _, err := websocket.Dial(ctx, u, opts)
	if err != nil {
		sh.Logger.Error().Err(err).Msg("Failed to dial websocket")
		return errorCh, messageCh, xerrors.Errorf("failed to connect to websocket: %w", err)
	}
	conn.SetReadLimit(websocketReadLimit)
	sh.wsConn = conn
	go func() {
		for {
			mt, buf, err := conn.Read(ctx)
			select {
			case <-ctx.Done():
				return
			default:
			}
			if err != nil {
				errorCh <- xerrors.Errorf("readMessage read: %w", err)
				return
			}
			if mt == websocket.MessageBinary {
				buf, err = czlib.Decompress(buf)
				if err != nil {
					errorCh <- xerrors.Errorf("readMessage decompress: %w", err)
					return
				}
			}
			now := time.Now().UTC()
			msg := discord.ReceivedPayload{
				TraceTime: now,
				Trace:     make(map[string]int),
			}
			err = json.Unmarshal(buf, &msg)
			if err != nil {
				sh.Logger.Error().Err(err).Msg("Failed to unmarshal message")
				continue
			}
			now = time.Now().UTC()
			msg.AddTrace("unmarshal", now)
			atomic.AddInt64(sh.events, 1)
			messageCh <- msg
		}
	}()
	return errorCh, messageCh, nil
}
func (sh *Shard) OnEvent(msg discord.ReceivedPayload) {
	var err error
	fin := make(chan void)
	go func() {
		since := time.Now()
		t := time.NewTimer(dispatchTimeout)
		for {
			select {
			case <-fin:
				return
			case <-t.C:
				sh.Logger.Warn().
					Str("type", msg.Type).
					Int("op", int(msg.Op)).
					Str("data", gotils.B2S(msg.Data)).
					Msgf("Event %s is taking too long. Been executing for %f seconds. Possible deadlock?",
						msg.Type, time.Since(since).
							Round(time.Second).Seconds(),
					)
				t.Reset(dispatchTimeout)
			}
		}
	}()
	defer close(fin)
	switch msg.Op {
	case discord.GatewayOpHeartbeat:
		sh.Logger.Debug().Msg("Received heartbeat request")
		err = sh.SendEvent(discord.GatewayOpHeartbeat, atomic.LoadInt64(sh.seq))
		if err != nil {
			go sh.PublishWebhook("Failed to send heartbeat to gateway", err.Error(), 16760839, false)
			sh.Logger.Error().Err(err).Msg("Failed to send heartbeat in response to gateway, reconnecting...")
			err = sh.Reconnect(websocket.StatusNormalClosure)
			if err != nil {
				sh.Logger.Error().Err(err).Msg("Failed to reconnect")
			}
			return
		}
	case discord.GatewayOpInvalidSession:
		resumable := json.Get(msg.Data, "d").ToBool()
		if !resumable {
			sh.sessionID = ""
			atomic.StoreInt64(sh.seq, 0)
		}
		go sh.PublishWebhook("Received invalid session from gateway", "", 16760839, false)
		sh.Logger.Warn().Bool("resumable", resumable).Msg("Received invalid session from gateway")
		err = sh.Reconnect(reconnectCloseCode)
		if err != nil {
			sh.Logger.Error().Err(err).Msg("Failed to reconnect")
		}
	case discord.GatewayOpHello:
		hello := discord.Hello{}
		err = sh.decodeContent(msg, &hello)
		sh.LastHeartbeatMu.Lock()
		sh.LastHeartbeatAck = time.Now().UTC()
		sh.LastHeartbeatSent = time.Now().UTC()
		sh.LastHeartbeatMu.Unlock()
		sh.Lock()
		sh.HeartbeatInterval = hello.HeartbeatInterval * time.Millisecond
		sh.MaxHeartbeatFailures = sh.HeartbeatInterval * time.Duration(sh.Manager.Configuration.Bot.MaxHeartbeatFailures)
		sh.Heartbeater = time.NewTicker(sh.HeartbeatInterval)
		sh.Unlock()
		sh.Logger.Debug().
			Dur("interval", sh.HeartbeatInterval).
			Int("maxfails", sh.Manager.Configuration.Bot.MaxHeartbeatFailures).
			Msg("Retrieved HELLO event from discord")
		return
	case discord.GatewayOpReconnect:
		sh.Logger.Info().Msg("Reconnecting in response to gateway")
		err = sh.Reconnect(reconnectCloseCode)
		if err != nil {
			sh.Logger.Error().Err(err).Msg("Failed to reconnect")
		}
		return
	case discord.GatewayOpDispatch:
		exec := func() {
			var ticket int
			atomic.AddInt64(sh.Manager.Sandwich.PoolWaiting, 1)
			ticket = sh.Manager.Sandwich.Pool.Wait()
			defer sh.Manager.Sandwich.Pool.FreeTicket(ticket)
			msg.AddTrace("ticket", time.Now().UTC())
			msg.Trace["ticket_id"] = ticket
			atomic.AddInt64(sh.Manager.Sandwich.PoolWaiting, -1)
			err = sh.OnDispatch(msg)
			if err != nil && !xerrors.Is(err, NoHandler) {
				sh.Logger.Error().Err(err).Msg("Failed to handle event")
			}
		}
		go exec()
	case discord.GatewayOpHeartbeatACK:
		sh.LastHeartbeatMu.Lock()
		sh.LastHeartbeatAck = time.Now().UTC()
		sh.Logger.Debug().
			Int64("RTT", sh.LastHeartbeatAck.Sub(sh.LastHeartbeatSent).Milliseconds()).
			Msg("Received heartbeat ACK")
		sh.LastHeartbeatMu.Unlock()
		return
	case discord.GatewayOpVoiceStateUpdate:
	case discord.GatewayOpIdentify,
		discord.GatewayOpRequestGuildMembers,
		discord.GatewayOpResume,
		discord.GatewayOpStatusUpdate:
	default:
		sh.Logger.Warn().
			Int("op", int(msg.Op)).
			Str("type", msg.Type).
			Msg("Gateway sent unknown packet")
		return
	}
	atomic.StoreInt64(sh.seq, msg.Sequence)
}
func (sh *Shard) OnDispatch(msg discord.ReceivedPayload) (err error) {
	start := time.Now().UTC()
	defer func() {
		now := time.Now().UTC()
		change := now.Sub(start)
		msg.AddTrace("publish", now)
		if change > time.Second {
			l := sh.Logger.Warn()
			if trcrslt, err := json.MarshalToString(msg.Trace); err == nil {
				l = l.Str("trace", trcrslt)
			}
			l.Msgf("%s took %d ms", msg.Type, change.Milliseconds())
		}
		if change > 15*time.Second {
			trcrslt := ""
			for tracer, tracetime := range msg.Trace {
				trcrslt += fmt.Sprintf("%s: **%d**ms\n", tracer, tracetime)
			}
			go sh.PublishWebhook(
				fmt.Sprintf("Packet `%s` took too long. Took `%dms`", msg.Type,
					change.Milliseconds()), trcrslt, 16760839, false)
		}
	}()
	if sh.Manager.ProducerClient == nil {
		return xerrors.Errorf("no producer client found")
	}
	sh.Manager.EventBlacklistMu.RLock()
	contains := gotils.StringSliceInclude(sh.Manager.EventBlacklist, msg.Type)
	sh.Manager.EventBlacklistMu.RUnlock()
	if contains {
		return
	}
	msg.AddTrace("dispatch", time.Now().UTC())
	results, ok, err := sh.Manager.Sandwich.StateDispatch(&StateCtx{
		Sg: sh.Manager.Sandwich,
		Mg: sh.Manager,
		Sh: sh,
	}, msg)
	msg.AddTrace("state", time.Now().UTC())
	if err != nil {
		return xerrors.Errorf("on dispatch failure for %s: %w", msg.Type, err)
	}
	if !ok {
		return
	}
	sh.Manager.ProduceBlacklistMu.RLock()
	contains = gotils.StringSliceInclude(sh.Manager.ProduceBlacklist, msg.Type)
	sh.Manager.ProduceBlacklistMu.RUnlock()
	if contains {
		return
	}
	packet := sh.pp.Get().(*structs.SandwichPayload)
	defer sh.pp.Put(packet)
	packet.ReceivedPayload = msg
	packet.Trace = msg.Trace
	packet.Data = results.Data
	packet.Extra = results.Extra
	err = sh.PublishEvent(packet)
	return err
}