WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-09-08 18:57:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

Refactoring code

Browse Source
This commit is contained in:
Leonid Kaganov
2025-08-19 13:41:22 +03:00
parent dc68a27f5f
commit df44eab1e3
9 changed files with 844 additions and 710 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/TEST.html
+2
View File
@@ -99,6 +99,7 @@
// let ws = new WebSocket(`ws://localhost:8095/ws`);
ws.onopen = () => {
output.textContent = "✅ WebSocket connected.";
};
@@ -122,6 +123,7 @@
} catch (e) {
output.textContent += "\n\n⚠️ Invalid JSON:\n" + e.message;
}
}
function place(event) { textarea.value = (event || window.event).target.getAttribute("data"); sendMessage(); }
scripts/TEST_HTTP_API.sh
+13 -100
View File
@@ -11,6 +11,19 @@ TOKEN=$(./token.sh claims.json)
ZP="00000000-0000-0000-0000-000000000001/TESTS"
echo "--------- authorization_test ----------"
TOKEN=""
put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2"
TOKEN=$(./token.sh claims_system.json)
put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2"
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
echo "--------- if-match ----------"
put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2"
@@ -23,106 +36,6 @@ echo "--------- if-match ----------"
get "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/"
exit
Key
Key is a string that consists of one or multiple segments separated by ‘/’. Example: foo/bar/baz.
Segment may not contain special characters (‘$’, ‘*’, ‘?’)
Key may not end with ‘/’
Segment may not be empty
Key segment may be private (prefixed with ‘$’)
Query
May not contain special characters (‘*’, ‘?’)
It is possible to use prefix, for listings / subscriptions (prefix ends with segment separator ‘/’)
GET/SUBSCRIBE/.. a/b → single key
GET/SUBSCRIBE/.. a/b/c/ → multiple
If multiple
select all keys starting with prefix
skip keys, containing private segments to the right from the prefix
example
1. /a/b/$c/$d, 2. /a/b/c, 3. /a/b/$c, 4. /a/b/$c/$d/e
/ → [2]
/a/b/ → [2]
/a/b/$c/ → [3]
/a/b/$c/$d/ → [4]
/a/b/$c/$d → (1)
exit
src/handlers_http.rs
+103 -66
View File
@@ -1,55 +1,61 @@
use redis::aio::MultiplexedConnection;
use std::sync::Arc;
use std::collections::HashMap;
use tokio::sync::Mutex;
//
// 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 crate::workspace_owner::workspace_check;
use anyhow::anyhow;
use redis::aio::MultiplexedConnection;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::Mutex;
use tracing::{error, trace};
use uuid::Uuid;
use crate::workspace_owner::workspace_check;
use crate::redis_lib::{
Ttl, SaveMode,
RedisArray,
redis_save,
redis_read,
redis_delete,
redis_list,
RedisArray, SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save,
};
use actix_web::{
HttpRequest, HttpResponse, error, Error,
Error, HttpRequest, HttpResponse, error,
web::{self, Data, Json, Query},
};
pub fn map_handler_error(err: impl std::fmt::Display) -> Error {
let msg = err.to_string();
if let Some(detail) = msg.split(" - ExtensionError: ").nth(1) {
if let Some((code, text)) = detail.split_once(": ") {
let text = format!("{} {}", code, text);
let text = format!("{} {}", code, text);
return match code {
"400" => actix_web::error::ErrorBadRequest(text),
"404" => actix_web::error::ErrorNotFound(text),
"412" => actix_web::error::ErrorPreconditionFailed(text),
"500" => actix_web::error::ErrorInternalServerError(text),
_ => actix_web::error::ErrorInternalServerError("unexpected error"),
_ => actix_web::error::ErrorInternalServerError("unexpected error"),
};
}
}
actix_web::error::ErrorInternalServerError("internal error")
}
/// list
pub async fn list(
req: HttpRequest,
path: web::Path<String>,
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
) -> Result<HttpResponse, actix_web::Error> {
workspace_check(&req)?; // Check workspace
let key = path.into_inner();
@@ -57,24 +63,22 @@ pub async fn list(
trace!(key, "list request");
async move || -> anyhow::Result<HttpResponse> {
let mut conn = redis.lock().await;
let entries = redis_list(&mut *conn, &key).await?;
let entries = redis_list(&mut *conn, &key).await?;
Ok(HttpResponse::Ok().json(entries))
}().await.map_err(map_handler_error)
}()
.await
.map_err(map_handler_error)
}
/// get
pub async fn get(
req: HttpRequest,
path: web::Path<String>,
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
) -> Result<HttpResponse, actix_web::error::Error> {
workspace_check(&req)?; // Check workspace
let key = path.into_inner();
@@ -82,21 +86,21 @@ pub async fn get(
// trace!(key, "get request");
async move || -> anyhow::Result<HttpResponse> {
let mut conn = redis.lock().await;
Ok(
redis_read(&mut *conn, &key).await?
.map(|entry| HttpResponse::Ok()
Ok(redis_read(&mut *conn, &key)
.await?
.map(|entry| {
HttpResponse::Ok()
.insert_header(("ETag", &*entry.etag))
.json(entry))
.unwrap_or_else(|| HttpResponse::NotFound().body("empty"))
)
}().await.map_err(map_handler_error)
.json(entry)
})
.unwrap_or_else(|| HttpResponse::NotFound().body("empty")))
}()
.await
.map_err(map_handler_error)
}
/// put
pub async fn put(
req: HttpRequest,
@@ -104,54 +108,72 @@ pub async fn put(
body: web::Bytes,
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
) -> Result<HttpResponse, actix_web::error::Error> {
workspace_check(&req)?; // Check workspace
let key: String = path.into_inner();
async move || -> anyhow::Result<HttpResponse> {
if !req.query_string().is_empty() { return Err(anyhow!("Query parameters are not allowed")); }
if !req.query_string().is_empty() {
return Err(anyhow!("Query parameters are not allowed"));
}
let mut conn = redis.lock().await;
// TTL logic
let mut ttl = None;
if let Some(x) = req.headers().get("HULY-TTL") {
let s = x.to_str().map_err(|_| anyhow!("Invalid HULY-TTL header"))?;
let secs = s.parse::<usize>().map_err(|_| anyhow!("Invalid TTL value in HULY-TTL header"))?;
ttl = Some(Ttl::Sec(secs));
} else if let Some(x) = req.headers().get("HULY-EXPIRE-AT") {
let s = x.to_str().map_err(|_| anyhow!("Invalid HULY-EXPIRE-AT header"))?;
let ts = s.parse::<u64>().map_err(|_| anyhow!("Invalid EXPIRE-AT value in HULY-EXPIRE-AT header"))?;
ttl = Some(Ttl::At(ts));
}
// TTL logic
let mut ttl = None;
if let Some(x) = req.headers().get("HULY-TTL") {
let s = x.to_str().map_err(|_| anyhow!("Invalid HULY-TTL header"))?;
let secs = s
.parse::<usize>()
.map_err(|_| anyhow!("Invalid TTL value in HULY-TTL header"))?;
ttl = Some(Ttl::Sec(secs));
} else if let Some(x) = req.headers().get("HULY-EXPIRE-AT") {
let s = x
.to_str()
.map_err(|_| anyhow!("Invalid HULY-EXPIRE-AT header"))?;
let ts = s
.parse::<u64>()
.map_err(|_| anyhow!("Invalid EXPIRE-AT value in HULY-EXPIRE-AT header"))?;
ttl = Some(Ttl::At(ts));
}
// MODE logic
let mut mode = Some(SaveMode::Upsert);
if let Some(h) = req.headers().get("If-Match") { // `If-Match: *` - update only if the key exists
let s = h.to_str().map_err(|_| anyhow!("Invalid If-Match header"))?;
if s == "*" { mode = Some(SaveMode::Update); } // `If-Match: *` — update only if exist
else { mode = Some(SaveMode::Equal(s.to_string())); } // `If-Match: <md5>` — update only if current
} else if let Some(h) = req.headers().get("If-None-Match") { // `If-None-Match: *` — insert only if does not exist
let s = h.to_str().map_err(|_| anyhow!("Invalid If-None-Match header"))?;
if s == "*" { mode = Some(SaveMode::Insert); } else { return Err(anyhow!("If-None-Match must be '*'")); }
}
// MODE logic
let mut mode = Some(SaveMode::Upsert);
if let Some(h) = req.headers().get("If-Match") {
// `If-Match: *` - update only if the key exists
let s = h.to_str().map_err(|_| anyhow!("Invalid If-Match header"))?;
if s == "*" {
mode = Some(SaveMode::Update);
}
// `If-Match: *` — update only if exist
else {
mode = Some(SaveMode::Equal(s.to_string()));
} // `If-Match: <md5>` — update only if current
} else if let Some(h) = req.headers().get("If-None-Match") {
// `If-None-Match: *` — insert only if does not exist
let s = h
.to_str()
.map_err(|_| anyhow!("Invalid If-None-Match header"))?;
if s == "*" {
mode = Some(SaveMode::Insert);
} else {
return Err(anyhow!("If-None-Match must be '*'"));
}
}
redis_save(&mut *conn, &key, &body[..], ttl, mode).await?;
return Ok(HttpResponse::Ok().body("DONE"));
}().await.map_err(map_handler_error)
return Ok(HttpResponse::Ok().body("DONE"));
}()
.await
.map_err(map_handler_error)
}
/// delete
pub async fn delete(
req: HttpRequest,
path: web::Path<String>,
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
) -> Result<HttpResponse, actix_web::error::Error> {
workspace_check(&req)?; // Check workspace
let key: String = path.into_inner();
@@ -161,7 +183,21 @@ pub async fn delete(
async move || -> anyhow::Result<HttpResponse> {
let mut conn = redis.lock().await;
let deleted = redis_delete(&mut *conn, &key).await?;
// MODE logic
let mut mode = Some(SaveMode::Upsert);
if let Some(h) = req.headers().get("If-Match") {
// `If-Match: *` - delete only if the key exists
let s = h.to_str().map_err(|_| anyhow!("Invalid If-Match header"))?;
if s == "*" {
mode = Some(SaveMode::Update);
}
// `If-Match: *` — return error if not exist
else {
mode = Some(SaveMode::Equal(s.to_string()));
} // `If-Match: <md5>` — delete only if current
}
let deleted = redis_delete(&mut *conn, &key, mode).await?;
let response = match deleted {
true => HttpResponse::NoContent().finish(),
@@ -169,6 +205,7 @@ pub async fn delete(
};
Ok(response)
}().await.map_err(map_handler_error)
}()
.await
.map_err(map_handler_error)
}
src/handlers_ws.rs
+365 -275
View File
@@ -1,43 +1,70 @@
//
// 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 actix::prelude::*;
use uuid::Uuid;
use actix::{prelude::*};
use crate::ws_hub::{
WsHub, ServerMessage, SessionId,
Connect, Disconnect,
Subscribe, Unsubscribe, UnsubscribeAll,
SubscribeList,
Connect, Disconnect, ServerMessage, SessionId, Subscribe, SubscribeList, Unsubscribe,
UnsubscribeAll, WsHub,
};
use actix::{
Actor, ActorContext, ActorFutureExt, AsyncContext, Handler, StreamHandler, WrapFuture, fut,
};
use actix_web::{Error, HttpRequest, HttpResponse, web};
use actix_web_actors::ws;
use redis::aio::MultiplexedConnection;
use serde::Deserialize;
use serde_json::{Map, Value, json};
use std::collections::HashSet;
use std::sync::Arc;
use tokio::sync::Mutex;
use serde_json::{Value, Map, json};
use actix::{
Actor,
StreamHandler,
AsyncContext,
ActorContext,
fut,
ActorFutureExt,
Handler, WrapFuture
};
use actix_web::{web, HttpRequest, HttpResponse, Error};
use actix_web_actors::ws;
use serde::Deserialize;
use std::collections::HashSet;
use crate::redis_lib::{
Ttl, SaveMode,
RedisArray,
RedisArray, SaveMode, Ttl, deprecated_symbol, error, redis_delete, redis_list, redis_read,
redis_save,
redis_read,
redis_delete,
redis_list,
error,
deprecated_symbol,
};
type JsonMap = Map<String, Value>;
use serde::Serialize;
#[derive(Serialize, Default)]
struct ReturnBase<'a> {
action: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
key: Option<&'a str>,
#[serde(skip_serializing_if = "Option::is_none")]
data: Option<&'a str>,
#[serde(skip_serializing_if = "Option::is_none")]
correlation: Option<&'a str>,
#[serde(rename = "TTL", skip_serializing_if = "Option::is_none")]
ttl: Option<u64>,
#[serde(rename = "expiresAt", skip_serializing_if = "Option::is_none")]
expires_at: Option<u64>,
#[serde(rename = "ifMatch", skip_serializing_if = "Option::is_none")]
if_match: Option<&'a str>,
#[serde(rename = "ifNoneMatch", skip_serializing_if = "Option::is_none")]
if_none_match: Option<&'a str>,
}
/// WsCommand - commands enum (put, delete, sub, unsub)
#[derive(Deserialize, Debug)]
@@ -66,6 +93,16 @@ pub enum WsCommand {
if_none_match: Option<String>,
},
Delete {
#[serde(default)]
correlation: Option<String>,
key: String,
#[serde(rename = "ifMatch")]
#[serde(default)]
if_match: Option<String>,
},
Get {
#[serde(default)]
correlation: Option<String>,
@@ -78,16 +115,6 @@ pub enum WsCommand {
key: String,
},
Delete {
#[serde(default)]
correlation: Option<String>,
key: String,
#[serde(rename = "ifMatch")]
#[serde(default)]
if_match: Option<String>,
},
Sub {
#[serde(default)]
correlation: Option<String>,
@@ -117,7 +144,6 @@ pub struct WsSession {
pub claims: Option<Claims>,
}
/// Actor External trait: must be in separate impl block
impl Actor for WsSession {
type Context = ws::WebsocketContext<Self>;
@@ -131,33 +157,35 @@ impl Actor for WsSession {
self.hub
.send(Connect { addr: recipient })
.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();
}
.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
}
fn stopped(&mut self, _ctx: &mut Self::Context) {
if self.id != 0 { self.hub.do_send(Disconnect { session_id: self.id }); }
tracing::info!("WebSocket disconnected: {:?}",&self.id);
if self.id != 0 {
self.hub.do_send(Disconnect {
session_id: self.id,
});
}
tracing::info!("WebSocket disconnected: {:?}", &self.id);
}
}
impl actix::Handler<ServerMessage> 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);
}
}
@@ -166,16 +194,12 @@ impl actix::Handler<ServerMessage> for WsSession {
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
match msg {
Ok(ws::Message::Text(text)) => {
// println!("Message: {}", text);
match serde_json::from_str::<WsCommand>(&text) {
Ok(cmd) => self.handle_command(cmd, ctx),
Err(err) => ctx.text(format!("Invalid JSON: {}", err)),
}
}
Ok(ws::Message::Text(text)) => match serde_json::from_str::<WsCommand>(&text) {
Ok(cmd) => self.handle_command(cmd, ctx),
Err(err) => ctx.text(format!("Invalid JSON: {}", err)),
},
Ok(ws::Message::Ping(msg)) => ctx.pong(&msg),
Ok(ws::Message::Close(reason)) => {
// println!("Closing WS: {:?}", reason);
ctx.close(reason);
ctx.stop();
}
@@ -184,287 +208,352 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
}
}
use crate::workspace_owner::check_workspace_core;
/// All logic in one impl
impl WsSession {
fn ws_error(&self, ctx: &mut ws::WebsocketContext<Self>, msg: &str) {
ctx.text(format!(r#"{{"error":"{}"}}"#, msg));
}
fn workspace_check_ws(&self, key: &str) -> Result<(), &'static str> {
let claims = self.claims.as_ref().ok_or("Missing auth claims")?;
if claims.is_system() { return Ok(()); }
let jwt_workspace = claims.workspace.as_ref().ok_or("Missing workspace in token")?;
let path_ws = key.split('/').next().ok_or("Invalid key: missing workspace")?;
if path_ws.is_empty() { return Err("Invalid key: missing workspace"); }
let path_ws_uuid = Uuid::parse_str(path_ws).map_err(|_| "Invalid workspace UUID in key")?;
if jwt_workspace != &path_ws_uuid { return Err("Workspace mismatch"); }
Ok(())
check_workspace_core(claims, key)
}
fn wait_and_send<F>(
fn fut_send(
&mut self,
ctx: &mut ws::WebsocketContext<Self>,
fut: F,
mut base: JsonMap,
)
where
F: std::future::Future<Output = Result<JsonMap, String>> + 'static,
{
fut: impl Future<Output = Result<Value, String>> + 'static,
mut base: Value,
) {
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(extra) => { base.extend(extra); }
Err(err) => { base.insert("error".into(), json!(err)); }
Ok(Value::Object(extra)) => {
obj.extend(extra);
}
Ok(v) => {
obj.insert("extra".into(), v);
}
Err(err) => {
obj.insert("error".into(), json!(err));
}
}
ctx.text(Value::Object(base).to_string());
})
ctx.text(base.to_string());
}),
);
}
/// When valid JSON recieved for WsSession
fn handle_command(&mut self, cmd: WsCommand, ctx: &mut ws::WebsocketContext<Self>) {
match cmd {
WsCommand::Put {
key,
data,
expires_at,
ttl,
if_match,
if_none_match,
correlation,
} => {
tracing::info!(
"PUT {} = {} (expires_at: {:?}) (ttl: {:?}) correlation: {:?}",
&key,
&data,
&expires_at,
&ttl,
&correlation
);
WsCommand::Put { key, data, expires_at, ttl, if_match, if_none_match, correlation } => {
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) {
self.ws_error(ctx, e);
return;
}
tracing::info!("PUT {} = {} (expires_at: {:?}) (ttl: {:?}) correlation: {:?}", &key, &data, &expires_at, &ttl, &correlation);
let redis = self.redis.clone();
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; }
let base = serde_json::json!(ReturnBase {
action: "put",
key: Some(key.as_str()),
data: Some(data.as_str()),
correlation: correlation.as_deref(),
ttl,
expires_at,
if_match: if_match.as_deref(),
if_none_match: if_none_match.as_deref(),
});
let redis = self.redis.clone();
let mut base = JsonMap::new();
base.insert("action".into(), json!("put"));
base.insert("key".into(), json!(&key));
base.insert("data".into(), json!(&data));
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
if let Some(x) = &expires_at { base.insert("expiresAt".into(), json!(x)); }
if let Some(x) = &ttl { base.insert("TTL".into(), json!(x)); }
if let Some(x) = &if_match { base.insert("ifMatch".into(), json!(x)); }
if let Some(x) = &if_none_match { base.insert("ifNoneMatch".into(),json!(x)); }
let fut = async move {
// TTL logic
let fut = async move {
// TTL logic
let real_ttl = if let Some(secs) = ttl {
Some(Ttl::Sec(secs as usize))
} else if let Some(timestamp) = expires_at {
Some(Ttl::At(timestamp))
} else {
None
};
Some(Ttl::Sec(secs as usize))
} else if let Some(timestamp) = expires_at {
Some(Ttl::At(timestamp))
} else {
None
};
// MODE logic
let mut mode = Some(SaveMode::Upsert);
if let Some(s) = if_match { // `If-Match: *` - update only if the key exists
if s == "*" { // `If-Match: *` — update only if exist
mode = Some(SaveMode::Update);
} else { // `If-Match: <md5>` — update only if current
mode = Some(SaveMode::Equal(s.to_string()));
}
} else if let Some(s) = if_none_match { // `If-None-Match: *` — insert only if does not exist
if s == "*" {
mode = Some(SaveMode::Insert);
} else {
return Err::<JsonMap, String>("ifNoneMatch must contain only '*'".into());
}
}
// MODE logic
let mut mode = Some(SaveMode::Upsert);
if let Some(s) = if_match {
// `If-Match: *` - update only if the key exists
if s == "*" {
// `If-Match: *` — update only if exist
mode = Some(SaveMode::Update);
} else {
// `If-Match: <md5>` — update only if current
mode = Some(SaveMode::Equal(s.to_string()));
}
} else if let Some(s) = if_none_match {
// `If-None-Match: *` — insert only if does not exist
if s == "*" {
mode = Some(SaveMode::Insert);
} else {
return Err("ifNoneMatch must contain only '*'".into());
}
}
let mut conn = redis.lock().await;
let mut conn = redis.lock().await;
redis_save(&mut *conn, &key, &data, real_ttl, mode)
.await
.map_err(|e| e.to_string())?;
redis_save(&mut *conn, &key, &data, real_ttl, mode)
.await
.map_err(|e| e.to_string())?;
let mut extra = JsonMap::new();
extra.insert("response".into(), json!("OK"));
Ok::<JsonMap, String>(extra)
Ok(json!({"result": "OK"}))
};
};
self.fut_send(ctx, fut, base);
}
self.wait_and_send(ctx, fut, base);
}
WsCommand::Delete { key, correlation, if_match } => {
WsCommand::Delete {
key,
correlation,
if_match,
} => {
tracing::info!("DELETE {} correlation:{:?}", &key, &correlation);
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; }
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) {
self.ws_error(ctx, e);
return;
}
let redis = self.redis.clone();
let redis = self.redis.clone();
let mut base = JsonMap::new();
base.insert("action".into(), json!("delete"));
base.insert("key".into(), json!(&key));
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
if let Some(x) = &if_match { base.insert("ifMatch".into(), json!(x)); }
let base = serde_json::json!(ReturnBase {
action: "delete",
key: Some(key.as_str()),
correlation: correlation.as_deref(),
if_match: if_match.as_deref(),
..Default::default()
});
let fut = async move {
let fut = async move {
let mut conn = redis.lock().await;
let mut conn = redis.lock().await;
// MODE logic
let mut mode = Some(SaveMode::Upsert);
if let Some(s) = if_match {
// `If-Match: *` - delete only if the key exists
if s == "*" {
// `If-Match: *` — return error if not exist
mode = Some(SaveMode::Update);
} else {
// `If-Match: <md5>` — update only if current
mode = Some(SaveMode::Equal(s.to_string()));
}
}
let deleted = redis_delete(&mut *conn, &key).await.map_err(|e| e.to_string())?;
let deleted = redis_delete(&mut *conn, &key, mode)
.await
.map_err(|e| e.to_string())?;
if deleted {
let mut extra = JsonMap::new();
extra.insert("response".into(), json!("OK"));
Ok::<JsonMap, String>(extra)
} else {
Err::<JsonMap, String>("not found".into())
}
if deleted {
Ok(json!({"result": "OK"}))
} else {
Err("not found".into())
}
};
};
self.wait_and_send(ctx, fut, base);
self.fut_send(ctx, fut, base);
}
WsCommand::Get { key, correlation } => {
tracing::info!("GET {} correlation:{:?}", &key, &correlation);
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; }
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) {
self.ws_error(ctx, e);
return;
}
let redis = self.redis.clone();
let redis = self.redis.clone();
let mut base = JsonMap::new();
base.insert("action".into(), json!("get"));
base.insert("key".into(), json!(&key));
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
let base = serde_json::json!(ReturnBase {
action: "get",
key: Some(key.as_str()),
correlation: correlation.as_deref(),
..Default::default()
});
let fut = async move {
let fut = async move {
let mut conn = redis.lock().await;
let mut conn = redis.lock().await;
let data_opt = redis_read(&mut *conn, &key)
.await
.map_err(|e| e.to_string())?;
let data_opt = redis_read(&mut *conn, &key)
.await
.map_err(|e| e.to_string())?;
match data_opt {
Some(data) => {
let data_value =
serde_json::to_value(&data).map_err(|e| e.to_string())?;
Ok(json!({"result": data_value}))
}
None => Err("not found".into()),
}
};
match data_opt {
Some(data) => {
let mut extra = JsonMap::new();
let data_value = serde_json::to_value(&data).map_err(|e| e.to_string())?;
extra.insert("response".into(), data_value);
Ok::<JsonMap, String>(extra)
}
None => Err::<JsonMap, String>("not found".into())
}
};
self.wait_and_send(ctx, fut, base);
self.fut_send(ctx, fut, base);
}
WsCommand::List { key, correlation } => {
tracing::info!("LIST {:?} correlation: {:?}", &key, &correlation);
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; }
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) {
self.ws_error(ctx, e);
return;
}
let redis = self.redis.clone();
let redis = self.redis.clone();
let mut base = JsonMap::new();
base.insert("action".into(), json!("get"));
base.insert("key".into(), json!(&key));
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
let base = serde_json::json!(ReturnBase {
action: "list",
key: Some(key.as_str()),
correlation: correlation.as_deref(),
..Default::default()
});
let fut = async move {
let fut = async move {
let mut conn = redis.lock().await;
let data = redis_list(&mut *conn, &key)
.await
.map_err(|e| e.to_string())?;
Ok(json!({ "result": data }))
};
let mut conn = redis.lock().await;
let data = redis_list(&mut *conn, &key).await.map_err(|e| e.to_string())?;
let mut extra = JsonMap::new();
let data_value = serde_json::to_value(&data).map_err(|e| e.to_string())?;
extra.insert("response".into(), data_value);
Ok::<JsonMap, String>(extra)
};
self.wait_and_send(ctx, fut, base);
self.fut_send(ctx, fut, base);
}
WsCommand::Sub { key, correlation } => {
// LEVENT 3
tracing::info!("SUB {} correlation: {:?}", &key, &correlation);
WsCommand::Sub { key, correlation } => {
// LEVENT 3
tracing::info!("SUB {} correlation: {:?}", &key, &correlation);
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) {
self.ws_error(ctx, e);
return;
}
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; }
let mut obj = serde_json::json!(ReturnBase {
action: "sub",
key: Some(key.as_str()),
correlation: correlation.as_deref(),
..Default::default()
});
let mut obj = JsonMap::new();
obj.insert("action".into(), json!("sub"));
obj.insert("key".into(), json!(key));
if let Some(c) = correlation { obj.insert("correlation".into(), json!(c)); }
let map = obj.as_object_mut().unwrap();
if deprecated_symbol(&key) {
obj.insert("error".into(), json!("Deprecated symbol in key"));
} else {
self.hub.do_send(Subscribe { session_id: self.id, key: key.clone() });
}
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(),
});
map.insert("result".into(), json!("OK"));
}
ctx.text(Value::Object(obj).to_string());
}
WsCommand::Unsub { key, correlation } => {
// LEVENT 4
tracing::info!("UNSUB {} correlation: {:?}", &key, &correlation);
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; }
let mut obj = JsonMap::new();
obj.insert("action".into(), json!("unsub"));
obj.insert("key".into(), json!(key));
if let Some(c) = correlation { obj.insert("correlation".into(), json!(c)); }
if key == "*" {
self.hub.do_send(UnsubscribeAll { session_id: self.id });
} else {
if deprecated_symbol(&key) {
obj.insert("error".into(), json!("Deprecated symbol in key"));
} else {
self.hub.do_send(Unsubscribe { session_id: self.id, key: key.clone() });
}
};
ctx.text(Value::Object(obj).to_string());
}
WsCommand::Sublist { correlation } => {
tracing::info!("SUBLIST correlation: {:?}", &correlation);
// w/o Check workspace!
let mut base = JsonMap::new();
base.insert("action".into(), json!("sublist"));
if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); }
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();
let mut extra = JsonMap::new();
extra.insert("response".into(), serde_json::to_value(&keys).map_err(|e| e.to_string())? );
Ok::<JsonMap, String>(extra)
};
self.wait_and_send(ctx, fut, base);
ctx.text(obj.to_string());
}
// End of commands
WsCommand::Unsub { key, correlation } => {
// LEVENT 4
tracing::info!("UNSUB {} correlation: {:?}", &key, &correlation);
let mut obj = serde_json::json!(ReturnBase {
action: "unsub",
key: Some(key.as_str()),
correlation: correlation.as_deref(),
..Default::default()
});
let map = obj.as_object_mut().unwrap();
if key == "*" {
self.hub.do_send(UnsubscribeAll {
session_id: self.id,
});
map.insert("result".into(), json!("OK"));
} else {
// Check workspace
if let Err(e) = self.workspace_check_ws(&key) {
self.ws_error(ctx, e);
return;
}
if deprecated_symbol(&key) {
map.insert("error".into(), json!("Deprecated symbol in key"));
} else {
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);
// w/o Check workspace!
let base = serde_json::json!(ReturnBase {
action: "list",
correlation: correlation.as_deref(),
..Default::default()
});
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, fut, base);
} // End of commands
}
}
}
// ---- auth
use actix_web::{HttpMessage,error};
use url::form_urlencoded;
use jsonwebtoken::{decode, DecodingKey, Validation, Algorithm};
use crate::CONFIG;
use actix_web::{HttpMessage, error};
use jsonwebtoken::{Algorithm, DecodingKey, Validation, decode};
use url::form_urlencoded;
pub async fn handler(
req: HttpRequest,
@@ -472,23 +561,24 @@ pub async fn handler(
redis: web::Data<Arc<Mutex<MultiplexedConnection>>>,
hub: web::Data<Addr<WsHub>>,
) -> Result<HttpResponse, Error> {
let token_opt = req.uri().query().and_then(|q| {
form_urlencoded::parse(q.as_bytes())
.find(|(k, _)| k == "token")
.map(|(_, v)| v.into_owned())
});
form_urlencoded::parse(q.as_bytes())
.find(|(k, _)| k == "token")
.map(|(_, v)| v.into_owned())
});
let claims = match token_opt {
Some(t) if !t.is_empty() => {
let mut validation = Validation::new(Algorithm::HS256);
validation.required_spec_claims = HashSet::new(); // no: exp/iat/nbf
let mut validation = Validation::new(Algorithm::HS256);
validation.required_spec_claims = HashSet::new(); // no: exp/iat/nbf
let c = decode::<Claims>(&t, &DecodingKey::from_secret(CONFIG.token_secret.as_bytes()), &validation )
.map(|td| td.claims)
.map_err(|_e| error::ErrorUnauthorized("Invalid token"))?;
let c = decode::<Claims>(
&t,
&DecodingKey::from_secret(CONFIG.token_secret.as_bytes()),
&validation,
)
.map(|td| td.claims)
.map_err(|_e| error::ErrorUnauthorized("Invalid token"))?;
Some(c)
}
src/main.rs
+58 -112
View File
@@ -23,10 +23,14 @@ use actix_web::{
App, Error, HttpMessage, HttpRequest, HttpResponse, HttpServer,
body::MessageBody,
dev::{ServiceRequest, ServiceResponse},
error::ErrorBadRequest,
http::header::{AUTHORIZATION, HeaderValue},
middleware::{self, Next},
web::{self, Data, PayloadConfig},
};
use url::form_urlencoded;
use actix_web_actors::ws;
use tracing::info;
@@ -41,15 +45,10 @@ use crate::redis_lib::redis_connect;
mod workspace_owner;
// == =hub ===
mod redis_events;
mod ws_hub;
use crate::ws_hub::{ServerMessage, TestGetSubs, WsHub};
use actix::prelude::*;
use crate::ws_hub::{WsHub, ServerMessage,
TestGetSubs,
};
// === /hub ===
use config::CONFIG;
@@ -70,54 +69,55 @@ fn initialize_tracing(level: tracing::Level) {
.init();
}
// #[allow(dead_code)]
async fn interceptor(
request: ServiceRequest,
mut request: ServiceRequest,
next: Next<impl MessageBody>,
) -> Result<ServiceResponse<impl MessageBody>, Error> {
// Authorization/token patch
if request.headers().get(AUTHORIZATION).is_none() {
if let Some(qs) = request.uri().query() {
if let Some(token) = form_urlencoded::parse(qs.as_bytes())
.find(|(k, _)| k == "token")
.map(|(_, v)| v.into_owned())
{
let auth_value = HeaderValue::from_str(&format!("Bearer {}", token))
.map_err(|_| ErrorBadRequest("Malformed token"))?;
request.headers_mut().insert(AUTHORIZATION, auth_value);
}
}
}
let secret = SecretString::new(CONFIG.token_secret.clone().into_boxed_str());
let claims = request.extract_claims(&secret)?;
// TODO: сделать это здесь
request.extensions_mut().insert(claims.to_owned());
// TODO потом исправить hulyrs: extract_claims
next.call(request).await
}
// =====================================================================================
// =====================================================================================
// =====================================================================================
// =====================================================================================
// =====================================================================================
// =====================================================================================
// =====================================================================================
// =====================================================================================
use crate::redis_events::RedisEventAction::*; // Set, Del, Unlink, Expired, Other
pub async fn start_redis_logger(redis_url: String, hub: Addr<WsHub>) {
let client = match redis::Client::open(redis_url) {
Ok(c) => c,
Err(e) => { eprintln!("[redis] bad url: {e}"); return; }
Err(e) => {
eprintln!("[redis] bad url: {e}");
return;
}
};
match crate::redis_events::make_pubsub_with_kea(&client).await {
Ok(pubsub) => {
let (mut rx, _handle) = crate::redis_events::start_keyevent_listener(pubsub);
while let Some(ev) = rx.recv().await {
match ev.action {
Set => println!("[redis] db{} SET {}", ev.db, ev.key),
Del | Unlink => println!("[redis] db{} DEL {}", ev.db, ev.key),
Expired => println!("[redis] db{} EXPIRED {}", ev.db, ev.key),
Other(ref k) => println!("[redis] db{} {} {}", ev.db, k, ev.key),
}
/*
match ev.action {
Set => println!("[redis] db{} SET {}", ev.db, ev.key),
Del | Unlink => println!("[redis] db{} DEL {}", ev.db, ev.key),
Expired => println!("[redis] db{} EXPIRED {}", ev.db, ev.key),
Other(ref k) => println!("[redis] db{} {} {}", ev.db, k, ev.key),
}
*/
hub.do_send(ev.clone());
}
@@ -126,14 +126,6 @@ pub async fn start_redis_logger(redis_url: String, hub: Addr<WsHub>) {
}
}
// use actix_web::http::header;
// use actix_web::http::header::HeaderValue;
// use actix_web::body::BoxBody;
// use url::form_urlencoded;
// #[tokio::main]
#[actix_web::main]
async fn main() -> anyhow::Result<()> {
@@ -146,12 +138,14 @@ async fn main() -> anyhow::Result<()> {
let redis_data = web::Data::new(redis.clone());
// starting Hub
// let hub = WsHub::default().start();
let hub = WsHub::new(redis.clone()).start();
let hub_data = web::Data::new(hub.clone());
// starting Logger
tokio::spawn(start_redis_logger("redis://127.0.0.1/".to_string(), hub.clone()));
tokio::spawn(start_redis_logger(
"redis://127.0.0.1/".to_string(),
hub.clone(),
));
let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port);
let payload_config = PayloadConfig::new(CONFIG.payload_size_limit.bytes() as usize);
@@ -167,85 +161,37 @@ async fn main() -> anyhow::Result<()> {
App::new()
.app_data(payload_config.clone())
.app_data(redis_data.clone())
.app_data(hub_data.clone())
.app_data(hub_data.clone())
.wrap(middleware::Logger::default())
.wrap(cors)
.service(
web::scope("/api")
.wrap(middleware::from_fn(interceptor))
.route("/{key:.+/}", web::get().to(handlers_http::list))
.route("/{key:.+/}", web::get().to(handlers_http::list))
.route("/{key:.+}", web::get().to(handlers_http::get))
.route("/{key:.+}", web::put().to(handlers_http::put))
.route("/{key:.+}", web::delete().to(handlers_http::delete))
.route("/{key:.+}", web::put().to(handlers_http::put))
.route("/{key:.+}", web::delete().to(handlers_http::delete)),
)
.route("/status", web::get().to(async || "ok"))
// .route("/stat", web::get().to(ws_hub::stat))
.route("/stat2", web::get().to(|hub: web::Data<Addr<WsHub>>| 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<Addr<WsHub>>| async move {
match hub.send(TestGetSubs).await {
Ok(subs) => HttpResponse::Ok().json(subs),
Err(_) => HttpResponse::InternalServerError().body("Failed to get subscriptions"),
}
}))
.route("/ws", web::get().to(handlers_ws::handler)) // WebSocket
/*
.service(
web::resource("/ws")
.wrap(middleware::from_fn(|mut req: ServiceRequest, next: Next<BoxBody>| async move {
// Уже есть Authorization?
let has_auth = req.headers().contains_key(header::AUTHORIZATION);
if !has_auth {
// ?token=...
if let Some(token) = form_urlencoded::parse(req.query_string().as_bytes())
.find(|(k, _)| k == "token")
.map(|(_, v)| v.into_owned())
{
if !token.is_empty() {
let value = format!("Bearer {}", token);
req.headers_mut().insert(
header::AUTHORIZATION,
HeaderValue::from_str(&value)
.map_err(|_| actix_web::error::ErrorBadRequest("Invalid token header"))?,
);
}
}
}
next.call(req).await
}))
// затем твой interceptor:
// .wrap(middleware::from_fn(interceptor))
.route(web::get().to(handlers_ws::handler))
)
*/
.route(
"/stat2",
web::get().to(|hub: web::Data<Addr<WsHub>>| 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<Addr<WsHub>>| async move {
match hub.send(TestGetSubs).await {
Ok(subs) => HttpResponse::Ok().json(subs),
Err(_) => {
HttpResponse::InternalServerError().body("Failed to get subscriptions")
}
}
}),
)
.route("/ws", web::get().to(handlers_ws::handler)) // WebSocket
})
.bind(socket)?
.run();
src/redis_events.rs
+59 -31
View File
@@ -1,24 +1,35 @@
//
// 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 tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_stream::StreamExt;
use serde::Serialize;
use redis::{
self,
AsyncCommands,
RedisResult,
Client,
aio::{PubSub, ConnectionLike},
self, AsyncCommands, Client, RedisResult,
aio::{ConnectionLike, PubSub},
};
#[derive(Debug, Clone, Serialize)]
pub enum RedisEventAction {
Set, // Insert or Update
Del, // Delete
Unlink, // async Delete
Expired, // TTL Delete
Set, // Insert or Update
Del, // Delete
Unlink, // async Delete
Expired, // TTL Delete
Other(String),
}
@@ -29,17 +40,21 @@ use actix::Message;
pub struct RedisEvent {
pub db: u32,
pub key: String,
// pub value: String,
// pub value: String,
pub action: RedisEventAction,
}
/// Notifications: keyevent + generic + expired = "Egx" (no keyspace)
async fn try_enable_keyspace_notifications<C>(conn: &mut C) -> RedisResult<()>
where
C: ConnectionLike + Send,
{
let _: String = redis::cmd("CONFIG").arg("SET").arg("notify-keyspace-events").arg("E$gx").query_async(conn).await?;
let _: String = redis::cmd("CONFIG")
.arg("SET")
.arg("notify-keyspace-events")
.arg("E$gx")
.query_async(conn)
.await?;
Ok(())
}
@@ -82,33 +97,46 @@ pub fn start_keyevent_listener(
while let Some(msg) = stream.next().await {
let channel = match msg.get_channel::<String>() {
Ok(c) => c,
Err(e) => { eprintln!("[redis_events] bad channel: {e}"); continue; }
Ok(c) => c,
Err(e) => {
eprintln!("[redis_events] bad channel: {e}");
continue;
}
};
let payload = match msg.get_payload::<String>() {
Ok(p) => p,
Err(e) => { eprintln!("[redis_events] bad payload: {e}"); continue; }
Ok(p) => p,
Err(e) => {
eprintln!("[redis_events] bad payload: {e}");
continue;
}
};
// "__keyevent@0__:set" → event="set", db=0; payload = key
let event = channel.rsplit(':').next().unwrap_or("");
let action = match event {
"set" => RedisEventAction::Set,
"del" => RedisEventAction::Del,
"unlink" => RedisEventAction::Unlink,
"expired" => RedisEventAction::Expired,
other => RedisEventAction::Other(other.to_string()),
};
let action = match event {
"set" => RedisEventAction::Set,
"del" => RedisEventAction::Del,
"unlink" => RedisEventAction::Unlink,
"expired" => RedisEventAction::Expired,
other => RedisEventAction::Other(other.to_string()),
};
let db = channel.find('@')
.and_then(|at| channel.get(at + 1..))
.and_then(|rest| rest.find("__:").map(|end| &rest[..end]))
.and_then(|s| s.parse::<u32>().ok())
.unwrap_or(0);
let db = channel
.find('@')
.and_then(|at| channel.get(at + 1..))
.and_then(|rest| rest.find("__:").map(|end| &rest[..end]))
.and_then(|s| s.parse::<u32>().ok())
.unwrap_or(0);
let ev = RedisEvent { db, key: payload.clone(), action };
let ev = RedisEvent {
db,
key: payload.clone(),
action,
};
if tx.send(ev).is_err() { break; } // closed
if tx.send(ev).is_err() {
break;
} // closed
}
});
src/redis_lib.rs
+158 -80
View File
@@ -1,3 +1,18 @@
//
// 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 crate::config::{CONFIG, RedisMode};
use std::time::{SystemTime, UNIX_EPOCH};
@@ -5,21 +20,20 @@ use std::time::{SystemTime, UNIX_EPOCH};
#[derive(serde::Serialize)]
pub enum Ttl {
Sec(usize), // EX
At(u64), // EXAT (timestamp in seconds)
At(u64), // EXAT (timestamp in seconds)
}
#[derive(Debug)]
pub enum SaveMode {
Upsert, // default: set or overwrite
Insert, // only if not exists (NX)
Update, // only if exists (XX)
Upsert, // default: set or overwrite
Insert, // only if not exists (NX)
Update, // only if exists (XX)
Equal(String), // only if md5 matches provided
}
use redis::{
AsyncCommands, RedisResult,
ToRedisArgs,
Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, aio::MultiplexedConnection
AsyncCommands, Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, RedisResult,
ToRedisArgs, aio::MultiplexedConnection,
};
use url::Url;
@@ -30,24 +44,28 @@ pub struct RedisArray {
pub key: String,
pub data: String,
pub expires_at: u64, // sec to expire TTL
pub etag: String, // md5 hash (data)
pub etag: String, // md5 hash (data)
}
/// return Error
pub fn error<T>(code: u16, msg: impl Into<String>) -> redis::RedisResult<T> {
let msg = msg.into();
let full = format!("{}: {}", code, msg);
Err(redis::RedisError::from(( redis::ErrorKind::ExtensionError, "", full )))
Err(redis::RedisError::from((
redis::ErrorKind::ExtensionError,
"",
full,
)))
}
/// Check for redis-deprecated symbols
pub fn deprecated_symbol(s: &str) -> bool {
s.chars().any(|c| matches!(
c,
'*' | '?' | '[' | ']' | '\\' |
'\0'..='\x1F' | '\x7F' |
'"' | '\''
))
s.chars().any(|c| {
matches!(
c,
'*' | '?' | '[' | ']' | '\\' | '\0'..='\x1F' | '\x7F' | '"' | '\''
)
})
}
pub fn deprecated_symbol_error(s: &str) -> redis::RedisResult<()> {
@@ -63,70 +81,84 @@ pub async fn redis_list(
conn: &mut MultiplexedConnection,
key: &str,
) -> redis::RedisResult<Vec<RedisArray>> {
deprecated_symbol_error(key)?;
if !key.ends_with('/') { return error(412, "Key must end with slash"); }
if !key.ends_with('/') {
return error(412, "Key must end with slash");
}
let pattern = format!("{key}*");
let mut cursor = 0u64;
let mut results = Vec::new();
loop {
let mut cmd = redis::cmd("SCAN");
let mut cmd = redis::cmd("SCAN");
cmd.arg(cursor);
cmd.arg("MATCH").arg(&pattern);
// cmd.arg("COUNT").arg(100); // Optionally adjust batch size
let (next_cursor, keys): (u64, Vec<String>) = cmd.query_async(conn).await?;
for k in keys {
for k in keys {
// Check for $-security path
if k.strip_prefix(key).map_or(false, |s| s.contains('$')) {
continue;
}
// Check for $-security path
if k.strip_prefix(key).map_or(false, |s| s.contains('$')) { continue; }
// Get value
let value: Option<String> = redis::cmd("GET").arg(&k).query_async(conn).await?;
let Some(value) = value else {
continue;
}; // Old and deleted
// Get value
let value: Option<String> = redis::cmd("GET").arg(&k).query_async(conn).await?;
let Some(value) = value else { continue; }; // Old and deleted
// Get TTL
let ttl: i64 = redis::cmd("TTL").arg(&k).query_async(conn).await?;
if ttl >= 0 {
results.push(RedisArray {
key: k,
data: value.clone(),
expires_at: ttl as u64,
etag: hex::encode(md5::compute(&value).0),
});
}
}
// Get TTL
let ttl: i64 = redis::cmd("TTL").arg(&k).query_async(conn).await?;
if ttl >= 0 {
results.push(RedisArray {
key: k,
data: value.clone(),
expires_at: ttl as u64,
etag: hex::encode(md5::compute(&value).0),
});
}
}
if next_cursor == 0 { break;}
cursor = next_cursor;
if next_cursor == 0 {
break;
}
cursor = next_cursor;
}
Ok(results)
}
/// redis_read(&connection,key)
#[allow(dead_code)]
pub async fn redis_read(
conn: &mut MultiplexedConnection,
key: &str,
) -> redis::RedisResult<Option<RedisArray>> {
deprecated_symbol_error(key)?;
if key.ends_with('/') { return error(412, "Key must not end with a slash"); }
if key.ends_with('/') {
return error(412, "Key must not end with a slash");
}
let data: Option<String> = redis::cmd("GET").arg(key).query_async(conn).await?;
let Some(data) = data else { return Ok(None); };
let Some(data) = data else {
return Ok(None);
};
let ttl: i64 = redis::cmd("TTL").arg(key).query_async(conn).await?;
if ttl == -1 { return error(500, "TTL not set"); }
if ttl == -2 { return error(500, "Key not found"); }
if ttl < 0 { return error(500, "Unknown TTL error"); }
if ttl == -1 {
return error(500, "TTL not set");
}
if ttl == -2 {
return error(500, "Key not found");
}
if ttl < 0 {
return error(500, "Unknown TTL error");
}
Ok(Some(RedisArray {
key: key.to_string(),
@@ -136,7 +168,6 @@ pub async fn redis_read(
}))
}
/// TTL sec
/// redis_save(&mut conn, "key", "val", Some(Ttl::Sec(300)), Some(SaveMode::Insert)).await?;
///
@@ -155,25 +186,33 @@ pub async fn redis_save<T: ToRedisArgs>(
ttl: Option<Ttl>,
mode: Option<SaveMode>,
) -> RedisResult<()> {
deprecated_symbol_error(&key)?;
if key.ends_with('/') { return error(412, "Key must not end with a slash"); }
if key.ends_with('/') {
return error(412, "Key must not end with a slash");
}
// TTL logic
let sec = match ttl {
Some(Ttl::Sec(secs)) => secs,
Some(Ttl::At(timestamp)) => {
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
if timestamp <= now {
return error(400, "TTL timestamp exceeds MAX_TTL limit");
}
(timestamp - now) as usize
}
None => CONFIG.max_ttl,
Some(Ttl::Sec(secs)) => secs,
Some(Ttl::At(timestamp)) => {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
if timestamp <= now {
return error(400, "TTL timestamp exceeds MAX_TTL limit");
}
(timestamp - now) as usize
}
None => CONFIG.max_ttl,
};
if sec == 0 { return error(400, "TTL must be > 0"); }
if sec > CONFIG.max_ttl { return error(412, "TTL exceeds MAX_TTL"); }
if sec == 0 {
return error(400, "TTL must be > 0");
}
if sec > CONFIG.max_ttl {
return error(412, "TTL exceeds MAX_TTL");
}
let mut cmd = redis::cmd("SET");
cmd.arg(key).arg(value).arg("EX").arg(sec);
@@ -182,25 +221,39 @@ pub async fn redis_save<T: ToRedisArgs>(
let mode = mode.unwrap_or(SaveMode::Upsert);
match mode {
SaveMode::Upsert => {} // none
SaveMode::Insert => { cmd.arg("NX"); } // if NOT Exist
SaveMode::Insert => {
cmd.arg("NX");
} // if NOT Exist
SaveMode::Update => { cmd.arg("XX"); } // if Exist
SaveMode::Update => {
cmd.arg("XX");
} // if Exist
SaveMode::Equal(ref expected_md5) => { // if md5 === actual_md5
let current_value: Option<String> = redis::cmd("GET").arg(key).query_async(conn).await?;
SaveMode::Equal(ref expected_md5) => {
// if md5 === actual_md5
let current_value: Option<String> =
redis::cmd("GET").arg(key).query_async(conn).await?;
if let Some(existing) = current_value {
let actual_md5 = hex::encode(md5::compute(&existing).0);
if &actual_md5 != expected_md5 { return error(412, format!("md5 mismatch, current: {}, expected: {}", actual_md5, expected_md5)); }
} else { return error(404, "Equal: key does not exist"); }
if &actual_md5 != expected_md5 {
return error(
412,
format!(
"md5 mismatch, current: {}, expected: {}",
actual_md5, expected_md5
),
);
}
} else {
return error(404, "Equal: key does not exist");
}
}
}
// execute
let result: Option<String> = cmd.query_async(conn).await?;
// // execute
// cmd.query_async::<i64>(&mut *conn).await?;
if result.is_none() {
match mode {
@@ -213,28 +266,54 @@ pub async fn redis_save<T: ToRedisArgs>(
Ok(())
}
/// redis_delete(&connection,key)
#[allow(dead_code)]
pub async fn redis_delete(
conn: &mut MultiplexedConnection,
key: &str,
) -> redis::RedisResult<bool> {
mode: Option<SaveMode>, // <— добавили
) -> RedisResult<bool> {
deprecated_symbol_error(key)?;
if key.ends_with('/') { return error(412, "Key must not end with a slash"); }
if key.ends_with('/') {
return error(412, "Key must not end with a slash");
}
let deleted: i32 = redis::cmd("DEL")
.arg(key)
.query_async(conn)
.await?;
let mode = mode.unwrap_or(SaveMode::Upsert);
match mode {
SaveMode::Equal(ref expected_md5) => {
let current: Option<String> = redis::cmd("GET").arg(key).query_async(conn).await?;
match current {
None => return error(404, "Equal: key does not exist"),
Some(val) => {
let actual_md5 = hex::encode(md5::compute(&val).0);
if &actual_md5 != expected_md5 {
return error(
412,
format!(
"md5 mismatch, current: {}, expected: {}",
actual_md5, expected_md5
),
);
}
}
}
}
SaveMode::Insert => {
return error(412, "Insert mode is not supported for delete");
}
SaveMode::Update | SaveMode::Upsert => {}
}
let deleted: i32 = redis::cmd("DEL").arg(key).query_async(conn).await?;
if deleted == 0 && matches!(mode, SaveMode::Equal(_)) {
return error(404, "Delete: key does not exist");
}
Ok(deleted > 0)
}
/// redis_connect()
pub async fn redis_connect() -> anyhow::Result<MultiplexedConnection> {
let default_port = match CONFIG.redis_mode {
@@ -292,4 +371,3 @@ pub async fn redis_connect() -> anyhow::Result<MultiplexedConnection> {
Ok(conn)
}
src/workspace_owner.rs
+46 -27
View File
@@ -1,43 +1,62 @@
//
// 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 actix_web::{Error, HttpMessage, HttpRequest};
use hulyrs::services::jwt::Claims;
use uuid::Uuid;
use actix_web::{ Error, HttpMessage, HttpRequest, error };
/// Checking workspace in Authorization
pub fn workspace_check(req: &HttpRequest) -> Result<(), Error> {
let extensions = req.extensions();
// Get key
let key = req
.match_info()
.get("key")
.ok_or_else(|| error::ErrorBadRequest("Missing key in URL path"))?;
// Get workspace
let path_ws = match key.find('/') {
Some(x) if x > 0 => &key[..x],
_ => return Err(error::ErrorBadRequest("Invalid key: missing workspace")),
};
let claims = extensions
.get::<Claims>()
.ok_or_else(|| error::ErrorUnauthorized("Missing auth claims"))?;
// is_system - allowed to all
// common checker
pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> {
if claims.is_system() {
return Ok(());
}
// else - check workplace
let jwt_workspace = claims
.workspace
.as_ref()
.ok_or_else(|| error::ErrorForbidden("Missing workspace in token"))?;
let path_ws_uuid = Uuid::parse_str(path_ws).map_err(|_| error::ErrorBadRequest("Invalid workspace UUID"))?;
.ok_or("Missing workspace in token")?;
let path_ws = key
.split('/')
.next()
.ok_or("Invalid key: missing workspace")?;
if path_ws.is_empty() {
return Err("Invalid key: missing workspace");
}
let path_ws_uuid = Uuid::parse_str(path_ws).map_err(|_| "Invalid workspace UUID in key")?;
if jwt_workspace != &path_ws_uuid {
return Err(error::ErrorForbidden("Workspace mismatch"));
return Err("Workspace mismatch");
}
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::<Claims>()
.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)),
}
}
src/ws_hub.rs
+40 -19
View File
@@ -1,7 +1,24 @@
use std::collections::{ HashMap, HashSet };
//
// 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};
fn subscription_matches(sub_key: &str, key: &str) -> bool {
if sub_key == key { return true; }
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('$');
@@ -9,16 +26,16 @@ fn subscription_matches(sub_key: &str, key: &str) -> bool {
false
}
use crate::redis_events::{ RedisEvent, RedisEventAction };
use crate::redis_events::{RedisEvent, RedisEventAction};
use serde::Serialize;
#[derive(Message, Clone, Serialize, Debug)]
#[rtype(result = "()")]
pub struct ServerMessage {
#[serde(flatten)]
pub event: RedisEvent, // поля RedisEvent «вливаются» в корень JSON
pub event: RedisEvent,
#[serde(skip_serializing_if = "Option::is_none")]
pub value: Option<String>, // будет только при Set
pub value: Option<String>,
}
/// Count of active sessions
@@ -61,7 +78,7 @@ impl Handler<Connect> for WsHub {
type Result = SessionId;
fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
// LEVENT 1
// LEVENT 1
let id = self.next_id;
self.next_id = self.next_id.wrapping_add(1);
self.sessions.insert(id, msg.addr);
@@ -81,7 +98,7 @@ impl Handler<Disconnect> for WsHub {
type Result = ();
fn handle(&mut self, msg: Disconnect, _ctx: &mut Context<Self>) {
// LEVENT 2
// LEVENT 2
// Delete all subscribes
self.subs.retain(|_key, session_ids| {
@@ -110,7 +127,8 @@ impl Handler<SubscribeList> for WsHub {
fn handle(&mut self, msg: SubscribeList, _ctx: &mut Context<Self>) -> Self::Result {
// Collect all keys with my session_id
let list = self.subs
let list = self
.subs
.iter()
.filter_map(|(key, sessions)| {
if sessions.contains(&msg.session_id) {
@@ -162,7 +180,9 @@ impl Handler<Unsubscribe> for WsHub {
fn handle(&mut self, msg: Unsubscribe, _ctx: &mut Context<Self>) {
if let Some(set) = self.subs.get_mut(&msg.key) {
set.remove(&msg.session_id);
if set.is_empty() { self.subs.remove(&msg.key); }
if set.is_empty() {
self.subs.remove(&msg.key);
}
}
}
}
@@ -183,9 +203,6 @@ impl Handler<UnsubscribeAll> for WsHub {
}
}
#[derive(Message)]
#[rtype(result = "HashMap<String, Vec<SessionId>>")]
pub struct TestGetSubs;
@@ -194,7 +211,8 @@ impl Handler<TestGetSubs> for WsHub {
type Result = MessageResult<TestGetSubs>;
fn handle(&mut self, _msg: TestGetSubs, _ctx: &mut Context<Self>) -> Self::Result {
let s: HashMap<String, Vec<SessionId>> = self.subs
let s: HashMap<String, Vec<SessionId>> = self
.subs
.iter()
.map(|(key, ids)| (key.clone(), ids.iter().copied().collect()))
.collect();
@@ -215,12 +233,12 @@ impl WsHub {
}
}
use actix::prelude::*;
use actix::ActorFutureExt;
use actix::fut::ready;
use actix::prelude::*;
use redis::aio::MultiplexedConnection;
use std::sync::Arc;
use tokio::sync::Mutex;
use redis::aio::MultiplexedConnection;
impl Handler<RedisEvent> for WsHub {
type Result = ResponseActFuture<Self, ()>;
@@ -231,7 +249,8 @@ impl Handler<RedisEvent> for WsHub {
return Box::pin(actix::fut::ready(()).into_actor(self));
}
let recipients: Vec<Recipient<ServerMessage>> = targets.into_iter()
let recipients: Vec<Recipient<ServerMessage>> = targets
.into_iter()
.filter_map(|sid| self.sessions.get(&sid).cloned())
.collect();
@@ -242,9 +261,11 @@ impl Handler<RedisEvent> for WsHub {
Box::pin(
async move {
let value = if need_get {
let mut conn = redis.lock().await;
match redis::cmd("GET").arg(&event.key).query_async::<Option<String>>(&mut *conn).await
match redis::cmd("GET")
.arg(&event.key)
.query_async::<Option<String>>(&mut *conn)
.await
{
Ok(v) => v,
Err(e) => {
@@ -262,7 +283,7 @@ impl Handler<RedisEvent> for WsHub {
let _ = rcpt.do_send(payload.clone());
}
}
.into_actor(self)
.into_actor(self),
)
}
}