Ports "Adds a Chat Reliability Layer" 2: The Reliabaling

code/__DEFINES/chat.dm

@@ -2,6 +2,11 @@
  * Copyright (c) 2020 Aleksej Komarov
  * 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"

code/controllers/subsystem/chat.dm

@@ -5,39 +5,97 @@
 
 SUBSYSTEM_DEF(chat)
 	name = "Chat"
-	flags = SS_TICKER
+	flags = SS_TICKER|SS_NO_INIT
 	wait = 1
 	priority = SS_PRIORITY_CHAT
 	init_order = SS_INIT_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()
 
-/datum/controller/subsystem/chat/Initialize()
-	// Just used by chat system to know that initialization is nearly finished.
-	// The to_chat checks could probably check the runlevel instead, but would require testing.
-	return SS_INIT_SUCCESS
+	/// 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()
+
+	/// Keeps track of resends to see how often chat bugs out
+	var/resends = 0
+
+/datum/controller/subsystem/chat/stat_entry(msg)
+	msg = "Messages resent: [resends]"
+	return ..()
+
+/datum/controller/subsystem/chat/Shutdown()
+	log_world("SSchat shutting down, Number of messages resent: [resends]")
+	return ..()
+
+/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/datum/chat_payload/payload = client_history[sequence]
+	if(payload.resends > CHAT_RELIABILITY_MAX_RESENDS)
 		return
-	var/client/client = CLIENT_FROM_VAR(target)
-	if(client)
-		LAZYADD(payload_by_client[client], list(message))
+
+	payload.resends += 1
+	resends += 1
+	send_payload_to_client(client, client_history[sequence])

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

@@ -43,23 +43,8 @@
 	if(text) message["text"] = text
 	if(html) message["html"] = html
 	if(avoid_highlighting) message["avoidHighlighting"] = avoid_highlighting
-	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).

code/modules/tgui/tgui_window.dm

@@ -384,6 +384,8 @@
 			client << link(href_list["url"])
 		if("cacheReloaded")
 			reinitialize()
+		if("chat/resend")
+			SSchat.handle_resend(client, payload)
 
 /*
 /datum/tgui_window/vv_edit_var(var_name, var_value)

colonialmarines.dme

@@ -341,6 +341,7 @@
 #include "code\datums\bug_report.dm"
 #include "code\datums\callback.dm"
 #include "code\datums\changelog.dm"
+#include "code\datums\chat_payload.dm"
 #include "code\datums\combat_personalized.dm"
 #include "code\datums\computerfiles.dm"
 #include "code\datums\crew_manifest.dm"

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

@@ -82,6 +82,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
@@ -103,9 +105,40 @@ 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 {
+        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++
+          ) {
+            sequences_requested.push(requesting);
+            Byond.sendMessage('chat/resend', requesting);
+          }
+        }
+      }
+
+      sequences.push(sequence);
+      chatRenderer.processBatch([payload_obj.content]);
       return;
     }
     if (type === loadChat.type) {