diff --git a/Cargo.lock b/Cargo.lock
index ac96a96a2e..7f17d70d92 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1181,7 +1181,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
[[package]]
name = "hulypulse"
-version = "0.1.8"
+version = "0.1.10"
dependencies = [
"actix",
"actix-cors",
diff --git a/Cargo.toml b/Cargo.toml
index 679318aeab..c79af425b2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "hulypulse"
-version = "0.1.8"
+version = "0.1.10"
edition = "2024"
[dependencies]
diff --git a/scripts/TEST_HTTP_API.sh b/scripts/TEST_HTTP_API.sh
index 04383691a1..a07394a694 100755
--- a/scripts/TEST_HTTP_API.sh
+++ b/scripts/TEST_HTTP_API.sh
@@ -6,10 +6,28 @@ source ./pulse_lib.sh
TOKEN=$(./token.sh claims.json)
#echo ${TOKEN}
#exit
-
-
ZP="00000000-0000-0000-0000-000000000001/TESTS"
+
+
+put "00000000-0000-0000-0000-000000000001/TESTS/val1" "value" "HULY-TTL: 1"
+put "00000000-0000-0000-0000-000000000001/TESTS/val2" "value" "HULY-TTL: 12"
+put "00000000-0000-0000-0000-000000000001/TESTS/val3" "value" "HULY-TTL: 1"
+
+get "00000000-0000-0000-0000-000000000001/TESTS/"
+sleep 2
+# get "00000000-0000-0000-0000-000000000001/TESTS/val2"
+get "00000000-0000-0000-0000-000000000001/TESTS/"
+
+
+
+exit
+
+
+
+
+
+
put "00000000-0000-0000-0000-000000000001/TESTS" "Value"
#exit
diff --git a/scripts/TEST_lleo.html b/scripts/TEST_lleo.html
new file mode 100644
index 0000000000..706cc56619
--- /dev/null
+++ b/scripts/TEST_lleo.html
@@ -0,0 +1,140 @@
+
+
+
+
+ WebSocket JSON Tester
+
+
+
+
+ WebSocket JSON Tester
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Waiting for server response...
+
+
+
+
+
diff --git a/scripts/TEST_no_auth.html b/scripts/TEST_no_auth.html
new file mode 100644
index 0000000000..26b34faca7
--- /dev/null
+++ b/scripts/TEST_no_auth.html
@@ -0,0 +1,141 @@
+
+
+
+
+ WebSocket JSON Tester
+
+
+
+
+ WebSocket JSON Tester
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Waiting for server response...
+
+
+
+
+
diff --git a/scripts/lleo_TEST_HTTP_API.sh b/scripts/lleo_TEST_HTTP_API.sh
new file mode 100755
index 0000000000..a30faf5608
--- /dev/null
+++ b/scripts/lleo_TEST_HTTP_API.sh
@@ -0,0 +1,149 @@
+#!/bin/bash
+
+clear
+source ./pulse_lib_lleo.sh
+
+#TOKEN=$(./token.sh claims.json)
+#echo ${TOKEN}
+
+
+
+put "dnevnik/onlline/admin" "oki" "HULY-TTL: 3"
+
+
+
+exit
+ZP="00000000-0000-0000-0000-000000000001/TESTS"
+
+put "00000000-0000-0000-0000-000000000001/TESTS/val1" "value" "HULY-TTL: 3"
+put "00000000-0000-0000-0000-000000000001/TESTS/val2" "value" "HULY-TTL: 12"
+put "00000000-0000-0000-0000-000000000001/TESTS/val3" "value" "HULY-TTL: 3"
+
+get "00000000-0000-0000-0000-000000000001/TESTS/"
+sleep 4
+# get "00000000-0000-0000-0000-000000000001/TESTS/val2"
+get "00000000-0000-0000-0000-000000000001/TESTS/"
+
+
+
+exit
+
+
+
+
+
+
+put "00000000-0000-0000-0000-000000000001/TESTS" "Value"
+
+#exit
+ delete "00000000-0000-0000-0000-000000000001/TESTS"
+put "00000000-0000-0000-0000-000000000001/TESTS" "Value"
+ delete "00000000-0000-0000-0000-000000000001/TESTS" "If-Match: *"
+put "00000000-0000-0000-0000-000000000001/TESTS" "Value"
+ delete "00000000-0000-0000-0000-000000000001/TESTS" "If-Match: dd358c74cb9cb897424838fbcb69c933"
+
+#exit
+
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/1" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/2" "Value_2" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/2/$/secret" "Value_secret" "HULY-TTL: 2"
+ get "00000000-0000-0000-0000-000000000001/TESTS/"
+
+#exit
+
+ delete "0000000/TESTS"
+ delete ${ZP}
+ put ${ZP} "Value_1" "HULY-TTL: 2"
+ delete ${ZP}
+
+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"
+
+
+
+echo "--------- if-match ----------"
+
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/1" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/2" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/3$" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/4" "Value_1" "HULY-TTL: 2"
+ get "00000000-0000-0000-0000-000000000001/TESTS"
+ get "00000000-0000-0000-0000-000000000001/TESTS/"
+ get "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/"
+
+
+echo "--------- Deprecated symbols ----------"
+
+ put "00000000-0000-0000-0000-000000000001/'TESTS" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TES?TS" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS*" "Value_1" "HULY-TTL: 2"
+ put "00000000-0000-0000-0000-000000000001/TESTS/" "Value_1" "HULY-TTL: 2"
+
+echo "--------- if-match ----------"
+
+ delete ${ZP}
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 1" "If-Match: *"
+ get ${ZP}
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value_2" "HULY-TTL: 1"
+ get ${ZP}
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value_3" "HULY-TTL: 1" "If-Match: dd358c74cb9cb897424838fbcb69c933"
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value_4" "HULY-TTL: 1" "If-Match: *"
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value_5" "HULY-TTL: 1" "If-Match: c7bcabf6b98a220f2f4888a18d01568d"
+ put "00000000-0000-0000-0000-000000000001/TESTS" "Value_6" "HULY-TTL: 1" "If-None-Match: *"
+
+echo "-- Expected OK: 201 Created (key was not exist)"
+
+ put ${ZP} "enother text" "If-None-Match" "*"
+
+ put ${ZP} "some text"
+ echo "-- Expected Error: 412 Precondition Failed (key was exist)"
+ put ${ZP} "enother text" "If-None-Match" "*"
+
+echo "================> UPDATE PUT If-Match"
+
+ get ${ZP}
+
+ echo "-- Expected OK: 204 No Content (right hash)"
+ put ${ZP} "some text" "If-Match" "552e21cd4cd9918678e3c1a0df491bc3"
+ get ${ZP}
+
+ echo "-- Expected OK: 204 No Content (hash still right)"
+ put ${ZP} "enother version" "If-Match" "552e21cd4cd9918678e3c1a0df491bc3"
+
+
+
+
+
+
+put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 3"
+echo "sleep 1 sec"
+sleep 1
+get "00000000-0000-0000-0000-000000000001/TESTS"
+echo "sleep 3 sec"
+sleep 2
+get "00000000-0000-0000-0000-000000000001/TESTS"
+
+echo "--------- delete ----------"
+put "00000000-0000-0000-0000-000000000001/TESTS" "Value_2" "HULY-TTL: 3"
+get "00000000-0000-0000-0000-000000000001/TESTS"
+delete "00000000-0000-0000-0000-000000000001/TESTS"
+get "00000000-0000-0000-0000-000000000001/TESTS"
+
+echo "--------- prefix ----------"
+put "00000000-0000-0000-0000-000000000001/TESTS1" "Value_1" "HULY-TTL: 3"
+put "00000000-0000-0000-0000-000000000001/TESTS2" "Value_1" "HULY-TTL: 3"
+put "00000000-0000-0000-0000-000000000001/HREST2" "Value_1" "HULY-TTL: 3"
+get "00000000-0000-0000-0000-000000000001?prefix=TES"
+sleep 1
+get "00000000-0000-0000-0000-000000000001?prefix="
+
+exit
diff --git a/scripts/pulse_lib_lleo.sh b/scripts/pulse_lib_lleo.sh
new file mode 100755
index 0000000000..4cf8832fca
--- /dev/null
+++ b/scripts/pulse_lib_lleo.sh
@@ -0,0 +1,79 @@
+#!/bin/bash
+
+clear
+
+# URL="http://localhost:8095/api"
+URL="https://hulypulse_mem.lleo.me/api"
+
+R='\033[0;31m' # Color red
+G='\033[0;32m' # Color green
+W='\033[0;33m' # Color ?
+S='\033[0;34m' # Color Blue
+F='\033[0;35m' # Color Fiolet
+L='\033[0;36m' # Color LightBlue
+N='\033[0m' # No Color
+GRAY='\033[90m' # bright black
+
+api() {
+ local tmpfile
+ tmpfile=$1
+ local status
+ status=$(head -n 1 "$tmpfile")
+ local status_code
+ status_code=$(echo "$status" | awk '{print $2}')
+ local etag
+ etag=$(grep -i "^ETag:" "${tmpfile}")
+ local body
+ body=$(awk 'found { print; next } NF == 0 { found = 1 }' "$tmpfile")
+ case "$status_code" in
+ 2*) echo -en "${G}${status}${N}" ;;
+ 3*) echo -en "${F}${status}${N}" ;;
+ 4*) echo -en "${R}${status}${N}" ;;
+ 5*) echo -en "${R}${status}${N}" ;;
+ *) echo -en "${GRAY}${status}${N}" ;;
+ esac
+ if [ -n "$etag" ]; then echo -n -e " ${F}${etag}${N}" ; fi
+
+ body=$(echo "$body" | sed 's/{/\\n{/g')
+
+ if [ -n "$body" ]; then echo -e "\n ${GRAY}[${body}]${N}" ; else echo -e " ${L}(no body)${N}" ; fi
+ rm -f "$tmpfile"
+}
+
+get() {
+ echo -n -e "📥 ${L}GET ${W}$1${N} > "
+ local tmpfile
+ tmpfile=$(mktemp)
+ curl -i -s -X GET "$URL/$1" -H "Authorization: Bearer ${TOKEN}" | tr -d '\r' > "$tmpfile"
+ api ${tmpfile}
+}
+
+put() { # If-None-Match If-Match
+ local match
+ local match_prn
+# if [ -n "$3" ]; then match=(-H "$3: $4") ; else match=() ; fi
+# if [ -n "$3" ]; then match_prn=" ${F}$3:$4${N}" ; else match_prn="" ; fi
+# echo -n -e "📥 ${L}PUT ${W}$1${N}${match_prn} > "
+
+ if [ -n "$3" ]; then match1=(-H "$3") ; else match1=() ; fi
+ if [ -n "$3" ]; then match1_prn=" ${F}$3${N}" ; else match1_prn="" ; fi
+ if [ -n "$4" ]; then match2=(-H "$4") ; else match2=() ; fi
+ if [ -n "$4" ]; then match2_prn=" ${F}$4${N}" ; else match2_prn="" ; fi
+ echo -n -e "📥 ${L}PUT ${W}$1${N}${match1_prn}${match2_prn} > "
+
+ local tmpfile
+ tmpfile=$(mktemp)
+# curl -v -i -s -X PUT "$URL/$1" "${match1[@]}" "${match2[@]}" -H "Content-Type: application/json" -d "$2" | tr -d '\r' > "$tmpfile"
+# curl -v -i -s -X PUT "$URL/$1" -H "Authorization: Bearer ${TOKEN}" "${match1[@]}" "${match2[@]}" -H "Content-Type: application/json" -d "$2" | tr -d '\r' > "$tmpfile"
+ curl -i -s -X PUT "$URL/$1" -H "Authorization: Bearer ${TOKEN}" "${match1[@]}" "${match2[@]}" -H "Content-Type: application/json" -d "$2" | tr -d '\r' > "$tmpfile"
+ api ${tmpfile}
+}
+
+delete() {
+ echo -n -e "📥 ${L}DELETE ${W}$1${N} > "
+ 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/config.rs b/src/config.rs
index dfe5037cba..feea94aa94 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -46,6 +46,9 @@ pub struct Config {
pub redis_service: String,
pub max_ttl: usize,
+
+ pub memory_mode: Option,
+ pub no_authorization: Option,
}
pub static CONFIG: LazyLock = LazyLock::new(|| {
diff --git a/src/config/default.toml b/src/config/default.toml
index a570fff0c7..b4ad885aed 100644
--- a/src/config/default.toml
+++ b/src/config/default.toml
@@ -9,3 +9,6 @@ redis_mode = "direct"
redis_service = "mymaster"
max_ttl = 3600
+
+memory_mode = true
+no_authorization = true
\ No newline at end of file
diff --git a/src/db.rs b/src/db.rs
new file mode 100644
index 0000000000..71f84a2bc4
--- /dev/null
+++ b/src/db.rs
@@ -0,0 +1,101 @@
+use std::sync::Arc;
+
+use crate::{
+ hub_service::{broadcast_event, HubState, RedisEvent, RedisEventAction},
+ memory::{memory_delete, memory_list, memory_read, memory_save, MemoryBackend},
+ redis::{redis_delete, redis_list, redis_read, redis_save, RedisArray, SaveMode, Ttl}
+};
+use ::redis::aio::MultiplexedConnection;
+use tokio::sync::RwLock;
+
+#[derive(Clone)]
+pub struct Db {
+ inner: DbInner,
+ hub: Arc>,
+}
+
+#[derive(Clone)]
+enum DbInner {
+ Memory(MemoryBackend),
+ Redis(MultiplexedConnection),
+}
+
+impl Db {
+
+ pub fn new_memory(m: MemoryBackend, hub: Arc>) -> Self {
+ Self { inner: DbInner::Memory(m), hub }
+ }
+ pub fn new_redis(c: MultiplexedConnection, hub: Arc>) -> Self {
+ Self { inner: DbInner::Redis(c), hub }
+ }
+
+
+ pub async fn list(&self, key: &str) -> redis::RedisResult> {
+ match &self.inner {
+ DbInner::Memory(m) => memory_list(m, key).await,
+ DbInner::Redis(conn) => {
+ let mut c = conn.clone();
+ redis_list(&mut c, key).await
+ }
+ }
+ }
+
+ pub async fn read(&self, key: &str) -> redis::RedisResult> {
+ match &self.inner {
+ DbInner::Memory(m) => memory_read(m, key).await,
+ DbInner::Redis(conn) => {
+ let mut c = conn.clone();
+ redis_read(&mut c, key).await
+ }
+ }
+ }
+
+
+ pub async fn save>(
+ &self,
+ key: &str,
+ value: V,
+ ttl: Option,
+ mode: Option,
+ ) -> redis::RedisResult<()> {
+ match &self.inner {
+ DbInner::Memory(m) => {
+ memory_save(m, key, value.as_ref(), ttl, mode).await?;
+ // Send events
+ let value_str = std::str::from_utf8(value.as_ref()).ok().map(|s| s.to_string());
+ broadcast_event(
+ &self.hub,
+ RedisEvent { message: RedisEventAction::Set, key: key.to_string() },
+ value_str,
+ ).await;
+ Ok(())
+ }
+ DbInner::Redis(conn) => {
+ let mut c = conn.clone();
+ redis_save(&mut c, key, value.as_ref(), ttl, mode).await
+ }
+ }
+ }
+
+
+ pub async fn delete(&self, key: &str, mode: Option) -> redis::RedisResult {
+ match &self.inner {
+ DbInner::Memory(m) => {
+ let deleted = memory_delete(m, key, mode).await?;
+ if deleted {
+ broadcast_event(
+ &self.hub,
+ RedisEvent { message: RedisEventAction::Del, key: key.to_string() },
+ None,
+ ).await;
+ }
+ Ok(deleted)
+ }
+ DbInner::Redis(conn) => {
+ let mut c = conn.clone();
+ redis_delete(&mut c, key, mode).await
+ }
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/src/handlers_http.rs b/src/handlers_http.rs
index 74766e218e..a577709184 100644
--- a/src/handlers_http.rs
+++ b/src/handlers_http.rs
@@ -14,7 +14,6 @@
//
use anyhow::anyhow;
-use redis::aio::MultiplexedConnection;
use serde::Deserialize;
use tracing::*;
@@ -23,7 +22,10 @@ use actix_web::{
web::{self},
};
-use crate::redis::{SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save};
+use crate::{
+ db::Db,
+ redis::{SaveMode, Ttl}
+};
pub fn map_handler_error(err: impl std::fmt::Display) -> Error {
let msg = err.to_string();
@@ -51,17 +53,14 @@ pub struct PathParams {
/// list
pub async fn list(
path: web::Path,
- redis: web::Data,
+ db: web::Data,
) -> Result {
let key = path.into_inner().key;
trace!(key, "list request");
async move || -> anyhow::Result {
- let mut redis = redis.get_ref().clone();
-
- let entries = redis_list(&mut redis, &key).await?;
-
+ let entries = db.list(&key).await?;
Ok(HttpResponse::Ok().json(entries))
}()
.await
@@ -71,23 +70,21 @@ pub async fn list(
/// get
pub async fn get(
path: web::Path,
- redis: web::Data,
+ db: web::Data,
) -> Result {
let key = path.into_inner().key;
trace!(key, "get request");
async move || -> anyhow::Result {
- let mut redis = redis.get_ref().clone();
-
- Ok(redis_read(&mut redis, &key)
- .await?
- .map(|entry| {
- HttpResponse::Ok()
- .insert_header(("ETag", &*entry.etag))
- .json(entry)
- })
- .unwrap_or_else(|| HttpResponse::NotFound().body("empty")))
+ let entry_opt = db.read(&key).await?;
+ let resp = match entry_opt {
+ Some(entry) => HttpResponse::Ok()
+ .insert_header(("ETag", entry.etag.clone()))
+ .json(entry),
+ None => HttpResponse::NotFound().body("empty"),
+ };
+ Ok(resp)
}()
.await
.map_err(map_handler_error)
@@ -98,14 +95,13 @@ pub async fn put(
req: HttpRequest,
path: web::Path,
body: web::Bytes,
- redis: web::Data,
+ db: web::Data,
) -> Result {
let key: String = path.into_inner().key;
trace!(key, "put request");
async move || -> anyhow::Result {
- let mut redis = redis.get_ref().clone();
// TTL logic
let mut ttl = None;
@@ -149,8 +145,8 @@ pub async fn put(
}
}
- redis_save(&mut redis, &key, &body[..], ttl, mode).await?;
- return Ok(HttpResponse::Ok().body("DONE"));
+ db.save(&key, &body[..], ttl, mode).await?;
+ Ok(HttpResponse::Ok().body("DONE"))
}()
.await
.map_err(map_handler_error)
@@ -160,14 +156,13 @@ pub async fn put(
pub async fn delete(
req: HttpRequest,
path: web::Path,
- redis: web::Data,
+ db: web::Data,
) -> Result {
let key: String = path.into_inner().key;
trace!(key, "delete request");
async move || -> anyhow::Result {
- let mut redis = redis.get_ref().clone();
// MODE logic
let mut mode = Some(SaveMode::Upsert);
@@ -183,8 +178,7 @@ pub async fn delete(
} // `If-Match: ` — delete only if current
}
- let deleted = redis_delete(&mut redis, &key, mode).await?;
-
+ let deleted = db.delete(&key, mode).await?;
let response = match deleted {
true => HttpResponse::NoContent().finish(),
false => HttpResponse::NotFound().body("not found"),
diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs
index 97385c5e7f..26ff558fe0 100644
--- a/src/handlers_ws.rs
+++ b/src/handlers_ws.rs
@@ -19,15 +19,12 @@ use tokio::sync::RwLock;
use actix::{Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut};
use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse, web};
use actix_web_actors::ws;
-use redis::aio::MultiplexedConnection;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
-use crate::redis::{ SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save };
-use crate::hub_service::{ServerMessage, SessionId, new_session_id};
-use crate::workspace_owner::check_workspace_core;
-use crate::hub_service::HubState;
-
+use crate::{
+ config::CONFIG, db::Db, hub_service::{new_session_id, HubState, ServerMessage, SessionId}, redis::{ SaveMode, Ttl }, workspace_owner::check_workspace_core
+};
#[derive(Serialize, Default)]
struct ReturnBase<'a> {
@@ -125,10 +122,10 @@ use hulyrs::services::jwt::Claims;
/// Session condition
pub struct WsSession {
- pub redis: MultiplexedConnection,
+ pub db: Db,
pub id: SessionId,
hub_state: Arc>,
- pub claims: Claims,
+ pub claims: Option,
}
/// Actor External trait: must be in separate impl block
@@ -240,12 +237,12 @@ impl WsSession {
..Default::default()
});
- let mut redis = self.redis.clone();
let claims = self.claims.clone();
+ let db = self.db.clone();
let fut = async move {
// Check workspace
- if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+ if let Err(e) = check_workspace_core(claims, &key) { return Err(e.into()); }
// TTL logic
let real_ttl = if let Some(secs) = ttl {
@@ -276,7 +273,8 @@ impl WsSession {
}
}
- redis_save(&mut redis, &key, &data, real_ttl, mode)
+ // Save
+ db.save(&key, &data, real_ttl, mode)
.await
.map_err(|e| e.to_string())?;
@@ -293,7 +291,7 @@ impl WsSession {
} => {
tracing::info!("DELETE {}", &key); // correlation:{:?} , &correlation
- let mut redis = self.redis.clone();
+ let db = self.db.clone();
let claims = self.claims.clone();
let base = serde_json::json!(ReturnBase {
@@ -306,7 +304,7 @@ impl WsSession {
let fut = async move {
// Check workspace
- if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+ if let Err(e) = check_workspace_core(claims, &key) { return Err(e.into()); }
// MODE logic
let mut mode = Some(SaveMode::Upsert);
@@ -320,7 +318,8 @@ impl WsSession {
}
}
- let deleted = redis_delete(&mut redis, &key, mode)
+ // Delete
+ let deleted = db.delete(&key, mode)
.await
.map_err(|e| e.to_string())?;
@@ -344,21 +343,18 @@ impl WsSession {
..Default::default()
});
- let mut redis = self.redis.clone();
+ let db = self.db.clone();
let claims = self.claims.clone();
let fut = async move {
// Check workspace
- if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
-
- let data_opt = redis_read(&mut redis, &key)
- .await
- .map_err(|e| e.to_string())?;
+ if let Err(e) = check_workspace_core(claims, &key) { return Err(e.into()); }
+ // Read
+ let data_opt = db.read(&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())?;
+ let data_value = serde_json::to_value(&data).map_err(|e| e.to_string())?;
Ok(json!({"result": data_value}))
}
None => Err("not found".into()),
@@ -378,16 +374,14 @@ impl WsSession {
..Default::default()
});
- let mut redis = self.redis.clone();
+ let db = self.db.clone();
let claims = self.claims.clone();
let fut = async move {
// Check workspace
- if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
-
- let data = redis_list(&mut redis, &key)
- .await
- .map_err(|e| e.to_string())?;
+ if let Err(e) = check_workspace_core(claims, &key) { return Err(e.into()); }
+ // List
+ let data = db.list(&key).await.map_err(|e| e.to_string())?;
Ok(json!({ "result": data }))
};
@@ -411,7 +405,7 @@ impl WsSession {
let fut = async move {
// Check workspace
- if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+ if let Err(e) = check_workspace_core(claims, &key) { return Err(e.into()); }
hub_state.write().await.subscribe(id, key);
Ok(json!({ "result": "OK" }))
@@ -441,7 +435,7 @@ impl WsSession {
Ok(json!({ "result": "OK" }))
} else {
// Check workspace
- if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+ if let Err(e) = check_workspace_core(claims, &key) { return Err(e.into()); }
hub_state.write().await.unsubscribe(id, key);
Ok(json!({ "result": "OK" }))
@@ -476,21 +470,21 @@ impl WsSession {
pub async fn handler(
req: HttpRequest,
payload: web::Payload,
- redis: web::Data,
+ db: web::Data,
hub_state: web::Data>>,
) -> Result {
- let claims = req
- .extensions()
- .get::()
- .expect("Missing claims")
- .to_owned();
+
+ let claims = if CONFIG.no_authorization == Some(true) {
+ None
+ } else {
+ Some(req.extensions().get::().expect("Missing claims").to_owned())
+ };
let session = WsSession {
- redis: redis.get_ref().clone(),
- // hub: hub.get_ref().clone(),
+ db: db.get_ref().clone(),
hub_state: hub_state.get_ref().clone(),
id: new_session_id(),
- claims,
+ claims: claims,
};
ws::start(session, &req, payload)
diff --git a/src/hub_service.rs b/src/hub_service.rs
index dc07686484..f44b9a0511 100644
--- a/src/hub_service.rs
+++ b/src/hub_service.rs
@@ -43,16 +43,12 @@ pub struct ServerMessage {
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,
@@ -69,8 +65,6 @@ pub struct RedisEvent {
pub key: String,
}
-// ==== Handle ====
-
#[derive(Debug, Default)]
pub struct HubState {
sessions: HashMap>,
@@ -134,14 +128,13 @@ impl HubState {
}
-
-// Send messages about new Redis events
-pub async fn push_event(
+// Send messages about new db events
+pub async fn broadcast_event(
hub_state: &Arc>,
- redis: &mut MultiplexedConnection,
ev: RedisEvent,
+ value: Option,
) {
- // Collect Addresses
+ // Collect
let recipients: Vec> = {
hub_state.read().await.recipients_for_key(&ev.key)
};
@@ -149,10 +142,22 @@ pub async fn push_event(
return;
}
- // Get value from Redis (only for `Set` event, not for `Delete`, `Expire`)
+ // Send
+ let payload = ServerMessage { event: ev, value };
+ for rcpt in recipients {
+ let _ = rcpt.do_send(payload.clone());
+ }
+}
+
+pub async fn push_event(
+ hub_state: &Arc>,
+ redis: &mut MultiplexedConnection,
+ ev: RedisEvent,
+) {
+ // Value only for Set
let mut value: Option = None;
if matches!(ev.message, RedisEventAction::Set) {
- match redis::cmd("GET")
+ match ::redis::cmd("GET")
.arg(&ev.key)
.query_async::>(redis)
.await
@@ -162,9 +167,5 @@ pub async fn push_event(
}
}
- // Sending
- let payload = ServerMessage { event: ev, value };
- for rcpt in recipients {
- let _ = rcpt.do_send(payload.clone());
- }
-}
+ broadcast_event(hub_state, ev, value).await;
+}
\ No newline at end of file
diff --git a/src/main.rs b/src/main.rs
index 6a1ed21a39..6ab19514df 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -38,6 +38,12 @@ use hub_service::{HubState};
use config::CONFIG;
+mod db;
+mod memory;
+
+use crate::db::Db;
+use crate::memory::MemoryBackend;
+
fn initialize_tracing(level: tracing::Level) {
use tracing_subscriber::{filter::targets::Targets, prelude::*};
@@ -61,6 +67,10 @@ async fn extract_claims(
token: Option,
}
+ if CONFIG.no_authorization == Some(true) {
+ return next.call(request).await;
+ }
+
let query = request.extract::>().await?.into_inner();
let claims = if let Some(token) = query.token {
@@ -77,6 +87,11 @@ async fn check_workspace(
mut request: ServiceRequest,
next: Next,
) -> Result, Error> {
+
+ if CONFIG.no_authorization.unwrap_or(false) {
+ return next.call(request).await;
+ }
+
let workspace = Uuid::parse_str(&request.extract::>().await?);
let claims = request.extensions().get::().cloned().unwrap();
@@ -98,14 +113,21 @@ async fn main() -> anyhow::Result<()> {
tracing::info!("{}/{}", env!("CARGO_BIN_NAME"), env!("CARGO_PKG_VERSION"));
- let redis_client = redis::client().await?;
- let redis_connection = redis_client.get_multiplexed_async_connection().await?;
-
// starting HubService
let hub_state = Arc::new(RwLock::new(HubState::default()));
- // starting Logger
- tokio::spawn(redis::receiver(redis_client, hub_state.clone()));
+ let db_backend = if CONFIG.memory_mode == Some(true) {
+ let memory = MemoryBackend::new();
+ memory.spawn_ticker(hub_state.clone());
+ tracing::info!("Memory mode enabled");
+ Db::new_memory(memory, hub_state.clone())
+ } else {
+ let redis_client = redis::client().await?;
+ let redis_connection = redis_client.get_multiplexed_async_connection().await?;
+ tokio::spawn(crate::redis::receiver(redis_client, hub_state.clone()));
+ tracing::info!("Redis mode enabled");
+ Db::new_redis(redis_connection, hub_state.clone())
+ };
let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port);
@@ -127,8 +149,7 @@ async fn main() -> anyhow::Result<()> {
.max_age(3600);
App::new()
- .app_data(web::Data::new(redis_connection.clone()))
- //.app_data(web::Data::new(hub.clone()))
+ .app_data(web::Data::new(db_backend.clone()))
.app_data(web::Data::new(hub_state.clone()))
.wrap(middleware::Logger::default())
.wrap(cors)
diff --git a/src/memory.rs b/src/memory.rs
new file mode 100644
index 0000000000..f375de329b
--- /dev/null
+++ b/src/memory.rs
@@ -0,0 +1,277 @@
+
+//
+// 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, sync::Arc, time::{SystemTime, UNIX_EPOCH}};
+use tokio::{sync::RwLock, time::{self, Duration}};
+use crate::{
+ config::CONFIG,
+ hub_service::{broadcast_event, HubState, RedisEvent, RedisEventAction},
+ redis::{deprecated_symbol_error, error, RedisArray, SaveMode, Ttl},
+};
+
+#[derive(Debug, Clone)]
+struct Entry {
+ data: String,
+ ttl: u8,
+}
+
+#[derive(Clone, Default)]
+pub struct MemoryBackend {
+ inner: Arc>>,
+}
+
+impl MemoryBackend {
+
+ pub fn new() -> Self {
+ Self { inner: Arc::new(RwLock::new(HashMap::new())) }
+ }
+
+ pub fn spawn_ticker(&self, hub_state: Arc>) {
+ let inner = self.inner.clone();
+ tokio::spawn(async move {
+ let mut ticker = time::interval(Duration::from_secs(1));
+ loop {
+ ticker.tick().await;
+
+ // lock
+ let expired_keys: Vec = {
+ let mut map = inner.write().await;
+
+ let mut expired: Vec = Vec::new();
+ for (k, v) in map.iter_mut() {
+ if v.ttl > 0 {
+ v.ttl = v.ttl.saturating_sub(1);
+ if v.ttl == 0 {
+ expired.push(k.clone());
+ }
+ } else {
+ expired.push(k.clone());
+ }
+ }
+
+ for k in &expired {
+ map.remove(k);
+ }
+
+ expired
+ }; // write-lock free
+
+ for k in expired_keys {
+ broadcast_event(&hub_state,RedisEvent { message: RedisEventAction::Expired, key: k },None).await;
+ }
+ }
+ });
+ }
+}
+
+
+/// memory_list(&backend, "prefix/") → Vec
+pub async fn memory_list(
+ backend: &MemoryBackend,
+ key_prefix: &str,
+) -> redis::RedisResult> {
+ deprecated_symbol_error(key_prefix)?;
+ if !key_prefix.ends_with('/') {
+ return error(412, "Key must end with slash");
+ }
+
+ let map = backend.inner.read().await;
+
+ let mut results = Vec::new();
+ for (k, v) in map.iter() {
+ if !k.starts_with(key_prefix) {
+ continue;
+ }
+
+ if k.strip_prefix(key_prefix).map_or(false, |s| s.contains('$')) {
+ continue;
+ }
+
+ if v.ttl == 0 {
+ continue;
+ }
+
+ results.push(RedisArray {
+ key: k.clone(),
+ data: v.data.clone(),
+ expires_at: v.ttl as u64,
+ etag: hex::encode(md5::compute(&v.data).0),
+ });
+ }
+
+ Ok(results)
+}
+
+/// memory_read(&backend, "key")
+pub async fn memory_read(
+ backend: &MemoryBackend,
+ key: &str,
+) -> redis::RedisResult> {
+ deprecated_symbol_error(key)?;
+ if key.ends_with('/') {
+ return error(412, "Key must not end with a slash");
+ }
+
+ let map = backend.inner.read().await;
+
+ match map.get(key) {
+ None => Ok(None),
+ Some(entry) => {
+ let data = entry.data.clone();
+ let ttl = entry.ttl as u64;
+
+ Ok(Some(RedisArray {
+ key: key.to_string(),
+ data: data.clone(),
+ expires_at: ttl,
+ etag: hex::encode(md5::compute(&data).0),
+ }))
+ }
+ }
+}
+
+/// TTL in sec
+fn compute_ttl_u8(ttl: Option) -> redis::RedisResult {
+
+ let sec_usize = 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,
+ };
+
+ if sec_usize == 0 {
+ return error(400, "TTL must be > 0");
+ }
+
+ if sec_usize > CONFIG.max_ttl {
+ return error(412, "TTL exceeds MAX_TTL");
+ }
+
+ let capped = sec_usize.min(u8::MAX as usize);
+ Ok(capped as u8)
+}
+
+/// memory_save(&backend, "key", value, ttl, mode)
+pub async fn memory_save>(
+ backend: &MemoryBackend,
+ key: &str,
+ bytes_value: V,
+ ttl: Option,
+ mode: Option,
+) -> redis::RedisResult<()> {
+
+ // u8 - String
+ let value = match std::str::from_utf8(bytes_value.as_ref()) {
+ Ok(s) => s.to_string(),
+ Err(_) => return error(400, "Value must be valid UTF-8"),
+ };
+
+ deprecated_symbol_error(key)?;
+ if key.ends_with('/') {
+ return error(412, "Key must not end with a slash");
+ }
+
+ let sec_u8 = compute_ttl_u8(ttl)?;
+ let val = value.to_string();
+
+ let mut map = backend.inner.write().await;
+
+ let mode = mode.unwrap_or(SaveMode::Upsert);
+
+ match mode {
+ SaveMode::Upsert => {
+ map.insert(key.to_string(), Entry { data: val, ttl: sec_u8 });
+ }
+ SaveMode::Insert => {
+ if map.contains_key(key) {
+ return error(412, "Insert: key already exists");
+ }
+ map.insert(key.to_string(), Entry { data: val, ttl: sec_u8 });
+ }
+ SaveMode::Update => {
+ let Some(existing) = map.get_mut(key) else {
+ return error(404, "Update: key does not exist");
+ };
+ *existing = Entry { data: val, ttl: sec_u8 };
+ }
+ SaveMode::Equal(ref expected_md5) => {
+ let Some(existing) = map.get_mut(key) else {
+ return error(404, "Equal: key does not exist");
+ };
+ let actual_md5 = hex::encode(md5::compute(&existing.data).0);
+ if &actual_md5 != expected_md5 {
+ return error(
+ 412,
+ format!("md5 mismatch, current: {}, expected: {}", actual_md5, expected_md5),
+ );
+ }
+ *existing = Entry { data: val, ttl: sec_u8 };
+ }
+ }
+
+ Ok(())
+}
+
+/// memory_delete(&backend, "key", mode)
+pub async fn memory_delete(
+ backend: &MemoryBackend,
+ key: &str,
+ mode: Option,
+) -> redis::RedisResult {
+ deprecated_symbol_error(key)?;
+ if key.ends_with('/') {
+ return error(412, "Key must not end with a slash");
+ }
+
+ let mut map = backend.inner.write().await;
+ let mode = mode.unwrap_or(SaveMode::Upsert);
+
+ match mode {
+ SaveMode::Insert => {
+ return error(412, "Insert mode is not supported for delete");
+ }
+ SaveMode::Update | SaveMode::Upsert => {
+ let existed = map.remove(key).is_some();
+ Ok(existed)
+ }
+ SaveMode::Equal(ref expected_md5) => {
+ match map.get(key) {
+ None => return error(404, "Equal: key does not exist"),
+ Some(existing) => {
+ let actual_md5 = hex::encode(md5::compute(&existing.data).0);
+ if &actual_md5 != expected_md5 {
+ return error(
+ 412,
+ format!("md5 mismatch, current: {}, expected: {}", actual_md5, expected_md5),
+ );
+ }
+ }
+ }
+ let existed = map.remove(key).is_some();
+ if !existed {
+ // WHF?!
+ return error(404, "Delete: key does not exist");
+ }
+ Ok(true)
+ }
+ }
+}
diff --git a/src/workspace_owner.rs b/src/workspace_owner.rs
index 2014953921..42423db154 100644
--- a/src/workspace_owner.rs
+++ b/src/workspace_owner.rs
@@ -16,15 +16,21 @@
use hulyrs::services::jwt::Claims;
use uuid::Uuid;
-use crate::redis::deprecated_symbol;
+use crate::{config::CONFIG, redis::deprecated_symbol};
// common checker
-pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> {
+pub fn check_workspace_core(claims_opt: Option, key: &str) -> Result<(), &'static str> {
if deprecated_symbol(key) {
return Err("Invalid key: deprecated symbols");
}
+ if CONFIG.no_authorization == Some(true) {
+ return Ok(());
+ }
+
+ let claims = claims_opt.ok_or("Missing authorization")?;
+
if claims.is_system() {
return Ok(());
}
@@ -40,11 +46,12 @@ pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static s
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(())
+
}