gocommon/wshandler/wshandler.go

package wshandler

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/gob"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"

	"go.mongodb.org/mongo-driver/bson/primitive"
	"repositories.action2quare.com/ayo/gocommon"
	"repositories.action2quare.com/ayo/gocommon/flagx"
	"repositories.action2quare.com/ayo/gocommon/logger"

	"github.com/go-redis/redis/v8"
	"github.com/gorilla/websocket"
)

var noAuthFlag = flagx.Bool("noauth", false, "")

type wsconn struct {
	*websocket.Conn
	sender      *Sender
	joinedRooms []*room
}

func (conn *wsconn) popRoom(r *room) int {
	for i, jr := range conn.joinedRooms {
		if jr == r {
			conn.joinedRooms = append(conn.joinedRooms[:i], conn.joinedRooms[i+1:]...)
			break
		}
	}
	return len(conn.joinedRooms)
}

func (conn *wsconn) pushRoom(r *room) {
	conn.joinedRooms = append(conn.joinedRooms, r)
}

func (conn *wsconn) isInRoom(roomname string) bool {
	for _, r := range conn.joinedRooms {
		if r.name == roomname {
			return true
		}
	}
	return false
}

type UpstreamMessage struct {
	Alias  string
	Accid  primitive.ObjectID
	Target string
	Body   any
	Tag    []string
}

type DownstreamMessage struct {
	Alias string   `json:",omitempty"`
	Body  any      `json:",omitempty"`
	Tag   []string `json:",omitempty"`
}

type commandType string

const (
	commandType_EnterRoom = commandType("enter_room")
	commandType_LeaveRoom = commandType("leave_room")
)

type commandMessage struct {
	Cmd  commandType
	Args []any
}

type WebSocketMessageType int

const (
	TextMessage   = WebSocketMessageType(websocket.TextMessage)
	BinaryMessage = WebSocketMessageType(websocket.BinaryMessage)
	CloseMessage  = WebSocketMessageType(websocket.CloseMessage)
	PingMessage   = WebSocketMessageType(websocket.PingMessage)
	PongMessage   = WebSocketMessageType(websocket.PongMessage)
	Connected     = WebSocketMessageType(100)
	Disconnected  = WebSocketMessageType(101)
)

type Sender struct {
	Region                string
	Accid                 primitive.ObjectID
	Alias                 string
	disconnectedCallbacks map[string]func()
}

func (s *Sender) RegistDisconnectedCallback(name string, f func()) (old func()) {
	if s.disconnectedCallbacks == nil {
		s.disconnectedCallbacks = make(map[string]func())
	}

	old = s.disconnectedCallbacks[name]
	s.disconnectedCallbacks[name] = f
	return
}

func (s *Sender) PopDisconnectedCallback(name string) func() {
	old := s.disconnectedCallbacks[name]
	delete(s.disconnectedCallbacks, name)
	return old
}

type EventReceiver interface {
	OnClientMessageReceived(sender *Sender, messageType WebSocketMessageType, body io.Reader)
	OnRoomCreated(region, name string)
	OnRoomDestroyed(region, name string)
}

type send_msg_queue_elem struct {
	to  *wsconn
	mt  int
	msg []byte
}

type subhandler struct {
	redisMsgChanName  string
	redisCmdChanName  string
	redisSync         *redis.Client
	connInOutChan     chan *wsconn
	deliveryChan      chan any
	localDeliveryChan chan any
	sendMsgChan       chan send_msg_queue_elem
	callReceiver      EventReceiver
	connWaitGroup     sync.WaitGroup
	region            string
	receiverChain     []EventReceiver
}

// WebsocketHandler  :
type WebsocketHandler struct {
	subhandlers map[string]*subhandler
}

type wsConfig struct {
	gocommon.RegionStorageConfig
	Maingate string `json:"maingate_service_url"`
}

var config wsConfig

func init() {
	gocommon.LoadConfig(&config)
	gob.Register(UpstreamMessage{})
	gob.Register(commandMessage{})
	gob.Register(map[string]any{})
	gob.Register(primitive.A{})
	gob.Register(primitive.M{})
	gob.Register(primitive.ObjectID{})
	gob.Register([]any{})
}

