cli: avoid data-dir initialization for version; create db parent dirs; redact generated passwords in CLI output
- Prevent 'nx9-wg version' from creating data directories by avoiding database initialization. - Create parent directories when an explicit --database path is provided. - Redact printed generated administrator passwords; announce file path or redact instead. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
commit
2ac6c81dfe
140 files changed
+31342
No files matched your search
@@ -0,0 +1,71 @@
|
||||
//! Real-time WebSocket event streaming.
|
||||
|
||||
use crate::error::{ApiError, ApiResult};
|
||||
use crate::state::AppState;
|
||||
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
||||
use axum::extract::{Query, State};
|
||||
use axum::response::IntoResponse;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use serde::Deserialize;
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct WsAuthQuery {
|
||||
pub token: Option<String>,
|
||||
pub session: Option<String>,
|
||||
}
|
||||
|
||||
/// GET /api/v1/ws
|
||||
pub async fn ws_handler(
|
||||
ws: WebSocketUpgrade,
|
||||
State(state): State<AppState>,
|
||||
Query(query): Query<WsAuthQuery>,
|
||||
) -> ApiResult<impl IntoResponse> {
|
||||
// Authenticate WebSocket connection via query parameters
|
||||
let authenticated = if let Some(ref raw_token) = query.token {
|
||||
state.auth.authenticate_token(raw_token).await.is_ok()
|
||||
} else if let Some(ref session_id) = query.session {
|
||||
state.auth.authenticate_session(session_id).await.is_ok()
|
||||
} else {
|
||||
false
|
||||
};
|
||||
|
||||
if !authenticated {
|
||||
return Err(ApiError::Unauthenticated(
|
||||
"WebSocket authentication required. Supply ?token=... or ?session=...".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
Ok(ws.on_upgrade(move |socket| handle_socket(socket, state)))
|
||||
}
|
||||
|
||||
async fn handle_socket(socket: WebSocket, state: AppState) {
|
||||
let (mut sender, mut receiver) = socket.split();
|
||||
let mut rx = state.event_tx.subscribe();
|
||||
|
||||
// Spawn background task to stream broadcast events to client
|
||||
let mut send_task = tokio::spawn(async move {
|
||||
while let Ok(event) = rx.recv().await {
|
||||
if let Ok(json) = serde_json::to_string(&event) {
|
||||
let msg = Message::Text(json.into());
|
||||
if sender.send(msg).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Client receive loop to handle close/ping/pong
|
||||
let mut recv_task = tokio::spawn(async move {
|
||||
while let Some(Ok(msg)) = receiver.next().await {
|
||||
if let Message::Close(_) = msg {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// If either task exits, abort the other
|
||||
tokio::select! {
|
||||
_ = (&mut send_task) => recv_task.abort(),
|
||||
_ = (&mut recv_task) => send_task.abort(),
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user