2020-06-23 03:11:56 +02:00
package chat
import (
2021-07-20 04:22:29 +02:00
"encoding/json"
2021-11-03 03:27:41 +01:00
"fmt"
2020-06-23 03:11:56 +02:00
"net/http"
2020-12-22 04:42:47 +01:00
"sync"
2020-06-23 03:11:56 +02:00
"time"
log "github.com/sirupsen/logrus"
2020-06-23 08:52:50 +02:00
2021-07-20 04:22:29 +02:00
"github.com/gorilla/websocket"
2022-03-07 08:26:24 +01:00
"github.com/owncast/owncast/config"
2021-07-20 04:22:29 +02:00
"github.com/owncast/owncast/core/chat/events"
2021-02-19 08:05:52 +01:00
"github.com/owncast/owncast/core/data"
2021-07-20 04:22:29 +02:00
"github.com/owncast/owncast/core/user"
2021-02-19 08:05:52 +01:00
"github.com/owncast/owncast/core/webhooks"
2021-08-12 23:55:21 +02:00
"github.com/owncast/owncast/geoip"
2021-07-20 04:22:29 +02:00
"github.com/owncast/owncast/utils"
2020-06-23 03:11:56 +02:00
)
2021-09-12 09:18:15 +02:00
var _server * Server
2020-06-23 22:11:01 +02:00
2021-09-21 23:06:23 +02:00
// a map of user IDs and when they last were active.
var _lastSeenCache = map [ string ] time . Time { }
2021-09-12 09:18:15 +02:00
// Server represents an instance of the chat server.
type Server struct {
2021-08-14 00:26:44 +02:00
mu sync . RWMutex
seq uint
2021-09-12 09:18:15 +02:00
clients map [ uint ] * Client
2021-08-14 00:26:44 +02:00
maxSocketConnectionLimit int64
2020-12-22 04:42:47 +01:00
2021-07-20 04:22:29 +02:00
// send outbound message payload to all clients
outbound chan [ ] byte
2020-06-23 03:11:56 +02:00
2021-07-20 04:22:29 +02:00
// receive inbound message payload from all clients
inbound chan chatClientEvent
2020-06-23 22:11:01 +02:00
2021-07-20 04:22:29 +02:00
// unregister requests from clients.
2021-07-27 02:30:28 +02:00
unregister chan uint // the ChatClient id
2021-10-12 22:21:37 +02:00
geoipClient * geoip . Client
2020-06-23 03:11:56 +02:00
}
2021-09-12 09:18:15 +02:00
// NewChat will return a new instance of the chat server.
func NewChat ( ) * Server {
2021-08-14 00:26:44 +02:00
maximumConcurrentConnectionLimit := getMaximumConcurrentConnectionLimit ( )
setSystemConcurrentConnectionLimit ( maximumConcurrentConnectionLimit )
2021-09-12 09:18:15 +02:00
server := & Server {
clients : map [ uint ] * Client { } ,
2021-08-14 00:26:44 +02:00
outbound : make ( chan [ ] byte ) ,
inbound : make ( chan chatClientEvent ) ,
unregister : make ( chan uint ) ,
maxSocketConnectionLimit : maximumConcurrentConnectionLimit ,
2021-10-12 22:21:37 +02:00
geoipClient : geoip . NewClient ( ) ,
2021-07-20 04:22:29 +02:00
}
2020-06-23 03:11:56 +02:00
2021-07-20 04:22:29 +02:00
return server
2020-06-23 03:11:56 +02:00
}
2021-09-12 09:18:15 +02:00
// Run will start the chat server.
func ( s * Server ) Run ( ) {
2021-07-20 04:22:29 +02:00
for {
select {
2021-09-12 09:18:15 +02:00
case clientID := <- s . unregister :
if _ , ok := s . clients [ clientID ] ; ok {
2021-07-20 04:22:29 +02:00
s . mu . Lock ( )
2021-09-12 09:18:15 +02:00
delete ( s . clients , clientID )
2021-07-20 04:22:29 +02:00
s . mu . Unlock ( )
}
case message := <- s . inbound :
s . eventReceived ( message )
}
}
2020-06-23 03:11:56 +02:00
}
2021-07-20 04:22:29 +02:00
// Addclient registers new connection as a User.
2021-09-12 09:18:15 +02:00
func ( s * Server ) Addclient ( conn * websocket . Conn , user * user . User , accessToken string , userAgent string , ipAddress string ) * Client {
client := & Client {
2021-07-20 04:22:29 +02:00
server : s ,
conn : conn ,
User : user ,
2022-03-07 05:34:49 +01:00
IPAddress : ipAddress ,
2021-07-20 04:22:29 +02:00
accessToken : accessToken ,
2021-07-31 21:48:42 +02:00
send : make ( chan [ ] byte , 256 ) ,
2021-07-20 04:22:29 +02:00
UserAgent : userAgent ,
ConnectedAt : time . Now ( ) ,
}
2021-09-21 23:06:23 +02:00
// Do not send user re-joined broadcast message if they've been active within 5 minutes.
2022-03-06 07:34:06 +01:00
shouldSendJoinedMessages := data . GetChatJoinMessagesEnabled ( )
2021-09-21 23:06:23 +02:00
if previouslyLastSeen , ok := _lastSeenCache [ user . ID ] ; ok && time . Since ( previouslyLastSeen ) < time . Minute * 5 {
shouldSendJoinedMessages = false
}
2021-07-20 04:22:29 +02:00
s . mu . Lock ( )
{
client . id = s . seq
s . clients [ client . id ] = client
s . seq ++
2021-09-21 23:06:23 +02:00
_lastSeenCache [ user . ID ] = time . Now ( )
2021-07-20 04:22:29 +02:00
}
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 {
2021-09-21 23:06:23 +02:00
if shouldSendJoinedMessages {
s . sendUserJoinedMessage ( client )
}
2021-07-20 04:22:29 +02:00
s . sendWelcomeMessageToClient ( client )
}
2021-08-12 23:55:21 +02:00
// Asynchronously, optionally, fetch GeoIP data.
2021-09-12 09:18:15 +02:00
go func ( client * Client ) {
2021-10-12 22:21:37 +02:00
client . Geo = s . geoipClient . GetGeoFromIP ( ipAddress )
2021-08-12 23:55:21 +02:00
} ( client )
2021-07-20 04:22:29 +02:00
return client
2020-06-23 03:11:56 +02:00
}
2021-09-12 09:18:15 +02:00
func ( s * Server ) sendUserJoinedMessage ( c * Client ) {
2021-07-20 04:22:29 +02:00
userJoinedEvent := events . UserJoinedEvent { }
userJoinedEvent . SetDefaults ( )
userJoinedEvent . User = c . User
2021-09-13 10:26:28 +02:00
userJoinedEvent . ClientID = c . id
2021-07-20 04:22:29 +02:00
if err := s . Broadcast ( userJoinedEvent . GetBroadcastPayload ( ) ) ; err != nil {
log . Errorln ( "error adding client to chat server" , err )
2020-06-23 03:11:56 +02:00
}
2021-07-20 04:22:29 +02:00
// Send chat user joined webhook
webhooks . SendChatEventUserJoined ( userJoinedEvent )
2020-06-23 03:11:56 +02:00
}
2021-09-12 09:18:15 +02:00
// ClientClosed is fired when a client disconnects or connection is dropped.
func ( s * Server ) ClientClosed ( c * Client ) {
2021-07-20 04:22:29 +02:00
s . mu . Lock ( )
defer s . mu . Unlock ( )
c . close ( )
2021-03-04 05:44:13 +01:00
2021-07-20 04:22:29 +02:00
if _ , ok := s . clients [ c . id ] ; ok {
log . Debugln ( "Deleting" , c . id )
delete ( s . clients , c . id )
2020-06-23 03:11:56 +02:00
}
}
2021-09-12 09:18:15 +02:00
// HandleClientConnection is fired when a single client connects to the websocket.
func ( s * Server ) HandleClientConnection ( w http . ResponseWriter , r * http . Request ) {
2021-07-20 04:22:29 +02:00
if data . GetChatDisabled ( ) {
_ , _ = w . Write ( [ ] byte ( events . ChatDisabled ) )
return
2020-07-29 06:30:03 +02:00
}
2021-02-19 08:05:52 +01:00
2022-03-07 05:34:49 +01:00
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." )
event := events . UserDisabledEvent { }
event . SetDefaults ( )
w . WriteHeader ( http . StatusForbidden )
// Send this disabled event specifically to this single connected client
// to let them know they've been banned.
// _server.Send(event.GetBroadcastPayload(), client)
return
} else if err != nil {
log . Errorln ( "error determining if IP address is blocked: " , err )
}
2021-07-20 04:22:29 +02:00
// Limit concurrent chat connections
2021-08-14 00:26:44 +02:00
if int64 ( len ( s . clients ) ) >= s . maxSocketConnectionLimit {
log . Warnln ( "rejecting incoming client connection as it exceeds the max client count of" , s . maxSocketConnectionLimit )
2021-07-20 04:22:29 +02:00
_ , _ = w . Write ( [ ] byte ( events . ErrorMaxConnectionsExceeded ) )
return
}
2021-02-19 08:05:52 +01:00
2021-07-20 04:22:29 +02:00
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
2021-09-12 09:18:15 +02:00
_ = conn . Close ( )
2021-07-20 04:22:29 +02:00
return
}
// A user is required to use the websocket
user := user . GetUserByToken ( accessToken )
if user == nil {
_ = conn . WriteJSON ( events . EventPayload {
"type" : events . ErrorNeedsRegistration ,
} )
// Send error that registration is required
2021-09-12 09:18:15 +02:00
_ = conn . Close ( )
2021-07-20 04:22:29 +02:00
return
}
// User is disabled therefore we should disconnect.
if user . DisabledAt != nil {
2021-09-12 09:18:15 +02:00
log . Traceln ( "Disabled user" , user . ID , user . DisplayName , "rejected" )
2021-07-20 04:22:29 +02:00
_ = conn . WriteJSON ( events . EventPayload {
"type" : events . ErrorUserDisabled ,
} )
2021-09-12 09:18:15 +02:00
_ = conn . Close ( )
2021-07-20 04:22:29 +02:00
return
2021-02-19 08:05:52 +01:00
}
2021-03-04 05:44:13 +01:00
2021-07-20 04:22:29 +02:00
userAgent := r . UserAgent ( )
2021-08-12 23:55:21 +02:00
s . Addclient ( conn , user , accessToken , userAgent , ipAddress )
2020-07-29 06:30:03 +02:00
}
2021-07-20 04:22:29 +02:00
// Broadcast sends message to all connected clients.
2021-09-12 09:18:15 +02:00
func ( s * Server ) Broadcast ( payload events . EventPayload ) error {
2021-07-20 04:22:29 +02:00
data , err := json . Marshal ( payload )
if err != nil {
return err
}
s . mu . Lock ( )
defer s . mu . Unlock ( )
2020-06-23 08:52:50 +02:00
2021-07-20 04:22:29 +02:00
for _ , client := range s . clients {
if client == nil {
continue
}
2020-06-23 08:52:50 +02:00
2021-07-20 04:22:29 +02:00
select {
case client . send <- data :
default :
2021-07-23 00:27:04 +02:00
client . close ( )
2021-07-20 04:22:29 +02:00
delete ( s . clients , client . id )
2020-06-23 08:52:50 +02:00
}
2021-07-20 04:22:29 +02:00
}
2020-06-23 08:52:50 +02:00
2021-07-20 04:22:29 +02:00
return nil
2020-06-23 08:52:50 +02:00
}
2021-09-12 09:18:15 +02:00
// Send will send a single payload to a single connected client.
func ( s * Server ) Send ( payload events . EventPayload , client * Client ) {
2021-07-20 04:22:29 +02:00
data , err := json . Marshal ( payload )
if err != nil {
log . Errorln ( err )
return
}
2020-06-23 03:11:56 +02:00
2021-07-20 04:22:29 +02:00
client . send <- data
}
2020-06-23 03:11:56 +02:00
2022-03-07 05:34:49 +01:00
// DisconnectClients will forcefully disconnect all clients belonging to a user by ID.
func ( s * Server ) DisconnectClients ( clients [ ] * Client ) {
2021-07-20 04:22:29 +02:00
for _ , client := range clients {
2021-09-12 09:18:15 +02:00
log . Traceln ( "Disconnecting client" , client . User . ID , "owned by" , client . User . DisplayName )
2021-03-14 19:46:27 +01:00
2021-09-12 09:18:15 +02:00
go func ( client * Client ) {
2021-07-20 04:22:29 +02:00
event := events . UserDisabledEvent { }
event . SetDefaults ( )
2021-03-10 20:07:47 +01:00
2021-07-20 04:22:29 +02:00
// Send this disabled event specifically to this single connected client
// to let them know they've been banned.
_server . Send ( event . GetBroadcastPayload ( ) , client )
2020-10-14 01:45:52 +02:00
2021-07-20 04:22:29 +02:00
// 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 )
2021-02-19 08:05:52 +01:00
2021-07-20 04:22:29 +02:00
// Forcefully disconnect if still valid.
if client != nil {
client . close ( )
2021-01-31 23:57:50 +01:00
}
2021-07-20 04:22:29 +02:00
} ( client )
}
}
2020-06-23 03:11:56 +02:00
2021-11-03 03:27:41 +01:00
// 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
}
// Get an updated reference to the user.
user := user . 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
}
2021-09-12 09:18:15 +02:00
func ( s * Server ) eventReceived ( event chatClientEvent ) {
2022-03-07 08:26:24 +01:00
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
}
2021-07-20 04:22:29 +02:00
var typecheck map [ string ] interface { }
if err := json . Unmarshal ( event . data , & typecheck ) ; err != nil {
log . Debugln ( err )
}
2020-06-23 03:11:56 +02:00
2021-07-20 04:22:29 +02:00
eventType := typecheck [ "type" ]
switch eventType {
case events . MessageSent :
s . userMessageSent ( event )
case events . UserNameChanged :
s . userNameChanged ( event )
default :
log . Debugln ( eventType , "event not found:" , typecheck )
2020-06-23 03:11:56 +02:00
}
}
2020-07-20 00:14:51 +02:00
2021-09-12 09:18:15 +02:00
func ( s * Server ) sendWelcomeMessageToClient ( c * Client ) {
2021-07-20 04:22:29 +02:00
// Add an artificial delay so people notice this message come in.
time . Sleep ( 7 * time . Second )
welcomeMessage := utils . RenderSimpleMarkdown ( data . GetServerWelcomeMessage ( ) )
2020-12-22 04:42:47 +01:00
2021-07-20 04:22:29 +02:00
if welcomeMessage != "" {
s . sendSystemMessageToClient ( c , welcomeMessage )
2020-12-22 04:42:47 +01:00
}
}
2021-09-12 09:18:15 +02:00
func ( s * Server ) sendAllWelcomeMessage ( ) {
2021-07-20 04:22:29 +02:00
welcomeMessage := utils . RenderSimpleMarkdown ( data . GetServerWelcomeMessage ( ) )
2020-07-20 00:14:51 +02:00
2021-07-20 04:22:29 +02:00
if welcomeMessage != "" {
clientMessage := events . SystemMessageEvent {
Event : events . Event { } ,
MessageEvent : events . MessageEvent {
Body : welcomeMessage ,
} ,
2021-03-21 22:10:56 +01:00
}
2021-07-20 04:22:29 +02:00
clientMessage . SetDefaults ( )
_ = s . Broadcast ( clientMessage . GetBroadcastPayload ( ) )
}
}
2021-09-12 09:18:15 +02:00
func ( s * Server ) sendSystemMessageToClient ( c * Client , message string ) {
2021-07-20 04:22:29 +02:00
clientMessage := events . SystemMessageEvent {
Event : events . Event { } ,
MessageEvent : events . MessageEvent {
Body : message ,
} ,
}
clientMessage . SetDefaults ( )
2021-09-13 10:26:28 +02:00
clientMessage . RenderBody ( )
2021-07-20 04:22:29 +02:00
s . Send ( clientMessage . GetBroadcastPayload ( ) , c )
}
2021-09-12 09:18:15 +02:00
func ( s * Server ) sendActionToClient ( c * Client , message string ) {
2021-07-20 04:22:29 +02:00
clientMessage := events . ActionEvent {
MessageEvent : events . MessageEvent {
Body : message ,
} ,
2021-11-03 03:27:41 +01:00
Event : events . Event {
Type : events . ChatActionSent ,
} ,
2021-07-20 04:22:29 +02:00
}
clientMessage . SetDefaults ( )
clientMessage . RenderBody ( )
s . Send ( clientMessage . GetBroadcastPayload ( ) , c )
2020-10-07 08:14:33 +02:00
}