diff --git a/Cargo.lock b/Cargo.lock index d31124600e..a79b93081e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1181,7 +1181,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.1.0" +version = "0.1.7" dependencies = [ "actix", "actix-cors", diff --git a/Cargo.toml b/Cargo.toml index 3ef483388b..8bb882c4bc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.1.0" +version = "0.1.7" edition = "2024" [dependencies] diff --git a/scripts/TEST_HTTP_API.sh b/scripts/TEST_HTTP_API.sh index b04a1fa74f..9ce7ee896d 100755 --- a/scripts/TEST_HTTP_API.sh +++ b/scripts/TEST_HTTP_API.sh @@ -10,6 +10,11 @@ TOKEN=$(./token.sh claims.json) ZP="00000000-0000-0000-0000-000000000001/TESTS" + delete "0000000/TESTS" + delete ${ZP} + put ${ZP} "Value_1" "HULY-TTL: 2" + delete ${ZP} + echo "--------- authorization_test ----------" TOKEN="" @@ -20,7 +25,6 @@ TOKEN=$(./token.sh claims_wrong_ws.json) put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2" TOKEN=$(./token.sh claims.json) put "00000000-0000-0000-0000-000000000002/TESTS" "Value_1" "HULY-TTL: 2" -exit @@ -36,9 +40,6 @@ echo "--------- if-match ----------" get "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/" - -exit - echo "--------- Deprecated symbols ----------" put "00000000-0000-0000-0000-000000000001/'TESTS" "Value_1" "HULY-TTL: 2" diff --git a/scripts/pulse_lib.sh b/scripts/pulse_lib.sh index 022d8cb87c..e1512a3169 100755 --- a/scripts/pulse_lib.sh +++ b/scripts/pulse_lib.sh @@ -72,5 +72,6 @@ delete() { local tmpfile tmpfile=$(mktemp) curl -i -s -X DELETE "$URL/$1" -H "Authorization: Bearer ${TOKEN}" | tr -d '\r' > "$tmpfile" +# curl -v -i -s -X DELETE "$URL/$1" -H "Authorization: Bearer ${TOKEN}" | tr -d '\r' > "$tmpfile" api ${tmpfile} } diff --git a/src/handlers_http.rs b/src/handlers_http.rs index 55c1f04338..dd31bed486 100644 --- a/src/handlers_http.rs +++ b/src/handlers_http.rs @@ -24,7 +24,6 @@ use actix_web::{ }; use crate::redis::{SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save}; -use crate::workspace_owner::workspace_check; pub fn map_handler_error(err: impl std::fmt::Display) -> Error { let msg = err.to_string(); @@ -46,7 +45,6 @@ pub fn map_handler_error(err: impl std::fmt::Display) -> Error { #[derive(Deserialize, Debug)] pub struct PathParams { - //workspace: Uuid, key: String, } @@ -95,14 +93,6 @@ pub async fn get( .map_err(map_handler_error) } -/* -#[derive(serde::Deserialize)] -struct MyHeaders { - #[serde(rename = "HULY-TTL")] - ttl: Option, -} -*/ - /// put pub async fn put( req: HttpRequest, @@ -172,7 +162,6 @@ pub async fn delete( path: web::Path, redis: web::Data, ) -> Result { - workspace_check(&req)?; // Check workspace let key: String = path.into_inner().key; diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index dd1e74f69c..154a838880 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -13,8 +13,10 @@ // limitations under the License. // +// https://github.com/hcengineering/hulypulse/ + use actix::{ - Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, WrapFuture, fut, prelude::*, + Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut, }; use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse, web}; use actix_web_actors::ws; @@ -25,10 +27,10 @@ use serde_json::{Value, json}; use crate::redis::{ SaveMode, Ttl, deprecated_symbol, redis_delete, redis_list, redis_read, redis_save, }; -use crate::ws_hub::{ - Connect, Disconnect, ServerMessage, SessionId, Subscribe, SubscribeList, Unsubscribe, - UnsubscribeAll, WsHub, -}; + +use crate::hub_service::{HubServiceHandle, ServerMessage, SessionId, new_session_id}; +use crate::workspace_owner::check_workspace_core; + #[derive(Serialize, Default)] struct ReturnBase<'a> { @@ -130,7 +132,7 @@ use hulyrs::services::jwt::Claims; pub struct WsSession { pub redis: MultiplexedConnection, pub id: SessionId, - pub hub: Addr, + pub hub: HubServiceHandle, pub claims: Claims, } @@ -139,35 +141,16 @@ impl Actor for WsSession { type Context = ws::WebsocketContext; fn started(&mut self, ctx: &mut Self::Context) { - // ask ID from Hub let addr = ctx.address(); - let recipient = addr.recipient::(); - // println!("WebSocket connected"); - self.hub - .send(Connect { - addr: recipient, - session_id: self.id, - }) - .into_actor(self) - .map(|res, act, _ctx| match res { - Ok(id) => { - act.id = id; - tracing::info!("WebSocket connected: {id}"); - } - Err(e) => { - tracing::error!("WebSocket failed connect to hub: {e}"); - _ctx.stop(); - } - }) - .wait(ctx); // waiting for ID + + self.hub.connect(self.id, recipient); + tracing::info!("WebSocket connected: {}", self.id); } fn stopped(&mut self, _ctx: &mut Self::Context) { if self.id != 0 { - self.hub.do_send(Disconnect { - session_id: self.id, - }); + self.hub.disconnect(self.id); } tracing::info!("WebSocket disconnected: {:?}", &self.id); } @@ -175,14 +158,13 @@ impl Actor for WsSession { impl actix::Handler for WsSession { type Result = (); - fn handle(&mut self, msg: ServerMessage, ctx: &mut Self::Context) { - let json = - serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into()); + let json = serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into()); ctx.text(json); } } + /// StreamHandler External trait: must be in separate impl block impl StreamHandler> for WsSession { fn handle(&mut self, msg: Result, ctx: &mut Self::Context) { @@ -201,8 +183,6 @@ impl StreamHandler> for WsSession { } } -use crate::workspace_owner::check_workspace_core; - /// All logic in one impl impl WsSession { fn ws_error(&self, ctx: &mut ws::WebsocketContext, msg: &str) { @@ -221,7 +201,6 @@ impl WsSession { ) { ctx.wait( fut::wrap_future(fut).map(move |res, _actor: &mut Self, ctx| { - // if !base.is_object() { base = json!({ "base": base }); } let obj = base.as_object_mut().unwrap(); match res { Ok(Value::Object(extra)) => { @@ -251,14 +230,7 @@ impl WsSession { if_none_match, correlation, } => { - tracing::info!( - "PUT {} = {} (expires_at: {:?}) (ttl: {:?}) correlation: {:?}", - &key, - &data, - &expires_at, - &ttl, - &correlation - ); + tracing::info!("PUT {} = {}", &key, &data); // (expires_at: {:?}) (ttl: {:?}) correlation: {:?} &expires_at, &ttl, &correlation // Check workspace if let Err(e) = self.workspace_check_ws(&key) { @@ -324,7 +296,7 @@ impl WsSession { correlation, if_match, } => { - tracing::info!("DELETE {} correlation:{:?}", &key, &correlation); + tracing::info!("DELETE {}", &key); // correlation:{:?} , &correlation // Check workspace if let Err(e) = self.workspace_check_ws(&key) { @@ -332,6 +304,8 @@ impl WsSession { return; } + tracing::info!("DELETE!!! {}", &key); + let mut redis = self.redis.clone(); let base = serde_json::json!(ReturnBase { @@ -371,7 +345,7 @@ impl WsSession { } WsCommand::Get { key, correlation } => { - tracing::info!("GET {} correlation:{:?}", &key, &correlation); + tracing::info!("GET {}", &key); // correlation:{:?} , &correlation // Check workspace if let Err(e) = self.workspace_check_ws(&key) { @@ -407,7 +381,7 @@ impl WsSession { } WsCommand::List { key, correlation } => { - tracing::info!("LIST {:?} correlation: {:?}", &key, &correlation); + tracing::info!("LIST {:?}", &key); // correlation: {:?} , &correlation // Check workspace if let Err(e) = self.workspace_check_ws(&key) { @@ -434,9 +408,10 @@ impl WsSession { self.fut_send(ctx, fut, base); } + WsCommand::Sub { key, correlation } => { // LEVENT 3 - tracing::info!("SUB {} correlation: {:?}", &key, &correlation); + tracing::info!("SUB {}", &key); // correlation: {:?} , &correlation // Check workspace if let Err(e) = self.workspace_check_ws(&key) { @@ -456,19 +431,16 @@ impl WsSession { if deprecated_symbol(&key) { map.insert("error".into(), json!("Deprecated symbol in key")); } else { - self.hub.do_send(Subscribe { - session_id: self.id, - key: key.clone(), - }); + self.hub.subscribe(self.id, key.clone()); map.insert("result".into(), json!("OK")); } - ctx.text(obj.to_string()); } + WsCommand::Unsub { key, correlation } => { // LEVENT 4 - tracing::info!("UNSUB {} correlation: {:?}", &key, &correlation); + tracing::info!("UNSUB {}", &key); // correlation: {:?} , &correlation let mut obj = serde_json::json!(ReturnBase { action: "unsub", @@ -480,9 +452,7 @@ impl WsSession { let map = obj.as_object_mut().unwrap(); if key == "*" { - self.hub.do_send(UnsubscribeAll { - session_id: self.id, - }); + self.hub.unsubscribe_all(self.id); map.insert("result".into(), json!("OK")); } else { // Check workspace @@ -490,26 +460,20 @@ impl WsSession { self.ws_error(ctx, e); return; } - if deprecated_symbol(&key) { map.insert("error".into(), json!("Deprecated symbol in key")); } else { + self.hub.unsubscribe(self.id, key.clone()); map.insert("result".into(), json!("OK")); - self.hub.do_send(Unsubscribe { - session_id: self.id, - key: key.clone(), - }); } - }; - + } ctx.text(obj.to_string()); } + WsCommand::Sublist { correlation } => { - tracing::info!("SUBLIST correlation: {:?}", &correlation); - + tracing::info!("SUBLIST"); // correlation: {:?} , &correlation // w/o Check workspace! - let base = serde_json::json!(ReturnBase { action: "list", correlation: correlation.as_deref(), @@ -519,27 +483,27 @@ impl WsSession { let hub = self.hub.clone(); let id = self.id; - let fut = async move { - let keys = hub - .send(SubscribeList { session_id: id }) - .await - .unwrap_or_default(); - Ok(json!({ "result": keys })) - }; + self.fut_send( + ctx, + async move { + let keys = hub.subscribe_list(id).await; + Ok(json!({ "result": keys })) + }, + base, + ); + } - self.fut_send(ctx, fut, base); - } // End of commands + // End of commands } } } -// ---- auth pub async fn handler( req: HttpRequest, payload: web::Payload, redis: web::Data, - hub: web::Data>, + hub: web::Data, // <-- было Addr ) -> Result { let claims = req .extensions() @@ -550,7 +514,7 @@ pub async fn handler( let session = WsSession { redis: redis.get_ref().clone(), hub: hub.get_ref().clone(), - id: crate::ws_hub::new_session_id(), + id: new_session_id(), claims, }; diff --git a/src/hub_service.rs b/src/hub_service.rs new file mode 100644 index 0000000000..2e79a468c4 --- /dev/null +++ b/src/hub_service.rs @@ -0,0 +1,266 @@ +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use actix::prelude::*; + +use redis::aio::MultiplexedConnection; +use serde::Serialize; +use tokio::sync::{mpsc, oneshot}; + +fn subscription_matches(sub_key: &str, key: &str) -> bool { + if sub_key == key { + return true; + } + if sub_key.ends_with('/') && key.starts_with(sub_key) { + let rest = &key[sub_key.len()..]; + return !rest.contains('$'); + } + false +} + +#[derive(Clone, Serialize, Debug, Message)] +#[rtype(result = "()")] +pub struct ServerMessage { + #[serde(flatten)] + pub event: RedisEvent, + #[serde(skip_serializing_if = "Option::is_none")] + pub value: Option, +} + +// ==== ID ==== + +pub type SessionId = u64; +static NEXT_ID: AtomicU64 = AtomicU64::new(1); +pub fn new_session_id() -> SessionId { + NEXT_ID.fetch_add(1, Ordering::SeqCst) +} + +// ==== Redis events ==== + +#[derive(Debug, Clone, Serialize)] +pub enum RedisEventAction { + Set, + Del, + Unlink, + Expired, + Other(String), +} + +#[derive(Debug, Clone, Serialize)] +pub struct RedisEvent { + pub db: u32, + pub key: String, + pub action: RedisEventAction, +} + +// ==== Commands for worker ==== + +enum Command { + Connect { + session_id: SessionId, + addr: Recipient, + }, + Disconnect { + session_id: SessionId, + }, + Subscribe { + session_id: SessionId, + key: String, + }, + Unsubscribe { + session_id: SessionId, + key: String, + }, + UnsubscribeAll { + session_id: SessionId, + }, + SubscribeList { + session_id: SessionId, + reply: oneshot::Sender>, + }, + Count { + reply: oneshot::Sender, + }, + DumpSubs { + reply: oneshot::Sender>>, + }, + RedisEvent(RedisEvent), +} + +// ==== Handle ==== + +#[derive(Clone)] +pub struct HubServiceHandle { + tx: mpsc::Sender, +} + +impl HubServiceHandle { + pub fn start(redis: MultiplexedConnection) -> Self { + let (tx, mut rx) = mpsc::channel::(1024); + + // Владелец состояния живёт внутри задачи + tokio::spawn(async move { + let mut sessions: HashMap> = HashMap::new(); + let mut subs: HashMap> = HashMap::new(); + let mut redis_conn = redis; + + fn subscribers_for( + subs: &HashMap>, + key: &str, + ) -> HashSet { + let mut out = HashSet::::new(); + for (sub_key, set) in subs.iter() { + if subscription_matches(sub_key, key) { + out.extend(set.iter().copied()); + } + } + out + } + + while let Some(cmd) = rx.recv().await { + match cmd { + + Command::Connect { session_id, addr } => { + sessions.insert(session_id, addr); + } + + Command::Disconnect { session_id } => { + subs.retain(|_, ids| { + ids.remove(&session_id); + !ids.is_empty() + }); + sessions.remove(&session_id); + } + + Command::Subscribe { session_id, key } => { + subs.entry(key).or_default().insert(session_id); + } + + Command::Unsubscribe { session_id, key } => { + if let Some(set) = subs.get_mut(&key) { + set.remove(&session_id); + if set.is_empty() { + subs.remove(&key); + } + } + } + + Command::UnsubscribeAll { session_id } => { + subs.retain(|_, ids| { + ids.remove(&session_id); + !ids.is_empty() + }); + } + + Command::SubscribeList { session_id, reply } => { + let list = subs + .iter() + .filter_map(|(key, ids)| { + if ids.contains(&session_id) { + Some(key.clone()) + } else { + None + } + }) + .collect::>(); + let _ = reply.send(list); + } + + Command::Count { reply } => { + let _ = reply.send(sessions.len()); + } + + Command::DumpSubs { reply } => { + let snapshot = subs + .iter() + .map(|(k, set)| (k.clone(), set.iter().copied().collect::>())) + .collect::>(); + let _ = reply.send(snapshot); + } + + Command::RedisEvent(event) => { + let targets = subscribers_for(&subs, &event.key); + if targets.is_empty() { + continue; + } + let recipients: Vec> = targets + .into_iter() + .filter_map(|sid| sessions.get(&sid).cloned()) + .collect(); + + // Inside: waiting GET + let need_get = matches!(event.action, RedisEventAction::Set); + let mut value: Option = None; + if need_get { + match redis::cmd("GET") + .arg(&event.key) + .query_async::>(&mut redis_conn) + .await + { + Ok(v) => value = v, + Err(e) => { + tracing::warn!("redis GET {} failed: {}", &event.key, e); + } + } + } + + let payload = ServerMessage { + event, + value, + }; + + for rcpt in recipients { + let _ = rcpt.do_send(payload.clone()); + } + } + } + } + }); + + Self { tx } + } + + // ---- API, ничего не выполняет параллельно внутри worker'а ---- + + pub fn connect(&self, session_id: SessionId, addr: Recipient) { + let _ = self.tx.try_send(Command::Connect { session_id, addr }); + } + + pub fn disconnect(&self, session_id: SessionId) { + let _ = self.tx.try_send(Command::Disconnect { session_id }); + } + + pub fn subscribe(&self, session_id: SessionId, key: String) { + let _ = self.tx.try_send(Command::Subscribe { session_id, key }); + } + + pub fn unsubscribe(&self, session_id: SessionId, key: String) { + let _ = self.tx.try_send(Command::Unsubscribe { session_id, key }); + } + + pub fn unsubscribe_all(&self, session_id: SessionId) { + let _ = self.tx.try_send(Command::UnsubscribeAll { session_id }); + } + + pub async fn subscribe_list(&self, session_id: SessionId) -> Vec { + let (tx, rx) = oneshot::channel(); + let _ = self.tx.send(Command::SubscribeList { session_id, reply: tx }).await; + rx.await.unwrap_or_default() + } + + pub async fn count(&self) -> usize { + let (tx, rx) = oneshot::channel(); + let _ = self.tx.send(Command::Count { reply: tx }).await; + rx.await.unwrap_or_default() + } + + pub async fn dump_subs(&self) -> std::collections::HashMap> { + let (tx, rx) = oneshot::channel(); + let _ = self.tx.send(Command::DumpSubs { reply: tx }).await; + rx.await.unwrap_or_default() + } + + pub fn push_event(&self, ev: RedisEvent) { + let _ = self.tx.try_send(Command::RedisEvent(ev)); + } +} diff --git a/src/main.rs b/src/main.rs index 5807a3169a..7067a6358c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,29 +13,26 @@ // limitations under the License. // -use actix::prelude::*; use actix_cors::Cors; use actix_web::{ - App, Error, HttpMessage, HttpResponse, HttpServer, - body::MessageBody, - dev::{ServiceRequest, ServiceResponse}, - middleware::{self, Next}, - web::{self, Path, Query}, + body::MessageBody, dev::{ServiceRequest, ServiceResponse}, middleware::{self, Next}, web::{self, Path, Query}, App, Error, HttpMessage, HttpResponse, HttpServer }; use hulyrs::services::jwt::{Claims, actix::ServiceRequestExt}; use secrecy::ExposeSecret; +use serde_json::json; use tracing::*; +use uuid::Uuid; mod config; mod handlers_http; mod handlers_ws; mod redis; mod workspace_owner; -mod ws_hub; + +mod hub_service; +use hub_service::HubServiceHandle; use config::CONFIG; -use uuid::Uuid; -use ws_hub::{TestGetSubs, WsHub}; fn initialize_tracing(level: tracing::Level) { use tracing_subscriber::{filter::targets::Targets, prelude::*}; @@ -100,14 +97,22 @@ async fn main() -> anyhow::Result<()> { let redis_client = redis::client().await?; let redis_connection = redis_client.get_multiplexed_async_connection().await?; - // starting Hub - let hub = WsHub::new(redis_connection.clone()).start(); + // starting HubService + let hub = HubServiceHandle::start(redis_connection.clone()); // starting Logger tokio::spawn(redis::receiver(redis_client, hub.clone())); let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port); + let url = format!("http://{}:{}", &CONFIG.bind_host, &CONFIG.bind_port); + tracing::info!("Server running at {}", &url); + tracing::info!("HTTP API: {}/api", &url); + tracing::info!("WebSocket API: {}/ws", &url); + tracing::info!("Status: {}/status", &url); + tracing::info!("Stats: {}/stat", &url); + tracing::info!("Subscriptions: {}/subs", &url); + let server = HttpServer::new(move || { let cors = Cors::default() .allow_any_origin() @@ -130,32 +135,21 @@ async fn main() -> anyhow::Result<()> { .route("/{key:.+}", web::put().to(handlers_http::put)) .route("/{key:.+}", web::delete().to(handlers_http::delete)), ) - .route( - "/ws", - web::get() - .to(handlers_ws::handler) + .route("/ws", web::get().to(handlers_ws::handler) .wrap(middleware::from_fn(extract_claims)), ) // WebSocket .route("/status", web::get().to(async || "ok")) - // - .route( - "/stat2", - web::get().to(|hub: web::Data>| async move { - let count = hub.send(crate::ws_hub::Count).await.unwrap_or(0); - HttpResponse::Ok().json(serde_json::json!({ "connections": count })) - }), - ) - .route( - "/subs", - web::get().to(|hub: web::Data>| async move { - match hub.send(TestGetSubs).await { - Ok(subs) => HttpResponse::Ok().json(subs), - Err(_) => { - HttpResponse::InternalServerError().body("Failed to get subscriptions") - } - } - }), - ) + + .route("/stat", web::get().to(|hub: web::Data| async move { + let count = hub.count().await; + Ok::<_, actix_web::Error>(HttpResponse::Ok().json(json!({ "connections": count }))) + })) + + .route("/subs", web::get().to(|hub: web::Data| async move { + let subs = hub.dump_subs().await; + Ok::<_, actix_web::Error>(HttpResponse::Ok().json(subs)) + })) + }) .bind(socket)? .run(); diff --git a/src/redis.rs b/src/redis.rs index 4bd6ca252b..f7f9ff3295 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -15,13 +15,11 @@ use std::time::{SystemTime, UNIX_EPOCH}; -use actix::Addr; use ::redis::Msg; use tokio_stream::StreamExt; use tracing::*; -use crate::config::{CONFIG, RedisMode}; -use crate::ws_hub::{RedisEvent, RedisEventAction, WsHub}; +use crate::{config::{RedisMode, CONFIG}, hub_service::{HubServiceHandle, RedisEvent, RedisEventAction}}; #[derive(serde::Serialize)] pub enum Ttl { @@ -358,7 +356,8 @@ impl TryFrom for RedisEvent { } } -pub async fn receiver(redis_client: Client, hub: Addr) -> anyhow::Result<()> { + +pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Result<()> { let mut redis = redis_client.get_multiplexed_async_connection().await?; let mut pubsub = redis_client.get_async_pubsub().await?; @@ -383,10 +382,10 @@ pub async fn receiver(redis_client: Client, hub: Addr) -> anyhow::Result< while let Some(message) = messages.next().await { match RedisEvent::try_from(message) { Ok(ev) => { - - debug!("redis event: {ev:#?}"); - hub.do_send(ev); + // debug!("redis event: {ev:#?}"); + + hub.push_event(ev); } Err(e) => { diff --git a/src/workspace_owner.rs b/src/workspace_owner.rs index 2f36abbe17..148b06740c 100644 --- a/src/workspace_owner.rs +++ b/src/workspace_owner.rs @@ -13,12 +13,12 @@ // limitations under the License. // -use actix_web::{HttpMessage, HttpRequest}; use hulyrs::services::jwt::Claims; use uuid::Uuid; // common checker pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> { + if claims.is_system() { return Ok(()); } @@ -41,22 +41,4 @@ pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static s } Ok(()) -} - -/// HTTP API -pub fn workspace_check(req: &HttpRequest) -> Result<(), actix_web::Error> { - let key = req - .match_info() - .get("key") - .ok_or_else(|| actix_web::error::ErrorBadRequest("Missing key in URL path"))?; - let claims = req - .extensions() - .get::() - .cloned() - .ok_or_else(|| actix_web::error::ErrorUnauthorized("Missing auth claims"))?; - - match check_workspace_core(&claims, key) { - Ok(()) => Ok(()), - Err(msg) => Err(actix_web::error::ErrorUnauthorized(msg)), - } -} +} \ No newline at end of file diff --git a/src/ws_hub.rs b/src/ws_hub.rs deleted file mode 100644 index 84a96fc215..0000000000 --- a/src/ws_hub.rs +++ /dev/null @@ -1,309 +0,0 @@ -// -// Copyright © 2025 Hardcore Engineering Inc. -// -// Licensed under the Eclipse Public License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. You may -// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// -// See the License for the specific language governing permissions and -// limitations under the License. -// - -use std::{ - collections::{HashMap, HashSet}, - sync::atomic::AtomicU64, -}; - -use actix::prelude::*; -use redis::aio::MultiplexedConnection; -use serde::Serialize; - -fn subscription_matches(sub_key: &str, key: &str) -> bool { - if sub_key == key { - return true; - } - if sub_key.ends_with('/') && key.starts_with(sub_key) { - let rest = &key[sub_key.len()..]; - return !rest.contains('$'); - } - false -} - -#[derive(Message, Clone, Serialize, Debug)] -#[rtype(result = "()")] -pub struct ServerMessage { - #[serde(flatten)] - pub event: RedisEvent, - #[serde(skip_serializing_if = "Option::is_none")] - pub value: Option, -} - -/// Count of active sessions -#[derive(Message)] -#[rtype(result = "usize")] -pub struct Count; - -pub type SessionId = u64; - -static NEXT_ID: AtomicU64 = AtomicU64::new(1); - -pub struct WsHub { - sessions: HashMap>, - subs: HashMap>, // Subscriptions array: key -> {id, id, id ...} - redis: MultiplexedConnection, -} - -impl WsHub { - pub fn new(redis: MultiplexedConnection) -> Self { - Self { - sessions: HashMap::new(), - subs: HashMap::new(), - redis, - } - } -} - -impl Actor for WsHub { - type Context = Context; -} - -/// Connect -#[derive(Message)] -#[rtype(result = "SessionId")] -pub struct Connect { - pub session_id: SessionId, - pub addr: Recipient, -} - -pub fn new_session_id() -> SessionId { - NEXT_ID.fetch_add(1, std::sync::atomic::Ordering::SeqCst) -} - -impl Handler for WsHub { - type Result = SessionId; - - fn handle(&mut self, msg: Connect, _ctx: &mut Context) -> Self::Result { - // LEVENT 1 - //let id = self.next_id; - //self.next_id = self.next_id.wrapping_add(1); - self.sessions.insert(msg.session_id, msg.addr); - // tracing::info!("session connected: id={id} (total={})", self.sessions.len()); - msg.session_id - } -} - -/// Disconnect -#[derive(Message)] -#[rtype(result = "()")] -pub struct Disconnect { - pub session_id: SessionId, -} - -impl Handler for WsHub { - type Result = (); - - fn handle(&mut self, msg: Disconnect, _ctx: &mut Context) { - // LEVENT 2 - - // Delete all subscribes - self.subs.retain(|_key, session_ids| { - session_ids.remove(&msg.session_id); - !session_ids.is_empty() - }); - - let existed = self.sessions.remove(&msg.session_id).is_some(); - if existed { - // tracing::info!("session disconnected: id={} (total={})", msg.session_id, self.sessions.len()); - } else { - tracing::warn!("disconnect for unknown id={}", msg.session_id); - } - } -} - -/// SubscribeList -#[derive(Message)] -#[rtype(result = "Vec")] -pub struct SubscribeList { - pub session_id: SessionId, -} - -impl Handler for WsHub { - type Result = MessageResult; - - fn handle(&mut self, msg: SubscribeList, _ctx: &mut Context) -> Self::Result { - // Collect all keys with my session_id - let list = self - .subs - .iter() - .filter_map(|(key, sessions)| { - if sessions.contains(&msg.session_id) { - Some(key.clone()) - } else { - None - } - }) - .collect::>(); - - MessageResult(list) - } -} - -/// Count of IDs -impl Handler for WsHub { - type Result = usize; - - fn handle(&mut self, _: Count, _: &mut Context) -> Self::Result { - self.sessions.len() - } -} - -/// Subscribe -#[derive(Message)] -#[rtype(result = "()")] -pub struct Subscribe { - pub session_id: SessionId, - pub key: String, -} - -impl Handler for WsHub { - type Result = (); - fn handle(&mut self, msg: Subscribe, _ctx: &mut Context) { - self.subs.entry(msg.key).or_default().insert(msg.session_id); - } -} - -/// Unsubscribe -#[derive(Message)] -#[rtype(result = "()")] -pub struct Unsubscribe { - pub session_id: SessionId, - pub key: String, -} - -impl Handler for WsHub { - type Result = (); - fn handle(&mut self, msg: Unsubscribe, _ctx: &mut Context) { - if let Some(set) = self.subs.get_mut(&msg.key) { - set.remove(&msg.session_id); - if set.is_empty() { - self.subs.remove(&msg.key); - } - } - } -} - -#[derive(Message)] -#[rtype(result = "()")] -pub struct UnsubscribeAll { - pub session_id: SessionId, -} - -impl Handler for WsHub { - type Result = (); - fn handle(&mut self, msg: UnsubscribeAll, _ctx: &mut Context) { - self.subs.retain(|_key, session_ids| { - session_ids.remove(&msg.session_id); - !session_ids.is_empty() - }); - } -} - -#[derive(Message)] -#[rtype(result = "HashMap>")] -pub struct TestGetSubs; - -impl Handler for WsHub { - type Result = MessageResult; - - fn handle(&mut self, _msg: TestGetSubs, _ctx: &mut Context) -> Self::Result { - let s: HashMap> = self - .subs - .iter() - .map(|(key, ids)| (key.clone(), ids.iter().copied().collect())) - .collect(); - MessageResult(s) - } -} - -// List of subscribers -impl WsHub { - fn subscribers_for(&self, key: &str) -> HashSet { - let mut out = HashSet::new(); - for (sub_key, set) in &self.subs { - if subscription_matches(sub_key, key) { - out.extend(set.iter().copied()); - } - } - out - } -} - -#[derive(Debug, Clone, Serialize)] -pub enum RedisEventAction { - Set, // Insert or Update - Del, // Delete - Unlink, // async Delete - Expired, // TTL Delete - Other(String), -} - -use actix::Message; - -#[derive(Debug, Clone, Serialize, Message)] -#[rtype(result = "()")] -pub struct RedisEvent { - pub db: u32, - pub key: String, - pub action: RedisEventAction, -} - -impl Handler for WsHub { - type Result = ResponseActFuture; - - fn handle(&mut self, msg: RedisEvent, _ctx: &mut Context) -> Self::Result { - let targets = self.subscribers_for(&msg.key); - if targets.is_empty() { - return Box::pin(actix::fut::ready(()).into_actor(self)); - } - - let recipients: Vec> = targets - .into_iter() - .filter_map(|sid| self.sessions.get(&sid).cloned()) - .collect(); - - let mut redis = self.redis.clone(); - let event = msg.clone(); - let need_get = matches!(msg.action, RedisEventAction::Set); - - Box::pin( - async move { - let value = if need_get { - match redis::cmd("GET") - .arg(&event.key) - .query_async::>(&mut redis) - .await - { - Ok(v) => v, - Err(e) => { - tracing::warn!("redis GET {} failed: {}", &event.key, e); - None - } - } - } else { - None - }; - - let payload = ServerMessage { event, value }; - - for rcpt in recipients { - let _ = rcpt.do_send(payload.clone()); - } - } - .into_actor(self), - ) - } -}