add: client streams api

fix: mobile chats ui

crates/app/src/assets.rs

@@ -29,6 +29,7 @@ pub const STYLE_CSS: &str = include_str!("./public/css/style.css");
 pub const LOADER_JS: &str = include_str!("./public/js/loader.js");
 pub const ATTO_JS: &str = include_str!("./public/js/atto.js");
 pub const ME_JS: &str = include_str!("./public/js/me.js");
+pub const STREAMS_JS: &str = include_str!("./public/js/streams.js");
 
 // html
 pub const ROOT: &str = include_str!("./public/html/root.html");

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

@@ -250,6 +250,7 @@
             height: calc(100dvh - 42px);
             overflow: auto;
             transition: left 0.15s;
+            z-index: 1;
         }
 
         .sidebar .title {
@@ -286,14 +287,8 @@
             display: flex !important;
         }
 
-        .floating_message_actions {
-            position: absolute;
-            top: 0.25rem;
-            right: 1rem;
-        }
-
         .message.grouped {
-            padding: 0.25rem 0 0.25rem calc(1rem + 0.5rem + 52px);
+            padding: 0.25rem 1rem 0.25rem calc(1rem + 0.5rem + 52px);
         }
         turbo-frame {
             display: contents;
@@ -301,7 +296,7 @@
 
         @media screen and (max-width: 900px) {
             .message.grouped {
-                padding: 0.25rem 0 0.25rem calc(1rem + 0.5rem + 39px);
+                padding: 0.25rem 1rem 0.25rem calc(1rem + 0.5rem + 39px);
             }
 
             body:not(.sidebars_shown) .sidebar {

crates/app/src/public/html/components.html

@@ -959,14 +959,18 @@ can_manage_message=false, grouped=false) -%}
                 can_manage_message=can_manage_message) }}
             </div>
         </div>
-        {% else %}
-        <div class="floating_message_actions hidden">
-            {{ self::message_actions(user=user, message=message,
-            can_manage_message=can_manage_message) }}
-        </div>
         {% endif %}
 
-        <span class="no_p_margin">{{ message.content|markdown|safe }}</span>
+        <div class="flex w-full gap-2 justify-between">
+            <span class="no_p_margin">{{ message.content|markdown|safe }}</span>
+
+            {% if grouped %}
+            <div class="hidden">
+                {{ self::message_actions(user=user, message=message,
+                can_manage_message=can_manage_message) }}
+            </div>
+            {% endif %}
+        </div>
     </div>
 </div>
 {%- endmacro %}

crates/app/src/public/html/root.html

