123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461 |
- package chat
- import (
- "encoding/json"
- "fmt"
- "net/http"
- "sync"
- "time"
- log "github.com/sirupsen/logrus"
- "github.com/gorilla/websocket"
- "github.com/owncast/owncast/config"
- "github.com/owncast/owncast/core/chat/events"
- "github.com/owncast/owncast/core/data"
- "github.com/owncast/owncast/core/webhooks"
- "github.com/owncast/owncast/models"
- "github.com/owncast/owncast/persistence/userrepository"
- "github.com/owncast/owncast/services/geoip"
- "github.com/owncast/owncast/utils"
- )
- var _server *Server
- // Server represents an instance of the chat server.
- type Server struct {
- clients map[uint]*Client
- // send outbound message payload to all clients
- outbound chan []byte
- // receive inbound message payload from all clients
- inbound chan chatClientEvent
- // unregister requests from clients.
- unregister chan uint // the ChatClient id
- geoipClient *geoip.Client
- // a map of user IDs and timers that fire for chat part messages.
- userPartedTimers map[string]*time.Ticker
- seq uint
- maxSocketConnectionLimit uint64
- mu sync.RWMutex
- }
- // NewChat will return a new instance of the chat server.
- func NewChat() *Server {
- maximumConcurrentConnectionLimit := getMaximumConcurrentConnectionLimit()
- setSystemConcurrentConnectionLimit(maximumConcurrentConnectionLimit)
- server := &Server{
- clients: map[uint]*Client{},
- outbound: make(chan []byte),
- inbound: make(chan chatClientEvent),
- unregister: make(chan uint),
- maxSocketConnectionLimit: maximumConcurrentConnectionLimit,
- geoipClient: geoip.NewClient(),
- userPartedTimers: map[string]*time.Ticker{},
- }
- return server
- }
- // Run will start the chat server.
- func (s *Server) Run() {
- for {
- select {
- case clientID := <-s.unregister:
- if client, ok := s.clients[clientID]; ok {
- s.handleClientDisconnected(client)
- s.mu.Lock()
- delete(s.clients, clientID)
- s.mu.Unlock()
- }
- case message := <-s.inbound:
- s.eventReceived(message)
- }
- }
- }
- // Addclient registers new connection as a User.
- func (s *Server) Addclient(conn *websocket.Conn, user *models.User, accessToken string, userAgent string, ipAddress string) *Client {
- client := &Client{
- server: s,
- conn: conn,
- User: user,
- IPAddress: ipAddress,
- accessToken: accessToken,
- send: make(chan []byte, 256),
- UserAgent: userAgent,
- ConnectedAt: time.Now(),
- }
- shouldSendJoinedMessages := data.GetChatJoinPartMessagesEnabled()
- // If there are existing clients connected for this user do not send
- // a user joined message. Do not put this under a mutex, as
- // GetClientsForUser already has a lock.
- if existingConnectedClients, _ := GetClientsForUser(user.ID); len(existingConnectedClients) > 0 {
- shouldSendJoinedMessages = false
- }
- s.mu.Lock()
- {
- // If there is a pending disconnect timer then clear it.
- // Do not send user joined message if enough time hasn't passed where the
- // user chat part message hasn't been sent yet.
- if ticker, ok := s.userPartedTimers[user.ID]; ok {
- ticker.Stop()
- delete(s.userPartedTimers, user.ID)
- shouldSendJoinedMessages = false
- }
- client.Id = s.seq
- s.clients[client.Id] = client
- s.seq++
- }
- s.mu.Unlock()
- log.Traceln("Adding client", client.Id, "total count:", len(s.clients))
- go client.writePump()
- go client.readPump()
- client.sendConnectedClientInfo()
- if getStatus().Online {
- if shouldSendJoinedMessages {
- s.sendUserJoinedMessage(client)
- }
- s.sendWelcomeMessageToClient(client)
- }
- // Asynchronously, optionally, fetch GeoIP data.
- go func(client *Client) {
- client.Geo = s.geoipClient.GetGeoFromIP(ipAddress)
- }(client)
- return client
- }
- func (s *Server) sendUserJoinedMessage(c *Client) {
- userJoinedEvent := events.UserJoinedEvent{}
- userJoinedEvent.SetDefaults()
- userJoinedEvent.User = c.User
- userJoinedEvent.ClientID = c.Id
- if err := s.Broadcast(userJoinedEvent.GetBroadcastPayload()); err != nil {
- log.Errorln("error adding client to chat server", err)
- }
- // Send chat user joined webhook
- webhooks.SendChatEventUserJoined(userJoinedEvent)
- }
- func (s *Server) handleClientDisconnected(c *Client) {
- if _, ok := s.clients[c.Id]; ok {
- log.Debugln("Deleting", c.Id)
- delete(s.clients, c.Id)
- }
- additionalClientCheck, _ := GetClientsForUser(c.User.ID)
- if len(additionalClientCheck) > 0 {
- // This user is still connected to chat with another client.
- return
- }
- s.userPartedTimers[c.User.ID] = time.NewTicker(10 * time.Second)
- go func() {
- <-s.userPartedTimers[c.User.ID].C
- s.sendUserPartedMessage(c)
- }()
- }
- func (s *Server) sendUserPartedMessage(c *Client) {
- s.userPartedTimers[c.User.ID].Stop()
- delete(s.userPartedTimers, c.User.ID)
- userPartEvent := events.UserPartEvent{}
- userPartEvent.SetDefaults()
- userPartEvent.User = c.User
- userPartEvent.ClientID = c.Id
- // If part messages are disabled.
- if data.GetChatJoinPartMessagesEnabled() {
- if err := s.Broadcast(userPartEvent.GetBroadcastPayload()); err != nil {
- log.Errorln("error sending chat part message", err)
- }
- }
- // Send chat user joined webhook
- webhooks.SendChatEventUserParted(userPartEvent)
- }
- // HandleClientConnection is fired when a single client connects to the websocket.
- func (s *Server) HandleClientConnection(w http.ResponseWriter, r *http.Request) {
- if data.GetChatDisabled() {
- _, _ = w.Write([]byte(events.ChatDisabled))
- return
- }
- ipAddress := utils.GetIPAddressFromRequest(r)
- // Check if this client's IP address is banned. If so send a rejection.
- if blocked, err := data.IsIPAddressBanned(ipAddress); blocked {
- log.Debugln("Client ip address has been blocked. Rejecting.")
- w.WriteHeader(http.StatusForbidden)
- return
- } else if err != nil {
- log.Errorln("error determining if IP address is blocked: ", err)
- }
- // Limit concurrent chat connections
- if uint64(len(s.clients)) >= s.maxSocketConnectionLimit {
- log.Warnln("rejecting incoming client connection as it exceeds the max client count of", s.maxSocketConnectionLimit)
- _, _ = w.Write([]byte(events.ErrorMaxConnectionsExceeded))
- return
- }
- // To allow dev web environments to connect.
- upgrader.CheckOrigin = func(r *http.Request) bool {
- return true
- }
- conn, err := upgrader.Upgrade(w, r, nil)
- if err != nil {
- log.Debugln(err)
- return
- }
- accessToken := r.URL.Query().Get("accessToken")
- if accessToken == "" {
- log.Errorln("Access token is required")
- // Return HTTP status code
- _ = conn.Close()
- return
- }
- userRepository := userrepository.Get()
- // A user is required to use the websocket
- user := userRepository.GetUserByToken(accessToken)
- if user == nil {
- // Send error that registration is required
- _ = conn.WriteJSON(events.EventPayload{
- "type": events.ErrorNeedsRegistration,
- })
- _ = conn.Close()
- return
- }
- // User is disabled therefore we should disconnect.
- if user.DisabledAt != nil {
- log.Traceln("Disabled user", user.ID, user.DisplayName, "rejected")
- _ = conn.WriteJSON(events.EventPayload{
- "type": events.ErrorUserDisabled,
- })
- _ = conn.Close()
- return
- }
- userAgent := r.UserAgent()
- s.Addclient(conn, user, accessToken, userAgent, ipAddress)
- }
- // Broadcast sends message to all connected clients.
- func (s *Server) Broadcast(payload events.EventPayload) error {
- data, err := json.Marshal(payload)
- if err != nil {
- return err
- }
- s.mu.RLock()
- defer s.mu.RUnlock()
- for _, client := range s.clients {
- if client == nil {
- continue
- }
- select {
- case client.send <- data:
- default:
- go client.close()
- }
- }
- return nil
- }
- // Send will send a single payload to a single connected client.
- func (s *Server) Send(payload events.EventPayload, client *Client) {
- data, err := json.Marshal(payload)
- if err != nil {
- log.Errorln(err)
- return
- }
- client.send <- data
- }
- // DisconnectClients will forcefully disconnect all clients belonging to a user by ID.
- func (s *Server) DisconnectClients(clients []*Client) {
- for _, client := range clients {
- log.Traceln("Disconnecting client", client.User.ID, "owned by", client.User.DisplayName)
- go func(client *Client) {
- event := events.UserDisabledEvent{}
- event.SetDefaults()
- // Send this disabled event specifically to this single connected client
- // to let them know they've been banned.
- _server.Send(event.GetBroadcastPayload(), client)
- // Give the socket time to send out the above message.
- // Unfortunately I don't know of any way to get a real callback to know when
- // the message was successfully sent, so give it a couple seconds.
- time.Sleep(2 * time.Second)
- // Forcefully disconnect if still valid.
- if client != nil {
- client.close()
- }
- }(client)
- }
- }
- // SendConnectedClientInfoToUser will find all the connected clients assigned to a user
- // and re-send each the connected client info.
- func SendConnectedClientInfoToUser(userID string) error {
- clients, err := GetClientsForUser(userID)
- if err != nil {
- return err
- }
- userRepository := userrepository.Get()
- // Get an updated reference to the user.
- user := userRepository.GetUserByID(userID)
- if user == nil {
- return fmt.Errorf("user not found")
- }
- if err != nil {
- return err
- }
- for _, client := range clients {
- // Update the client's reference to its user.
- client.User = user
- // Send the update to the client.
- client.sendConnectedClientInfo()
- }
- return nil
- }
- // SendActionToUser will send system action text to all connected clients
- // assigned to a user ID.
- func SendActionToUser(userID string, text string) error {
- clients, err := GetClientsForUser(userID)
- if err != nil {
- return err
- }
- for _, client := range clients {
- _server.sendActionToClient(client, text)
- }
- return nil
- }
- func (s *Server) eventReceived(event chatClientEvent) {
- c := event.client
- u := c.User
- // If established chat user only mode is enabled and the user is not old
- // enough then reject this event and send them an informative message.
- if u != nil && data.GetChatEstbalishedUsersOnlyMode() && time.Since(event.client.User.CreatedAt) < config.GetDefaults().ChatEstablishedUserModeTimeDuration && !u.IsModerator() {
- s.sendActionToClient(c, "You have not been an established chat participant long enough to take part in chat. Please enjoy the stream and try again later.")
- return
- }
- var typecheck map[string]interface{}
- if err := json.Unmarshal(event.data, &typecheck); err != nil {
- log.Debugln(err)
- }
- eventType := typecheck["type"]
- switch eventType {
- case events.MessageSent:
- s.userMessageSent(event)
- case events.UserNameChanged:
- s.userNameChanged(event)
- case events.UserColorChanged:
- s.userColorChanged(event)
- default:
- log.Debugln(logSanitize(fmt.Sprint(eventType)), "event not found:", logSanitize(fmt.Sprint(typecheck)))
- }
- }
- func (s *Server) sendWelcomeMessageToClient(c *Client) {
- // Add an artificial delay so people notice this message come in.
- time.Sleep(7 * time.Second)
- welcomeMessage := utils.RenderSimpleMarkdown(data.GetServerWelcomeMessage())
- if welcomeMessage != "" {
- s.sendSystemMessageToClient(c, welcomeMessage)
- }
- }
- func (s *Server) sendAllWelcomeMessage() {
- welcomeMessage := utils.RenderSimpleMarkdown(data.GetServerWelcomeMessage())
- if welcomeMessage != "" {
- clientMessage := events.SystemMessageEvent{
- Event: events.Event{},
- MessageEvent: events.MessageEvent{
- Body: welcomeMessage,
- },
- }
- clientMessage.SetDefaults()
- _ = s.Broadcast(clientMessage.GetBroadcastPayload())
- }
- }
- func (s *Server) sendSystemMessageToClient(c *Client, message string) {
- clientMessage := events.SystemMessageEvent{
- Event: events.Event{},
- MessageEvent: events.MessageEvent{
- Body: message,
- },
- }
- clientMessage.SetDefaults()
- clientMessage.RenderBody()
- s.Send(clientMessage.GetBroadcastPayload(), c)
- }
- func (s *Server) sendActionToClient(c *Client, message string) {
- clientMessage := events.ActionEvent{
- MessageEvent: events.MessageEvent{
- Body: message,
- },
- Event: events.Event{
- Type: events.ChatActionSent,
- },
- }
- clientMessage.SetDefaults()
- clientMessage.RenderBody()
- s.Send(clientMessage.GetBroadcastPayload(), c)
- }
|