LogalDeveloper/owncast
0
Fork 0
Code Activity

Chat refactor + persistent backing chat users (#1163)

Browse Source 
* First pass at chat user registration and validation

* Disable chat if the user is disabled/blocked or the server hits max connections

* Handle dropping sockets if chat is disabled

* Fix origin in automated chat test

* Work for updated chat moderation

* Chat message markdown rendering and fix tests

* Put /api/chat behind a chat user access token. Closes #1085

* Reject blocked username changes

* More WIP moderation

* Defer configuring chat until we know if it is enabled. Closes #1135

* chat user blocking. Closes #1096

* Add tests around user access for #1096

* Add external integration chat message API + update integration auth middleware to pass along integration name. Closes #1092

* Delete old chat messages from db as to not hold on to excessive data. Closes #1152

* Add schema migration for messages. Closes #1155

* Commit updated API documentation

* Add chat load test

* Shared db mutex and db optimizations

* Simplify past display name handling

* Use a new test db for each test run

* Wire up the external messages actions + add tests for them

* Move access tokens to be actual users

* Run message pruning at launch + fix comparison

* Do not return API users in disabled users response

* Fix incorrect highlighting. Closes #1160

* Consolidate user table statements

* Set the max process connection limit to 70% of maximum

* Fix wrong old display name being returned in name change event

* Delete the old chat server files

* Wire back up the webhooks

* Remove unused

* Invalidate user cache on changes

* Do not send rendered body as RawBody

* Some cleanup

* Standardize names for external API users to ExternalAPIUser

* Do not log token

* Checkout branch when building admin for testing

* Bundle in dev admin for testing

* Some cleanup

* Cleanup js logs

* Cleanup and standardize event names

* Clean up some logging

* Update API spec. Closes #1133

* Commit updated API documentation

* Change paths to be better named

* Commit updated API documentation

* Update admin bundle

* Fix duplicate event name

* Rename scope var

* Update admin bundle

* Move connected clients controller into admin package

* Fix collecting usernames for autocomplete purposes

* No longer generate username when it is empty

* Sort clients and users by timestamp

* Move file to admin controller package

* Swap, so the comments stay correct

Co-authored-by: Jannik <jannik@outlook.com>

* Use explicit type alias

Co-authored-by: Jannik <jannik@outlook.com>

* Remove commented code.

Co-authored-by: Jannik <jannik@outlook.com>

* Cleanup test

* Remove some extra logging

* Add some clarity

* Update dev instance of admin for testing

* Consolidate lines

Co-authored-by: Jannik <jannik@outlook.com>

* Remove commented unused vars

Co-authored-by: Jannik <jannik@outlook.com>

* Until needed do not return IP address with client list

* Fix typo of wrong var

* Typo led to a bad test. Fix typo and fix test.

* Guard against the socket reconnecting on error if previously set to shutdown

* Do not log access tokens

* Return success message on enable/disable user

* Clean up some inactionable error messages. Sent ban message. Sort banned users.

* fix styling for when chat is completely disabled

* Unused

* guard against nil clients

* Update dev admin bundle

* Do not unhide messages when unblocking user just to be safe. Send removal action from the controller

* Add convinience function for getting active connections for a single user

* Lock db on these mutations

* Cleanup force disconnect using GetClientsForUser and capture client reference explicitly

* No longer re-showing banned user messages for safety. Removing this test.

* Remove no longer needed comment

* Tweaks to forbidden username handling.

- Standardize naming to not use "block" but "forbidden" instead.
- Pass array over the wire instead of string.
- Add API test
- Fix default list incorrectly being appended to custom list.

* Logging cleanup

* Update dev admin bundle

* Add an artificial delay in order to visually see message being hidden when testing

* Remove the user cache as it is a premature optimization

* When connected to chat let the user know their current user details to sync the username in the UI

* On connected send current display name back to client.
- Move name change out of chat component.
- Add additional event type constants.

* Fix broken workflow due to typo

* Troubleshoot workflow

* Bump htm from 3.0.4 to 3.1.0 in /build/javascript (#1181)

* Bump htm from 3.0.4 to 3.1.0 in /build/javascript

Bumps [htm](https://github.com/developit/htm) from 3.0.4 to 3.1.0.
- [Release notes](https://github.com/developit/htm/releases)
- [Commits](https://github.com/developit/htm/compare/3.0.4...3.1.0)

---
updated-dependencies:
- dependency-name: htm
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

* Run npm run build and update libraries

Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Gabe Kangas <gabek@real-ity.com>

* Commit updated Javascript packages

* Re-send current user info when a rejected name change takes place

* All socket writes should be through the send chan and not directly

* Seed the random generator

* Add keys and indexes to users table

* a util to generate consistent emoji markup

* console clean up

* mod tidy

* Commit updated API documentation

* Handle the max payload size of a socket message.
- Only close socket if x2 greater than the max size.
- Send the user a message if a message is too large.
- Surface the max size in bytes in the config.

* Update admin bundle

* Force all events to be sent in their own socket message and do not concatinate in a single message

* Update chat embed to register for access token

* Use a different access token for embed chat

* Update the chat message bubble background color to be bolder

* add base tag to open links in new window, closes #1220

* Support text input of :emoji: in chat (#1190)

* Initial implementation of emoji injection

* fix bookkeeping with multiple emoji

* make the emoji lookup case-insensitive

* try another solution for Caretposition

* add title to emojis

minor refactoring

* bind moji injection to InputKeyUp

* simplify the code

replace all found emojis

* inject emoji if the modifer is released earlier

* more efficient emoji tag search

* use json emoji.emoji as url

* use createEmojiMarkup()

* move emojify() to chat.js

* emojify on paste

* cleanup emoji titles in paste

* update inputText in InputKeyup

* mark emoji titles with 2*zwnj

this way paste cleanup will not interfere with text which include zwnj

* emoji should not change the inputText

* Do not show join messages when chat is offline. Closes #1224
- Show stream starting/ending messages in chat.
- When stream starts show everyone the welcome message.

* Force scrolling chat to bottom after history is populated regardless of scroll position. Closes https://github.com/owncast/owncast/issues/1222

* use maxSocketPayloadSize to calculate total bytes of message payload (#1221)

* utilize maxSocketPayloadSize from config; update chatInput to calculate based on that value instead of text value; remove usage of inputText for counting

* add a buffer to account for entire websocket payload for message char counting; trim nbsp;'s from ends of messages when calculating count

Co-authored-by: Gabe Kangas <gabek@real-ity.com>

Co-authored-by: Owncast <owncast@owncast.online>
Co-authored-by: Jannik <jannik@outlook.com>
Co-authored-by: Ginger Wong <omqmail@gmail.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Meisam <39205857+MFTabriz@users.noreply.github.com>
This commit is contained in:
Gabe Kangas
2021-07-19 19:22:29 -07:00
committed by GitHub
parent e3dc736cf4
commit b6f68628c0
88 changed files with 10691 additions and 2281 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
core/chat/chat.go
View File

@@ -2,84 +2,113 @@ package chat
import (
"errors"
"time"
"net/http"
"sort"
"github.com/owncast/owncast/core/chat/events"
"github.com/owncast/owncast/models"
log "github.com/sirupsen/logrus"
)
// Setup sets up the chat server.
func Setup(listener models.ChatListener) {
var getStatus func() models.Status
func Start(getStatusFunc func() models.Status) error {
setupPersistence()
clients := make(map[string]*Client)
addCh := make(chan *Client)
delCh := make(chan *Client)
sendAllCh := make(chan models.ChatEvent)
pingCh := make(chan models.PingMessage)
doneCh := make(chan bool)
errCh := make(chan error)
getStatus = getStatusFunc
_server = NewChat()
_server = &server{
clients,
"/entry", //hardcoded due to the UI requiring this and it is not configurable
listener,
addCh,
delCh,
sendAllCh,
pingCh,
doneCh,
errCh,
}
}
go _server.Run()
// Start starts the chat server.
func Start() error {
if _server == nil {
return errors.New("chat server is nil")
}
log.Traceln("Chat server started with max connection count of", _server.maxClientCount)
ticker := time.NewTicker(30 * time.Second)
go func() {
for range ticker.C {
_server.ping()
}
}()
_server.Listen()
return errors.New("chat server failed to start")
}
// SendMessage sends a message to all.
func SendMessage(message models.ChatEvent) {
if _server == nil {
return
}
_server.SendToAll(message)
}
// GetMessages gets all of the messages.
func GetMessages() []models.ChatEvent {
if _server == nil {
return []models.ChatEvent{}
}
return getChatHistory()
}
func GetModerationChatMessages() []models.ChatEvent {
return getChatModerationHistory()
}
func GetClient(clientID string) *Client {
l.RLock()
defer l.RUnlock()
for _, client := range _server.Clients {
if client.ClientID == clientID {
return client
}
}
return nil
}
// GetClientsForUser will return chat connections that are owned by a specific user.
func GetClientsForUser(userID string) ([]*ChatClient, error) {
clients := map[string][]*ChatClient{}
for _, client := range _server.clients {
clients[client.User.Id] = append(clients[client.User.Id], client)
}
if _, exists := clients[userID]; !exists {
return nil, errors.New("no connections for user found")
}
return clients[userID], nil
}
func GetClients() []*ChatClient {
clients := []*ChatClient{}
// Convert the keyed map to a slice.
for _, client := range _server.clients {
clients = append(clients, client)
}
sort.Slice(clients, func(i, j int) bool {
return clients[i].ConnectedAt.Before(clients[j].ConnectedAt)
})
return clients
}
func SendSystemMessage(text string, ephemeral bool) error {
message := events.SystemMessageEvent{
MessageEvent: events.MessageEvent{
Body: text,
},
}
message.SetDefaults()
message.RenderBody()
if err := Broadcast(&message); err != nil {
log.Errorln("error sending system message", err)
}
if !ephemeral {
saveEvent(message.Id, "system", message.Body, message.GetMessageType(), nil, message.Timestamp)
}
return nil
}
func SendSystemAction(text string, ephemeral bool) error {
message := events.ActionEvent{
MessageEvent: events.MessageEvent{
Body: text,
},
}
message.SetDefaults()
message.RenderBody()
if err := Broadcast(&message); err != nil {
log.Errorln("error sending system chat action")
}
if !ephemeral {
saveEvent(message.Id, "action", message.Body, message.GetMessageType(), nil, message.Timestamp)
}
return nil
}
func SendAllWelcomeMessage() {
_server.sendAllWelcomeMessage()
}
func Broadcast(event events.OutboundEvent) error {
return _server.Broadcast(event.GetBroadcastPayload())
}
func HandleClientConnection(w http.ResponseWriter, r *http.Request) {
_server.HandleClientConnection(w, r)
}
// DisconnectUser will forcefully disconnect all clients belonging to a user by ID.
func DisconnectUser(userID string) {
_server.DisconnectUser(userID)
}

190
core/chat/chatclient.go Normal file
View File

@@ -0,0 +1,190 @@
package chat
import (
"bytes"
"encoding/json"
"time"
log "github.com/sirupsen/logrus"
"golang.org/x/time/rate"
"github.com/gorilla/websocket"
"github.com/owncast/owncast/config"
"github.com/owncast/owncast/core/chat/events"
"github.com/owncast/owncast/core/user"
"github.com/owncast/owncast/geoip"
)
type ChatClient struct {
id uint
accessToken string
conn *websocket.Conn
User *user.User `json:"user"`
server *ChatServer
ipAddress string `json:"-"`
// Buffered channel of outbound messages.
send chan []byte
rateLimiter *rate.Limiter
Geo *geoip.GeoDetails `json:"geo"`
MessageCount int `json:"messageCount"`
UserAgent string `json:"userAgent"`
ConnectedAt time.Time `json:"connectedAt"`
}
type chatClientEvent struct {
data []byte
client *ChatClient
}
const (
// Time allowed to write a message to the peer.
writeWait = 10 * time.Second
// Time allowed to read the next pong message from the peer.
pongWait = 60 * time.Second
// Send pings to peer with this period. Must be less than pongWait.
pingPeriod = (pongWait * 9) / 10
// Maximum message size allowed from peer.
// Larger messages get thrown away.
// Messages > *2 the socket gets closed.
maxMessageSize = config.MaxSocketPayloadSize
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
}
var (
newline = []byte{'\n'}
space = []byte{' '}
)
func (c *ChatClient) sendConnectedClientInfo() {
payload := events.EventPayload{
"type": events.ConnectedUserInfo,
"user": c.User,
}
c.sendPayload(payload)
}
func (c *ChatClient) readPump() {
c.rateLimiter = rate.NewLimiter(0.6, 5)
defer func() {
c.close()
}()
// If somebody is sending 2x the max message size they're likely a bad actor
// and should be disconnected. Below we throw away messages > max size.
c.conn.SetReadLimit(maxMessageSize * 2)
_ = c.conn.SetReadDeadline(time.Now().Add(pongWait))
c.conn.SetPongHandler(func(string) error { _ = c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
for {
_, message, err := c.conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
c.close()
}
break
}
// Throw away messages greater than max message size.
if len(message) > maxMessageSize {
c.sendAction("Sorry, that message exceeded the maximum size and can't be delivered.")
continue
}
// Guard against floods.
if !c.passesRateLimit() {
continue
}
message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))
c.handleEvent(message)
}
}
func (c *ChatClient) writePump() {
ticker := time.NewTicker(pingPeriod)
defer func() {
ticker.Stop()
c.conn.Close()
}()
for {
select {
case message, ok := <-c.send:
_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if !ok {
// The server closed the channel.
_ = c.conn.WriteMessage(websocket.CloseMessage, []byte{})
return
}
w, err := c.conn.NextWriter(websocket.TextMessage)
if err != nil {
return
}
if _, err := w.Write(message); err != nil {
log.Debugln(err)
}
if err := w.Close(); err != nil {
return
}
case <-ticker.C:
_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}
func (c *ChatClient) handleEvent(data []byte) {
c.server.inbound <- chatClientEvent{data: data, client: c}
}
func (c *ChatClient) close() {
log.Traceln("client closed:", c.User.DisplayName, c.id, c.ipAddress)
c.conn.Close()
c.server.unregister <- c
}
func (c *ChatClient) passesRateLimit() bool {
if !c.rateLimiter.Allow() {
log.Debugln("Client", c.id, c.User.DisplayName, "has exceeded the messaging rate limiting thresholds.")
return false
}
return true
}
func (c *ChatClient) sendPayload(payload events.EventPayload) {
var data []byte
data, err := json.Marshal(payload)
if err != nil {
log.Errorln(err)
return
}
c.send <- data
}
func (c *ChatClient) sendAction(message string) {
clientMessage := events.ActionEvent{
MessageEvent: events.MessageEvent{
Body: message,
},
}
clientMessage.SetDefaults()
clientMessage.RenderBody()
c.sendPayload(clientMessage.GetBroadcastPayload())
}

241
core/chat/client.go
View File

@@ -1,241 +0,0 @@
package chat
import (
"encoding/json"
"fmt"
"io"
"time"
log "github.com/sirupsen/logrus"
"golang.org/x/net/websocket"
"github.com/owncast/owncast/geoip"
"github.com/owncast/owncast/models"
"github.com/owncast/owncast/utils"
"github.com/teris-io/shortid"
"golang.org/x/time/rate"
)
const channelBufSize = 100
//Client represents a chat client.
type Client struct {
ConnectedAt time.Time
MessageCount int
UserAgent string
IPAddress string
Username *string
ClientID string // How we identify unique viewers when counting viewer counts.
Geo *geoip.GeoDetails `json:"geo"`
Ignore bool // If set to true this will not be treated as a viewer
socketID string // How we identify a single websocket client.
ws *websocket.Conn
ch chan models.ChatEvent
pingch chan models.PingMessage
usernameChangeChannel chan models.NameChangeEvent
userJoinedChannel chan models.UserJoinedEvent
doneCh chan bool
rateLimiter *rate.Limiter
}
// NewClient creates a new chat client.
func NewClient(ws *websocket.Conn) *Client {
if ws == nil {
log.Panicln("ws cannot be nil")
}
var ignoreClient = false
for _, extraData := range ws.Config().Protocol {
if extraData == "IGNORE_CLIENT" {
ignoreClient = true
}
}
ch := make(chan models.ChatEvent, channelBufSize)
doneCh := make(chan bool)
pingch := make(chan models.PingMessage)
usernameChangeChannel := make(chan models.NameChangeEvent)
userJoinedChannel := make(chan models.UserJoinedEvent)
ipAddress := utils.GetIPAddressFromRequest(ws.Request())
userAgent := ws.Request().UserAgent()
socketID, _ := shortid.Generate()
clientID := socketID
rateLimiter := rate.NewLimiter(0.6, 5)
return &Client{time.Now(), 0, userAgent, ipAddress, nil, clientID, nil, ignoreClient, socketID, ws, ch, pingch, usernameChangeChannel, userJoinedChannel, doneCh, rateLimiter}
}
func (c *Client) write(msg models.ChatEvent) {
select {
case c.ch <- msg:
default:
_server.removeClient(c)
_server.err(fmt.Errorf("client %s is disconnected", c.ClientID))
}
}
// Listen Write and Read request via channel.
func (c *Client) listen() {
go c.listenWrite()
c.listenRead()
}
// Listen write request via channel.
func (c *Client) listenWrite() {
for {
select {
// Send a PING keepalive
case msg := <-c.pingch:
if err := websocket.JSON.Send(c.ws, msg); err != nil {
c.handleClientSocketError(err)
}
// send message to the client
case msg := <-c.ch:
if err := websocket.JSON.Send(c.ws, msg); err != nil {
c.handleClientSocketError(err)
}
case msg := <-c.usernameChangeChannel:
if err := websocket.JSON.Send(c.ws, msg); err != nil {
c.handleClientSocketError(err)
}
case msg := <-c.userJoinedChannel:
if err := websocket.JSON.Send(c.ws, msg); err != nil {
c.handleClientSocketError(err)
}
// receive done request
case <-c.doneCh:
_server.removeClient(c)
c.doneCh <- true // for listenRead method
return
}
}
}
func (c *Client) handleClientSocketError(err error) {
_server.removeClient(c)
}
func (c *Client) passesRateLimit() bool {
if !c.rateLimiter.Allow() {
log.Debugln("Client", c.ClientID, "has exceeded the messaging rate limiting thresholds.")
return false
}
return true
}
// Listen read request via channel.
func (c *Client) listenRead() {
for {
select {
// receive done request
case <-c.doneCh:
_server.remove(c)
c.doneCh <- true // for listenWrite method
return
// read data from websocket connection
default:
var data []byte
if err := websocket.Message.Receive(c.ws, &data); err != nil {
if err == io.EOF {
c.doneCh <- true
return
}
c.handleClientSocketError(err)
}
if !c.passesRateLimit() {
continue
}
var messageTypeCheck map[string]interface{}
// Bad messages should be thrown away
if err := json.Unmarshal(data, &messageTypeCheck); err != nil {
log.Debugln("Badly formatted message received from", c.Username, c.ws.Request().RemoteAddr)
continue
}
// If we can't tell the type of message, also throw it away.
if messageTypeCheck == nil {
log.Debugln("Untyped message received from", c.Username, c.ws.Request().RemoteAddr)
continue
}
messageType := messageTypeCheck["type"].(string)
if messageType == models.MessageSent {
c.chatMessageReceived(data)
} else if messageType == models.UserNameChanged {
c.userChangedName(data)
} else if messageType == models.UserJoined {
c.userJoined(data)
}
}
}
}
func (c *Client) userJoined(data []byte) {
var msg models.UserJoinedEvent
if err := json.Unmarshal(data, &msg); err != nil {
log.Errorln(err)
return
}
msg.ID = shortid.MustGenerate()
msg.Type = models.UserJoined
msg.Timestamp = time.Now()
c.Username = &msg.Username
_server.userJoined(msg)
}
func (c *Client) userChangedName(data []byte) {
var msg models.NameChangeEvent
if err := json.Unmarshal(data, &msg); err != nil {
log.Errorln(err)
}
msg.Type = models.UserNameChanged
msg.ID = shortid.MustGenerate()
_server.usernameChanged(msg)
c.Username = &msg.NewName
}
func (c *Client) chatMessageReceived(data []byte) {
var msg models.ChatEvent
if err := json.Unmarshal(data, &msg); err != nil {
log.Errorln(err)
}
msg.SetDefaults()
c.MessageCount++
c.Username = &msg.Author
msg.ClientID = c.ClientID
msg.RenderAndSanitizeMessageBody()
_server.SendToAll(msg)
}
// GetViewerClientFromChatClient returns a general models.Client from a chat websocket client.
func (c *Client) GetViewerClientFromChatClient() models.Client {
return models.Client{
ConnectedAt: c.ConnectedAt,
MessageCount: c.MessageCount,
UserAgent: c.UserAgent,
IPAddress: c.IPAddress,
Username: c.Username,
ClientID: c.ClientID,
Geo: geoip.GetGeoFromIP(c.IPAddress),
}
}

102
core/chat/events.go Normal file
View File

@@ -0,0 +1,102 @@
package chat
import (
"encoding/json"
"fmt"
"strings"
"github.com/owncast/owncast/core/chat/events"
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/core/user"
"github.com/owncast/owncast/core/webhooks"
log "github.com/sirupsen/logrus"
)
func (s *ChatServer) userNameChanged(eventData chatClientEvent) {
var receivedEvent events.NameChangeEvent
if err := json.Unmarshal(eventData.data, &receivedEvent); err != nil {
log.Errorln("error unmarshalling to NameChangeEvent", err)
return
}
proposedUsername := receivedEvent.NewName
blocklist := data.GetForbiddenUsernameList()
for _, blockedName := range blocklist {
normalizedName := strings.TrimSpace(blockedName)
normalizedName = strings.ToLower(normalizedName)
if strings.Contains(normalizedName, proposedUsername) {
// Denied.
log.Debugln(eventData.client.User.DisplayName, "blocked from changing name to", proposedUsername, "due to blocked name", normalizedName)
message := fmt.Sprintf("You cannot change your name to **%s**.", proposedUsername)
s.sendActionToClient(eventData.client, message)
// Resend the client's user so their username is in sync.
eventData.client.sendConnectedClientInfo()
return
}
}
savedUser := user.GetUserByToken(eventData.client.accessToken)
oldName := savedUser.DisplayName
// Save the new name
user.ChangeUsername(eventData.client.User.Id, receivedEvent.NewName)
// Update the connected clients associated user with the new name
eventData.client.User = savedUser
// Send chat event letting everyone about about the name change
savedUser.DisplayName = receivedEvent.NewName
broadcastEvent := events.NameChangeBroadcast{
Oldname: oldName,
}
broadcastEvent.User = savedUser
broadcastEvent.SetDefaults()
payload := broadcastEvent.GetBroadcastPayload()
if err := s.Broadcast(payload); err != nil {
log.Errorln("error broadcasting NameChangeEvent", err)
return
}
// Send chat user name changed webhook
receivedEvent.User = savedUser
webhooks.SendChatEventUsernameChanged(receivedEvent)
}
func (s *ChatServer) userMessageSent(eventData chatClientEvent) {
var event events.UserMessageEvent
if err := json.Unmarshal(eventData.data, &event); err != nil {
log.Errorln("error unmarshalling to UserMessageEvent", err)
return
}
event.SetDefaults()
// Ignore empty messages
if event.Empty() {
return
}
event.User = user.GetUserByToken(eventData.client.accessToken)
// Guard against nil users
if event.User == nil {
return
}
payload := event.GetBroadcastPayload()
if err := s.Broadcast(payload); err != nil {
log.Errorln("error broadcasting UserMessageEvent payload", err)
return
}
// Send chat message sent webhook
webhooks.SendChatEvent(&event)
SaveUserMessage(event)
eventData.client.MessageCount = eventData.client.MessageCount + 1
}

20
core/chat/events/actionEvent.go Normal file
View File

@@ -0,0 +1,20 @@
package events
type ActionEvent struct {
Event
MessageEvent
}
// ActionEvent will return the object to send to all chat users.
func (e *ActionEvent) GetBroadcastPayload() EventPayload {
return EventPayload{
"id": e.Id,
"timestamp": e.Timestamp,
"body": e.Body,
"type": e.GetMessageType(),
}
}
func (e *ActionEvent) GetMessageType() EventType {
return ChatActionSent
}

156
core/chat/events/events.go Normal file
View File

@@ -0,0 +1,156 @@
package events
import (
"bytes"
"regexp"
"strings"
"time"
"github.com/microcosm-cc/bluemonday"
"github.com/teris-io/shortid"
"github.com/yuin/goldmark"
"github.com/yuin/goldmark/extension"
"github.com/yuin/goldmark/renderer/html"
"mvdan.cc/xurls"
"github.com/owncast/owncast/core/user"
log "github.com/sirupsen/logrus"
)
// EventPayload is a generic key/value map for sending out to chat clients.
type EventPayload map[string]interface{}
type OutboundEvent interface {
GetBroadcastPayload() EventPayload
GetMessageType() EventType
}
// Event is any kind of event. A type is required to be specified.
type Event struct {
Type EventType `json:"type"`
Id string `json:"id"`
Timestamp time.Time `json:"timestamp"`
}
type UserEvent struct {
User *user.User `json:"user"`
HiddenAt *time.Time `json:"hiddenAt,omitempty"`
}
// MessageEvent is an event that has a message body.
type MessageEvent struct {
OutboundEvent `json:"-"`
Body string `json:"body"`
RawBody string `json:"-"`
}
type SystemActionEvent struct {
Event
MessageEvent
}
// SetDefaults will set default properties of all inbound events.
func (e *Event) SetDefaults() {
e.Id = shortid.MustGenerate()
e.Timestamp = time.Now()
}
// SetDefaults will set default properties of all inbound events.
func (e *UserMessageEvent) SetDefaults() {
e.Id = shortid.MustGenerate()
e.Timestamp = time.Now()
e.RenderAndSanitizeMessageBody()
}
// RenderAndSanitizeMessageBody will turn markdown into HTML, sanitize raw user-supplied HTML and standardize
// the message into something safe and renderable for clients.
func (m *MessageEvent) RenderAndSanitizeMessageBody() {
m.RawBody = m.Body
// Set the new, sanitized and rendered message body
m.Body = RenderAndSanitize(m.RawBody)
}
// Empty will return if this message's contents is empty.
func (m *MessageEvent) Empty() bool {
return m.Body == ""
}
// RenderBody will render markdown to html without any sanitization.
func (m *MessageEvent) RenderBody() {
m.RawBody = m.Body
m.Body = RenderMarkdown(m.RawBody)
}
// RenderAndSanitize will turn markdown into HTML, sanitize raw user-supplied HTML and standardize
// the message into something safe and renderable for clients.
func RenderAndSanitize(raw string) string {
rendered := RenderMarkdown(raw)
safe := sanitize(rendered)
// Set the new, sanitized and rendered message body
return strings.TrimSpace(safe)
}
func RenderMarkdown(raw string) string {
markdown := goldmark.New(
goldmark.WithRendererOptions(
html.WithUnsafe(),
),
goldmark.WithExtensions(
extension.NewLinkify(
extension.WithLinkifyAllowedProtocols([][]byte{
[]byte("http:"),
[]byte("https:"),
}),
extension.WithLinkifyURLRegexp(
xurls.Strict,
),
),
),
)
trimmed := strings.TrimSpace(raw)
var buf bytes.Buffer
if err := markdown.Convert([]byte(trimmed), &buf); err != nil {
log.Debugln(err)
}
return buf.String()
}
func sanitize(raw string) string {
p := bluemonday.StrictPolicy()
// Require URLs to be parseable by net/url.Parse
p.AllowStandardURLs()
p.RequireParseableURLs(true)
// Allow links
p.AllowAttrs("href").OnElements("a")
// Force all URLs to have "noreferrer" in their rel attribute.
p.RequireNoReferrerOnLinks(true)
// Links will get target="_blank" added to them.
p.AddTargetBlankToFullyQualifiedLinks(true)
// Allow breaks
p.AllowElements("br", "p")
// Allow img tags from the the local emoji directory only
p.AllowAttrs("src", "alt", "class", "title").Matching(regexp.MustCompile(`(?i)/img/emoji`)).OnElements("img")
p.AllowAttrs("class").OnElements("img")
// Allow bold
p.AllowElements("strong")
// Allow emphasis
p.AllowElements("em")
// Allow code blocks
p.AllowElements("code")
p.AllowElements("pre")
return p.Sanitize(raw)
}

34
core/chat/events/eventtype.go Normal file
View File

@@ -0,0 +1,34 @@
package events
// EventType is the type of a websocket event.
type EventType = string
const (
// MessageSent is the event sent when a chat event takes place.
MessageSent EventType = "CHAT"
// UserJoined is the event sent when a chat user join action takes place.
UserJoined EventType = "USER_JOINED"
// UserNameChanged is the event sent when a chat username change takes place.
UserNameChanged EventType = "NAME_CHANGE"
// VisibiltyToggled is the event sent when a chat message's visibility changes.
VisibiltyToggled EventType = "VISIBILITY-UPDATE"
// PING is a ping message.
PING EventType = "PING"
// PONG is a pong message.
PONG EventType = "PONG"
// StreamStarted represents a stream started event.
StreamStarted EventType = "STREAM_STARTED"
// StreamStopped represents a stream stopped event.
StreamStopped EventType = "STREAM_STOPPED"
// SystemMessageSent is the event sent when a system message is sent.
SystemMessageSent EventType = "SYSTEM"
// ChatDisabled is when a user is explicitly disabled and blocked from using chat.
ChatDisabled EventType = "CHAT_DISABLED"
// ConnectedUserInfo is a private event to a user letting them know their user details.
ConnectedUserInfo EventType = "CONNECTED_USER_INFO"
// ChatActionSent is a generic chat action that can be used for anything that doesn't need specific handling or formatting.
ChatActionSent EventType = "CHAT_ACTION"
ErrorNeedsRegistration EventType = "ERROR_NEEDS_REGISTRATION"
ErrorMaxConnectionsExceeded EventType = "ERROR_MAX_CONNECTIONS_EXCEEDED"
ErrorUserDisabled EventType = "ERROR_USER_DISABLED"
)

26
core/chat/events/nameChangeEvent.go Normal file
View File

@@ -0,0 +1,26 @@
package events
// NameChangeEvent is received when a user changes their chat display name.
type NameChangeEvent struct {
Event
UserEvent
NewName string `json:"newName"`
}
// NameChangeEventBroadcast is fired when a user changes their chat display name.
type NameChangeBroadcast struct {
Event
UserEvent
Oldname string `json:"oldName"`
}
// GetBroadcastPayload will return the object to send to all chat users.
func (e *NameChangeBroadcast) GetBroadcastPayload() EventPayload {
return EventPayload{
"id": e.Id,
"timestamp": e.Timestamp,
"user": e.User,
"oldName": e.Oldname,
"type": UserNameChanged,
}
}

26
core/chat/events/systemMessageEvent.go Normal file
View File

@@ -0,0 +1,26 @@
package events
import "github.com/owncast/owncast/core/data"
// SystemMessageEvent is a message displayed in chat on behalf of the server.
type SystemMessageEvent struct {
Event
MessageEvent
}
// SystemMessageEvent will return the object to send to all chat users.
func (e *SystemMessageEvent) GetBroadcastPayload() EventPayload {
return EventPayload{
"id": e.Id,
"timestamp": e.Timestamp,
"body": e.Body,
"type": SystemMessageSent,
"user": EventPayload{
"displayName": data.GetServerName(),
},
}
}
func (e *SystemMessageEvent) GetMessageType() EventType {
return SystemMessageSent
}

17
core/chat/events/userDisabledEvent.go Normal file
View File

@@ -0,0 +1,17 @@
package events
// UserDisabledEvent is the event fired when a user is banned/blocked and disconnected from chat.
type UserDisabledEvent struct {
Event
UserEvent
}
// GetBroadcastPayload will return the object to send to all chat users.
func (e *UserDisabledEvent) GetBroadcastPayload() EventPayload {
return EventPayload{
"type": ErrorUserDisabled,
"id": e.Id,
"timestamp": e.Timestamp,
"user": e.User,
}
}

17
core/chat/events/userJoinedEvent.go Normal file
View File

@@ -0,0 +1,17 @@
package events
// UserJoinedEvent is the event fired when a user joins chat.
type UserJoinedEvent struct {
Event
UserEvent
}
// GetBroadcastPayload will return the object to send to all chat users.
func (e *UserJoinedEvent) GetBroadcastPayload() EventPayload {
return EventPayload{
"type": UserJoined,
"id": e.Id,
"timestamp": e.Timestamp,
"user": e.User,
}
}

24
core/chat/events/userMessageEvent.go Normal file
View File

@@ -0,0 +1,24 @@
package events
// UserMessageEvent is an inbound message from a user.
type UserMessageEvent struct {
Event
UserEvent
MessageEvent
}
// GetBroadcastPayload will return the object to send to all chat users.
func (e *UserMessageEvent) GetBroadcastPayload() EventPayload {
return EventPayload{
"id": e.Id,
"timestamp": e.Timestamp,
"body": e.Body,
"user": e.User,
"type": MessageSent,
"visible": e.HiddenAt == nil,
}
}
func (e *UserMessageEvent) GetMessageType() EventType {
return MessageSent
}

10
core/chat/messageRendering_test.go
View File

@@ -3,7 +3,7 @@ package chat
import (
"testing"
"github.com/owncast/owncast/models"
"github.com/owncast/owncast/core/chat/events"
)
// Test a bunch of arbitrary markup and markdown to make sure we get sanitized
@@ -25,7 +25,7 @@ blah blah blah
<p><a href="http://owncast.online" rel="nofollow noreferrer noopener" target="_blank">test link</a>
<img class="emoji" src="/img/emoji/bananadance.gif"></p>`
result := models.RenderAndSanitize(messageContent)
result := events.RenderAndSanitize(messageContent)
if result != expected {
t.Errorf("message rendering/sanitation does not match expected. Got\n%s, \n\n want:\n%s", result, expected)
}
@@ -35,7 +35,7 @@ blah blah blah
func TestBlockRemoteImages(t *testing.T) {
messageContent := `<img src="https://via.placeholder.com/350x150"> test ![](https://via.placeholder.com/350x150)`
expected := `<p> test </p>`
result := models.RenderAndSanitize(messageContent)
result := events.RenderAndSanitize(messageContent)
if result != expected {
t.Errorf("message rendering/sanitation does not match expected. Got\n%s, \n\n want:\n%s", result, expected)
@@ -46,7 +46,7 @@ func TestBlockRemoteImages(t *testing.T) {
func TestAllowEmojiImages(t *testing.T) {
messageContent := `<img src="/img/emoji/beerparrot.gif"> test ![](/img/emoji/beerparrot.gif)`
expected := `<p><img src="/img/emoji/beerparrot.gif"> test <img src="/img/emoji/beerparrot.gif"></p>`
result := models.RenderAndSanitize(messageContent)
result := events.RenderAndSanitize(messageContent)
if result != expected {
t.Errorf("message rendering/sanitation does not match expected. Got\n%s, \n\n want:\n%s", result, expected)
@@ -57,7 +57,7 @@ func TestAllowEmojiImages(t *testing.T) {
func TestAllowHTML(t *testing.T) {
messageContent := `<img src="/img/emoji/beerparrot.gif"><ul><li>**test thing**</li></ul>`
expected := "<p><img src=\"/img/emoji/beerparrot.gif\"><ul><li><strong>test thing</strong></li></ul></p>\n"
result := models.RenderMarkdown(messageContent)
result := events.RenderMarkdown(messageContent)
if result != expected {
t.Errorf("message rendering does not match expected. Got\n%s, \n\n want:\n%s", result, expected)

9
core/chat/messages.go
View File

@@ -1,8 +1,8 @@
package chat
import (
"github.com/owncast/owncast/core/chat/events"
"github.com/owncast/owncast/core/webhooks"
"github.com/owncast/owncast/models"
log "github.com/sirupsen/logrus"
)
@@ -22,8 +22,11 @@ func SetMessagesVisibility(messageIDs []string, visibility bool) error {
log.Errorln(err)
continue
}
message.MessageType = models.VisibiltyToggled
_server.sendAll(message)
payload := message.GetBroadcastPayload()
payload["type"] = events.VisibiltyToggled
if err := _server.Broadcast(payload); err != nil {
log.Debugln(err)
}
go webhooks.SendChatEvent(message)
}

280
core/chat/persistence.go
View File

@@ -1,171 +1,309 @@
package chat
import (
"database/sql"
"fmt"
"strings"
"time"
_ "github.com/mattn/go-sqlite3"
"github.com/owncast/owncast/core/chat/events"
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/core/user"
"github.com/owncast/owncast/models"
log "github.com/sirupsen/logrus"
)
var _db *sql.DB
var _datastore *data.Datastore
const (
maxBacklogHours = 5 // Keep backlog max hours worth of messages
maxBacklogNumber = 50 // Return max number of messages in history request
)
func setupPersistence() {
_db = data.GetDatabase()
createTable()
_datastore = data.GetDatastore()
createMessagesTable()
chatDataPruner := time.NewTicker(5 * time.Minute)
go func() {
runPruner()
for range chatDataPruner.C {
runPruner()
}
}()
}
func createTable() {
func createMessagesTable() {
createTableSQL := `CREATE TABLE IF NOT EXISTS messages (
"id" string NOT NULL PRIMARY KEY,
"author" TEXT,
"user_id" INTEGER,
"body" TEXT,
"messageType" TEXT,
"visible" INTEGER,
"timestamp" DATE
"eventType" TEXT,
"hidden_at" DATETIME,
"timestamp" DATETIME
);`
stmt, err := _db.Prepare(createTableSQL)
stmt, err := _datastore.DB.Prepare(createTableSQL)
if err != nil {
log.Fatal(err)
log.Fatal("error creating chat messages table", err)
}
defer stmt.Close()
if _, err := stmt.Exec(); err != nil {
log.Warnln(err)
log.Fatal("error creating chat messages table", err)
}
}
func addMessage(message models.ChatEvent) {
tx, err := _db.Begin()
if err != nil {
log.Fatal(err)
}
stmt, err := tx.Prepare("INSERT INTO messages(id, author, body, messageType, visible, timestamp) values(?, ?, ?, ?, ?, ?)")
func SaveUserMessage(event events.UserMessageEvent) {
saveEvent(event.Id, event.User.Id, event.Body, event.Type, event.HiddenAt, event.Timestamp)
}
func saveEvent(id string, userId string, body string, eventType string, hidden *time.Time, timestamp time.Time) {
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
tx, err := _datastore.DB.Begin()
if err != nil {
log.Fatal(err)
log.Errorln("error saving", eventType, err)
return
}
defer tx.Rollback() // nolint
stmt, err := tx.Prepare("INSERT INTO messages(id, user_id, body, eventType, hidden_at, timestamp) values(?, ?, ?, ?, ?, ?)")
if err != nil {
log.Errorln("error saving", eventType, err)
return
}
defer stmt.Close()
if _, err := stmt.Exec(message.ID, message.Author, message.Body, message.MessageType, 1, message.Timestamp); err != nil {
log.Fatal(err)
if _, err = stmt.Exec(id, userId, body, eventType, hidden, timestamp); err != nil {
log.Errorln("error saving", eventType, err)
return
}
if err := tx.Commit(); err != nil {
log.Fatal(err)
if err = tx.Commit(); err != nil {
log.Errorln("error saving", eventType, err)
return
}
}
func getChat(query string) []models.ChatEvent {
history := make([]models.ChatEvent, 0)
rows, err := _db.Query(query)
func getChat(query string) []events.UserMessageEvent {
history := make([]events.UserMessageEvent, 0)
rows, err := _datastore.DB.Query(query)
if err != nil {
log.Fatal(err)
log.Errorln("error fetching chat history", err)
return history
}
defer rows.Close()
for rows.Next() {
var id string
var author string
var userId string
var body string
var messageType models.EventType
var visible int
var hiddenAt *time.Time
var timestamp time.Time
err = rows.Scan(&id, &author, &body, &messageType, &visible, &timestamp)
var userDisplayName *string
var userDisplayColor *int
var userCreatedAt *time.Time
var userDisabledAt *time.Time
var previousUsernames *string
var userNameChangedAt *time.Time
// Convert a database row into a chat event
err = rows.Scan(&id, &userId, &body, &messageType, &hiddenAt, &timestamp, &userDisplayName, &userDisplayColor, &userCreatedAt, &userDisabledAt, &previousUsernames, &userNameChangedAt)
if err != nil {
log.Debugln(err)
log.Error("There is a problem with the chat database. Restore a backup of owncast.db or remove it and start over.")
log.Errorln("There is a problem converting query to chat objects. Please report this:", query)
break
}
message := models.ChatEvent{}
message.ID = id
message.Author = author
message.Body = body
message.MessageType = messageType
message.Visible = visible == 1
message.Timestamp = timestamp
// System messages and chat actions are special and are not from real users
if messageType == events.SystemMessageSent || messageType == events.ChatActionSent {
name := "Owncast"
userDisplayName = &name
color := 200
userDisplayColor = &color
}
if previousUsernames == nil {
previousUsernames = userDisplayName
}
if userCreatedAt == nil {
now := time.Now()
userCreatedAt = &now
}
user := user.User{
Id: userId,
AccessToken: "",
DisplayName: *userDisplayName,
DisplayColor: *userDisplayColor,
CreatedAt: *userCreatedAt,
DisabledAt: userDisabledAt,
NameChangedAt: userNameChangedAt,
PreviousNames: strings.Split(*previousUsernames, ","),
}
message := events.UserMessageEvent{
Event: events.Event{
Type: messageType,
Id: id,
Timestamp: timestamp,
},
UserEvent: events.UserEvent{
User: &user,
HiddenAt: hiddenAt,
},
MessageEvent: events.MessageEvent{
Body: body,
RawBody: body,
},
}
history = append(history, message)
}
if err := rows.Err(); err != nil {
log.Fatal(err)
}
return history
}
func getChatModerationHistory() []models.ChatEvent {
var query = "SELECT * FROM messages WHERE messageType == 'CHAT' AND datetime(timestamp) >=datetime('now', '-5 Hour')"
func GetChatModerationHistory() []events.UserMessageEvent {
// Get all messages regardless of visibility
var query = "SELECT messages.id, user_id, body, eventType, hidden_at, timestamp, display_name, display_color, created_at, disabled_at, previous_names, namechanged_at FROM messages INNER JOIN users ON messages.user_id = users.id ORDER BY timestamp DESC"
return getChat(query)
}
func getChatHistory() []models.ChatEvent {
// Get all messages sent within the past 5hrs, max 50
var query = "SELECT * FROM (SELECT * FROM messages WHERE datetime(timestamp) >=datetime('now', '-5 Hour') AND visible = 1 ORDER BY timestamp DESC LIMIT 50) ORDER BY timestamp asc"
func GetChatHistory() []events.UserMessageEvent {
// Get all visible messages
var query = fmt.Sprintf("SELECT id, user_id, body, eventType, hidden_at, timestamp, display_name, display_color, created_at, disabled_at, previous_names, namechanged_at FROM (SELECT * FROM messages LEFT OUTER JOIN users ON messages.user_id = users.id WHERE hidden_at IS NULL ORDER BY timestamp DESC LIMIT %d) ORDER BY timestamp asc", maxBacklogNumber)
return getChat(query)
}
// SetMessageVisibilityForUserId will bulk change the visibility of messages for a user
// and then send out visibility changed events to chat clients.
func SetMessageVisibilityForUserId(userID string, visible bool) error {
// Get a list of IDs from this user within the 5hr window to send to the connected clients to hide
ids := make([]string, 0)
query := fmt.Sprintf("SELECT messages.id, user_id, body, eventType, hidden_at, timestamp, display_name, display_color, created_at, disabled_at, previous_names, namechanged_at FROM messages INNER JOIN users ON messages.user_id = users.id WHERE user_id IS '%s'", userID)
messages := getChat(query)
if len(messages) == 0 {
return nil
}
for _, message := range messages {
ids = append(ids, message.Id)
}
// Tell the clients to hide/show these messages.
return SetMessagesVisibility(ids, visible)
}
func saveMessageVisibility(messageIDs []string, visible bool) error {
tx, err := _db.Begin()
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
tx, err := _datastore.DB.Begin()
if err != nil {
log.Fatal(err)
return err
}
stmt, err := tx.Prepare("UPDATE messages SET visible=? WHERE id IN (?" + strings.Repeat(",?", len(messageIDs)-1) + ")")
stmt, err := tx.Prepare("UPDATE messages SET hidden_at=? WHERE id IN (?" + strings.Repeat(",?", len(messageIDs)-1) + ")")
if err != nil {
log.Fatal(err)
return err
}
defer stmt.Close()
var hiddenAt *time.Time
if !visible {
now := time.Now()
hiddenAt = &now
} else {
hiddenAt = nil
}
args := make([]interface{}, len(messageIDs)+1)
args[0] = visible
args[0] = hiddenAt
for i, id := range messageIDs {
args[i+1] = id
}
if _, err := stmt.Exec(args...); err != nil {
log.Fatal(err)
if _, err = stmt.Exec(args...); err != nil {
return err
}
if err := tx.Commit(); err != nil {
log.Fatal(err)
if err = tx.Commit(); err != nil {
return err
}
return nil
}
func getMessageById(messageID string) (models.ChatEvent, error) {
func getMessageById(messageID string) (*events.UserMessageEvent, error) {
var query = "SELECT * FROM messages WHERE id = ?"
row := _db.QueryRow(query, messageID)
row := _datastore.DB.QueryRow(query, messageID)
var id string
var author string
var userId string
var body string
var messageType models.EventType
var visible int
var eventType models.EventType
var hiddenAt *time.Time
var timestamp time.Time
err := row.Scan(&id, &author, &body, &messageType, &visible, &timestamp)
err := row.Scan(&id, &userId, &body, &eventType, &hiddenAt, &timestamp)
if err != nil {
log.Errorln(err)
return models.ChatEvent{}, err
return nil, err
}
return models.ChatEvent{
ID: id,
Author: author,
Body: body,
MessageType: messageType,
Visible: visible == 1,
Timestamp: timestamp,
user := user.GetUserById(userId)
return &events.UserMessageEvent{
events.Event{
Type: eventType,
Id: id,
Timestamp: timestamp,
},
events.UserEvent{
User: user,
HiddenAt: hiddenAt,
},
events.MessageEvent{
Body: body,
},
}, nil
}
// Only keep recent messages so we don't keep more chat data than needed
// for privacy and efficiency reasons.
func runPruner() {
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
log.Traceln("Removing chat messages older than", maxBacklogHours, "hours")
deleteStatement := `DELETE FROM messages WHERE timestamp <= datetime('now', 'localtime', ?)`
tx, err := _datastore.DB.Begin()
if err != nil {
log.Debugln(err)
return
}
stmt, err := tx.Prepare(deleteStatement)
if err != nil {
log.Debugln(err)
return
}
defer stmt.Close()
if _, err = stmt.Exec(fmt.Sprintf("-%d hours", maxBacklogHours)); err != nil {
log.Debugln(err)
return
}
if err = tx.Commit(); err != nil {
log.Debugln(err)
return
}
}

436
core/chat/server.go
View File

@@ -1,191 +1,317 @@
package chat
import (
"fmt"
"encoding/json"
"net/http"
"sync"
"time"
log "github.com/sirupsen/logrus"
"golang.org/x/net/websocket"
"github.com/gorilla/websocket"
"github.com/owncast/owncast/core/chat/events"
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/core/user"
"github.com/owncast/owncast/core/webhooks"
"github.com/owncast/owncast/models"
"github.com/owncast/owncast/utils"
)
var (
_server *server
)
var _server *ChatServer
var l = &sync.RWMutex{}
type ChatServer struct {
mu sync.RWMutex
seq uint
clients map[uint]*ChatClient
maxClientCount uint
// Server represents the server which handles the chat.
type server struct {
Clients map[string]*Client
// send outbound message payload to all clients
outbound chan []byte
pattern string
listener models.ChatListener
// receive inbound message payload from all clients
inbound chan chatClientEvent
addCh chan *Client
delCh chan *Client
sendAllCh chan models.ChatEvent
pingCh chan models.PingMessage
doneCh chan bool
errCh chan error
// unregister requests from clients.
unregister chan *ChatClient
}
// Add adds a client to the server.
func (s *server) add(c *Client) {
s.addCh <- c
}
// Remove removes a client from the server.
func (s *server) remove(c *Client) {
s.delCh <- c
}
// SendToAll sends a message to all of the connected clients.
func (s *server) SendToAll(msg models.ChatEvent) {
s.sendAllCh <- msg
}
// Err handles an error.
func (s *server) err(err error) {
s.errCh <- err
}
func (s *server) sendAll(msg models.ChatEvent) {
l.RLock()
for _, c := range s.Clients {
c.write(msg)
func NewChat() *ChatServer {
server := &ChatServer{
clients: map[uint]*ChatClient{},
outbound: make(chan []byte),
inbound: make(chan chatClientEvent),
unregister: make(chan *ChatClient),
maxClientCount: handleMaxConnectionCount(),
}
l.RUnlock()
return server
}
func (s *server) ping() {
ping := models.PingMessage{MessageType: models.PING}
l.RLock()
for _, c := range s.Clients {
c.pingch <- ping
}
l.RUnlock()
}
func (s *server) usernameChanged(msg models.NameChangeEvent) {
l.RLock()
for _, c := range s.Clients {
c.usernameChangeChannel <- msg
}
l.RUnlock()
go webhooks.SendChatEventUsernameChanged(msg)
}
func (s *server) userJoined(msg models.UserJoinedEvent) {
l.RLock()
if s.listener.IsStreamConnected() {
for _, c := range s.Clients {
c.userJoinedChannel <- msg
}
}
l.RUnlock()
go webhooks.SendChatEventUserJoined(msg)
}
func (s *server) onConnection(ws *websocket.Conn) {
client := NewClient(ws)
defer func() {
s.removeClient(client)
if err := ws.Close(); err != nil {
log.Debugln(err)
//s.errCh <- err
}
}()
s.add(client)
client.listen()
}
// Listen and serve.
// It serves client connection and broadcast request.
func (s *server) Listen() {
http.Handle(s.pattern, websocket.Handler(s.onConnection))
log.Tracef("Starting the websocket listener on: %s", s.pattern)
func (s *ChatServer) Run() {
for {
select {
// add new a client
case c := <-s.addCh:
l.Lock()
s.Clients[c.socketID] = c
if !c.Ignore {
s.listener.ClientAdded(c.GetViewerClientFromChatClient())
s.sendWelcomeMessageToClient(c)
}
l.Unlock()
// remove a client
case c := <-s.delCh:
s.removeClient(c)
case msg := <-s.sendAllCh:
if data.GetChatDisabled() {
break
case client := <-s.unregister:
if _, ok := s.clients[client.id]; ok {
s.mu.Lock()
delete(s.clients, client.id)
close(client.send)
s.mu.Unlock()
}
if !msg.Empty() {
// set defaults before sending msg to anywhere
msg.SetDefaults()
s.listener.MessageSent(msg)
s.sendAll(msg)
// Store in the message history
if !msg.Ephemeral {
addMessage(msg)
}
// Send webhooks
go webhooks.SendChatEvent(msg)
}
case ping := <-s.pingCh:
fmt.Println("PING?", ping)
case err := <-s.errCh:
log.Trace("Error: ", err.Error())
case <-s.doneCh:
return
case message := <-s.inbound:
s.eventReceived(message)
}
}
}
func (s *server) removeClient(c *Client) {
l.Lock()
if _, ok := s.Clients[c.socketID]; ok {
delete(s.Clients, c.socketID)
s.listener.ClientRemoved(c.socketID)
log.Tracef("The client was connected for %s and sent %d messages (%s)", time.Since(c.ConnectedAt), c.MessageCount, c.ClientID)
// Addclient registers new connection as a User.
func (s *ChatServer) Addclient(conn *websocket.Conn, user *user.User, accessToken string, userAgent string) *ChatClient {
client := &ChatClient{
server: s,
conn: conn,
User: user,
ipAddress: conn.RemoteAddr().String(),
accessToken: accessToken,
send: make(chan []byte, 256),
UserAgent: userAgent,
ConnectedAt: time.Now(),
}
l.Unlock()
s.mu.Lock()
{
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 {
s.sendUserJoinedMessage(client)
s.sendWelcomeMessageToClient(client)
}
return client
}
func (s *server) sendWelcomeMessageToClient(c *Client) {
go func() {
// Add an artificial delay so people notice this message come in.
time.Sleep(7 * time.Second)
func (s *ChatServer) sendUserJoinedMessage(c *ChatClient) {
userJoinedEvent := events.UserJoinedEvent{}
userJoinedEvent.SetDefaults()
userJoinedEvent.User = c.User
welcomeMessage := data.GetServerWelcomeMessage()
if welcomeMessage != "" {
initialMessage := models.ChatEvent{ClientID: "owncast-server", Author: data.GetServerName(), Body: welcomeMessage, ID: "initial-message-1", MessageType: "SYSTEM", Visible: true, Timestamp: time.Now()}
c.write(initialMessage)
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 *ChatServer) ClientClosed(c *ChatClient) {
s.mu.Lock()
defer s.mu.Unlock()
c.close()
if _, ok := s.clients[c.id]; ok {
log.Debugln("Deleting", c.id)
delete(s.clients, c.id)
}
}
func (s *ChatServer) HandleClientConnection(w http.ResponseWriter, r *http.Request) {
if data.GetChatDisabled() {
_, _ = w.Write([]byte(events.ChatDisabled))
return
}
// Limit concurrent chat connections
if uint(len(s.clients)) >= s.maxClientCount {
log.Warnln("rejecting incoming client connection as it exceeds the max client count of", s.maxClientCount)
_, _ = w.Write([]byte(events.ErrorMaxConnectionsExceeded))
return
}
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
}
// 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
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)
}
// Broadcast sends message to all connected clients.
func (s *ChatServer) Broadcast(payload events.EventPayload) error {
data, err := json.Marshal(payload)
if err != nil {
return err
}
s.mu.Lock()
defer s.mu.Unlock()
for _, client := range s.clients {
if client == nil {
continue
}
}()
select {
case client.send <- data:
default:
close(client.send)
delete(s.clients, client.id)
}
}
return nil
}
func (s *ChatServer) Send(payload events.EventPayload, client *ChatClient) {
data, err := json.Marshal(payload)
if err != nil {
log.Errorln(err)
return
}
client.send <- data
}
// DisconnectUser will forcefully disconnect all clients belonging to a user by ID.
func (s *ChatServer) DisconnectUser(userID string) {
s.mu.Lock()
clients, err := GetClientsForUser(userID)
s.mu.Unlock()
if err != nil || clients == nil || len(clients) == 0 {
log.Debugln("Requested to disconnect user", userID, err)
return
}
for _, client := range clients {
log.Traceln("Disconnecting client", client.User.Id, "owned by", client.User.DisplayName)
go func(client *ChatClient) {
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)
}
}
func (s *ChatServer) eventReceived(event chatClientEvent) {
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)
default:
log.Debugln(eventType, "event not found:", typecheck)
}
}
func (s *ChatServer) sendWelcomeMessageToClient(c *ChatClient) {
// 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 *ChatServer) 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 *ChatServer) sendSystemMessageToClient(c *ChatClient, message string) {
clientMessage := events.SystemMessageEvent{
Event: events.Event{},
MessageEvent: events.MessageEvent{
Body: message,
},
}
clientMessage.SetDefaults()
s.Send(clientMessage.GetBroadcastPayload(), c)
}
func (s *ChatServer) sendActionToClient(c *ChatClient, message string) {
clientMessage := events.ActionEvent{
MessageEvent: events.MessageEvent{
Body: message,
},
}
clientMessage.SetDefaults()
clientMessage.RenderBody()
s.Send(clientMessage.GetBroadcastPayload(), c)
}

29
core/chat/utils.go Normal file
View File

@@ -0,0 +1,29 @@
package chat
import (
"syscall"
log "github.com/sirupsen/logrus"
)
// Set the soft file handler limit as 70% of
// the max as the client connection limit.
func handleMaxConnectionCount() uint {
var rLimit syscall.Rlimit
if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit); err != nil {
panic(err)
}
originalLimit := rLimit.Cur
// Set the limit to 70% of max so the machine doesn't die even if it's maxed out for some reason.
proposedLimit := int(float32(rLimit.Max) * 0.7)
rLimit.Cur = uint64(proposedLimit)
if err := syscall.Setrlimit(syscall.RLIMIT_NOFILE, &rLimit); err != nil {
panic(err)
}
log.Traceln("Max process connection count increased from", originalLimit, "to", proposedLimit)
return uint(float32(rLimit.Cur))
}

44
core/chatListener.go
View File

@@ -1,44 +0,0 @@
package core
import (
"github.com/owncast/owncast/core/chat"
"github.com/owncast/owncast/models"
)
// ChatListenerImpl the implementation of the chat client.
type ChatListenerImpl struct{}
// ClientAdded is for when a client is added the system.
func (cl ChatListenerImpl) ClientAdded(client models.Client) {
SetChatClientActive(client)
}
// ClientRemoved is for when a client disconnects/is removed.
func (cl ChatListenerImpl) ClientRemoved(clientID string) {
RemoveChatClient(clientID)
}
// MessageSent is for when a message is sent.
func (cl ChatListenerImpl) MessageSent(message models.ChatEvent) {
}
// IsStreamConnected will return if the stream is connected.
func (cl ChatListenerImpl) IsStreamConnected() bool {
return IsStreamConnected()
}
// SendMessageToChat sends a message to the chat server.
func SendMessageToChat(message models.ChatEvent) error {
chat.SendMessage(message)
return nil
}
// GetAllChatMessages gets all of the chat messages.
func GetAllChatMessages() []models.ChatEvent {
return chat.GetMessages()
}
func GetModerationChatMessages() []models.ChatEvent {
return chat.GetModerationChatMessages()
}

7
core/core.go
View File

@@ -12,6 +12,7 @@ import (
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/core/rtmp"
"github.com/owncast/owncast/core/transcoder"
"github.com/owncast/owncast/core/user"
"github.com/owncast/owncast/models"
"github.com/owncast/owncast/utils"
"github.com/owncast/owncast/yp"
@@ -53,6 +54,8 @@ func Start() error {
log.Errorln("storage error", err)
}
user.SetupUsers()
fileWriter.SetupFileWriterReceiverService(&handler)
if err := createInitialOfflineState(); err != nil {
@@ -62,7 +65,9 @@ func Start() error {
_yp = yp.NewYP(GetStatus)
chat.Setup(ChatListenerImpl{})
if err := chat.Start(GetStatus); err != nil {
log.Errorln(err)
}
// start the rtmp server
go rtmp.Start(setStreamAsConnected, setBroadcaster)

198
core/data/accessTokens.go
View File

@@ -1,198 +0,0 @@
package data
import (
"errors"
"strings"
"time"
"github.com/owncast/owncast/models"
log "github.com/sirupsen/logrus"
)
func createAccessTokensTable() {
log.Traceln("Creating access_tokens table...")
createTableSQL := `CREATE TABLE IF NOT EXISTS access_tokens (
"token" string NOT NULL PRIMARY KEY,
"name" string,
"scopes" TEXT,
"timestamp" DATETIME DEFAULT CURRENT_TIMESTAMP,
"last_used" DATETIME
);`
stmt, err := _db.Prepare(createTableSQL)
if err != nil {
log.Fatal(err)
}
defer stmt.Close()
if _, err := stmt.Exec(); err != nil {
log.Warnln(err)
}
}
// InsertToken will add a new token to the database.
func InsertToken(token string, name string, scopes []string) error {
log.Println("Adding new access token:", name)
scopesString := strings.Join(scopes, ",")
tx, err := _db.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare("INSERT INTO access_tokens(token, name, scopes) values(?, ?, ?)")
if err != nil {
return err
}
defer stmt.Close()
if _, err := stmt.Exec(token, name, scopesString); err != nil {
return err
}
if err = tx.Commit(); err != nil {
return err
}
return nil
}
// DeleteToken will delete a token from the database.
func DeleteToken(token string) error {
log.Println("Deleting access token:", token)
tx, err := _db.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare("DELETE FROM access_tokens WHERE token = ?")
if err != nil {
return err
}
defer stmt.Close()
result, err := stmt.Exec(token)
if err != nil {
return err
}
if rowsDeleted, _ := result.RowsAffected(); rowsDeleted == 0 {
tx.Rollback() //nolint
return errors.New(token + " not found")
}
if err = tx.Commit(); err != nil {
return err
}
return nil
}
// DoesTokenSupportScope will determine if a specific token has access to perform a scoped action.
func DoesTokenSupportScope(token string, scope string) (bool, error) {
// This will split the scopes from comma separated to individual rows
// so we can efficiently find if a token supports a single scope.
// This is SQLite specific, so if we ever support other database
// backends we need to support other methods.
var query = `SELECT count(*) FROM (
WITH RECURSIVE split(token, scope, rest) AS (
SELECT token, '', scopes || ',' FROM access_tokens
UNION ALL
SELECT token,
substr(rest, 0, instr(rest, ',')),
substr(rest, instr(rest, ',')+1)
FROM split
WHERE rest <> '')
SELECT token, scope
FROM split
WHERE scope <> ''
ORDER BY token, scope
) AS token WHERE token.token = ? AND token.scope = ?;`
row := _db.QueryRow(query, token, scope)
var count = 0
err := row.Scan(&count)
return count > 0, err
}
// GetAccessTokens will return all access tokens.
func GetAccessTokens() ([]models.AccessToken, error) { //nolint
tokens := make([]models.AccessToken, 0)
// Get all messages sent within the past day
var query = "SELECT * FROM access_tokens"
rows, err := _db.Query(query)
if err != nil {
return tokens, err
}
defer rows.Close()
for rows.Next() {
var token string
var name string
var scopes string
var timestampString string
var lastUsedString *string
if err := rows.Scan(&token, &name, &scopes, &timestampString, &lastUsedString); err != nil {
log.Error("There is a problem reading the database.", err)
return tokens, err
}
timestamp, err := time.Parse(time.RFC3339, timestampString)
if err != nil {
return tokens, err
}
var lastUsed *time.Time = nil
if lastUsedString != nil {
lastUsedTime, _ := time.Parse(time.RFC3339, *lastUsedString)
lastUsed = &lastUsedTime
}
singleToken := models.AccessToken{
Name: name,
Token: token,
Scopes: strings.Split(scopes, ","),
Timestamp: timestamp,
LastUsed: lastUsed,
}
tokens = append(tokens, singleToken)
}
if err := rows.Err(); err != nil {
return tokens, err
}
return tokens, nil
}
// SetAccessTokenAsUsed will update the last used timestamp for a token.
func SetAccessTokenAsUsed(token string) error {
tx, err := _db.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare("UPDATE access_tokens SET last_used = CURRENT_TIMESTAMP WHERE token = ?")
if err != nil {
return err
}
defer stmt.Close()
if _, err := stmt.Exec(token); err != nil {
return err
}
if err = tx.Commit(); err != nil {
return err
}
return nil
}

26
core/data/config.go
View File

@@ -539,7 +539,9 @@ func VerifySettings() error {
if err := utils.Copy(defaultLogo, filepath.Join(config.DataDirectory, "logo.svg")); err != nil {
log.Errorln("error copying default logo: ", err)
}
SetLogoPath("logo.svg")
if err := SetLogoPath("logo.svg"); err != nil {
log.Errorln("unable to set default logo to logo.svg", err)
}
}
return nil
@@ -577,19 +579,25 @@ func FindHighestVideoQualityIndex(qualities []models.StreamOutputVariant) int {
return indexedQualities[0].index
}
// GetUsernameBlocklist will return the blocked usernames as a comma separated string.
func GetUsernameBlocklist() string {
// GetForbiddenUsernameList will return the blocked usernames as a comma separated string.
func GetForbiddenUsernameList() []string {
usernameString, err := _datastore.GetString(blockedUsernamesKey)
if err != nil {
log.Traceln(blockedUsernamesKey, err)
return ""
return config.DefaultForbiddenUsernames
}
return usernameString
if usernameString == "" {
return config.DefaultForbiddenUsernames
}
blocklist := strings.Split(usernameString, ",")
return blocklist
}
// SetUsernameBlocklist set the username blocklist as a comma separated string.
func SetUsernameBlocklist(usernames string) error {
return _datastore.SetString(blockedUsernamesKey, usernames)
// SetForbiddenUsernameList set the username blocklist as a comma separated string.
func SetForbiddenUsernameList(usernames []string) error {
usernameListString := strings.Join(usernames, ",")
return _datastore.SetString(blockedUsernamesKey, usernameListString)
}

20
core/data/data.go
View File

@@ -17,7 +17,7 @@ import (
)
const (
schemaVersion = 0
schemaVersion = 1
)
var _db *sql.DB
@@ -45,7 +45,13 @@ func SetupPersistence(file string) error {
}
}
db, err := sql.Open("sqlite3", file)
db, err := sql.Open("sqlite3", fmt.Sprintf("file:%s", file))
db.SetMaxOpenConns(1)
_db = db
createWebhooksTable()
createUsersTable(db)
if err != nil {
return err
}
@@ -86,11 +92,6 @@ func SetupPersistence(file string) error {
}
}
_db = db
createWebhooksTable()
createAccessTokensTable()
_datastore = &Datastore{}
_datastore.Setup()
@@ -106,13 +107,14 @@ func SetupPersistence(file string) error {
}
func migrateDatabase(db *sql.DB, from, to int) error {
log.Printf("Migrating database from version %d to %d\n", from, to)
log.Printf("Migrating database from version %d to %d", from, to)
dbBackupFile := filepath.Join(config.BackupDirectory, fmt.Sprintf("owncast-v%d.bak", from))
utils.Backup(db, dbBackupFile)
for v := from; v < to; v++ {
switch v {
case 0:
log.Printf("Migration step from %d to %d\n", v, v+1)
log.Printf("Migration step from %d to %d", v, v+1)
migrateToSchema1(db)
default:
panic("missing database migration step")
}

9
core/data/data_test.go
View File

@@ -2,13 +2,18 @@ package data
import (
"fmt"
"io/ioutil"
"os"
"testing"
)
func TestMain(m *testing.M) {
dbFile := "../../test/test.db"
dbFile, err := ioutil.TempFile(os.TempDir(), "owncast-test-db.db")
if err != nil {
panic(err)
}
SetupPersistence(dbFile)
SetupPersistence(dbFile.Name())
m.Run()
}

118
core/data/migrations.go Normal file
View File

@@ -0,0 +1,118 @@
package data
import (
"database/sql"
"time"
"github.com/owncast/owncast/utils"
log "github.com/sirupsen/logrus"
"github.com/teris-io/shortid"
)
func migrateToSchema1(db *sql.DB) {
// Since it's just a backlog of chat messages let's wipe the old messages
// and recreate the table.
// Drop the old messages table
stmt, err := db.Prepare("DROP TABLE messages")
if err != nil {
log.Fatal(err)
}
defer stmt.Close()
_, err = stmt.Exec()
if err != nil {
log.Warnln(err)
}
// Recreate it
createUsersTable(db)
// Migrate access tokens to become chat users
type oldAccessToken struct {
accessToken string
displayName string
scopes string
createdAt time.Time
lastUsedAt *time.Time
}
oldAccessTokens := make([]oldAccessToken, 0)
query := `SELECT * FROM access_tokens`
rows, err := db.Query(query)
if err != nil || rows.Err() != nil {
log.Errorln("error migrating access tokens to schema v1", err, rows.Err())
return
}
defer rows.Close()
for rows.Next() {
var token string
var name string
var scopes string
var timestampString string
var lastUsedString *string
if err := rows.Scan(&token, &name, &scopes, &timestampString, &lastUsedString); err != nil {
log.Error("There is a problem reading the database.", err)
return
}
timestamp, err := time.Parse(time.RFC3339, timestampString)
if err != nil {
return
}
var lastUsed *time.Time = nil
if lastUsedString != nil {
lastUsedTime, _ := time.Parse(time.RFC3339, *lastUsedString)
lastUsed = &lastUsedTime
}
oldToken := oldAccessToken{
accessToken: token,
displayName: name,
scopes: scopes,
createdAt: timestamp,
lastUsedAt: lastUsed,
}
oldAccessTokens = append(oldAccessTokens, oldToken)
}
// Recreate them as users
for _, token := range oldAccessTokens {
color := utils.GenerateRandomDisplayColor()
if err := insertAPIToken(db, token.accessToken, token.displayName, color, token.scopes); err != nil {
log.Errorln("Error migrating access token", err)
}
}
}
func insertAPIToken(db *sql.DB, token string, name string, color int, scopes string) error {
log.Debugln("Adding new access token:", name)
id := shortid.MustGenerate()
tx, err := db.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare("INSERT INTO users(id, access_token, display_name, display_color, scopes, type) values(?, ?, ?, ?, ?, ?)")
if err != nil {
return err
}
defer stmt.Close()
if _, err = stmt.Exec(id, token, name, color, scopes, "API"); err != nil {
return err
}
if err = tx.Commit(); err != nil {
return err
}
return nil
}

45
core/data/persistence.go
View File

@@ -4,6 +4,7 @@ import (
"bytes"
"database/sql"
"encoding/gob"
"sync"
// sqlite requires a blank import.
_ "github.com/mattn/go-sqlite3"
@@ -12,14 +13,15 @@ import (
// Datastore is the global key/value store for configuration values.
type Datastore struct {
db *sql.DB
cache map[string][]byte
DB *sql.DB
cache map[string][]byte
DbLock *sync.Mutex
}
func (ds *Datastore) warmCache() {
log.Traceln("Warming config value cache")
res, err := ds.db.Query("SELECT key, value FROM datastore")
res, err := ds.DB.Query("SELECT key, value FROM datastore")
if err != nil || res.Err() != nil {
log.Errorln("error warming config cache", err, res.Err())
}
@@ -48,7 +50,7 @@ func (ds *Datastore) Get(key string) (ConfigEntry, error) {
var resultKey string
var resultValue []byte
row := ds.db.QueryRow("SELECT key, value FROM datastore WHERE key = ? LIMIT 1", key)
row := ds.DB.QueryRow("SELECT key, value FROM datastore WHERE key = ? LIMIT 1", key)
if err := row.Scan(&resultKey, &resultValue); err != nil {
return ConfigEntry{}, err
}
@@ -63,36 +65,26 @@ func (ds *Datastore) Get(key string) (ConfigEntry, error) {
// Save will save the ConfigEntry to the database.
func (ds *Datastore) Save(e ConfigEntry) error {
ds.DbLock.Lock()
defer ds.DbLock.Unlock()
var dataGob bytes.Buffer
enc := gob.NewEncoder(&dataGob)
if err := enc.Encode(e.Value); err != nil {
return err
}
tx, err := ds.db.Begin()
tx, err := ds.DB.Begin()
if err != nil {
return err
}
var stmt *sql.Stmt
var count int
row := ds.db.QueryRow("SELECT COUNT(*) FROM datastore WHERE key = ? LIMIT 1", e.Key)
if err := row.Scan(&count); err != nil {
stmt, err = tx.Prepare("INSERT INTO datastore (key, value) VALUES(?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value")
if err != nil {
return err
}
_, err = stmt.Exec(e.Key, dataGob.Bytes())
if count == 0 {
stmt, err = tx.Prepare("INSERT INTO datastore(key, value) values(?, ?)")
if err != nil {
return err
}
_, err = stmt.Exec(e.Key, dataGob.Bytes())
} else {
stmt, err = tx.Prepare("UPDATE datastore SET value=? WHERE key=?")
if err != nil {
return err
}
_, err = stmt.Exec(dataGob.Bytes(), e.Key)
}
if err != nil {
return err
}
@@ -110,7 +102,8 @@ func (ds *Datastore) Save(e ConfigEntry) error {
// Setup will create the datastore table and perform initial initialization.
func (ds *Datastore) Setup() {
ds.cache = make(map[string][]byte)
ds.db = GetDatabase()
ds.DB = GetDatabase()
ds.DbLock = &sync.Mutex{}
createTableSQL := `CREATE TABLE IF NOT EXISTS datastore (
"key" string NOT NULL PRIMARY KEY,
@@ -118,7 +111,7 @@ func (ds *Datastore) Setup() {
"timestamp" DATE DEFAULT CURRENT_TIMESTAMP NOT NULL
);`
stmt, err := ds.db.Prepare(createTableSQL)
stmt, err := ds.DB.Prepare(createTableSQL)
if err != nil {
log.Fatal(err)
}
@@ -137,7 +130,7 @@ func (ds *Datastore) Setup() {
// Reset will delete all config entries in the datastore and start over.
func (ds *Datastore) Reset() {
sql := "DELETE FROM datastore"
stmt, err := ds.db.Prepare(sql)
stmt, err := ds.DB.Prepare(sql)
if err != nil {
log.Fatalln(err)
}
@@ -150,3 +143,7 @@ func (ds *Datastore) Reset() {
PopulateDefaults()
}
func GetDatastore() *Datastore {
return _datastore
}

37
core/data/users.go Normal file
View File

@@ -0,0 +1,37 @@
package data
import (
"database/sql"
log "github.com/sirupsen/logrus"
)
func createUsersTable(db *sql.DB) {
log.Traceln("Creating users table...")
createTableSQL := `CREATE TABLE IF NOT EXISTS users (
"id" TEXT,
"access_token" string NOT NULL,
"display_name" TEXT NOT NULL,
"display_color" NUMBER NOT NULL,
"created_at" TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
"disabled_at" TIMESTAMP,
"previous_names" TEXT DEFAULT '',
"namechanged_at" TIMESTAMP,
"scopes" TEXT,
"type" TEXT DEFAULT 'STANDARD',
"last_used" DATETIME DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (id, access_token),
UNIQUE(id, access_token)
);CREATE INDEX index ON users (id, access_token)`
stmt, err := db.Prepare(createTableSQL)
if err != nil {
log.Fatal(err)
}
defer stmt.Close()
_, err = stmt.Exec()
if err != nil {
log.Warnln(err)
}
}

6
core/data/webhooks.go
View File

@@ -33,7 +33,7 @@ func createWebhooksTable() {
// InsertWebhook will add a new webhook to the database.
func InsertWebhook(url string, events []models.EventType) (int, error) {
log.Println("Adding new webhook:", url)
log.Traceln("Adding new webhook:", url)
eventsString := strings.Join(events, ",")
@@ -67,7 +67,7 @@ func InsertWebhook(url string, events []models.EventType) (int, error) {
// DeleteWebhook will delete a webhook from the database.
func DeleteWebhook(id int) error {
log.Println("Deleting webhook:", id)
log.Traceln("Deleting webhook:", id)
tx, err := _db.Begin()
if err != nil {
@@ -86,7 +86,7 @@ func DeleteWebhook(id int) error {
}
if rowsDeleted, _ := result.RowsAffected(); rowsDeleted == 0 {
tx.Rollback() //nolint
_ = tx.Rollback()
return errors.New(fmt.Sprint(id) + " not found")
}

17
core/stats.go
View File

@@ -7,7 +7,6 @@ import (
log "github.com/sirupsen/logrus"
"github.com/owncast/owncast/core/chat"
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/geoip"
"github.com/owncast/owncast/models"
@@ -86,22 +85,6 @@ func RemoveChatClient(clientID string) {
l.Unlock()
}
func GetChatClients() []models.Client {
l.RLock()
clients := make([]models.Client, 0)
for _, client := range _stats.ChatClients {
chatClient := chat.GetClient(client.ClientID)
if chatClient != nil {
clients = append(clients, chatClient.GetViewerClientFromChatClient())
} else {
clients = append(clients, client)
}
}
l.RUnlock()
return clients
}
// SetViewerIdActive sets a client as active and connected.
func SetViewerIdActive(id string) {
l.Lock()

6
core/streamState.go
View File

@@ -11,6 +11,7 @@ import (
log "github.com/sirupsen/logrus"
"github.com/owncast/owncast/config"
"github.com/owncast/owncast/core/chat"
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/core/rtmp"
"github.com/owncast/owncast/core/transcoder"
@@ -72,10 +73,15 @@ func setStreamAsConnected(rtmpOut *io.PipeReader) {
go webhooks.SendStreamStatusEvent(models.StreamStarted)
transcoder.StartThumbnailGenerator(segmentPath, data.FindHighestVideoQualityIndex(_currentBroadcast.OutputSettings))
_ = chat.SendSystemAction("Stay tuned, the stream is starting!", true)
chat.SendAllWelcomeMessage()
}
// SetStreamAsDisconnected sets the stream as disconnected.
func SetStreamAsDisconnected() {
_ = chat.SendSystemAction("The stream is ending.", true)
_stats.StreamConnected = false
_stats.LastDisconnectTime = utils.NullTime{Time: time.Now(), Valid: true}
_broadcaster = nil

264
core/user/externalAPIUser.go Normal file
View File

@@ -0,0 +1,264 @@
package user
import (
"database/sql"
"errors"
"strings"
"time"
"github.com/owncast/owncast/utils"
log "github.com/sirupsen/logrus"
"github.com/teris-io/shortid"
)
// ExternalAPIUser represents a single 3rd party integration that uses an access token.
// This struct mostly matches the User struct so they can be used interchangeably.
type ExternalAPIUser struct {
Id string `json:"id"`
AccessToken string `json:"accessToken"`
DisplayName string `json:"displayName"`
DisplayColor int `json:"displayColor"`
CreatedAt time.Time `json:"createdAt"`
Scopes []string `json:"scopes"`
Type string `json:"type,omitempty"` // Should be API
LastUsedAt *time.Time `json:"lastUsedAt,omitempty"`
}
const (
// ScopeCanSendChatMessages will allow sending chat messages as itself.
ScopeCanSendChatMessages = "CAN_SEND_MESSAGES"
// ScopeCanSendSystemMessages will allow sending chat messages as the system.
ScopeCanSendSystemMessages = "CAN_SEND_SYSTEM_MESSAGES"
// ScopeHasAdminAccess will allow performing administrative actions on the server.
ScopeHasAdminAccess = "HAS_ADMIN_ACCESS"
)
// For a scope to be seen as "valid" it must live in this slice.
var validAccessTokenScopes = []string{
ScopeCanSendChatMessages,
ScopeCanSendSystemMessages,
ScopeHasAdminAccess,
}
// InsertToken will add a new token to the database.
func InsertExternalAPIUser(token string, name string, color int, scopes []string) error {
log.Traceln("Adding new API user:", name)
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
scopesString := strings.Join(scopes, ",")
id := shortid.MustGenerate()
tx, err := _datastore.DB.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare("INSERT INTO users(id, access_token, display_name, display_color, scopes, type, previous_names) values(?, ?, ?, ?, ?, ?, ?)")
if err != nil {
return err
}
defer stmt.Close()
if _, err = stmt.Exec(id, token, name, color, scopesString, "API", name); err != nil {
return err
}
if err = tx.Commit(); err != nil {
return err
}
return nil
}
// DeleteExternalAPIUser will delete a token from the database.
func DeleteExternalAPIUser(token string) error {
log.Traceln("Deleting access token:", token)
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
tx, err := _datastore.DB.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare("UPDATE users SET disabled_at = ? WHERE access_token = ?")
if err != nil {
return err
}
defer stmt.Close()
result, err := stmt.Exec(time.Now(), token)
if err != nil {
return err
}
if rowsDeleted, _ := result.RowsAffected(); rowsDeleted == 0 {
tx.Rollback() //nolint
return errors.New(token + " not found")
}
if err = tx.Commit(); err != nil {
return err
}
return nil
}
// GetExternalAPIUserForAccessTokenAndScope will determine if a specific token has access to perform a scoped action.
func GetExternalAPIUserForAccessTokenAndScope(token string, scope string) (*ExternalAPIUser, error) {
// This will split the scopes from comma separated to individual rows
// so we can efficiently find if a token supports a single scope.
// This is SQLite specific, so if we ever support other database
// backends we need to support other methods.
var query = `SELECT id, access_token, scopes, display_name, display_color, created_at, last_used FROM (
WITH RECURSIVE split(id, access_token, scopes, display_name, display_color, created_at, last_used, disabled_at, scope, rest) AS (
SELECT id, access_token, scopes, display_name, display_color, created_at, last_used, disabled_at, '', scopes || ',' FROM users
UNION ALL
SELECT id, access_token, scopes, display_name, display_color, created_at, last_used, disabled_at,
substr(rest, 0, instr(rest, ',')),
substr(rest, instr(rest, ',')+1)
FROM split
WHERE rest <> '')
SELECT id, access_token, scopes, display_name, display_color, created_at, last_used, disabled_at, scope
FROM split
WHERE scope <> ''
ORDER BY access_token, scope
) AS token WHERE token.access_token = ? AND token.scope = ?`
row := _datastore.DB.QueryRow(query, token, scope)
integration, err := makeExternalAPIUserFromRow(row)
return integration, err
}
// GetIntegrationNameForAccessToken will return the integration name associated with a specific access token.
func GetIntegrationNameForAccessToken(token string) *string {
query := "SELECT display_name FROM users WHERE access_token IS ? AND disabled_at IS NULL"
row := _datastore.DB.QueryRow(query, token)
var name string
err := row.Scan(&name)
if err != nil {
log.Warnln(err)
return nil
}
return &name
}
// GetExternalAPIUser will return all access tokens.
func GetExternalAPIUser() ([]ExternalAPIUser, error) { //nolint
// Get all messages sent within the past day
var query = "SELECT id, access_token, display_name, display_color, scopes, created_at, last_used FROM users WHERE type IS 'API' AND disabled_at IS NULL"
rows, err := _datastore.DB.Query(query)
if err != nil {
return []ExternalAPIUser{}, err
}
defer rows.Close()
integrations, err := makeExternalAPIUsersFromRows(rows)
return integrations, err
}
// SetExternalAPIUserAccessTokenAsUsed will update the last used timestamp for a token.
func SetExternalAPIUserAccessTokenAsUsed(token string) error {
tx, err := _datastore.DB.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare("UPDATE users SET last_used = CURRENT_TIMESTAMP WHERE access_token = ?")
if err != nil {
return err
}
defer stmt.Close()
if _, err := stmt.Exec(token); err != nil {
return err
}
if err = tx.Commit(); err != nil {
return err
}
return nil
}
func makeExternalAPIUserFromRow(row *sql.Row) (*ExternalAPIUser, error) {
var id string
var accessToken string
var displayName string
var displayColor int
var scopes string
var createdAt time.Time
var lastUsedAt *time.Time
err := row.Scan(&id, &accessToken, &scopes, &displayName, &displayColor, &createdAt, &lastUsedAt)
if err != nil {
log.Errorln(err)
return nil, err
}
integration := ExternalAPIUser{
Id: id,
AccessToken: accessToken,
DisplayName: displayName,
DisplayColor: displayColor,
CreatedAt: createdAt,
Scopes: strings.Split(scopes, ","),
LastUsedAt: lastUsedAt,
}
return &integration, nil
}
func makeExternalAPIUsersFromRows(rows *sql.Rows) ([]ExternalAPIUser, error) {
integrations := make([]ExternalAPIUser, 0)
for rows.Next() {
var id string
var accessToken string
var displayName string
var displayColor int
var scopes string
var createdAt time.Time
var lastUsedAt *time.Time
err := rows.Scan(&id, &accessToken, &displayName, &displayColor, &scopes, &createdAt, &lastUsedAt)
if err != nil {
log.Errorln(err)
return nil, err
}
integration := ExternalAPIUser{
Id: id,
AccessToken: accessToken,
DisplayName: displayName,
DisplayColor: displayColor,
CreatedAt: createdAt,
Scopes: strings.Split(scopes, ","),
LastUsedAt: lastUsedAt,
}
integrations = append(integrations, integration)
}
return integrations, nil
}
// HasValidScopes will verify that all the scopes provided are valid.
func HasValidScopes(scopes []string) bool {
for _, scope := range scopes {
_, foundInSlice := utils.FindInSlice(validAccessTokenScopes, scope)
if !foundInSlice {
log.Errorln("Invalid scope", scope)
return false
}
}
return true
}

261
core/user/user.go Normal file
View File

@@ -0,0 +1,261 @@
package user
import (
"database/sql"
"fmt"
"sort"
"strings"
"time"
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/utils"
"github.com/teris-io/shortid"
log "github.com/sirupsen/logrus"
)
var _datastore *data.Datastore
type User struct {
Id string `json:"id"`
AccessToken string `json:"-"`
DisplayName string `json:"displayName"`
DisplayColor int `json:"displayColor"`
CreatedAt time.Time `json:"createdAt"`
DisabledAt *time.Time `json:"disabledAt,omitempty"`
PreviousNames []string `json:"previousNames"`
NameChangedAt *time.Time `json:"nameChangedAt,omitempty"`
}
func (u *User) IsEnabled() bool {
return u.DisabledAt == nil
}
func SetupUsers() {
_datastore = data.GetDatastore()
}
func CreateAnonymousUser(username string) (*User, error) {
id := shortid.MustGenerate()
accessToken, err := utils.GenerateAccessToken()
if err != nil {
log.Errorln("Unable to create access token for new user")
return nil, err
}
var displayName = username
if displayName == "" {
displayName = utils.GeneratePhrase()
}
displayColor := utils.GenerateRandomDisplayColor()
user := &User{
Id: id,
AccessToken: accessToken,
DisplayName: displayName,
DisplayColor: displayColor,
CreatedAt: time.Now(),
}
if err := create(user); err != nil {
return nil, err
}
return user, nil
}
func ChangeUsername(userId string, username string) {
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
tx, err := _datastore.DB.Begin()
if err != nil {
log.Debugln(err)
}
defer func() {
if err := tx.Rollback(); err != nil {
log.Debugln(err)
}
}()
stmt, err := tx.Prepare("UPDATE users SET display_name = ?, previous_names = previous_names || ?, namechanged_at = ? WHERE id = ?")
if err != nil {
log.Debugln(err)
}
defer stmt.Close()
_, err = stmt.Exec(username, fmt.Sprintf(",%s", username), time.Now(), userId)
if err != nil {
log.Errorln(err)
}
if err := tx.Commit(); err != nil {
log.Errorln("error changing display name of user", userId, err)
}
}
func create(user *User) error {
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
tx, err := _datastore.DB.Begin()
if err != nil {
log.Debugln(err)
}
defer func() {
_ = tx.Rollback()
}()
stmt, err := tx.Prepare("INSERT INTO users(id, access_token, display_name, display_color, previous_names, created_at) values(?, ?, ?, ?, ?, ?)")
if err != nil {
log.Debugln(err)
}
defer stmt.Close()
_, err = stmt.Exec(user.Id, user.AccessToken, user.DisplayName, user.DisplayColor, user.DisplayName, user.CreatedAt)
if err != nil {
log.Errorln("error creating new user", err)
}
return tx.Commit()
}
func SetEnabled(userID string, enabled bool) error {
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
tx, err := _datastore.DB.Begin()
if err != nil {
return err
}
defer tx.Rollback() //nolint
var stmt *sql.Stmt
if !enabled {
stmt, err = tx.Prepare("UPDATE users SET disabled_at=DATETIME('now', 'localtime') WHERE id IS ?")
} else {
stmt, err = tx.Prepare("UPDATE users SET disabled_at=null WHERE id IS ?")
}
if err != nil {
return err
}
defer stmt.Close()
if _, err := stmt.Exec(userID); err != nil {
return err
}
return tx.Commit()
}
// GetUserByToken will return a user by an access token.
func GetUserByToken(token string) *User {
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
query := "SELECT id, display_name, display_color, created_at, disabled_at, previous_names, namechanged_at FROM users WHERE access_token = ?"
row := _datastore.DB.QueryRow(query, token)
return getUserFromRow(row)
}
// GetUserById will return a user by a user ID.
func GetUserById(id string) *User {
_datastore.DbLock.Lock()
defer _datastore.DbLock.Unlock()
query := "SELECT id, display_name, display_color, created_at, disabled_at, previous_names, namechanged_at FROM users WHERE id = ?"
row := _datastore.DB.QueryRow(query, id)
if row == nil {
log.Errorln(row)
return nil
}
return getUserFromRow(row)
}
// GetDisabledUsers will return back all the currently disabled users that are not API users.
func GetDisabledUsers() []*User {
query := "SELECT id, display_name, display_color, created_at, disabled_at, previous_names, namechanged_at FROM users WHERE disabled_at IS NOT NULL AND type IS NOT 'API'"
rows, err := _datastore.DB.Query(query)
if err != nil {
log.Errorln(err)
return nil
}
defer rows.Close()
users := getUsersFromRows(rows)
sort.Slice(users, func(i, j int) bool {
return users[i].DisabledAt.Before(*users[j].DisabledAt)
})
return users
}
func getUsersFromRows(rows *sql.Rows) []*User {
users := make([]*User, 0)
for rows.Next() {
var id string
var displayName string
var displayColor int
var createdAt time.Time
var disabledAt *time.Time
var previousUsernames string
var userNameChangedAt *time.Time
if err := rows.Scan(&id, &displayName, &displayColor, &createdAt, &disabledAt, &previousUsernames, &userNameChangedAt); err != nil {
log.Errorln("error creating collection of users from results", err)
return nil
}
user := &User{
Id: id,
DisplayName: displayName,
DisplayColor: displayColor,
CreatedAt: createdAt,
DisabledAt: disabledAt,
PreviousNames: strings.Split(previousUsernames, ","),
NameChangedAt: userNameChangedAt,
}
users = append(users, user)
}
sort.Slice(users, func(i, j int) bool {
return users[i].CreatedAt.Before(users[j].CreatedAt)
})
return users
}
func getUserFromRow(row *sql.Row) *User {
var id string
var displayName string
var displayColor int
var createdAt time.Time
var disabledAt *time.Time
var previousUsernames string
var userNameChangedAt *time.Time
if err := row.Scan(&id, &displayName, &displayColor, &createdAt, &disabledAt, &previousUsernames, &userNameChangedAt); err != nil {
return nil
}
return &User{
Id: id,
DisplayName: displayName,
DisplayColor: displayColor,
CreatedAt: createdAt,
DisabledAt: disabledAt,
PreviousNames: strings.Split(previousUsernames, ","),
NameChangedAt: userNameChangedAt,
}
}

15
core/webhooks/chat.go
View File

@@ -1,18 +1,19 @@
package webhooks
import (
"github.com/owncast/owncast/core/chat/events"
"github.com/owncast/owncast/models"
)
func SendChatEvent(chatEvent models.ChatEvent) {
func SendChatEvent(chatEvent *events.UserMessageEvent) {
webhookEvent := WebhookEvent{
Type: chatEvent.MessageType,
Type: chatEvent.GetMessageType(),
EventData: &WebhookChatMessage{
Author: chatEvent.Author,
User: chatEvent.User,
Body: chatEvent.Body,
RawBody: chatEvent.RawBody,
ID: chatEvent.ID,
Visible: chatEvent.Visible,
ID: chatEvent.Id,
Visible: chatEvent.HiddenAt == nil,
Timestamp: &chatEvent.Timestamp,
},
}
@@ -20,7 +21,7 @@ func SendChatEvent(chatEvent models.ChatEvent) {
SendEventToWebhooks(webhookEvent)
}
func SendChatEventUsernameChanged(event models.NameChangeEvent) {
func SendChatEventUsernameChanged(event events.NameChangeEvent) {
webhookEvent := WebhookEvent{
Type: models.UserNameChanged,
EventData: event,
@@ -29,7 +30,7 @@ func SendChatEventUsernameChanged(event models.NameChangeEvent) {
SendEventToWebhooks(webhookEvent)
}
func SendChatEventUserJoined(event models.UserJoinedEvent) {
func SendChatEventUserJoined(event events.UserJoinedEvent) {
webhookEvent := WebhookEvent{
Type: models.UserNameChanged,
EventData: event,

4
core/webhooks/webhooks.go
View File

@@ -8,6 +8,8 @@ import (
log "github.com/sirupsen/logrus"
"github.com/owncast/owncast/core/user"
"github.com/owncast/owncast/core/data"
"github.com/owncast/owncast/models"
)
@@ -18,7 +20,7 @@ type WebhookEvent struct {
}
type WebhookChatMessage struct {
Author string `json:"author,omitempty"`
User *user.User `json:"user,omitempty"`
Body string `json:"body,omitempty"`
RawBody string `json:"rawBody,omitempty"`
ID string `json:"id,omitempty"`