Update chat message visibility for moderation (#524)

* update message viz in db

* create admin endpoint to update message visibility

* convert UpdateMessageVisibility api to take in an array of IDs to change visibility on instead

* Support requesting filtered or unfiltered chat messages

* Handle UPDATE chat events on front and backend for toggling messages

* Return entire message with UPDATE events

* Remove the UPDATE message type

* Revert "Remove the UPDATE message type"

This reverts commit 3a83df3d492f7ecf2bab65e845aa2b0365d3a7f6.

* update -> visibility update

* completely remove messages when they turn hidden on VISIBILITY-UPDATEs, and insert them if they turn visible

* Explicitly set visibility

* Fix multi-id sql updates

* increate scroll buffer a bit so chat scrolls when new large messages come in

* Add automated test around chat moderation

* Add new chat admin APIs to api spec

* Commit updated API documentation

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

.gitignore

@@ -25,7 +25,6 @@ webroot/static/content.md
 hls/
 dist/
 data/
-admin/
 transcoder.log
 chat.db
 .yp.key

controllers/admin/chat.go

@@ -0,0 +1,60 @@
+package admin
+
+// this is endpoint logic
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/owncast/owncast/controllers"
+	"github.com/owncast/owncast/core"
+	"github.com/owncast/owncast/core/chat"
+	log "github.com/sirupsen/logrus"
+)
+
+// UpdateMessageVisibility updates an array of message IDs to have the same visiblity.
+func UpdateMessageVisibility(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		controllers.WriteSimpleResponse(w, false, r.Method+" not supported")
+		return
+	}
+
+	decoder := json.NewDecoder(r.Body)
+	var request messageVisibilityUpdateRequest // creates an empty struc
+
+	err := decoder.Decode(&request) // decode the json into `request`
+	if err != nil {
+		log.Errorln(err)
+		controllers.WriteSimpleResponse(w, false, "")
+		return
+	}
+
+	// // make sql update call here.
+	// // := means create a new var
+	// _db := data.GetDatabase()
+	// updateMessageVisibility(_db, request)
+
+	if err := chat.SetMessagesVisibility(request.IDArray, request.Visible); err != nil {
+		controllers.WriteSimpleResponse(w, false, err.Error())
+		return
+	}
+
+	controllers.WriteSimpleResponse(w, true, "changed")
+}
+
+type messageVisibilityUpdateRequest struct {
+	IDArray []string `json:"idArray"`
+	Visible bool     `json:"visible"`
+}
+
+// GetChatMessages returns all of the chat messages, unfiltered.
+func GetChatMessages(w http.ResponseWriter, r *http.Request) {
+	// middleware.EnableCors(&w)
+	w.Header().Set("Content-Type", "application/json")
+
+	messages := core.GetAllChatMessages(false)
+
+	if err := json.NewEncoder(w).Encode(messages); err != nil {
+		log.Errorln(err)
+	}
+}

controllers/chat.go

