update

services/config/valkey_config.py

@@ -7,15 +7,59 @@ load_dotenv()
 # Valkey Configuration
 VALKEY_HOST = os.getenv('VALKEY_HOST', os.getenv('REDIS_HOST', 'localhost'))
 VALKEY_PORT = int(os.getenv('VALKEY_PORT', os.getenv('REDIS_PORT', 6379)))
-VALKEY_USER = os.getenv('VALKEY_USER', os.getenv('REDIS_USER', 'Default'))
+VALKEY_USER = os.getenv('VALKEY_USER', os.getenv('REDIS_USER', 'default'))
 VALKEY_PASSWORD = os.getenv('VALKEY_PASSWORD', os.getenv('REDIS_PASSWORD', None))
 VALKEY_DB = int(os.getenv('VALKEY_DB', os.getenv('REDIS_DB', 0)))
 
-VALKEY_URL = f"rediss://{VALKEY_USER}:{VALKEY_PASSWORD}@{VALKEY_HOST}:{VALKEY_PORT}"
+# Try to create Valkey client with fallback options
+valkey_client = None
+connection_available = False
 
+def create_valkey_client():
+    """Create Valkey client with fallback connection options"""
+    global valkey_client, connection_available
+    
+    try:
+        # First try with SSL if password is provided
+        if VALKEY_PASSWORD:
+            valkey_client = valkey.Valkey(
+                host=VALKEY_HOST,
+                port=VALKEY_PORT,
+                username=VALKEY_USER,
+                password=VALKEY_PASSWORD,
+                db=VALKEY_DB,
+                decode_responses=True,
+                ssl=True,
+                socket_timeout=5,
+                socket_connect_timeout=5
+            )
+        else:
+            # Try without SSL for local development
+            valkey_client = valkey.Valkey(
+                host=VALKEY_HOST,
+                port=VALKEY_PORT,
+                db=VALKEY_DB,
+                decode_responses=True,
+                ssl=False,
+                socket_timeout=5,
+                socket_connect_timeout=5
+            )
+        
+        # Test the connection
+        valkey_client.ping()
+        connection_available = True
+        print("✅ Successfully connected to Valkey")
+        
+    except Exception as e:
+        print(f"⚠️  Valkey connection failed: {e}")
+        print("🔄 Falling back to non-cached operations")
+        connection_available = False
+        valkey_client = None
+
+# Initialize the client
+create_valkey_client()
 
 # Maintain backward compatibility - alias for existing code
-valkey_client = valkey.from_url(VALKEY_URL)
 redis_client = valkey_client
 
 def get_redis_client():
@@ -26,12 +70,22 @@ def get_valkey_client():
     """Returns the Valkey client instance"""
     return valkey_client
 
+def is_connection_available():
+    """Check if Valkey connection is available"""
+    return connection_available
+
 # Test Valkey connection
 def test_connection():
+    global connection_available
+    if not valkey_client:
+        return False
+        
     try:
         valkey_client.ping()
+        connection_available = True
         print("Successfully connected to Valkey")
         return True
-    except valkey.ConnectionError:
-        print("Failed to connect to Valkey")
+    except Exception as e:
+        connection_available = False
+        print(f"Failed to connect to Valkey: {e}")
         return False

services/utils/api_cache.py

@@ -3,7 +3,7 @@ import json
 import zlib
 import base64
 import time
-from services.config.valkey_config import get_redis_client
+from services.config.valkey_config import get_redis_client, is_connection_available
 
 redis_client = get_redis_client()
 