func NewWebsocketHandler() (*WebsocketHandler, error) {
	subhandlers := make(map[string]*subhandler)
	for region, cfg := range config.RegionStorage {
		redisSync, err := gocommon.NewRedisClient(cfg.Redis["wshandler"])
		if err != nil {
			return nil, err
		}

		sendchan := make(chan send_msg_queue_elem, 1000)
		go func() {
			sender := func(elem *send_msg_queue_elem) {
				defer func() {
					r := recover()
					if r != nil {
						logger.Println(r)
					}
				}()
				elem.to.WriteMessage(elem.mt, elem.msg)
			}

			for elem := range sendchan {
				sender(&elem)
			}
		}()

		sh := &subhandler{
			redisMsgChanName:  fmt.Sprintf("_wsh_msg_%s_%d", region, redisSync.Options().DB),
			redisCmdChanName:  fmt.Sprintf("_wsh_cmd_%s_%d", region, redisSync.Options().DB),
			redisSync:         redisSync,
			connInOutChan:     make(chan *wsconn),
			deliveryChan:      make(chan any, 1000),
			localDeliveryChan: make(chan any, 100),
			sendMsgChan:       sendchan,
			region:            region,
		}

		subhandlers[region] = sh
	}

	return &WebsocketHandler{
		subhandlers: subhandlers,
	}, nil
}

func (ws *WebsocketHandler) RegisterReceiver(region string, receiver EventReceiver) {
	if sh := ws.subhandlers[region]; sh != nil {
		sh.receiverChain = append(sh.receiverChain, receiver)
	}
}

type nilReceiver struct{}

func (r *nilReceiver) OnClientMessageReceived(sender *Sender, messageType WebSocketMessageType, body io.Reader) {
}
func (r *nilReceiver) OnRoomCreated(region, name string)   {}
func (r *nilReceiver) OnRoomDestroyed(region, name string) {}

type chainReceiver struct {
	chain []EventReceiver
}

func (r *chainReceiver) OnClientMessageReceived(sender *Sender, messageType WebSocketMessageType, body io.Reader) {
	for _, cr := range r.chain {
		cr.OnClientMessageReceived(sender, messageType, body)
	}
}

func (r *chainReceiver) OnRoomCreated(region, name string) {
	for _, cr := range r.chain {
		cr.OnRoomCreated(region, name)
	}
}

func (r *chainReceiver) OnRoomDestroyed(region, name string) {
	for _, cr := range r.chain {
		cr.OnRoomDestroyed(region, name)
	}
}

func (ws *WebsocketHandler) Start(ctx context.Context) {
	for _, sh := range ws.subhandlers {
		chain := sh.receiverChain
		if len(chain) == 0 {
			sh.callReceiver = &nilReceiver{}
		} else if len(chain) == 1 {
			sh.callReceiver = chain[0]
		} else {
			sh.callReceiver = &chainReceiver{chain: sh.receiverChain}
		}

		sh.connWaitGroup.Add(1)
		go sh.mainLoop(ctx)
	}
}

func (ws *WebsocketHandler) Cleanup() {
	for _, sh := range ws.subhandlers {
		sh.connWaitGroup.Wait()
	}
}

func (ws *WebsocketHandler) RegisterHandlers(serveMux *http.ServeMux, prefix string) error {
	for region, sh := range ws.subhandlers {
		if region == "default" {
			region = ""
		}
		url := gocommon.MakeHttpHandlerPattern(prefix, region, "ws")
		if *noAuthFlag {
			serveMux.HandleFunc(url, sh.upgrade_nosession)
		} else {
			serveMux.HandleFunc(url, sh.upgrade)
		}
	}

	return nil
}

func (ws *WebsocketHandler) GetState(region string, accid primitive.ObjectID) string {
	if sh := ws.subhandlers[region]; sh != nil {
		state, _ := sh.redisSync.Get(context.Background(), accid.Hex()).Result()
		return state
	}
	return ""
}