@@ -17,14 +17,14 @@ func GetChatMessages(w http.ResponseWriter, r *http.Request) {
 
 	switch r.Method {
 	case http.MethodGet:
-		messages := core.GetAllChatMessages()
+		messages := core.GetAllChatMessages(true)
 
 		err := json.NewEncoder(w).Encode(messages)
 		if err != nil {
 			log.Errorln(err)
 		}
 	case http.MethodPost:
-		var message models.ChatMessage
+		var message models.ChatEvent
 		if err := json.NewDecoder(r.Body).Decode(&message); err != nil {
 			internalErrorHandler(w, err)
 			return

core/chat/chat.go

@@ -14,7 +14,7 @@ func Setup(listener models.ChatListener) {
 	clients := make(map[string]*Client)
 	addCh := make(chan *Client)
 	delCh := make(chan *Client)
-	sendAllCh := make(chan models.ChatMessage)
+	sendAllCh := make(chan models.ChatEvent)
 	pingCh := make(chan models.PingMessage)
 	doneCh := make(chan bool)
 	errCh := make(chan error)
@@ -51,7 +51,7 @@ func Start() error {
 }
 
 // SendMessage sends a message to all.
-func SendMessage(message models.ChatMessage) {
+func SendMessage(message models.ChatEvent) {
 	if _server == nil {
 		return
 	}
@@ -60,12 +60,12 @@ func SendMessage(message models.ChatMessage) {
 }
 
 // GetMessages gets all of the messages.
-func GetMessages() []models.ChatMessage {
+func GetMessages(filtered bool) []models.ChatEvent {
 	if _server == nil {
-		return []models.ChatMessage{}
+		return []models.ChatEvent{}
 	}
 
-	return getChatHistory()
+	return getChatHistory(filtered)
 }
 
 func GetClient(clientID string) *Client {

core/chat/client.go

@@ -30,7 +30,7 @@ type Client struct {
 
 	socketID              string // How we identify a single websocket client.
 	ws                    *websocket.Conn
-	ch                    chan models.ChatMessage
+	ch                    chan models.ChatEvent
 	pingch                chan models.PingMessage
 	usernameChangeChannel chan models.NameChangeEvent
 
@@ -38,10 +38,11 @@ type Client struct {
 }
 
 const (
-	CHAT       = "CHAT"
+	CHAT             = "CHAT"
-	NAMECHANGE = "NAME_CHANGE"
+	NAMECHANGE       = "NAME_CHANGE"
-	PING       = "PING"
+	PING             = "PING"
-	PONG       = "PONG"
+	PONG             = "PONG"
+	VISIBILITYUPDATE = "VISIBILITY-UPDATE"
 )
 
 // NewClient creates a new chat client.
@@ -50,7 +51,7 @@ func NewClient(ws *websocket.Conn) *Client {
 		log.Panicln("ws cannot be nil")
 	}
 
-	ch := make(chan models.ChatMessage, channelBufSize)
+	ch := make(chan models.ChatEvent, channelBufSize)
 	doneCh := make(chan bool)
 	pingch := make(chan models.PingMessage)
 	usernameChangeChannel := make(chan models.NameChangeEvent)
@@ -68,7 +69,7 @@ func (c *Client) GetConnection() *websocket.Conn {
 	return c.ws
 }
 
-func (c *Client) Write(msg models.ChatMessage) {
+func (c *Client) Write(msg models.ChatEvent) {
 	select {
 	case c.ch <- msg:
 	default:
@@ -176,7 +177,7 @@ func (c *Client) userChangedName(data []byte) {
 }
 
 func (c *Client) chatMessageReceived(data []byte) {
-	var msg models.ChatMessage
+	var msg models.ChatEvent
 	err := json.Unmarshal(data, &msg)
 	if err != nil {
 		log.Errorln(err)

core/chat/messages.go

@@ -0,0 +1,28 @@
+package chat
+
+import (
+	log "github.com/sirupsen/logrus"
+)
+
+func SetMessagesVisibility(messageIDs []string, visibility bool) error {
+	// Save new message visibility
+	if err := saveMessageVisibility(messageIDs, visibility); err != nil {
+		log.Errorln(err)
+		return err
+	}
+
+	// Send an update event to all clients for each message.
+	// Note: Our client expects a single message at a time, so we can't just
+	// send an array of messages in a single update.
+	for _, id := range messageIDs {
+		message, err := getMessageById(id)
+		if err != nil {
+			log.Errorln(err)
+			continue
+		}
+		message.MessageType = VISIBILITYUPDATE
+		_server.sendAll(message)
+	}
+
+	return nil
+}

core/chat/persistence.go

@@ -2,6 +2,7 @@ package chat
 
 import (
 	"database/sql"
+	"strings"
 	"time"
 
 	_ "github.com/mattn/go-sqlite3"
@@ -38,7 +39,7 @@ func createTable() {
 	}
 }
 
-func addMessage(message models.ChatMessage) {
+func addMessage(message models.ChatEvent) {
 	tx, err := _db.Begin()
 	if err != nil {
 		log.Fatal(err)
@@ -60,11 +61,16 @@ func addMessage(message models.ChatMessage) {
 	}
 }
 
-func getChatHistory() []models.ChatMessage {
+func getChatHistory(filtered bool) []models.ChatEvent {
-	history := make([]models.ChatMessage, 0)
+	history := make([]models.ChatEvent, 0)
 
 	// Get all messages sent within the past day
-	rows, err := _db.Query("SELECT * FROM messages WHERE visible = 1 AND messageType != 'SYSTEM' AND datetime(timestamp) >=datetime('now', '-1 Day')")
+	var query = "SELECT * FROM messages WHERE messageType != 'SYSTEM' AND datetime(timestamp) >=datetime('now', '-1 Day')"
+	if filtered {
+		query = query + " AND visible = 1"
+	}
+
+	rows, err := _db.Query(query)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -85,7 +91,7 @@ func getChatHistory() []models.ChatMessage {
 			break
 		}
 
-		message := models.ChatMessage{}
+		message := models.ChatEvent{}
 		message.ID = id
 		message.Author = author
 		message.Body = body
@@ -102,3 +108,64 @@ func getChatHistory() []models.ChatMessage {
 
 	return history
 }
+
+func saveMessageVisibility(messageIDs []string, visible bool) error {
+	tx, err := _db.Begin()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	stmt, err := tx.Prepare("UPDATE messages SET visible=? WHERE id IN (?" + strings.Repeat(",?", len(messageIDs)-1) + ")")
+
+	if err != nil {
+		log.Fatal(err)
+		return err
+	}
+	defer stmt.Close()
+
+	args := make([]interface{}, len(messageIDs)+1)
+	args[0] = visible
+	for i, id := range messageIDs {
+		args[i+1] = id
+	}
+
+	_, err = stmt.Exec(args...)
+	if err != nil {
+		log.Fatal(err)
+		return err
+	}
+
+	if err = tx.Commit(); err != nil {
+		log.Fatal(err)
+		return err
+	}
+
+	return nil
+}
+
+func getMessageById(messageID string) (models.ChatEvent, error) {
+	var query = "SELECT * FROM messages WHERE id = ?"
+	row := _db.QueryRow(query, messageID)
+
+	var id string
+	var author string
+	var body string
+	var messageType string
+	var visible int
+	var timestamp time.Time
+
+	err := row.Scan(&id, &author, &body, &messageType, &visible, &timestamp)
+	if err != nil {
+		log.Errorln(err)
+		return models.ChatEvent{}, err
+	}
+
+	return models.ChatEvent{
+		ID:          id,
+		Author:      author,
+		Body:        body,
+		MessageType: messageType,
+		Visible:     visible == 1,
+		Timestamp:   timestamp,
+	}, nil
+}

core/chat/server.go

@@ -28,7 +28,7 @@ type server struct {
 
 	addCh     chan *Client
 	delCh     chan *Client
-	sendAllCh chan models.ChatMessage
+	sendAllCh chan models.ChatEvent
 	pingCh    chan models.PingMessage
 	doneCh    chan bool
 	errCh     chan error
@@ -45,7 +45,7 @@ func (s *server) remove(c *Client) {
 }
 
 // SendToAll sends a message to all of the connected clients.
-func (s *server) SendToAll(msg models.ChatMessage) {
+func (s *server) SendToAll(msg models.ChatEvent) {
 	s.sendAllCh <- msg
 }
 
@@ -54,7 +54,7 @@ func (s *server) err(err error) {
 	s.errCh <- err
 }
 
-func (s *server) sendAll(msg models.ChatMessage) {
+func (s *server) sendAll(msg models.ChatEvent) {
 	for _, c := range s.Clients {
 		c.Write(msg)
 	}
@@ -153,7 +153,7 @@ func (s *server) sendWelcomeMessageToClient(c *Client) {
 		time.Sleep(7 * time.Second)
 
 		initialChatMessageText := fmt.Sprintf("Welcome to %s! %s", config.Config.InstanceDetails.Title, config.Config.InstanceDetails.Summary)
-		initialMessage := models.ChatMessage{ClientID: "owncast-server", Author: config.Config.InstanceDetails.Name, Body: initialChatMessageText, ID: "initial-message-1", MessageType: "SYSTEM", Visible: true, Timestamp: time.Now()}
+		initialMessage := models.ChatEvent{ClientID: "owncast-server", Author: config.Config.InstanceDetails.Name, Body: initialChatMessageText, ID: "initial-message-1", MessageType: "SYSTEM", Visible: true, Timestamp: time.Now()}
 		c.Write(initialMessage)
 	}()
 }

core/chatListener.go

@@ -21,11 +21,11 @@ func (cl ChatListenerImpl) ClientRemoved(clientID string) {
 }
 
 // MessageSent is for when a message is sent.
-func (cl ChatListenerImpl) MessageSent(message models.ChatMessage) {
+func (cl ChatListenerImpl) MessageSent(message models.ChatEvent) {
 }
 
 // SendMessageToChat sends a message to the chat server.
-func SendMessageToChat(message models.ChatMessage) error {
+func SendMessageToChat(message models.ChatEvent) error {
 	if !message.Valid() {
 		return errors.New("invalid chat message; id, author, and body are required")
 	}
@@ -36,6 +36,6 @@ func SendMessageToChat(message models.ChatMessage) error {
 }
 
 // GetAllChatMessages gets all of the chat messages.
-func GetAllChatMessages() []models.ChatMessage {
+func GetAllChatMessages(filtered bool) []models.ChatEvent {
-	return chat.GetMessages()
+	return chat.GetMessages(filtered)
 }

models/chatListener.go

@@ -4,5 +4,5 @@ package models
 type ChatListener interface {
 	ClientAdded(client Client)
 	ClientRemoved(clientID string)
-	MessageSent(message ChatMessage)
+	MessageSent(message ChatEvent)
 }

models/chatMessage.go

@@ -12,26 +12,26 @@ import (
 	"mvdan.cc/xurls"
 )
 
-// ChatMessage represents a single chat message.
+// ChatEvent represents a single chat message.
-type ChatMessage struct {
+type ChatEvent struct {
 	ClientID string `json:"-"`
 
-	Author      string    `json:"author"`
+	Author      string    `json:"author,omitempty"`
-	Body        string    `json:"body"`
+	Body        string    `json:"body,omitempty"`
 	ID          string    `json:"id"`
 	MessageType string    `json:"type"`
 	Visible     bool      `json:"visible"`
-	Timestamp   time.Time `json:"timestamp"`
+	Timestamp   time.Time `json:"timestamp,omitempty"`
 }
 
 // Valid checks to ensure the message is valid.
-func (m ChatMessage) Valid() bool {
+func (m ChatEvent) Valid() bool {
 	return m.Author != "" && m.Body != "" && m.ID != ""
 }
 
 // RenderAndSanitizeMessageBody will turn markdown into HTML, sanitize raw user-supplied HTML and standardize
 // the message into something safe and renderable for clients.
-func (m *ChatMessage) RenderAndSanitizeMessageBody() {
+func (m *ChatEvent) RenderAndSanitizeMessageBody() {
 	raw := m.Body
 
 	// Set the new, sanitized and rendered message body

openapi.yaml

@@ -257,10 +257,10 @@ components:
           examples:
             success:
               summary: Operation succeeded.
-              value: {"success": true, "message": "inbound stream disconnected"}
+              value: {"success": true, "message": "context specific success message"}
             failure:
               summary: Operation failed.
-              value: {"success": false, "message": "no inbound stream connected"}
+              value: {"success": false, "message": "context specific failure message"}
 
 paths:
 
@@ -648,6 +648,65 @@ paths:
                         description: The maximum number of HLS video segments we will keep referenced in the playlist.
                   yp:
                     $ref: "#/components/schemas/YP"
+              
+  /api/admin/chat/messages:
+    get:
+      summary: Chat messages, unfiltered.
+      description: Get a list of all chat messages with no filters applied.
+      tags: ["Admin"]
+      security:
+        - AdminBasicAuth: []
+      responses:
+        '200':
+          description: ""
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: object
+                  properties:
+                    author:
+                      type: string
+                      description: Username of the chat message poster.
+                    body:
+                      type: string
+                      description: Escaped HTML of the chat message content.
+                    id:
+                      type: string
+                      description: Unique ID of the chat message.
+                    visible:
+                      type: boolean
+                      description: "Should chat message be visibly rendered."
+                    timestamp:
+                      type: string
+                      format: date-time
+
+
+  /api/admin/chat/updatemessagevisibility:
+    post:
+      summary: Update the visibility of chat messages.
+      description: Pass an array of IDs you want to change the chat visibility of.
+      requestBody:
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                visible:
+                  type: boolean
+                  description: Are these messages visible in "Get the CPU, Memory and Disk utilization levels over the collected period."
+                idArray:
+                  type: array
+                  items:
+                    type: string
+                    description: IDs of the chat messages you wish to change the visibility of.
+      tags: ["Admin"]
+      security:
+        - AdminBasicAuth: []
+      responses:
+        '200':
+          $ref: "#/components/responses/BasicResponse"
 
   /api/admin/viewersOverTime:
     get:

router/router.go

@@ -82,6 +82,12 @@ func Start() error {
 	// Get warning/error logs
 	http.HandleFunc("/api/admin/logs/warnings", middleware.RequireAdminAuth(admin.GetWarnings))
 
+	// Get all chat messages for the admin, unfiltered.
+	http.HandleFunc("/api/admin/chat/messages", middleware.RequireAdminAuth(admin.GetChatMessages))
+
+	// Update chat message visibilty
+	http.HandleFunc("/api/admin/chat/updatemessagevisibility", middleware.RequireAdminAuth(admin.UpdateMessageVisibility))
+
 	port := config.Config.GetPublicWebServerPort()
 
 	log.Tracef("Web server running on port: %d", port)

test/automated/chatmoderation.test.js

@@ -0,0 +1,54 @@
+const { test } = require('@jest/globals');
+var request = require('supertest');
+request = request('http://127.0.0.1:8080');
+
+const WebSocket = require('ws');
+var ws;
+
+const testVisibilityMessage = {
+    author: "username",
+    body: "message " + Math.floor(Math.random() * 100),
+    type: 'CHAT',
+    visible: true,
+    timestamp: new Date().toISOString()
+};
+
+test('can send a chat message', (done) => {
+    ws = new WebSocket('ws://127.0.0.1:8080/entry', {
+        origin: 'http://localhost',
+    });
+
+    function onOpen() {
+        ws.send(JSON.stringify(testVisibilityMessage), function () {
+            ws.close();
+            done();
+        });
+    }
+
+    ws.on('open', onOpen);
+});
+
+var messageId;
+
+test('verify we can make API call to mark message as hidden', async (done) => {
+    const res = await request.get('/api/chat').expect(200);
+    const message = res.body[0];
+    messageId = message.id;
+    await request.post('/api/admin/chat/updatemessagevisibility')
+        .auth('admin', 'abc123')
+        .send({ "idArray": [messageId], "visible": false }).expect(200);
+    done();
+});
+
+test('verify message has become hidden', async (done) => {
+    const res = await request.get('/api/admin/chat/messages')
+        .expect(200)
+        .auth('admin', 'abc123')
+
+    const message = res.body.filter(obj => {
+        return obj.id === messageId;
+    });
+    expect(message.length).toBe(1);
+    expect(message[0].visible).toBe(false);
+    done();
+});

webroot/js/components/chat/chat-message-view.js

@@ -39,7 +39,7 @@ export default class ChatMessageView extends Component {
 
   render() {
     const { message } = this.props;
-    const { author, timestamp } = message;
+    const { author, timestamp, visible } = message;
 
     const { formattedMessage } = this.state;
     if (!formattedMessage) {

webroot/js/components/chat/chat.js

@@ -26,6 +26,7 @@ export default class Chat extends Component {
 
     this.websocket = null;
     this.receivedFirstMessages = false;
+    this.receivedMessageUpdate = false;
 
     this.windowBlurred = false;
     this.numMessagesSinceBlur = 0;
@@ -88,7 +89,7 @@ export default class Chat extends Component {
     }
 
     // scroll to bottom of messages list when new ones come in
-    if (messages.length > prevMessages.length) {
+    if (messages.length !== prevMessages.length) {
       this.setState({
         newMessagesReceived: true,
       });
@@ -144,7 +145,7 @@ export default class Chat extends Component {
   }
 
   receivedWebsocketMessage(message) {
-    this.addMessage(message);
+    this.handleMessage(message);
   }
 
   handleNetworkingError(error) {
@@ -152,16 +153,48 @@ export default class Chat extends Component {
     console.log(error);
   }
 
-  addMessage(message) {
+  // handle any incoming message
+  handleMessage(message) {
+    const {
+      id: messageId,
+      type: messageType,
+      timestamp: messageTimestamp,
+      visible: messageVisible,
+    } = message;
     const { messages: curMessages } = this.state;
     const { messagesOnly } = this.props;
 
-    // if incoming message has same id as existing message, don't add it
+    const existingIndex = curMessages.findIndex(item => item.id === messageId);
-    const existing = curMessages.filter(function (item) {
-      return item.id === message.id;
-    })
 
-    if (existing.length === 0 || !existing) {
+    // If the message already exists and this is an update event
+    // then update it.
+    if (messageType === 'VISIBILITY-UPDATE') {
+      const updatedMessageList = [...curMessages];
+      const convertedMessage = {
+        ...message,
+        type: 'CHAT',
+      };
+      // if message exists and should now hide, take it out.
+      if (existingIndex >= 0 && !messageVisible) {
+        this.setState({
+          messages: curMessages.filter(item => item.id !== messageId),
+        });
+      } else if (existingIndex === -1 && messageVisible) {
+        // insert message at timestamp
+        const insertAtIndex = curMessages.findIndex((item, index) => {
+          const time = item.timestamp || messageTimestamp;
+          const nextMessage = index < curMessages.length - 1 && curMessages[index + 1];
+          const nextTime = nextMessage.timestamp || messageTimestamp;
+          const messageTimestampDate = new Date(messageTimestamp);
+          return messageTimestampDate > (new Date(time)) && messageTimestampDate <= (new Date(nextTime));
+        });
+        updatedMessageList.splice(insertAtIndex + 1, 0, convertedMessage);
+        this.setState({
+          messages: updatedMessageList,
+        });
+      }
+    } else if (existingIndex === -1) {
+      // else if message doesn't exist, add it and extra username
       const newState = {
         messages: [...curMessages, message],
       };
@@ -173,7 +206,7 @@ export default class Chat extends Component {
     }
 
     // if window is blurred and we get a new message, add 1 to title
-    if (!messagesOnly && message.type === 'CHAT' && this.windowBlurred) {
+    if (!messagesOnly && messageType === 'CHAT' && this.windowBlurred) {
       this.numMessagesSinceBlur += 1;
     }
   }
@@ -279,7 +312,7 @@ export default class Chat extends Component {
     const { username, messagesOnly, chatInputEnabled } = props;
     const { messages, chatUserNames, webSocketConnected } = state;
 
-    const messageList = messages.map(
+    const messageList = messages.filter(message => message.visible !== false).map(
       (message) =>
         html`<${Message}
           message=${message}

webroot/js/utils/constants.js

@@ -48,7 +48,7 @@ export const CHAT_KEY_MODIFIERS = [
   'Meta',
   'Alt',
 ];
-export const MESSAGE_JUMPTOBOTTOM_BUFFER = 260;
+export const MESSAGE_JUMPTOBOTTOM_BUFFER = 300;
 
 // app styling
 export const WIDTH_SINGLE_COL = 730;