Function handle_sql_ws

src/environmentd/src/http/sql.rs:291–317

(
    State(state): State<WsState>,
    existing_user: Option<Extension<AuthedUser>>,
    ws: WebSocketUpgrade,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    tower_session: Option<Extension<Towe

}

pub(crate) async fn handle_sql_ws(
    State(state): State<WsState>,
    existing_user: Option<Extension<AuthedUser>>,
    ws: WebSocketUpgrade,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    tower_session: Option<Extension<TowerSession>>,
) -> Result<impl IntoResponse, AuthError> {
    let session = tower_session.map(|Extension(session)| session);
    // The `x_materialize_user_header_auth` middleware may have already provided the user for us
    let user = match existing_user {
        Some(Extension(user)) => Some(ExistingUser::XMaterializeUserHeader(user)),
        None => {
            let session = maybe_get_authenticated_session(session.as_ref()).await;
            if let Some((session, session_data)) = session {
                let user = ensure_session_unexpired(session, session_data).await?;
                Some(ExistingUser::Session(user))
            } else {
                None
            }
        }
    };

    let addr = Box::new(addr.ip());
    Ok(ws
        .max_message_size(MAX_REQUEST_SIZE)
        .on_upgrade(|ws| async move { run_ws(state, user, *addr, ws).await }))
}

#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
#[serde(untagged)]