func (ws *WebsocketHandler) SetState(region string, accid primitive.ObjectID, state string) {
	if sh := ws.subhandlers[region]; sh != nil {
		sh.redisSync.SetArgs(context.Background(), accid.Hex(), state, redis.SetArgs{Mode: "XX"}).Result()
	}
}

func (ws *WebsocketHandler) SendUpstreamMessage(region string, msg *UpstreamMessage) {
	sh := ws.subhandlers[region]
	if sh != nil {
		sh.localDeliveryChan <- msg
	}
}

func (ws *WebsocketHandler) EnterRoom(region string, room string, accid primitive.ObjectID) {
	sh := ws.subhandlers[region]
	if sh != nil {
		sh.localDeliveryChan <- &commandMessage{
			Cmd:  commandType_EnterRoom,
			Args: []any{room, accid},
		}
	}
}

func (ws *WebsocketHandler) LeaveRoom(region string, room string, accid primitive.ObjectID) {
	sh := ws.subhandlers[region]
	if sh != nil {
		sh.localDeliveryChan <- &commandMessage{
			Cmd:  commandType_LeaveRoom,
			Args: []any{room, accid},
		}
	}
}

func (sh *subhandler) mainLoop(ctx context.Context) {
	defer func() {
		sh.connWaitGroup.Done()
		s := recover()
		if s != nil {
			logger.Error(s)
		}
	}()

	// redis channel에서 유저가 보낸 메시지를 읽는 go rountine
	go func() {
		var pubsub *redis.PubSub
		for {
			if pubsub == nil {
				pubsub = sh.redisSync.Subscribe(ctx, sh.redisMsgChanName, sh.redisCmdChanName)
			}

			raw, err := pubsub.ReceiveMessage(ctx)
			if err == nil {
				buffer := bytes.NewBuffer([]byte(raw.Payload))
				dec := gob.NewDecoder(buffer)

				if raw.Channel == sh.redisMsgChanName {
					var msg UpstreamMessage
					if err := dec.Decode(&msg); err == nil {
						sh.deliveryChan <- &msg
					} else {
						logger.Println("decode UpstreamMessage failed :", err)
					}
				} else if raw.Channel == sh.redisCmdChanName {
					var cmd commandMessage
					if err := dec.Decode(&cmd); err == nil {
						sh.deliveryChan <- &cmd
					} else {
						logger.Println("decode UpstreamMessage failed :", err)
					}
				}
			} else {
				logger.Println("pubsub.ReceiveMessage failed :", err)
				pubsub.Close()
				pubsub = nil

				if ctx.Err() != nil {
					break
				}
			}
		}
	}()

	entireConns := make(map[string]*wsconn)
	rooms := make(map[string]*room)
	roomDestroyChan := make(chan string, 1000)
	findRoom := func(name string, create bool) *room {
		room := rooms[name]
		if room == nil && create {
			room = makeRoom(name, roomDestroyChan, sh.sendMsgChan)
			rooms[name] = room
			room.start(ctx)
			go sh.callReceiver.OnRoomCreated(sh.region, name)
		}
		return room
	}

	defer func() {
		for _, conn := range entireConns {
			var roomnames []string
			for _, room := range conn.joinedRooms {
				roomnames = append(roomnames, room.name)
			}
			bt, _ := json.Marshal(roomnames)
			sh.callReceiver.OnClientMessageReceived(conn.sender, Disconnected, bytes.NewBuffer(bt))
			conn.Close()
		}
	}()

	errProcessFailed_NotInRoom := errors.New("not in room")
	processPrivateUpstreamMessage := func(usermsg *UpstreamMessage) bool {
		target := usermsg.Target
		if target[0] == '#' {
			return false
		}
		roomindex := strings.IndexByte(target, '@')
		var accid string
		var roomid string
		if roomindex > 0 {
			accid = target[:roomindex]
			roomid = target[roomindex+1:]
		} else {
			accid = target
		}

		conn := entireConns[accid]
		if conn == nil {
			return false
		}

		if len(roomid) > 0 {
			if !conn.isInRoom(roomid) {
				return false
			}
		}

		// conn에게 메시지 보내면 처리 완료
		ds, _ := json.Marshal(DownstreamMessage{
			Alias: usermsg.Alias,
			Body:  usermsg.Body,
			Tag:   usermsg.Tag,
		})
		sh.sendMsgChan <- send_msg_queue_elem{
			to:  conn,
			mt:  websocket.TextMessage,
			msg: ds,
		}

		return true
	}
	errCommandArgumentNotCorrect := errors.New("command arguments are not correct")
	processCommandMessage := func(usermsg *commandMessage) (bool, error) {
		switch usermsg.Cmd {
		case commandType_EnterRoom:
			if len(usermsg.Args) != 2 {
				return false, errCommandArgumentNotCorrect
			}
			roomName := usermsg.Args[0].(string)
			accid := usermsg.Args[1].(primitive.ObjectID)
			conn := entireConns[accid.Hex()]
			if conn == nil {
				return false, nil
			}
			conn.pushRoom(findRoom(roomName, true).in(conn))

		case commandType_LeaveRoom:
			if len(usermsg.Args) != 2 {
				return false, errCommandArgumentNotCorrect
			}
			roomName := usermsg.Args[0].(string)
			accid := usermsg.Args[1].(primitive.ObjectID)
			conn := entireConns[accid.Hex()]
			if conn == nil {
				return false, nil
			}

			room := findRoom(roomName, false)
			if room == nil {
				return false, errProcessFailed_NotInRoom
			}

			if conn.popRoom(room.out(conn)) == 0 {
				closeMsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")
				conn.WriteControl(websocket.CloseMessage, closeMsg, time.Time{})
			}
		}
		return true, nil
	}

	// 유저에게서 온 메세지, 소켓 연결/해체 처리
	for {
		buffer := bytes.NewBuffer(make([]byte, 0, 1024))
		buffer.Reset()

		select {
		case <-ctx.Done():
			return

		case destroyedRoom := <-roomDestroyChan:
			delete(rooms, destroyedRoom)
			go sh.callReceiver.OnRoomDestroyed(sh.region, destroyedRoom)

		case usermsg := <-sh.localDeliveryChan:
			// 로컬에 connection이 있는지 먼저 확인해 보기 위한 채널
			// 없으면 publish한다.
			switch usermsg := usermsg.(type) {
			case *UpstreamMessage:
				if !processPrivateUpstreamMessage(usermsg) {
					// private메시지가 처리가 안됐으므로 publish
					enc := gob.NewEncoder(buffer)
					var err error
					if err = enc.Encode(usermsg); err == nil {
						_, err = sh.redisSync.Publish(context.Background(), sh.redisMsgChanName, buffer.Bytes()).Result()
					}

					if err != nil {
						logger.Println("gob.Encode or publish failed :", err)
					}
				}

			case *commandMessage:
				processed, err := processCommandMessage(usermsg)
				if err != nil {
					logger.Println("processCommandMessage returns err :", err)
					break
				}

				if !processed {
					var err error
					enc := gob.NewEncoder(buffer)
					if err = enc.Encode(usermsg); err == nil {
						_, err = sh.redisSync.Publish(context.Background(), sh.redisCmdChanName, buffer.Bytes()).Result()
					}
					if err != nil {
						logger.Println("gob.Encode or Publish failed :", err)
					}
				}
			}

		case usermsg := <-sh.deliveryChan:
			switch usermsg := usermsg.(type) {
			case *UpstreamMessage:
				target := usermsg.Target
				if target[0] == '#' {
					// 룸에 브로드캐스팅
					roomName := target[1:]
					if room := findRoom(roomName, false); room != nil {
						room.broadcast(usermsg)
					}
				} else {
					processPrivateUpstreamMessage(usermsg)
				}

			case *commandMessage:
				_, err := processCommandMessage(usermsg)
				if err != nil {
					logger.Println("processCommandMessage returns err :", err)
					break
				}

			default:
				logger.Println("usermsg is unknown type")
			}

		case c := <-sh.connInOutChan:
			if c.Conn == nil {
				delete(entireConns, c.sender.Accid.Hex())
				var roomnames []string
				for _, room := range c.joinedRooms {
					roomnames = append(roomnames, room.name)
					room.out(c)
				}
				c.joinedRooms = nil

				bt, _ := json.Marshal(roomnames)
				go sh.callReceiver.OnClientMessageReceived(c.sender, Disconnected, bytes.NewBuffer(bt))
			} else {
				entireConns[c.sender.Accid.Hex()] = c
				go sh.callReceiver.OnClientMessageReceived(c.sender, Connected, nil)
			}
		}
	}
}