@@ -29,49 +29,75 @@ async def get_cached_data(key, db_fetch_func, ttl=3600, use_compression=False):
     """
     global cache_hits, cache_misses, cached_data_size, compressed_data_size
     
-    # Check if data is in cache
-    cached_data = redis_client.get(key)
-    if cached_data:
-        # Increment hit counter
-        cache_hits += 1
-        print(f"Cache HIT: {key}")
-        
-        # Check if data is compressed (starts with special prefix)
-        if isinstance(cached_data, bytes) and cached_data.startswith(b'COMPRESSED:'):
-            # Remove prefix and decompress
-            compressed_data = base64.b64decode(cached_data[11:])
-            decompressed_data = zlib.decompress(compressed_data)
-            return json.loads(decompressed_data.decode('utf-8'))
-        else:
-            # Regular non-compressed data
-            return json.loads(cached_data)
-    
-    # Increment miss counter
-    cache_misses += 1
-    print(f"Cache MISS: {key}")
-    
-    # Fetch data from database
-    data = await db_fetch_func()
-    
-    # Serialize the data
-    serialized_data = json.dumps(data)
-    serialized_bytes = serialized_data.encode('utf-8')
+    # If Valkey is not available, fetch directly from database
+    if not is_connection_available() or not redis_client:
+        print(f"Cache DISABLED: {key} - fetching from database")
+        cache_misses += 1
+        return await db_fetch_func()
     
-    # Track original size
-    original_size = len(serialized_bytes)
-    cached_data_size += original_size
-    
-    # Decide whether to compress based on size and flag
-    if use_compression and original_size > COMPRESSION_THRESHOLD:
-        # Compress data
-        compressed_data = zlib.compress(serialized_bytes)
-        encoded_data = base64.b64encode(compressed_data)
+    try:
+        # Check if data is in cache
+        cached_data = redis_client.get(key)
+        if cached_data:
+            # Increment hit counter
+            cache_hits += 1
+            print(f"Cache HIT: {key}")
+            
+            # Check if data is compressed (starts with special prefix)
+            if isinstance(cached_data, bytes) and cached_data.startswith(b'COMPRESSED:'):
+                # Remove prefix and decompress
+                compressed_data = base64.b64decode(cached_data[11:])
+                decompressed_data = zlib.decompress(compressed_data)
+                return json.loads(decompressed_data.decode('utf-8'))
+            else:
+                # Regular non-compressed data
+                return json.loads(cached_data)
+        
+        # Increment miss counter
+        cache_misses += 1
+        print(f"Cache MISS: {key}")
+        
+        # Fetch data from database
+        data = await db_fetch_func()
+        
+        # Serialize the data
+        serialized_data = json.dumps(data)
+        serialized_bytes = serialized_data.encode('utf-8')
         
-        # Store with a prefix to indicate compression
-        redis_value = b'COMPRESSED:' + encoded_data
+        # Track original size
+        original_size = len(serialized_bytes)
+        cached_data_size += original_size
         
-        # Track compressed size
-        compressed_size = len(redis_value)
+        # Decide whether to compress based on size and flag
+        if use_compression and original_size > COMPRESSION_THRESHOLD:
+            # Compress data
+            compressed_data = zlib.compress(serialized_bytes)
+            encoded_data = base64.b64encode(compressed_data)
+            
+            # Store with a prefix to indicate compression
+            redis_value = b'COMPRESSED:' + encoded_data
+            
+            # Track compressed size
+            compressed_size = len(redis_value)
+            compressed_data_size += compressed_size
+            
+            # Calculate compression ratio
+            ratio = (compressed_size / original_size) * 100
+            print(f"Compressed {key}: {original_size} -> {compressed_size} bytes ({ratio:.2f}%)")
+            
+            # Store compressed data in Valkey
+            redis_client.setex(key, ttl, redis_value)
+        else:
+            # Store regular data
+            redis_client.setex(key, ttl, serialized_data)
+        
+        return data
+    
+    except Exception as e:
+        print(f"Cache ERROR for {key}: {e}")
+        print("Falling back to database")
+        cache_misses += 1
+        return await db_fetch_func()
         compressed_data_size += compressed_size
         
         # Calculate compression ratio
@@ -88,21 +114,35 @@ async def get_cached_data(key, db_fetch_func, ttl=3600, use_compression=False):
 
 def invalidate_cache(keys):
     """Delete multiple cache keys"""
-    if keys:
-        redis_client.delete(*keys)
+    if not is_connection_available() or not redis_client:
+        print("Cache DISABLED: Cannot invalidate cache keys")
+        return
+    
+    try:
+        if keys:
+            redis_client.delete(*keys)
+    except Exception as e:
+        print(f"Cache invalidation ERROR: {e}")
 
 def invalidate_cache_pattern(pattern):
     """
     Delete all cache keys matching a pattern
     For example: 'user:profile:*' to delete all user profiles
     """
-    cursor = 0
-    while True:
-        cursor, keys = redis_client.scan(cursor, match=pattern, count=100)
-        if keys:
-            redis_client.delete(*keys)
-        if cursor == 0:
-            break
+    if not is_connection_available() or not redis_client:
+        print("Cache DISABLED: Cannot invalidate cache pattern")
+        return
+    
+    try:
+        cursor = 0
+        while True:
+            cursor, keys = redis_client.scan(cursor, match=pattern, count=100)
+            if keys:
+                redis_client.delete(*keys)
+            if cursor == 0:
+                break
+    except Exception as e:
+        print(f"Cache pattern invalidation ERROR: {e}")
 
 def get_cache_metrics():
     """Get cache hit/miss metrics and efficiency statistics"""

services/utils/cache_utils.py

@@ -1,6 +1,6 @@
 import json
 from functools import wraps
-from services.config.valkey_config import get_redis_client
+from services.config.valkey_config import get_redis_client, is_connection_available
 import time
 
 redis_client = get_redis_client()
@@ -16,25 +16,37 @@ def cache_data(key_prefix, expire_time=3600):
     def decorator(func):
         @wraps(func)
         async def wrapper(*args, **kwargs):
-            # Create a unique key based on the function arguments
-            cache_key = f"{key_prefix}:{str(args)}:{str(kwargs)}"
+            # If Valkey is not available, execute function directly
+            if not is_connection_available() or not redis_client:
+                print(f"Cache DISABLED for {key_prefix} - executing function directly")
+                return await func(*args, **kwargs)
             
-            # Try to get the cached result
-            cached_result = redis_client.get(cache_key)
-            if cached_result:
-                return json.loads(cached_result)
-            
-            # If not cached, execute the function
-            result = await func(*args, **kwargs)
-            
-            # Cache the result
-            redis_client.setex(
-                cache_key,
-                expire_time,
-                json.dumps(result)
-            )
-            
-            return result
+            try:
+                # Create a unique key based on the function arguments
+                cache_key = f"{key_prefix}:{str(args)}:{str(kwargs)}"
+                
+                # Try to get the cached result
+                cached_result = redis_client.get(cache_key)
+                if cached_result:
+                    print(f"Cache HIT: {cache_key}")
+                    return json.loads(cached_result)
+                
+                # If not cached, execute the function
+                print(f"Cache MISS: {cache_key}")
+                result = await func(*args, **kwargs)
+                
+                # Cache the result
+                redis_client.setex(
+                    cache_key,
+                    expire_time,
+                    json.dumps(result)
+                )
+                
+                return result
+            except Exception as e:
+                print(f"Cache ERROR for {key_prefix}: {e}")
+                print("Falling back to direct function execution")
+                return await func(*args, **kwargs)
         return wrapper
     return decorator
 
@@ -42,13 +54,21 @@ def clear_cache(pattern="*"):
     """
     Clear cache entries matching the given pattern
     """
-    cursor = 0
-    while True:
-        cursor, keys = redis_client.scan(cursor, match=pattern)
-        if keys:
-            redis_client.delete(*keys)
-        if cursor == 0:
-            break
+    if not is_connection_available() or not redis_client:
+        print("Cache DISABLED: Cannot clear cache")
+        return
+    
+    try:
+        cursor = 0
+        while True:
+            cursor, keys = redis_client.scan(cursor, match=pattern)
+            if keys:
+                redis_client.delete(*keys)
+            if cursor == 0:
+                break
+        print(f"Cache cleared for pattern: {pattern}")
+    except Exception as e:
+        print(f"Cache clear ERROR: {e}")
 
 def cache_with_fallback(func):
     """
