Port chat reliability subsystem from /tg/ (#10367)

* Port chat reliability subsystem from /tg/

* This crash isn't needed, seems to just be a /tg/ "cleanup" runtime according to git blame

beestation.dme

@@ -503,6 +503,7 @@
 #include "code\datums\beam.dm"
 #include "code\datums\browser.dm"
 #include "code\datums\callback.dm"
+#include "code\datums\chat_payload.dm"
 #include "code\datums\chatmessage.dm"
 #include "code\datums\cinematic.dm"
 #include "code\datums\dash_weapon.dm"

code/__DEFINES/chat.dm

@@ -3,6 +3,11 @@
  * SPDX-License-Identifier: MIT
  */
 
+/// How many chat payloads to keep in history
+#define CHAT_RELIABILITY_HISTORY_SIZE 5
+/// How many resends to allow before giving up
+#define CHAT_RELIABILITY_MAX_RESENDS 3
+
 #define MESSAGE_TYPE_SYSTEM "system"
 #define MESSAGE_TYPE_LOCALCHAT "localchat"
 #define MESSAGE_TYPE_RADIO "radio"

code/controllers/subsystem/chat.dm

@@ -1,38 +1,98 @@
-/**
+/*!
  * Copyright (c) 2020 Aleksej Komarov
  * SPDX-License-Identifier: MIT
  */
 
 SUBSYSTEM_DEF(chat)
 	name = "Chat"
-	flags = SS_TICKER
+	flags = SS_TICKER | SS_NO_INIT
 	wait = 1
 	priority = FIRE_PRIORITY_CHAT
 	init_order = INIT_ORDER_CHAT
 
-	var/list/payload_by_client = list()
+	/// Assosciates a ckey with a list of messages to send to them.
+	var/list/list/datum/chat_payload/client_to_payloads = list()
+
+	/// Associates a ckey with an assosciative list of their last CHAT_RELIABILITY_HISTORY_SIZE messages.
+	var/list/list/datum/chat_payload/client_to_reliability_history = list()
+
+	/// Assosciates a ckey with their next sequence number.
+	var/list/client_to_sequence_number = list()
+
+/datum/controller/subsystem/chat/proc/generate_payload(client/target, message_data)
+	var/sequence = client_to_sequence_number[target.ckey]
+	client_to_sequence_number[target.ckey] += 1
+
+	var/datum/chat_payload/payload = new
+	payload.sequence = sequence
+	payload.content = message_data
+
+	if(!(target.ckey in client_to_reliability_history))
+		client_to_reliability_history[target.ckey] = list()
+	var/list/client_history = client_to_reliability_history[target.ckey]
+	client_history["[sequence]"] = payload
+
+	if(length(client_history) > CHAT_RELIABILITY_HISTORY_SIZE)
+		var/oldest = text2num(client_history[1])
+		for(var/index in 2 to length(client_history))
+			var/test = text2num(client_history[index])
+			if(test < oldest)
+				oldest = test
+		client_history -= "[oldest]"
+	return payload
+
+/datum/controller/subsystem/chat/proc/send_payload_to_client(client/target, datum/chat_payload/payload)
+	target.tgui_panel.window.send_message("chat/message", payload.into_message())
+	SEND_TEXT(target, payload.get_content_as_html())
 
 /datum/controller/subsystem/chat/fire()
-	for(var/key in payload_by_client)
-		var/client/client = key
-		var/payload = payload_by_client[key]
-		payload_by_client -= key
-		if(client)
-			// Send to tgchat
-			client.tgui_panel?.window.send_message("chat/message", payload)
-			// Send to old chat
-			for(var/message in payload)
-				SEND_TEXT(client, message_to_html(message))
+	for(var/ckey in client_to_payloads)
+		var/client/target = GLOB.directory[ckey]
+		if(isnull(target)) // verify client still exists
+			LAZYREMOVE(client_to_payloads, ckey)
+			continue
+
+		for(var/datum/chat_payload/payload as anything in client_to_payloads[ckey])
+			send_payload_to_client(target, payload)
+		LAZYREMOVE(client_to_payloads, ckey)
+
 		if(MC_TICK_CHECK)
 			return
 
-/datum/controller/subsystem/chat/proc/queue(target, message)
-	if(islist(target))
-		for(var/_target in target)
-			var/client/client = CLIENT_FROM_VAR(_target)
-			if(client)
-				LAZYADD(payload_by_client[client], list(message))
+/datum/controller/subsystem/chat/proc/queue(queue_target, list/message_data)
+	var/list/targets = islist(queue_target) ? queue_target : list(queue_target)
+	for(var/target in targets)
+		var/client/client = CLIENT_FROM_VAR(target)
+		if(isnull(client))
+			continue
+		LAZYADDASSOCLIST(client_to_payloads, client.ckey, generate_payload(client, message_data))
+
+/datum/controller/subsystem/chat/proc/send_immediate(send_target, list/message_data)
+	var/list/targets = islist(send_target) ? send_target : list(send_target)
+	for(var/target in targets)
+		var/client/client = CLIENT_FROM_VAR(target)
+		if(isnull(client))
+			continue
+		send_payload_to_client(client, generate_payload(client, message_data))
+
+/datum/controller/subsystem/chat/proc/handle_resend(client/client, sequence)
+	var/list/client_history = client_to_reliability_history[client.ckey]
+	sequence = "[sequence]"
+	if(isnull(client_history) || !(sequence in client_history))
 		return
-	var/client/client = CLIENT_FROM_VAR(target)
-	if(client)
-		LAZYADD(payload_by_client[client], list(message))
+
+	var/datum/chat_payload/payload = client_history[sequence]
+	if(payload.resends > CHAT_RELIABILITY_MAX_RESENDS)
+		return // we tried but byond said no
+
+	payload.resends += 1
+	send_payload_to_client(client, client_history[sequence])
+	SSblackbox.record_feedback(
+		"nested tally",
+		"chat_resend_byond_version",
+		1,
+		list(
+			"[client.byond_version]",
+			"[client.byond_build]",
+		),
+	)

code/datums/chat_payload.dm

@@ -0,0 +1,16 @@
+/// Stores information about a chat payload
+/datum/chat_payload
+	/// Sequence number of this payload
+	var/sequence = 0
+	/// Message we are sending
+	var/list/content
+	/// Resend count
+	var/resends = 0
+
+/// Converts the chat payload into a JSON string
+/datum/chat_payload/proc/into_message()
+	return "{\"sequence\":[sequence],\"content\":[json_encode(content)]}"
+
+/// Returns an HTML-encoded message from our contents.
+/datum/chat_payload/proc/get_content_as_html()
+	return message_to_html(content)

code/modules/tgchat/to_chat.dm

@@ -1,4 +1,4 @@
-/**
+/*!
  * Copyright (c) 2020 Aleksej Komarov
  * SPDX-License-Identifier: MIT
  */
@@ -7,66 +7,71 @@
  * Circumvents the message queue and sends the message
  * to the recipient (target) as soon as possible.
  */
-/proc/to_chat_immediate(target, html,
-		type = null,
-		text = null,
-		avoid_highlighting = FALSE,
-		allow_linkify = FALSE,
-		// FIXME: These flags are now pointless and have no effect
-		handle_whitespace = TRUE,
-		trailing_newline = TRUE)
+/proc/to_chat_immediate(
+	target,
+	html,
+	type = null,
+	text = null,
+	avoid_highlighting = FALSE,
+	allow_linkify = FALSE,
+	// FIXME: These flags are now pointless and have no effect
+	handle_whitespace = TRUE,
+	trailing_newline = TRUE
+)
+	// Useful where the integer 0 is the entire message. Use case is enabling to_chat(target, some_boolean) while preventing to_chat(target, "")
+	html = "[html]"
+	text = "[text]"
+
 	if(!target || (!html && !text))
 		return
 	if(target == world)
 		target = GLOB.clients
+
 	// Build a message
 	var/message = list()
 	if(type) message["type"] = type
 	if(text) message["text"] = text
 	if(html) message["html"] = html
 	if(avoid_highlighting) message["avoidHighlighting"] = avoid_highlighting
 	if(allow_linkify) message["allowLinkify"] = allow_linkify
-	var/message_blob = TGUI_CREATE_MESSAGE("chat/message", message)
-	var/message_html = message_to_html(message)
-	if(islist(target))
-		for(var/_target in target)
-			var/client/client = CLIENT_FROM_VAR(_target)
-			if(client)
-				// Send to tgchat
-				client.tgui_panel?.window.send_raw_message(message_blob)
-				// Send to old chat
-				SEND_TEXT(client, message_html)
-		return
-	var/client/client = CLIENT_FROM_VAR(target)
-	if(client)
-		// Send to tgchat
-		client.tgui_panel?.window.send_raw_message(message_blob)
-		// Send to old chat
-		SEND_TEXT(client, message_html)
+
+	// send it immediately
+	SSchat.send_immediate(target, message)
 
 /**
  * Sends the message to the recipient (target).
  *
  * Recommended way to write to_chat calls:
+ * ```
  * to_chat(client,
  *     type = MESSAGE_TYPE_INFO,
  *     html = "You have found <strong>[object]</strong>")
+ * ```
  */
-/proc/to_chat(target, html,
-		type = null,
-		text = null,
-		avoid_highlighting = FALSE,
-		allow_linkify = FALSE,
-		// FIXME: These flags are now pointless and have no effect
-		handle_whitespace = TRUE,
-		trailing_newline = TRUE)
-	if(isnull(Master) || Master.current_runlevel == RUNLEVEL_INIT || !SSchat?.initialized)
-		to_chat_immediate(target, html, type, text)
+/proc/to_chat(
+	target,
+	html,
+	type = null,
+	text = null,
+	avoid_highlighting = FALSE,
+	allow_linkify = FALSE,
+	// FIXME: These flags are now pointless and have no effect
+	handle_whitespace = TRUE,
+	trailing_newline = TRUE
+)
+	if(isnull(Master) || !Master.current_runlevel)
+		to_chat_immediate(target, html, type, text, avoid_highlighting, allow_linkify)
 		return
+
+	// Useful where the integer 0 is the entire message. Use case is enabling to_chat(target, some_boolean) while preventing to_chat(target, "")
+	html = "[html]"
+	text = "[text]"
+
 	if(!target || (!html && !text))
 		return
 	if(target == world)
 		target = GLOB.clients
+
 	// Build a message
 	var/message = list()
 	if(type) message["type"] = type

tgui/packages/tgui-panel/chat/middleware.js

@@ -56,6 +56,8 @@ const loadChatFromStorage = async (store) => {
 export const chatMiddleware = (store) => {
   let initialized = false;
   let loaded = false;
+  const sequences = [];
+  const sequences_requested = [];
   chatRenderer.events.on('batchProcessed', (countByType) => {
     // Use this flag to workaround unread messages caused by
     // loading them from storage. Side effect of that, is that
@@ -77,9 +79,37 @@ export const chatMiddleware = (store) => {
       loadChatFromStorage(store);
     }
     if (type === 'chat/message') {
-      // Normalize the payload
-      const batch = Array.isArray(payload) ? payload : [payload];
-      chatRenderer.processBatch(batch);
+      let payload_obj;
+      try {
+        payload_obj = JSON.parse(payload);
+      } catch (err) {
+        return;
+      }
+
+      const sequence = payload_obj.sequence;
+      if (sequences.includes(sequence)) {
+        return;
+      }
+
+      const sequence_count = sequences.length;
+      seq_check: if (sequence_count > 0) {
+        if (sequences_requested.includes(sequence)) {
+          sequences_requested.splice(sequences_requested.indexOf(sequence), 1);
+          // if we are receiving a message we requested, we can stop reliability checks
+          break seq_check;
+        }
+
+        // cannot do reliability if we don't have any messages
+        const expected_sequence = sequences[sequence_count - 1] + 1;
+        if (sequence !== expected_sequence) {
+          for (let requesting = expected_sequence; requesting < sequence; requesting++) {
+            requested_sequences.push(requesting);
+            Byond.sendMessage('chat/resend', requesting);
+          }
+        }
+      }
+
+      chatRenderer.processBatch([payload_obj.content]);
       return;
     }
     if (type === loadChat.type) {