behest 0.3.3

A Rust-native cloud agent runtime with typed tools, pluggable memory, queues, and observability.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

//! Redis-backed implementation of [`SessionDataStore`].
//!
//! Session data is stored as Redis hashes with key pattern
//! `behest:session_data:{session_id}`. Each field in the hash is a JSON-
//! serialised [`Value`].

use async_trait::async_trait;
use redis::AsyncCommands;
use serde_json::Value;
use uuid::Uuid;

use crate::runtime::invocation::{SessionDataError, SessionDataStore};

/// Redis-backed implementation of [`SessionDataStore`].
///
/// Each session's temporary KV data lives in a single Redis hash keyed by
/// `behest:session_data:{session_id}`. Values are stored as JSON strings.
///
/// # Example
///
/// ```rust,no_run
/// # async fn demo() -> Result<(), Box<dyn std::error::Error>> {
/// use behest::runtime::RedisSessionDataStore;
/// use behest::runtime::SessionDataStore;
/// use serde_json::json;
/// use uuid::Uuid;
///
/// let store = RedisSessionDataStore::new("redis://127.0.0.1:6379")?;
/// let sid = Uuid::new_v4();
/// store.set(sid, "key".into(), json!({"x": 1})).await?;
/// # Ok(())
/// # }
/// ```
pub struct RedisSessionDataStore {
    client: redis::Client,
}

impl RedisSessionDataStore {
    /// Creates a Redis session data store from a connection URL.
    ///
    /// # Errors
    ///
    /// Returns [`SessionDataError::Storage`] when the URL is malformed.
    pub fn new(url: &str) -> Result<Self, SessionDataError> {
        let client = redis::Client::open(url).map_err(|e| SessionDataError::Storage {
            message: format!("redis client error: {e}"),
        })?;
        Ok(Self { client })
    }

    /// Creates a Redis session data store from an existing `redis::Client`.
    #[must_use]
    pub fn from_client(client: redis::Client) -> Self {
        Self { client }
    }

    async fn conn(&self) -> Result<redis::aio::MultiplexedConnection, SessionDataError> {
        self.client
            .get_multiplexed_async_connection()
            .await
            .map_err(|e| SessionDataError::Storage {
                message: format!("redis connection error: {e}"),
            })
    }

    fn hash_key(session_id: Uuid) -> String {
        format!("behest:session_data:{session_id}")
    }
}

#[async_trait]
impl SessionDataStore for RedisSessionDataStore {
    async fn set(
        &self,
        session_id: Uuid,
        key: String,
        value: Value,
    ) -> Result<(), SessionDataError> {
        let mut conn = self.conn().await?;
        let hash = Self::hash_key(session_id);
        let json = serde_json::to_string(&value).map_err(|e| SessionDataError::Storage {
            message: format!("serialization error: {e}"),
        })?;
        conn.hset::<_, _, _, ()>(&hash, &key, &json)
            .await
            .map_err(|e| SessionDataError::Storage {
                message: format!("redis HSET error: {e}"),
            })
    }

    async fn get(&self, session_id: Uuid, key: &str) -> Result<Option<Value>, SessionDataError> {
        let mut conn = self.conn().await?;
        let hash = Self::hash_key(session_id);
        let raw: Option<String> =
            conn.hget(&hash, key)
                .await
                .map_err(|e| SessionDataError::Storage {
                    message: format!("redis HGET error: {e}"),
                })?;
        match raw {
            Some(s) => {
                let val: Value =
                    serde_json::from_str(&s).map_err(|e| SessionDataError::Storage {
                        message: format!("deserialization error: {e}"),
                    })?;
                Ok(Some(val))
            }
            None => Ok(None),
        }
    }

    async fn delete(&self, session_id: Uuid, key: &str) -> Result<(), SessionDataError> {
        let mut conn = self.conn().await?;
        let hash = Self::hash_key(session_id);
        conn.hdel::<_, _, ()>(&hash, key)
            .await
            .map_err(|e| SessionDataError::Storage {
                message: format!("redis HDEL error: {e}"),
            })
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;

    #[test]
    fn hash_key_format() {
        let sid = Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
        assert_eq!(
            RedisSessionDataStore::hash_key(sid),
            "behest:session_data:550e8400-e29b-41d4-a716-446655440000"
        );
    }
}