From 4c2a468689918e0aa3a10a841d48cb285e771bf8 Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Fri, 22 Aug 2025 12:08:53 +0300 Subject: [PATCH] Readme updated --- Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 173 +++++++++++++++++++++++---------------- scripts/TEST_HTTP_API.sh | 20 ++++- src/handlers_ws.rs | 44 +++++----- src/hub_service.rs | 32 ++++---- src/main.rs | 27 +++--- src/redis.rs | 14 ++-- 8 files changed, 179 insertions(+), 135 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a79b93081e..ac96a96a2e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1181,7 +1181,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.1.7" +version = "0.1.8" dependencies = [ "actix", "actix-cors", diff --git a/Cargo.toml b/Cargo.toml index 8bb882c4bc..679318aeab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.1.7" +version = "0.1.8" edition = "2024" [dependencies] diff --git a/README.md b/README.md index 2132c06479..76edec749e 100644 --- a/README.md +++ b/README.md @@ -24,8 +24,8 @@ Key segment may be private (prefixed with ‘$’) May not contain special characters (‘*’, ‘?’, ‘[’, ‘]’,‘\’,‘\x00..\xF1’,‘\x7F’,‘"’,‘'’) 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 +- GET/SUBSCRIBE/.. a/b → single key +- GET/SUBSCRIBE/.. a/b/c/ → multiple If multiple @@ -33,47 +33,34 @@ 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] + +- 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] ## Data “Data” is an arbitrary JSON document. Size of data is limited to some reasonable size -## API -Methods - -GET - returns values of one key - -LIST - returns values with given prefix until the “sentinel” - -PUT - put value to the key -- Support CAS -- Support If-* headers - -DELETE - delete value of the key - -SUB - subscribe to key data + get initial state -Behavior identical to LIST - -UNSUB - unsubscribe to key data - - ## HTTP API -```PUT /{workspace}/{key}``` +```GET /status``` - server status and websockets count +- Answer: {"status":"OK","websockets":2} + + +```PUT /{workspace}/{key}``` - Save key - Input Body - data - Content-Type: application/json (do we need something else?) + Content-Type: application/json Content-Length: optional Headers: TTL or absolute expiration time - HULY-TTL - HULY-EXPIRE-AT + - `HULY-TTL` — autodelete in N seconds + - or `HULY-EXPIRE-AT` — autodelete in UnixTime + - default max_ttl = 3600 (settings in config/default.toml) ** Conditional Headers If-*: ** - `If-Match: *` — update only if the key exists - `If-Match: ` — update only if current value's MD5 matches @@ -84,73 +71,118 @@ UNSUB - unsubscribe to key data - `204` on successful insert or update - `412` if the condition is not met - `400` if headers are invalid - - No body + - Body: `DONE` -```PATCH /{workspace}/{key}``` -- TODO (not in v1) - -```DELETE /{workspace}/{key}``` +```DELETE /{workspace}/{key}``` - Delete key - Output - Status: 204 + - Status: `204 No content`, no body + - `404 Not Found` if nothing to do -```GET /{workspace}/{key}``` +```GET /{workspace}/{key}``` - Read one key - Output - Status 200 - Content-type: application/json + - Header: `Etag: ` - Body: - - workspace - - key - - data - - expiresAt ? + - workspace (copy of input) + - key (copy of input) + - data (copy of input) + - expiresAt / TTL (copy of input, optional) + - etag -```GET /{workspace}?prefix={key}``` +```GET /{workspace}/{key}/``` - Read array of keys - Output - Status 200 - Content-type: application/json - Body (array): - - workspace - - key - - data - - expiresAt ? + - [{"key","data","expires_at","etag"}, ...] ## WebSocket API **Client to Server** ```PUT``` - - correlation id (optional) - type: "put" - - key: “foo/bar“ + - correlation id (optional) + - key: + - “workspace/foo/bar“ - shared key + - “workspace/foo/bar/$/secret“ - secret key - data - - TTL / expiresAt + ** time control (optional) ** + - `TTL` — autodelete in N seconds + - `ExpireAt` — autodelete in UnixTime + - or default max_ttl = 3600 (settings in config/default.toml) + ** Conditional (optional) ** + - `ifMatch: *` — update only if the key exists + - `ifMatch: ` — update only if current value's MD5 matches + - `ifNoneMatch: *` — insert only if the key does not exist + +- Answer: {"action":"put","correlation":"abc123","result":"OK"} + + +```GET``` + - type: "get" + - correlation id (optional) + - key: + - “workspace/foo/bar“ - one shared key + - “workspace/foo/bar/$/secret“ - one secret key + +- Answer: {"action":"get","result":{"data":"hello","etag":"5d41402abc4b2a76b9719d911017c592","expires_at":3599,"key":"00000000-0000-0000-0000-000000000001/foo/bar"}} + + +```LIST``` + - type: "list" + - correlation id (optional) + - key: + - “workspace/foo/bar/“ - keys from public space + - “workspace/foo/bar/$/secret/“ - keys from secret space + +- Answer: {"action":"list","result":[{"data":"hello 1","etag":"df0649bc4f1be901c85b6183091c1d83","expires_at":3570,"key":"00000000-0000-0000-0000-000000000001/foo/bar1"},{"data":"hello 2","etag":"bb21ec8394b75795622f61613a777a8b","expires_at":3555,"key":"00000000-0000-0000-0000-000000000001/foo/bar2"}]} + ```DELETE``` - - correlation id (optional) - type: "delete" - - key: “foo/bar“ + - correlation id (optional) + - key: “workspace/foo/bar“ + ** Conditional (optional) ** + - `ifMatch: ` — delete only if current value's MD5 matches + - `ifMatch: *` — return error if key does not exist -```SUB``` +- Answer: {"action":"delete","result":"OK"} + + +```SUBSCRIBE``` type: "sub" - key: “foo/bar“ + key: + - “workspace/foo/bar“ - subscribe one shared key + - “workspace/foo/bar/“ - subscribe all keys started with + - “workspace/foo/bar/$/my_secret“ - subscribe one secret key + - “workspace/foo/bar/$/my_secret/“ - subscribe all keys started with secret -```UNSUB``` +- Answer: {"action":"sub","result":"OK"} + + +```UNSUBSCRIBE``` - type: "unsub" - - key: “foo/bar“ + - key: + - “workspace/foo/bar“ - unsubscribe subscribed key + - “*“ - unsubscribe all -**Server to Client** +- Answer: {"action":"unsub","result":"OK"} -```PUT``` - - correlation id (optional) - - type: "put" - - ?? TODO: user? workspace: "11111111-2222-3333-4444-555555555555" - - key: “foo/bar“ - - data - - expiresAt +```MY SUBSCRIBES``` + - type: "sublist" -```DELETE``` - - correlation id (optional) - - type: "delete" - - key: “foo/bar“ +- Answer: {"action":"list","result":["00000000-0000-0000-0000-000000000001/foo/bar1","00000000-0000-0000-0000-000000000001/foo/bar2"]} + + +** Server to Client ** subscribed events: + + - {"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Set","value":"hello"} + + - {"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Expired"} + + - {"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Del"} ## Running @@ -179,22 +211,19 @@ Hulypulse uses bearer JWT token authetication. At the moment, it will accept any The following environment variables are used to configure hulypulse: - ```HULY_BIND_HOST```: host to bind the server to (default: 0.0.0.0) - ```HULY_BIND_PORT```: port to bind the server to (default: 8094) - - ```HULY_PAYLOAD_SIZE_LIMIT```: maximum size of the payload (default: 2Mb) - ```HULY_TOKEN_SECRET```: secret used to sign JWT tokens (default: secret) - ```HULY_REDIS_URLS```: redis connection string (default: redis://huly.local:6379) - ```HULY_REDIS_PASSWORD```: redis password (default: "<invalid>") - ```HULY_REDIS_MODE```: redis mode "direct" or "sentinel" (default: "direct") - ```HULY_REDIS_SERVICE```: redis service (default: "mymaster") - ```HULY_MAX_TTL```: maximum storage time (default: 3600) + - TODO: ```HULY_PAYLOAD_SIZE_LIMIT```: maximum size of the payload (default: 2Mb) ## Todo (in no particular order) - [ ] Optional value encryption -- [ ] HEAD request -- [ ] Conditional update (optimistic locking) - [ ] Support for open telemetry - [ ] Concurrency control for database migration (several instances of hulypulse are updated at the same time) - [ ] TLS support -- [ ] Namespacee based access control - [ ] Liveness/readiness probe endpoint ## Contributing diff --git a/scripts/TEST_HTTP_API.sh b/scripts/TEST_HTTP_API.sh index 9ce7ee896d..e0fc3918eb 100755 --- a/scripts/TEST_HTTP_API.sh +++ b/scripts/TEST_HTTP_API.sh @@ -10,12 +10,30 @@ TOKEN=$(./token.sh claims.json) ZP="00000000-0000-0000-0000-000000000001/TESTS" +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" diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 154a838880..6eff578291 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -13,8 +13,6 @@ // limitations under the License. // -// https://github.com/hcengineering/hulypulse/ - use actix::{ Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut, }; @@ -48,14 +46,14 @@ struct ReturnBase<'a> { #[serde(rename = "TTL", skip_serializing_if = "Option::is_none")] ttl: Option, - #[serde(rename = "expiresAt", skip_serializing_if = "Option::is_none")] - expires_at: Option, + // #[serde(rename = "expiresAt", skip_serializing_if = "Option::is_none")] + // expires_at: Option, - #[serde(rename = "ifMatch", skip_serializing_if = "Option::is_none")] - if_match: Option<&'a str>, + // #[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>, + // #[serde(rename = "ifNoneMatch", skip_serializing_if = "Option::is_none")] + // if_none_match: Option<&'a str>, } /// WsCommand - commands enum (put, delete, sub, unsub) @@ -242,13 +240,14 @@ impl WsSession { let base = serde_json::json!(ReturnBase { action: "put", - key: Some(key.as_str()), - data: Some(data.as_str()), + // 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(), + // ttl, + // expires_at, + // if_match: if_match.as_deref(), + // if_none_match: if_none_match.as_deref(), + ..Default::default() }); let fut = async move { @@ -304,15 +303,13 @@ impl WsSession { return; } - tracing::info!("DELETE!!! {}", &key); - let mut redis = self.redis.clone(); let base = serde_json::json!(ReturnBase { action: "delete", - key: Some(key.as_str()), + // key: Some(key.as_str()), correlation: correlation.as_deref(), - if_match: if_match.as_deref(), + // if_match: if_match.as_deref(), ..Default::default() }); @@ -320,12 +317,11 @@ impl WsSession { // 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: ` — update only if current + // `If-Match: ` — delete only if current mode = Some(SaveMode::Equal(s.to_string())); } } @@ -357,7 +353,7 @@ impl WsSession { let base = serde_json::json!(ReturnBase { action: "get", - key: Some(key.as_str()), + // key: Some(key.as_str()), correlation: correlation.as_deref(), ..Default::default() }); @@ -393,7 +389,7 @@ impl WsSession { let base = serde_json::json!(ReturnBase { action: "list", - key: Some(key.as_str()), + // key: Some(key.as_str()), correlation: correlation.as_deref(), ..Default::default() }); @@ -421,7 +417,7 @@ impl WsSession { let mut obj = serde_json::json!(ReturnBase { action: "sub", - key: Some(key.as_str()), + // key: Some(key.as_str()), correlation: correlation.as_deref(), ..Default::default() }); @@ -444,7 +440,7 @@ impl WsSession { let mut obj = serde_json::json!(ReturnBase { action: "unsub", - key: Some(key.as_str()), + // key: Some(key.as_str()), correlation: correlation.as_deref(), ..Default::default() }); diff --git a/src/hub_service.rs b/src/hub_service.rs index 2e79a468c4..39e0b71fff 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -48,7 +48,7 @@ pub enum RedisEventAction { #[derive(Debug, Clone, Serialize)] pub struct RedisEvent { - pub db: u32, +// pub db: u32, pub key: String, pub action: RedisEventAction, } @@ -81,9 +81,9 @@ enum Command { Count { reply: oneshot::Sender, }, - DumpSubs { - reply: oneshot::Sender>>, - }, + // DumpSubs { + // reply: oneshot::Sender>>, + // }, RedisEvent(RedisEvent), } @@ -170,13 +170,13 @@ impl HubServiceHandle { 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::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); @@ -254,11 +254,11 @@ impl HubServiceHandle { 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 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 7067a6358c..13b8c9f204 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,6 +13,8 @@ // limitations under the License. // +// https://github.com/hcengineering/hulypulse/ + use actix_cors::Cors; use actix_web::{ body::MessageBody, dev::{ServiceRequest, ServiceResponse}, middleware::{self, Next}, web::{self, Path, Query}, App, Error, HttpMessage, HttpResponse, HttpServer @@ -110,8 +112,6 @@ async fn main() -> anyhow::Result<()> { 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() @@ -138,18 +138,19 @@ async fn main() -> anyhow::Result<()> { .route("/ws", web::get().to(handlers_ws::handler) .wrap(middleware::from_fn(extract_claims)), ) // WebSocket - .route("/status", web::get().to(async || "ok")) - - .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)) - })) + // .route("/status", web::get().to(async || "ok")) + + .route("/status", web::get().to(|hub: web::Data| async move { + let count = hub.count().await; + Ok::<_, actix_web::Error>(HttpResponse::Ok().json(json!({ "websockets": count, "status": "OK" }))) + })) + + // .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 f7f9ff3295..bd090d6d4f 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -341,15 +341,15 @@ impl TryFrom for RedisEvent { 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::().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::().ok()) + // .unwrap_or(0); Ok(RedisEvent { - db, + // db, key: payload.clone(), action, })