@@ -57,14 +77,21 @@ def cache_with_fallback(func):
     """
     @wraps(func)
     async def wrapper(*args, **kwargs):
+        # If Valkey is not available, execute function directly
+        if not is_connection_available() or not redis_client:
+            print(f"Cache DISABLED for {func.__name__} - executing function directly")
+            return await func(*args, **kwargs)
+        
         cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
         try:
             # Try to get from cache
             cached_result = redis_client.get(cache_key)
             if cached_result:
+                print(f"Cache HIT: {cache_key}")
                 return json.loads(cached_result)
             
             # If not in cache, execute function
+            print(f"Cache MISS: {cache_key}")
             result = await func(*args, **kwargs)
             
             # Cache the result
@@ -76,6 +103,6 @@ def cache_with_fallback(func):
             return result
         except Exception as e:
             # If Valkey fails, execute function directly
-            print(f"Cache error: {str(e)}, executing function directly")
+            print(f"Cache error for {func.__name__}: {str(e)}, executing function directly")
             return await func(*args, **kwargs)
     return wrapper

services/utils/chat_cache.py

@@ -1,4 +1,4 @@
-from services.config.valkey_config import get_redis_client
+from services.config.valkey_config import get_redis_client, is_connection_available
 import json
 
 redis_client = get_redis_client()
@@ -6,29 +6,52 @@ CHAT_HISTORY_TTL = 3600  # 1 hour session expiry
 
 def get_chat_history(username: str) -> list:
     """Get cached chat history for a user"""
-    key = f"chat:history:{username}"
-    history = redis_client.get(key)
-    if history:
-        try:
-            parsed = json.loads(history)
-            print(f"Retrieved chat history for {username}: {len(parsed)} messages")
-            return parsed
-        except Exception as e:
-            print(f"Error parsing chat history: {str(e)}")
-            return []
-    return []
+    if not is_connection_available() or not redis_client:
+        print(f"Chat cache DISABLED for {username} - returning empty history")
+        return []
+    
+    try:
+        key = f"chat:history:{username}"
+        history = redis_client.get(key)
+        if history:
+            try:
+                parsed = json.loads(history)
+                print(f"Retrieved chat history for {username}: {len(parsed)} messages")
+                return parsed
+            except Exception as e:
+                print(f"Error parsing chat history: {str(e)}")
+                return []
+        return []
+    except Exception as e:
+        print(f"Chat cache ERROR for {username}: {e}")
+        return []
 
 def append_chat_message(username: str, message: str, is_user: bool):
     """Add a new message to user's chat history"""
-    key = f"chat:history:{username}"
-    history = get_chat_history(username)
-    history.append({
-        "content": message,
-        "is_user": is_user
-    })
-    redis_client.setex(key, CHAT_HISTORY_TTL, json.dumps(history))
+    if not is_connection_available() or not redis_client:
+        print(f"Chat cache DISABLED for {username} - cannot save message")
+        return
+    
+    try:
+        key = f"chat:history:{username}"
+        history = get_chat_history(username)
+        history.append({
+            "content": message,
+            "is_user": is_user
+        })
+        redis_client.setex(key, CHAT_HISTORY_TTL, json.dumps(history))
+    except Exception as e:
+        print(f"Chat cache ERROR saving message for {username}: {e}")
 
 def clear_user_chat_history(username: str):
     """Clear chat history for a user"""
-    key = f"chat:history:{username}"
-    redis_client.delete(key)
+    if not is_connection_available() or not redis_client:
+        print(f"Chat cache DISABLED for {username} - cannot clear history")
+        return
+    
+    try:
+        key = f"chat:history:{username}"
+        redis_client.delete(key)
+        print(f"Cleared chat history for {username}")
+    except Exception as e:
+        print(f"Chat cache ERROR clearing history for {username}: {e}")