func upgrade_core(sh *subhandler, conn *websocket.Conn, accid primitive.ObjectID, alias string) {
	newconn := &wsconn{
		Conn: conn,
		sender: &Sender{
			Region: sh.region,
			Alias:  alias,
			Accid:  accid,
		},
	}
	sh.connInOutChan <- newconn

	sh.connWaitGroup.Add(1)
	go func(c *wsconn, accid primitive.ObjectID, deliveryChan chan<- any) {
		sh.redisSync.Set(context.Background(), accid.Hex(), "online", 0)
		for {
			messageType, r, err := c.NextReader()
			if err != nil {
				c.Close()
				break
			}

			if messageType == websocket.CloseMessage {
				break
			}

			if messageType == websocket.TextMessage {
				// 유저가 직접 보낸 메시지
				sh.callReceiver.OnClientMessageReceived(c.sender, TextMessage, r)
			} else if messageType == websocket.BinaryMessage {
				sh.callReceiver.OnClientMessageReceived(c.sender, BinaryMessage, r)
			}
		}
		if c.sender.disconnectedCallbacks != nil {
			for _, f := range c.sender.disconnectedCallbacks {
				f()
			}
		}

		sh.redisSync.Del(context.Background(), accid.Hex())
		sh.connWaitGroup.Done()

		c.Conn = nil
		sh.connInOutChan <- c
	}(newconn, accid, sh.deliveryChan)
}

