feat: 收口 Rust Web 3000 主链

rust/crates/mnote-web/src/routes/gateway.rs

@@ -13,12 +13,15 @@ use crate::workspace_shell::{
     build_workspace_shell_projection, render_workspace_shell_sidebar_html,
 };
 use axum::body::Body;
+use axum::extract::ws::{Message as AxumWsMessage, WebSocket, WebSocketUpgrade};
 use axum::extract::{Extension, Query, State};
-use axum::http::{header, HeaderName, HeaderValue, Request, StatusCode};
+use axum::http::{header, HeaderName, HeaderValue, Request, StatusCode, Uri};
 use axum::response::{Html, IntoResponse, Response};
+use futures_util::{SinkExt, StreamExt};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use std::time::Duration;
+use tokio_tungstenite::tungstenite::Message as TungsteniteMessage;
 
 const HEADER_MNOTE_WEB_OWNER: &str = "x-mnote-web-owner";
 const HEADER_MNOTE_LEGACY_UPSTREAM: &str = "x-mnote-legacy-upstream";
@@ -482,6 +485,132 @@ pub async fn legacy_next_proxy(
     Ok(response)
 }
 
+#[allow(dead_code)]
+pub async fn legacy_next_websocket_proxy(
+    ws: WebSocketUpgrade,
+    State(state): State<AppState>,
+    Extension(context): Extension<RequestContext>,
+    uri: Uri,
+) -> Result<Response, WebError> {
+    if !state.config().enable_legacy_next_compat {
+        return Err(WebError::service_unavailable_code(
+            "legacy_next_compat_disabled",
+            "Next App Router legacy compat 已关闭，当前路径未迁到 Rust Web gateway。",
+        )
+        .with_context(&context)
+        .with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web"));
+    }
+
+    let Some(base_url) = state.config().legacy_next_base_url.as_deref() else {
+        return Err(WebError::service_unavailable_code(
+            "legacy_next_upstream_missing",
+            "未配置 MNOTE_WEB_LEGACY_NEXT_BASE_URL，无法代理 legacy Next WebSocket。",
+        )
+        .with_context(&context)
+        .with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web"));
+    };
+
+    let upstream_url = build_legacy_next_ws_url(base_url, &uri)?;
+    Ok(ws.on_upgrade(move |socket| async move {
+        if let Err(error) = proxy_legacy_next_websocket(socket, upstream_url).await {
+            tracing::warn!(error = %error, "legacy Next WebSocket 代理已断开");
+        }
+    }))
+}
+
+#[allow(dead_code)]
+fn build_legacy_next_ws_url(base_url: &str, uri: &Uri) -> Result<String, WebError> {
+    let upstream = reqwest::Url::parse(base_url)
+        .map_err(|error| WebError::internal(format!("legacy Next upstream URL 非法: {error}")))?;
+    let scheme = match upstream.scheme() {
+        "https" => "wss",
+        _ => "ws",
+    };
+    let host = upstream
+        .host_str()
+        .filter(|value| !value.trim().is_empty())
+        .ok_or_else(|| WebError::internal("legacy Next upstream URL 缺少 host"))?;
+    let host_with_port = match upstream.port() {
+        Some(port) => format!("{host}:{port}"),
+        None => host.to_string(),
+    };
+    let path_and_query = uri
+        .path_and_query()
+        .map(|value| value.as_str())
+        .unwrap_or("/");
+    Ok(format!("{scheme}://{host_with_port}{path_and_query}"))
+}
+
+#[allow(dead_code)]
+async fn proxy_legacy_next_websocket(
+    socket: WebSocket,
+    upstream_url: String,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let (upstream, _) = tokio_tungstenite::connect_async(upstream_url.as_str()).await?;
+    let (mut client_tx, mut client_rx) = socket.split();
+    let (mut upstream_tx, mut upstream_rx) = upstream.split();
+
+    let client_to_upstream = async {
+        while let Some(message) = client_rx.next().await {
+            let Ok(message) = message else {
+                break;
+            };
+            if upstream_tx
+                .send(axum_ws_to_tungstenite(message))
+                .await
+                .is_err()
+            {
+                break;
+            }
+        }
+    };
+
+    let upstream_to_client = async {
+        while let Some(message) = upstream_rx.next().await {
+            let Ok(message) = message else {
+                break;
+            };
+            if client_tx
+                .send(tungstenite_to_axum_ws(message))
+                .await
+                .is_err()
+            {
+                break;
+            }
+        }
+    };
+
+    tokio::select! {
+        _ = client_to_upstream => {}
+        _ = upstream_to_client => {}
+    }
+
+    Ok(())
+}
+
+#[allow(dead_code)]
+fn axum_ws_to_tungstenite(message: AxumWsMessage) -> TungsteniteMessage {
+    match message {
+        AxumWsMessage::Text(value) => TungsteniteMessage::Text(value.to_string().into()),
+        AxumWsMessage::Binary(value) => TungsteniteMessage::Binary(value),
+        AxumWsMessage::Ping(value) => TungsteniteMessage::Ping(value),
+        AxumWsMessage::Pong(value) => TungsteniteMessage::Pong(value),
+        AxumWsMessage::Close(_) => TungsteniteMessage::Close(None),
+    }
+}
+
+#[allow(dead_code)]
+fn tungstenite_to_axum_ws(message: TungsteniteMessage) -> AxumWsMessage {
+    match message {
+        TungsteniteMessage::Text(value) => AxumWsMessage::Text(value.to_string().into()),
+        TungsteniteMessage::Binary(value) => AxumWsMessage::Binary(value),
+        TungsteniteMessage::Ping(value) => AxumWsMessage::Ping(value),
+        TungsteniteMessage::Pong(value) => AxumWsMessage::Pong(value),
+        TungsteniteMessage::Close(_) => AxumWsMessage::Close(None),
+        TungsteniteMessage::Frame(_) => AxumWsMessage::Close(None),
+    }
+}
+
 async fn resolve_root_workspace_id(
     state: &AppState,
     context: &RequestContext,
@@ -947,6 +1076,18 @@ mod tests {
         format!("http://{addr}")
     }
 
+    async fn spawn_legacy_unmatched_upstream() -> String {
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("legacy listener");
+        let addr = listener.local_addr().expect("legacy addr");
+        let app = axum::Router::new().route("/unmigrated", get(|| async { "legacy-ok" }));
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.expect("legacy server");
+        });
+        format!("http://{addr}")
+    }
+
     #[tokio::test]
     async fn gateway_health_declares_mnote_web_owner() {
         let response = app()
@@ -1377,4 +1518,27 @@ mod tests {
             .iter()
             .any(|value| value.contains("__convexAuthRefreshToken=refresh-demo")));
     }
+
+    #[tokio::test]
+    async fn unmigrated_route_returns_not_found_instead_of_proxying_to_legacy_next() {
+        let legacy_base_url = spawn_legacy_unmatched_upstream().await;
+        let response = app_with_legacy_next_base_url(legacy_base_url)
+            .oneshot(
+                Request::builder()
+                    .uri("/unmigrated")
+                    .body(Body::empty())
+                    .expect("request"),
+            )
+            .await
+            .expect("response");
+
+        assert_eq!(response.status(), StatusCode::NOT_FOUND);
+        assert_eq!(
+            response
+                .headers()
+                .get("x-mnote-legacy-upstream")
+                .and_then(|value| value.to_str().ok()),
+            None
+        );
+    }
 }