fix: resolve semver variables and invalidate cache on flag updates

Signed-off-by: Eren Atas <[email protected]>

Author: erenatas

crates/flagd/src/cache/service.rs

@@ -245,6 +245,13 @@ where
         cache.add(cache_key, entry)
     }
 
+    pub async fn purge(&self) {
+        if self.enabled {
+            let mut cache = self.cache.write().await;
+            cache.purge();
+        }
+    }
+
     pub fn disable(&mut self) {
         if self.enabled {
             self.enabled = false;

crates/flagd/src/resolver/in_process/resolver/file.rs

@@ -1,20 +1,22 @@
 use crate::resolver::in_process::model::value_converter::ValueConverter;
 use crate::resolver::in_process::storage::connector::file::FileConnector;
-use crate::resolver::in_process::storage::{FlagStore, StorageState};
+use crate::resolver::in_process::storage::{FlagStore, StorageState, StorageStateChange};
 use crate::resolver::in_process::targeting::Operator;
 use crate::{CacheService, CacheSettings};
 use anyhow::Result;
 use async_trait::async_trait;
 use open_feature::provider::{FeatureProvider, ProviderMetadata, ResolutionDetails};
 use open_feature::{EvaluationContext, EvaluationError, EvaluationErrorCode, StructValue, Value};
 use std::sync::Arc;
+use tokio::sync::Mutex;
 use tracing::debug;
 
 pub struct FileResolver {
     store: Arc<FlagStore>,
     operator: Operator,
     metadata: ProviderMetadata,
     cache: Option<Arc<CacheService<Value>>>,
+    state_receiver: Arc<Mutex<tokio::sync::mpsc::Receiver<StorageStateChange>>>,
 }
 impl FileResolver {
     pub async fn new(source_path: String, cache_settings: Option<CacheSettings>) -> Result<Self> {
@@ -35,21 +37,51 @@ impl FileResolver {
             return Err(anyhow::anyhow!("Timeout waiting for initial flag state"));
         }
 
+        let cache = cache_settings.map(|settings| Arc::new(CacheService::new(settings)));
+
         Ok(Self {
             store,
             operator: Operator::new(),
             metadata: ProviderMetadata::new("flagd"),
-            cache: cache_settings.map(|settings| Arc::new(CacheService::new(settings))),
+            cache,
+            state_receiver: Arc::new(Mutex::new(state_receiver)),
         })
     }
 
+    /// Check for flag updates and clear cache if needed (non-blocking)
+    async fn check_for_updates(&self) {
+        if self.cache.is_none() {
+            return;
+        }
+
+        let mut receiver = self.state_receiver.lock().await;
+
+        // Drain all pending state changes (non-blocking)
+        let mut should_clear = false;
+        while let Ok(state_change) = receiver.try_recv() {
+            if state_change.storage_state == StorageState::Ok {
+                should_clear = true;
+            }
+        }
+
+        if should_clear {
+            debug!("Flag store updated, clearing cache");
+            if let Some(cache) = &self.cache {
+                cache.purge().await;
+            }
+        }
+    }
+
     async fn resolve_value<T>(
         &self,
         flag_key: &str,
         context: &EvaluationContext,
         value_converter: impl Fn(&serde_json::Value) -> Option<T>,
         type_name: &str,
     ) -> Result<ResolutionDetails<T>, EvaluationError> {
+        // Check for flag updates and clear cache if needed
+        self.check_for_updates().await;
+
         if let Some(cache) = &self.cache
             && let Some(cached_value) = cache.get(flag_key, context).await
         {

crates/flagd/src/resolver/in_process/resolver/grpc.rs

@@ -8,59 +8,83 @@ use open_feature::provider::{FeatureProvider, ProviderMetadata, ResolutionDetail
 use open_feature::{EvaluationContext, EvaluationError, EvaluationErrorCode, StructValue, Value};
 use serde_json::Value as JsonValue;
 use std::sync::Arc;
+use tokio::sync::Mutex;
+use tracing::debug;
 
-use crate::resolver::in_process::storage::FlagStore;
 use crate::resolver::in_process::storage::connector::grpc::GrpcStreamConnector;
+use crate::resolver::in_process::storage::{FlagStore, StorageState, StorageStateChange};
 
 pub struct InProcessResolver {
     store: Arc<FlagStore>,
     operator: Operator,
     metadata: ProviderMetadata,
     cache: Option<Arc<CacheService<Value>>>,
+    state_receiver: Arc<Mutex<tokio::sync::mpsc::Receiver<StorageStateChange>>>,
 }
 
 impl InProcessResolver {
     pub async fn new(options: &FlagdOptions) -> Result<Self> {
-        let store = match &options.socket_path {
+        let (store, state_receiver) = match &options.socket_path {
             Some(_) => Self::create_unix_socket_store(options).await?,
             None => Self::create_tcp_store(options).await?,
         };
 
+        let cache = options
+            .cache_settings
+            .clone()
+            .map(|settings| Arc::new(CacheService::new(settings)));
+
         Ok(Self {
             store,
             operator: Operator::new(),
             metadata: ProviderMetadata::new("flagd"),
-            cache: options
-                .cache_settings
-                .clone()
-                .map(|settings| Arc::new(CacheService::new(settings))),
+            cache,
+            state_receiver: Arc::new(Mutex::new(state_receiver)),
         })
     }
 
-    async fn create_unix_socket_store(_options: &FlagdOptions) -> Result<Arc<FlagStore>> {
-        // let socket_path = options.socket_path.as_ref().unwrap().clone();
-        // let socket_path_for_connector = socket_path.clone();
+    /// Check for flag updates and clear cache if needed (non-blocking)
+    async fn check_for_updates(&self) {
+        if self.cache.is_none() {
+            return;
+        }
+
+        let mut receiver = self.state_receiver.lock().await;
 
-        // let _channel = Endpoint::try_from("http://[::]:50051")?
-        //     .connect_with_connector(service_fn(move |_: Uri| {
-        //         let path = socket_path.clone();
-        //         async move {
-        //             let stream = UnixStream::connect(path).await?;
-        //             Ok::<_, std::io::Error>(TokioIo::new(stream))
-        //         }
-        //     }))
-        //     .await?;
+        // Drain all pending state changes (non-blocking)
+        let mut should_clear = false;
+        while let Ok(state_change) = receiver.try_recv() {
+            if state_change.storage_state == StorageState::Ok {
+                should_clear = true;
+            }
+        }
 
-        // let connector =
-        //     GrpcStreamConnector::new(socket_path_for_connector, options.selector.clone(), options);
-        // let (store, _state_receiver) = FlagStore::new(Arc::new(connector));
-        // let store = Arc::new(store);
-        // store.init().await?;
-        // Ok(store)
-        todo!("Unix socket store for in-process is not implemented")
+        if should_clear {
+            debug!("Flag store updated, clearing cache");
+            if let Some(cache) = &self.cache {
+                cache.purge().await;
+            }
+        }
     }
 
-    async fn create_tcp_store(options: &FlagdOptions) -> Result<Arc<FlagStore>> {
+    async fn create_unix_socket_store(
+        _options: &FlagdOptions,
+    ) -> Result<(
+        Arc<FlagStore>,
+        tokio::sync::mpsc::Receiver<crate::resolver::in_process::storage::StorageStateChange>,
+    )> {
+        // Unix socket store for in-process is not implemented
+        Err(anyhow::anyhow!(
+            "Unix socket store for in-process is not implemented"
+        ))
+    }
+
+    async fn create_tcp_store(
+        options: &FlagdOptions,
+    ) -> Result<(
+        Arc<FlagStore>,
+        tokio::sync::mpsc::Receiver<crate::resolver::in_process::storage::StorageStateChange>,
+    )> {
         let target = options
             .target_uri
             .clone()
@@ -73,10 +97,10 @@ impl InProcessResolver {
             upstream_config.authority().to_string(),
         );
 
-        let (store, _state_receiver) = FlagStore::new(Arc::new(connector));
+        let (store, state_receiver) = FlagStore::new(Arc::new(connector));
         let store = Arc::new(store);
         store.init().await?;
-        Ok(store)
+        Ok((store, state_receiver))
     }
 
     async fn get_cached_value<T>(
@@ -100,6 +124,9 @@ impl InProcessResolver {
         value_converter: impl Fn(&JsonValue) -> Option<T>,
         type_name: &str,
     ) -> Result<ResolutionDetails<T>, EvaluationError> {
+        // Check for flag updates and clear cache if needed
+        self.check_for_updates().await;
+
         // Try cache first
         if let Some(cached_value) = self
             .get_cached_value(flag_key, context, |v| match v {

crates/flagd/src/resolver/in_process/targeting/semver.rs

@@ -9,24 +9,66 @@ impl Operator for SemVerOperator {
     fn evaluate(
         &self,
         args: &[Value],
-        _context: &mut ContextStack,
+        context: &mut ContextStack,
         _evaluator: &dyn Evaluator,
     ) -> datalogic_rs::Result<Value> {
         if args.len() != 3 {
             debug!("SemVer requires exactly 3 arguments, got {}", args.len());
             return Ok(Value::Null);
         }
 
-        let version1 = match args[0].as_str() {
-            Some(s) => match Version::parse(s) {
-                Ok(v) => v,
-                Err(e) => {
-                    debug!("Failed to parse first version: {:?}: {}", s, e);
-                    return Ok(Value::Null);
+        // Helper function to resolve a value (either a string literal or a variable reference)
+        let resolve_value = |arg: &Value| -> Option<String> {
+            match arg {
+                Value::String(s) => Some(s.clone()),
+                Value::Object(obj) => {
+                    // Check if it's a variable reference: {"var": "variable_name"}
+                    if let Some(var_name) = obj.get("var").and_then(|v| v.as_str()) {
+                        // Get the current data from context
+                        let frame = context.root();
+                        let data = frame.data();
+                        // Try to resolve from context (check both direct key and nested in $flagd)
+                        data.get(var_name)
+                            .and_then(|v| v.as_str())
+                            .map(|s| s.to_string())
+                            .or_else(|| {
+                                // Also check in $flagd.flagKey format
+                                if let Some(parts) = var_name.split_once('.') {
+                                    if parts.0 == "$flagd" {
+                                        data.get("$flagd")
+                                            .and_then(|v| v.get(parts.1))
+                                            .and_then(|v| v.as_str())
+                                            .map(|s| s.to_string())
+                                    } else {
+                                        None
+                                    }
+                                } else {
+                                    None
+                                }
+                            })
+                    } else {
+                        None
+                    }
                 }
-            },
+                _ => None,
+            }
+        };
+
+        let version1_str = match resolve_value(&args[0]) {
+            Some(s) => s,
             None => {
-                debug!("First argument must be a string: {:?}", args[0]);
+                debug!(
+                    "First argument must be a string or variable reference: {:?}",
+                    args[0]
+                );
+                return Ok(Value::Null);
+            }
+        };
+
+        let version1 = match Version::parse(&version1_str) {
+            Ok(v) => v,
+            Err(e) => {
+                debug!("Failed to parse first version: {:?}: {}", version1_str, e);
                 return Ok(Value::Null);
             }
         };
@@ -39,16 +81,21 @@ impl Operator for SemVerOperator {
             }
         };
 
-        let version2 = match args[2].as_str() {
-            Some(s) => match Version::parse(s) {
-                Ok(v) => v,
-                Err(e) => {
-                    debug!("Failed to parse second version: {:?}: {}", s, e);
-                    return Ok(Value::Null);
-                }
-            },
+        let version2_str = match resolve_value(&args[2]) {
+            Some(s) => s,
             None => {
-                debug!("Second argument must be a string: {:?}", args[2]);
+                debug!(
+                    "Second argument must be a string or variable reference: {:?}",
+                    args[2]
+                );
+                return Ok(Value::Null);
+            }
+        };
+
+        let version2 = match Version::parse(&version2_str) {
+            Ok(v) => v,
+            Err(e) => {
+                debug!("Failed to parse second version: {:?}: {}", version2_str, e);
                 return Ok(Value::Null);
             }
         };