func (sh *subhandler) upgrade_nosession(w http.ResponseWriter, r *http.Request) {
	// 클라이언트 접속
	defer func() {
		s := recover()
		if s != nil {
			logger.Error(s)
		}
		io.Copy(io.Discard, r.Body)
		r.Body.Close()
	}()

	auth := strings.Split(r.Header.Get("Authorization"), " ")
	if len(auth) != 2 {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	temp, err := hex.DecodeString(auth[1])
	if err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	if len(temp) != len(primitive.NilObjectID) {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	raw := (*[12]byte)(temp)
	accid := primitive.ObjectID(*raw)

	var upgrader = websocket.Upgrader{} // use default options
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	var alias string
	if v := r.Header.Get("AS-X-ALIAS"); len(v) > 0 {
		vt, _ := base64.StdEncoding.DecodeString(v)
		alias = string(vt)
	} else {
		alias = accid.Hex()
	}

	upgrade_core(sh, conn, accid, alias)
}

func (sh *subhandler) upgrade(w http.ResponseWriter, r *http.Request) {
	// 클라이언트 접속
	defer func() {
		s := recover()
		if s != nil {
			logger.Error(s)
		}
		io.Copy(io.Discard, r.Body)
		r.Body.Close()
	}()

	sk := r.Header.Get("AS-X-SESSION")
	auth := r.Header.Get("Authorization")

	req, _ := http.NewRequest("GET", fmt.Sprintf("%s/query?sk=%s", config.Maingate, sk), nil)
	req.Header.Add("Authorization", auth)

	client := http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		logger.Error("authorize query failed :", err)
		return
	}
	defer resp.Body.Close()

	var authinfo gocommon.Authinfo
	dec := json.NewDecoder(resp.Body)
	if err = dec.Decode(&authinfo); err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		logger.Error("authorize query failed :", err)
		return
	}

	var upgrader = websocket.Upgrader{} // use default options
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	var alias string
	if v := r.Header.Get("AS-X-ALIAS"); len(v) > 0 {
		vt, _ := base64.StdEncoding.DecodeString(v)
		alias = string(vt)
	} else {
		alias = authinfo.Accid.Hex()
	}

	upgrade_core(sh, conn, authinfo.Accid, alias)
}