add: websocket pings

crates/app/src/public/html/chats/app.html

@@ -135,7 +135,6 @@
         <a
             class="w-full justify-start button {% if selected_channel == channel.id %}quaternary{% else %}camo{% endif %}"
             href="/chats/{{ selected_community }}/{{ channel.id }}"
-            data-turbo="false"
         >
             {{ icon "rss" }}
             <b class="name shortest">{{ channel.title }}</b>
@@ -494,6 +493,7 @@
                 const data = JSON.parse(msg.data);
 
                 if (msg.method === "Message" && window.CURRENT_PAGE === 0) {
+                    if (document.getElementById("stream_body")) {
                         const element = document.createElement("div");
                         element.style.display = "contents";
                         element.innerHTML = await (
@@ -511,6 +511,10 @@
 
                         document.getElementById("stream_body").prepend(element);
                         clean_text();
+                    } else {
+                        console.log("abandoned remote");
+                        socket.close();
+                    }
                 } else if (msg.method === "Delete") {
                     if (document.getElementById(`message-${data.id}`)) {
                         document.getElementById(`message-${data.id}`).remove();

crates/app/src/routes/api/v1/channels/messages.rs

@@ -16,7 +16,7 @@ use tetratto_core::{
         ApiReturn, Error,
     },
 };
-use std::sync::mpsc;
+use std::{sync::mpsc, time::Duration};
 use crate::{get_user_from_token, routes::api::v1::CreateMessage, State};
 use serde::Deserialize;
 use futures_util::{sink::SinkExt, stream::StreamExt};
@@ -47,7 +47,7 @@ pub async fn handle_socket(socket: WebSocket, state: State, channel_id: usize) {
     tokio::spawn(async move {
         while let Ok(message) = receiver.recv() {
             if message == "Close" {
-                sink.close().await.unwrap();
+                let _ = sink.close().await;
                 drop(receiver);
                 break;
             }
@@ -58,6 +58,20 @@ pub async fn handle_socket(socket: WebSocket, state: State, channel_id: usize) {
         }
     });
 
+    // ping
+    let ping_sender = sender.clone();
+    tokio::spawn(async move {
+        let mut heartbeat = tokio::time::interval(Duration::from_secs(30));
+
+        loop {
+            heartbeat.tick().await;
+            if ping_sender.send("Ping".to_string()).is_err() {
+                // remote has abandoned us
+                break;
+            }
+        }
+    });
+
     // ...
     let mut user: Option<User> = None;
     let mut con = db.2.clone().get_con().await;