From 3e4ae86c38c4cef9f2af6a808de9c777c2affa69 Mon Sep 17 00:00:00 2001
From: Natty <natty.sh.git@gmail.com>
Date: Thu, 19 Dec 2024 16:58:10 +0100
Subject: [PATCH 1/2] Optimizations of RPC and delivery

---
 Cargo.lock                      |  1 +
 Cargo.toml                      |  1 +
 ext_federation/Cargo.toml       |  1 +
 ext_federation/src/ap_client.rs | 19 +++++----
 ext_federation/src/crypto.rs    |  6 +--
 ext_federation/src/lib.rs       |  2 +-
 src/rpc_v1/mod.rs               | 10 ++---
 src/rpc_v1/proto.rs             | 74 +++++++++++++++++----------------
 8 files changed, 60 insertions(+), 54 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 343d5ad..d210d42 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2028,6 +2028,7 @@ dependencies = [
  "miette 7.2.0",
  "percent-encoding",
  "quick-xml",
+ "rand",
  "reqwest",
  "rsa",
  "serde",
diff --git a/Cargo.toml b/Cargo.toml
index 50145eb..068f26f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -62,6 +62,7 @@ quick-xml = "0.36"
 redis = "0.26"
 regex = "1.9"
 rmp-serde = "1.3"
+rand = "0.8"
 rsa = "0.9"
 reqwest = "0.12"
 sea-orm = "1"
diff --git a/ext_federation/Cargo.toml b/ext_federation/Cargo.toml
index 15e3f58..fa9e8e7 100644
--- a/ext_federation/Cargo.toml
+++ b/ext_federation/Cargo.toml
@@ -40,6 +40,7 @@ hyper = { workspace = true, features = ["full"] }
 percent-encoding = { workspace = true }
 reqwest = { workspace = true, features = ["stream", "hickory-dns"] }
 
+rand = { workspace = true }
 ed25519-dalek = { workspace = true, features = [
     "pem",
     "pkcs8",
diff --git a/ext_federation/src/ap_client.rs b/ext_federation/src/ap_client.rs
index 264d38d..4e42b10 100644
--- a/ext_federation/src/ap_client.rs
+++ b/ext_federation/src/ap_client.rs
@@ -4,6 +4,7 @@ use http::{HeaderMap, HeaderName, HeaderValue, Method};
 use indexmap::IndexSet;
 use serde_json::Value;
 use sha2::Digest;
+use std::fmt::Write;
 use std::{fmt::Display, string::FromUtf8Error, sync::Arc};
 use thiserror::Error;
 use tokio::task;
@@ -257,16 +258,17 @@ impl ApClientService for ApClientServiceDefaultProvider {
 
         let message = components
             .iter()
-            .map(|(k, v)| format!("{}: {}", k.as_ref(), v))
-            .collect::<Vec<_>>()
-            .join("\n");
+            .fold(String::new(), |mut acc, (k, v)| {
+                writeln!(&mut acc, "{}: {}", k.as_ref(), v).unwrap();
+                acc
+            });
 
         let key_id = signing_key.key_id.clone().into_owned();
         let key = signing_key.into_owned();
         let signature = task::spawn_blocking(move || {
             key
                 .key
-                .sign_base64(signing_algorithm, &message.into_bytes())
+                .sign_base64(signing_algorithm, message.trim_end().as_bytes())
         }).await??;
 
         Ok(ApSignature {
@@ -329,7 +331,7 @@ impl ApClientService for ApClientServiceDefaultProvider {
         }
 
         headers.insert(
-            HeaderName::from_lowercase(b"signature").unwrap(),
+            HeaderName::from_static("signature"),
             HeaderValue::try_from(signed.to_string())?,
         );
 
@@ -350,10 +352,9 @@ impl ApClientService for ApClientServiceDefaultProvider {
         signing_algorithm: SigningAlgorithm,
         expires: Option<chrono::DateTime<Utc>>,
         url: &str,
-        body: &Value,
+        body_bytes: Vec<u8>,
     ) -> Result<String, Self::Error> {
         let url = url.parse()?;
-        let body_bytes = serde_json::to_vec(body)?;
         // Move in, move out :3
         let (digest_raw, body_bytes) = task::spawn_blocking(move || {
             let mut sha = sha2::Sha256::new();
@@ -406,12 +407,12 @@ impl ApClientService for ApClientServiceDefaultProvider {
         }
 
         headers.insert(
-            HeaderName::from_lowercase(b"digest").unwrap(),
+            HeaderName::from_static("digest"),
             HeaderValue::try_from(digest_base64)?,
         );
 
         headers.insert(
-            HeaderName::from_lowercase(b"signature").unwrap(),
+            HeaderName::from_static("signature"),
             HeaderValue::try_from(signed.to_string())?,
         );
 
diff --git a/ext_federation/src/crypto.rs b/ext_federation/src/crypto.rs
index 1b25886..e8af944 100644
--- a/ext_federation/src/crypto.rs
+++ b/ext_federation/src/crypto.rs
@@ -2,7 +2,7 @@ use rsa::pkcs1::DecodeRsaPrivateKey;
 use rsa::pkcs1::DecodeRsaPublicKey;
 use rsa::pkcs8::DecodePrivateKey;
 use rsa::pkcs8::DecodePublicKey;
-use rsa::signature::Verifier;
+use rsa::signature::{RandomizedSigner, Verifier};
 use rsa::{
     sha2::{Sha256, Sha512},
     signature::Signer,
@@ -268,10 +268,10 @@ impl ApHttpSigningKey<'_> {
     ) -> Result<Vec<u8>, ApSigningError> {
         match (self, algorithm) {
             (Self::RsaSha256(key), SigningAlgorithm::RsaSha256 | SigningAlgorithm::Hs2019) => {
-                Ok(Box::<[u8]>::from(key.sign(message)).into_vec())
+                Ok(Box::<[u8]>::from(key.sign_with_rng(&mut rand::thread_rng(), message)).into_vec())
             }
             (Self::RsaSha512(key), SigningAlgorithm::Hs2019) => {
-                Ok(Box::<[u8]>::from(key.sign(message)).into_vec())
+                Ok(Box::<[u8]>::from(key.sign_with_rng(&mut rand::thread_rng(), message)).into_vec())
             }
             (Self::Ed25519(key), SigningAlgorithm::Hs2019) => {
                 Ok(key.sign(message).to_bytes().to_vec())
diff --git a/ext_federation/src/lib.rs b/ext_federation/src/lib.rs
index 15852c1..0136ffe 100644
--- a/ext_federation/src/lib.rs
+++ b/ext_federation/src/lib.rs
@@ -171,6 +171,6 @@ pub trait ApClientService: Send + Sync {
         signing_algorithm: SigningAlgorithm,
         expires: Option<chrono::DateTime<Utc>>,
         url: &str,
-        body: &Value,
+        body: Vec<u8>,
     ) -> Result<String, Self::Error>;
 }
diff --git a/src/rpc_v1/mod.rs b/src/rpc_v1/mod.rs
index 6ddb1ce..4e0f8fe 100644
--- a/src/rpc_v1/mod.rs
+++ b/src/rpc_v1/mod.rs
@@ -23,7 +23,7 @@ struct RpcApGet {
 struct RpcApPost {
     user_id: String,
     url: String,
-    body: serde_json::Value,
+    body: String,
 }
 
 pub fn create_rpc_router() -> MagRpc {
@@ -43,9 +43,9 @@ pub fn create_rpc_router() -> MagRpc {
                     attachments: true,
                     with_context: true,
                 }
-                .fetch_single(&ctx, &id)
-                .await?
-                .ok_or(ObjectNotFound(id))?;
+                    .fetch_single(&ctx, &id)
+                    .await?
+                    .ok_or(ObjectNotFound(id))?;
 
                 Result::<_, ApiError>::Ok(note)
             },
@@ -90,7 +90,7 @@ pub fn create_rpc_router() -> MagRpc {
                     .create_signing_key(&key_id, SigningAlgorithm::RsaSha256)?;
                 let result = service
                     .ap_client
-                    .signed_post(signing_key, SigningAlgorithm::RsaSha256, None, &url, &body)
+                    .signed_post(signing_key, SigningAlgorithm::RsaSha256, None, &url, body.into_bytes())
                     .await?;
                 Result::<_, DeliveryError>::Ok(result)
             },
diff --git a/src/rpc_v1/proto.rs b/src/rpc_v1/proto.rs
index 9e9cb59..41c99ec 100644
--- a/src/rpc_v1/proto.rs
+++ b/src/rpc_v1/proto.rs
@@ -12,7 +12,7 @@ use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
 use std::sync::Arc;
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter};
 use tokio::net::{TcpListener, UnixSocket};
 use tokio::select;
 use tokio::task::JoinSet;
@@ -57,7 +57,7 @@ pub struct RpcResult<T> {
 }
 
 impl<T: Serialize + Send + 'static, E: Serialize + Debug + Send + 'static> IntoRpcResponse
-    for Result<T, E>
+for Result<T, E>
 {
     fn into_rpc_response(self) -> Option<RpcResponse> {
         match self {
@@ -65,21 +65,21 @@ impl<T: Serialize + Send + 'static, E: Serialize + Debug + Send + 'static> IntoR
                 success: true,
                 data,
             })
-            .into_rpc_response(),
+                .into_rpc_response(),
             Err(data) => {
                 warn!("{:?}", data);
                 RpcMessage(RpcResult {
                     success: false,
                     data,
                 })
-                .into_rpc_response()
+                    .into_rpc_response()
             }
         }
     }
 }
 
 impl<L: IntoRpcResponse + Send + 'static, R: IntoRpcResponse + Send + 'static> IntoRpcResponse
-    for Either<L, R>
+for Either<L, R>
 {
     fn into_rpc_response(self) -> Option<RpcResponse> {
         match self {
@@ -97,14 +97,14 @@ where
         &self,
         context: Arc<MagnetarService>,
         message: RpcMessage<T>,
-    ) -> impl Future<Output = Option<RpcResponse>> + Send;
+    ) -> impl Future<Output=Option<RpcResponse>> + Send;
 }
 
 impl<T, F, Fut, RR> RpcHandler<T> for F
 where
     T: Send + 'static,
     F: Fn(Arc<MagnetarService>, RpcMessage<T>) -> Fut + Send + Sync + 'static,
-    Fut: Future<Output = RR> + Send,
+    Fut: Future<Output=RR> + Send,
     RR: IntoRpcResponse,
 {
     async fn process(
@@ -119,15 +119,15 @@ where
 type MessageRaw = Box<dyn Any + Send + 'static>;
 
 type MagRpcHandlerMapped = dyn Fn(
-        Arc<MagnetarService>,
-        MessageRaw,
-    ) -> Pin<Box<dyn Future<Output = Option<RpcResponse>> + Send + 'static>>
-    + Send
-    + Sync
-    + 'static;
+    Arc<MagnetarService>,
+    MessageRaw,
+) -> Pin<Box<dyn Future<Output=Option<RpcResponse>> + Send + 'static>>
++ Send
++ Sync
++ 'static;
 
 type MagRpcDecoderMapped =
-    dyn (Fn(&'_ [u8]) -> Result<MessageRaw, rmp_serde::decode::Error>) + Send + Sync + 'static;
+dyn (Fn(&'_ [u8]) -> Result<MessageRaw, rmp_serde::decode::Error>) + Send + Sync + 'static;
 
 pub struct MagRpc {
     listeners: HashMap<String, Arc<MagRpcHandlerMapped>>,
@@ -158,7 +158,7 @@ impl MagRpc {
                         .process(ctx, RpcMessage(*data.downcast().unwrap()))
                         .await
                 }
-                .boxed()
+                    .boxed()
             }),
         );
         self.payload_decoders.insert(
@@ -173,7 +173,7 @@ impl MagRpc {
         self,
         context: Arc<MagnetarService>,
         addr: RpcSockAddr,
-        graceful_shutdown: Option<impl Future<Output = ()>>,
+        graceful_shutdown: Option<impl Future<Output=()>>,
     ) -> miette::Result<()> {
         match addr {
             RpcSockAddr::Ip(sock_addr) => {
@@ -187,7 +187,7 @@ impl MagRpc {
         self,
         context: Arc<MagnetarService>,
         sock_addr: &SocketAddr,
-        graceful_shutdown: Option<impl Future<Output = ()>>,
+        graceful_shutdown: Option<impl Future<Output=()>>,
     ) -> miette::Result<()> {
         debug!("Binding RPC TCP socket to {}", sock_addr);
         let listener = TcpListener::bind(sock_addr).await.into_diagnostic()?;
@@ -223,15 +223,16 @@ impl MagRpc {
             debug!("RPC TCP connection accepted: {:?}", remote_addr);
 
             let (cancel_send, cancel_recv) = tokio::sync::oneshot::channel::<()>();
-            let (read_half, mut write_half) = stream.into_split();
+            let (read_half, write_half) = stream.into_split();
             let buf_read = BufReader::new(read_half);
+            let mut buf_write = BufWriter::new(write_half);
             let context = context.clone();
             let rx_dec = rx_dec.clone();
             let fut = async move {
                 let src = rx_dec
                     .stream_decode(buf_read, cancel_recv)
                     .map_ok(process(context))
-                    .try_buffer_unordered(100)
+                    .try_buffer_unordered(400)
                     .boxed();
 
                 futures::pin_mut!(src);
@@ -241,19 +242,19 @@ impl MagRpc {
                         continue;
                     };
 
-                    write_half.write_u8(b'M').await.into_diagnostic()?;
-                    write_half.write_u64(serial).await.into_diagnostic()?;
-                    write_half
+                    buf_write.write_u8(b'M').await.into_diagnostic()?;
+                    buf_write.write_u64(serial).await.into_diagnostic()?;
+                    buf_write
                         .write_u32(bytes.len() as u32)
                         .await
                         .into_diagnostic()?;
-                    write_half.write_all(&bytes).await.into_diagnostic()?;
-                    write_half.flush().await.into_diagnostic()?;
+                    buf_write.write_all(&bytes).await.into_diagnostic()?;
+                    buf_write.flush().await.into_diagnostic()?;
                 }
 
                 Ok(remote_addr)
             }
-            .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
+                .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
 
             connections.spawn(fut);
 
@@ -276,7 +277,7 @@ impl MagRpc {
         self,
         context: Arc<MagnetarService>,
         addr: &Path,
-        graceful_shutdown: Option<impl Future<Output = ()>>,
+        graceful_shutdown: Option<impl Future<Output=()>>,
     ) -> miette::Result<()> {
         let sock = UnixSocket::new_stream().into_diagnostic()?;
         debug!("Binding RPC Unix socket to {}", addr.display());
@@ -313,15 +314,16 @@ impl MagRpc {
             debug!("RPC Unix connection accepted: {:?}", remote_addr);
 
             let (cancel_send, cancel_recv) = tokio::sync::oneshot::channel::<()>();
-            let (read_half, mut write_half) = stream.into_split();
+            let (read_half, write_half) = stream.into_split();
             let buf_read = BufReader::new(read_half);
+            let mut buf_write = BufWriter::new(write_half);
             let context = context.clone();
             let rx_dec = rx_dec.clone();
             let fut = async move {
                 let src = rx_dec
                     .stream_decode(buf_read, cancel_recv)
                     .map_ok(process(context))
-                    .try_buffer_unordered(100)
+                    .try_buffer_unordered(400)
                     .boxed();
 
                 futures::pin_mut!(src);
@@ -331,19 +333,19 @@ impl MagRpc {
                         continue;
                     };
 
-                    write_half.write_u8(b'M').await.into_diagnostic()?;
-                    write_half.write_u64(serial).await.into_diagnostic()?;
-                    write_half
+                    buf_write.write_u8(b'M').await.into_diagnostic()?;
+                    buf_write.write_u64(serial).await.into_diagnostic()?;
+                    buf_write
                         .write_u32(bytes.len() as u32)
                         .await
                         .into_diagnostic()?;
-                    write_half.write_all(&bytes).await.into_diagnostic()?;
-                    write_half.flush().await.into_diagnostic()?;
+                    buf_write.write_all(&bytes).await.into_diagnostic()?;
+                    buf_write.flush().await.into_diagnostic()?;
                 }
 
                 miette::Result::<()>::Ok(())
             }
-            .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
+                .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
 
             connections.spawn(fut.boxed());
 
@@ -368,7 +370,7 @@ fn process(
 ) -> impl Fn(
     (u64, MessageRaw, Arc<MagRpcHandlerMapped>),
 ) -> Pin<
-    Box<dyn Future<Output = miette::Result<Option<(u64, RpcResponse)>>> + Send + 'static>,
+    Box<dyn Future<Output=miette::Result<Option<(u64, RpcResponse)>>> + Send + 'static>,
 > {
     move |(serial, payload, listener)| {
         let ctx = context.clone();
@@ -389,7 +391,7 @@ impl RpcCallDecoder {
         &self,
         mut buf_read: BufReader<R>,
         mut cancel: tokio::sync::oneshot::Receiver<()>,
-    ) -> impl Stream<Item = miette::Result<(u64, MessageRaw, Arc<MagRpcHandlerMapped>)>> + Send + 'static
+    ) -> impl Stream<Item=miette::Result<(u64, MessageRaw, Arc<MagRpcHandlerMapped>)>> + Send + 'static
     {
         let decoders = self.payload_decoders.clone();
         let listeners = self.listeners.clone();
-- 
2.40.1


From 647006d04f0035c1b810cb0bd26918b6b179561a Mon Sep 17 00:00:00 2001
From: Natty <natty.sh.git@gmail.com>
Date: Fri, 20 Dec 2024 00:23:37 +0100
Subject: [PATCH 2/2] Frontend: Fixed note header rendering

---
 .../client/src/components/MagNoteHeader.vue   |  5 +++--
 .../src/components/MkInstanceTicker.vue       | 19 +++++++++----------
 2 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/fe_calckey/frontend/client/src/components/MagNoteHeader.vue b/fe_calckey/frontend/client/src/components/MagNoteHeader.vue
index dcad883..b721592 100644
--- a/fe_calckey/frontend/client/src/components/MagNoteHeader.vue
+++ b/fe_calckey/frontend/client/src/components/MagNoteHeader.vue
@@ -74,10 +74,10 @@ const showTicker =
     display: flex;
     align-items: center;
     white-space: nowrap;
-    justify-self: flex-end;
     border-radius: 100px;
     font-size: 0.8em;
     text-shadow: 0 2px 2px var(--shadow);
+    gap: 0 0.1em;
 
     > .avatar {
         width: 3.7em;
@@ -99,6 +99,8 @@ const showTicker =
                 overflow: hidden;
                 text-overflow: ellipsis;
                 gap: 0.1em 0;
+                display: flex;
+                flex-wrap: wrap;
             }
 
             &:last-child {
@@ -151,7 +153,6 @@ const showTicker =
             margin: 0 0.5em 0 0;
             overflow: hidden;
             text-overflow: ellipsis;
-            align-self: flex-start;
             font-size: 0.9em;
         }
 
diff --git a/fe_calckey/frontend/client/src/components/MkInstanceTicker.vue b/fe_calckey/frontend/client/src/components/MkInstanceTicker.vue
index 21da639..98ef556 100644
--- a/fe_calckey/frontend/client/src/components/MkInstanceTicker.vue
+++ b/fe_calckey/frontend/client/src/components/MkInstanceTicker.vue
@@ -4,7 +4,7 @@
         v-tooltip="
             capitalize(
                 magTransProperty(instance, 'software_name', 'softwareName') ??
-                    '?',
+                    '?'
             )
         "
         ref="ticker"
@@ -40,7 +40,7 @@ const instance = props.instance ?? {
     name: instanceName,
     themeColor: (
         document.querySelector(
-            'meta[name="theme-color-orig"]',
+            'meta[name="theme-color-orig"]'
         ) as HTMLMetaElement
     )?.content,
     softwareName: (Instance.softwareName || "Magnetar") as string | null,
@@ -61,18 +61,18 @@ const bg = {
 };
 
 function getInstanceIcon(
-    instance?: Misskey.entities.User["instance"] | types.InstanceTicker | null,
+    instance?: Misskey.entities.User["instance"] | types.InstanceTicker | null
 ): string {
     if (!instance) return "/client-assets/dummy.png";
 
     return (
         getProxiedImageUrlNullable(
             magTransProperty(instance, "icon_url", "iconUrl"),
-            "preview",
+            "preview"
         ) ??
         getProxiedImageUrlNullable(
             magTransProperty(instance, "favicon_url", "faviconUrl"),
-            "preview",
+            "preview"
         ) ??
         "/client-assets/dummy.png"
     );
@@ -90,6 +90,7 @@ function getInstanceIcon(
     font-size: 0.8em;
     text-shadow: 0 2px 2px var(--shadow);
     overflow: hidden;
+
     .header > .body & {
         width: max-content;
         max-width: 100%;
@@ -108,11 +109,9 @@ function getInstanceIcon(
         font-weight: bold;
         text-overflow: ellipsis;
         white-space: nowrap;
-        text-shadow:
-            -1px -1px 0 var(--bg),
-            1px -1px 0 var(--bg),
-            -1px 1px 0 var(--bg),
-            1px 1px 0 var(--bg);
+        text-shadow: -1px -1px 0 var(--bg), 1px -1px 0 var(--bg),
+            -1px 1px 0 var(--bg), 1px 1px 0 var(--bg);
+
         .article > .main &,
         .header > .body & {
             display: unset;
-- 
2.40.1