@@ -125,6 +125,7 @@ macros -%}
         <script data-turbo-permanent="true" id="update-seen-script">
             document.documentElement.addEventListener("turbo:load", () => {
                 trigger("me::seen");
+                trigger("streams::user", ["{{ user.id }}"]);
 
                 if (!window.location.pathname.startsWith("/chats/")) {
                     if (window.socket) {

crates/app/src/public/js/atto.js

@@ -38,6 +38,7 @@ media_theme_pref();
 
     // init
     use("me", () => {});
+    use("streams", () => {});
 
     // env
     self.DEBOUNCE = [];

crates/app/src/public/js/streams.js

@@ -0,0 +1,85 @@
+(() => {
+    const self = reg_ns("streams");
+
+    self.STREAMS = {};
+    self.USER = null;
+
+    self.define("user", ({ $ }, user_id) => {
+        $.USER = user_id;
+    });
+
+    self.define("sock", ({ $ }, stream) => {
+        return $.STREAMS[stream];
+    });
+
+    self.define("subscribe", ({ $ }, stream) => {
+        if (!$.USER) {
+            console.warn("cannot subscribe without user id");
+            return;
+        }
+
+        if ($.STREAMS[stream]) {
+            console.warn("stream already exists");
+            return $.STREAMS[stream];
+        }
+
+        const endpoint = `${window.location.origin.replace("http", "ws")}/api/v1/auth/user/${$.USER}/_connect/${stream}`;
+        const socket = new WebSocket(endpoint);
+
+        $.STREAMS[stream] = {
+            socket,
+            events: {
+                message: () => {},
+            },
+        };
+
+        socket.addEventListener("message", (event) => {
+            if (event.data === "Ping") {
+                return socket.send("Pong");
+            }
+
+            return $.sock(stream).events.message(event.data);
+        });
+
+        return socket;
+    });
+
+    self.define("close", ({ $ }, stream) => {
+        const socket = $.sock(stream);
+
+        if (!socket) {
+            console.warn("no such stream to close");
+            return;
+        }
+
+        socket.socket.send("Close");
+        socket.socket.close();
+    });
+
+    self.define("event", ({ $ }, stream, event, handler) => {
+        const socket = $.sock(stream);
+
+        if (!socket) {
+            console.warn("no such stream to add event to");
+            return;
+        }
+
+        socket.events[event] = handler;
+        socket.socket.addEventListener(event, handler);
+    });
+
+    self.define("send_packet", async ({ $ }, stream, method, data) => {
+        await (
+            await fetch(`/api/v1/auth/user/${$.USER}/_connect/${stream}/send`, {
+                method: "POST",
+                headers: {
+                    "Content-Type": "application/json",
+                },
+                body: JSON.stringify({
+                    method,
+                    data: JSON.stringify(data),
+                }),
+            })
+        ).json();
+    });
+})();

crates/app/src/routes/api/v1/auth/profile.rs

@@ -454,7 +454,7 @@ pub async fn handle_socket(socket: WebSocket, db: DataManager, user_id: String,
     let (mut sink, mut stream) = socket.split();
 
     // get channel permissions
-    let channel = format!("{user_id}_{stream_id}");
+    let channel = format!("{user_id}/{stream_id}");
 
     // ...
     let mut recv_task = tokio::spawn(async move {
@@ -510,8 +510,8 @@ pub async fn handle_socket(socket: WebSocket, db: DataManager, user_id: String,
         let mut heartbeat = tokio::time::interval(Duration::from_secs(10));
 
         loop {
-            con.publish::<String, String, ()>(
-                format!("{channel_c}_heartbeat"),
+            con.publish::<&str, String, ()>(
+                &channel_c,
                 serde_json::to_string(&SocketMessage {
                     method: SocketMethod::Forward(PacketType::Ping),
                     data: "Ping".to_string(),
@@ -532,3 +532,33 @@ pub async fn handle_socket(socket: WebSocket, db: DataManager, user_id: String,
     heartbeat_task.abort(); // kill
     tracing::info!("socket terminate");
 }
+
+pub async fn post_to_socket_request(
+    jar: CookieJar,
+    Extension(data): Extension<State>,
+    Path((user_id, id)): Path<(String, String)>,
+    Json(msg): Json<SocketMessage>,
+) -> impl IntoResponse {
+    let data = &(data.read().await).0;
+    let user = match get_user_from_token!(jar, data) {
+        Some(ua) => ua,
+        None => return Json(Error::NotAllowed.into()),
+    };
+
+    if user.id.to_string() != user_id {
+        return Json(Error::NotAllowed.into());
+    }
+
+    let mut con = data.2.get_con().await;
+    con.publish::<String, String, ()>(
+        format!("{user_id}/{id}"),
+        serde_json::to_string(&msg).unwrap(),
+    )
+    .unwrap();
+
+    Json(ApiReturn {
+        ok: true,
+        message: "Data sent to socket".to_string(),
+        payload: (),
+    })
+}

crates/app/src/routes/api/v1/mod.rs

@@ -193,7 +193,11 @@ pub fn routes() -> Router {
         .route("/auth/ip/{ip}/block", post(auth::social::ip_block_request))
         .route(
             "/auth/user/{id}/_connect/{stream}",
-            get(auth::profile::subscription_handler),
+            any(auth::profile::subscription_handler),
+        )
+        .route(
+            "/auth/user/{id}/_connect/{stream}/send",
+            post(auth::profile::post_to_socket_request),
         )
         // warnings
         .route("/warnings/{id}", post(auth::user_warnings::create_request))

crates/app/src/routes/assets.rs

@@ -14,3 +14,4 @@ serve_asset!(style_css_request: STYLE_CSS("text/css"));
 serve_asset!(loader_js_request: LOADER_JS("text/javascript"));
 serve_asset!(atto_js_request: ATTO_JS("text/javascript"));
 serve_asset!(me_js_request: ME_JS("text/javascript"));
+serve_asset!(streams_js_request: STREAMS_JS("text/javascript"));

crates/app/src/routes/mod.rs

@@ -15,6 +15,7 @@ pub fn routes(config: &Config) -> Router {
         .route("/js/loader.js", get(assets::loader_js_request))
         .route("/js/atto.js", get(assets::atto_js_request))
         .route("/js/me.js", get(assets::me_js_request))
+        .route("/js/streams.js", get(assets::streams_js_request))
         .nest_service(
             "/public",
             get_service(tower_http::services::ServeDir::new(&config.dirs.assets)),

crates/app/src/routes/pages/chats.rs

@@ -158,7 +158,7 @@ pub async fn stream_request(
     let ignore_users = data.0.get_userblocks_receivers(user.id).await;
     let messages = match data
         .0
-        .get_messages_by_channel(channel, 12, props.page)
+        .get_messages_by_channel(channel, 24, props.page)
         .await
     {
         Ok(p) => match data.0.fill_messages(p, &ignore_users).await {

crates/core/src/model/socket.rs

@@ -4,6 +4,8 @@ use serde::{Serialize, Deserialize, de::DeserializeOwned};
 pub enum PacketType {
     /// A regular check to ensure the connection is still alive.
     Ping,
+    /// General text which can be ignored.
+    Text,
 }
 
 #[derive(Serialize, Deserialize, PartialEq, Eq)]
@@ -16,6 +18,8 @@ pub enum SocketMethod {
     Delete,
     /// Forward message from server to client. (Redis pubsub to ws)
     Forward(PacketType),
+    /// A general packet from client to server. (ws to Redis pubsub)
+    Misc(PacketType),
 }
 
 #[derive(Serialize, Deserialize)]
@@ -29,3 +33,9 @@ impl SocketMessage {
         serde_json::from_str(&self.data).unwrap()
     }
 }
+
+/// [`PacketType::Text`]
+#[derive(Serialize, Deserialize, PartialEq, Eq)]
+pub struct TextMessage {
+    pub text: String,
+}