import redis
import numpy as np
from redis.commands.search.field import VectorField, TextField, NumericField, TagField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query

client = redis.Redis(host='localhost', port=6379, decode_responses=False)

# Create index
schema = (
    TextField("name", weight=5.0),
    TextField("description"),
    TagField("category", sortable=True),
    NumericField("price", sortable=True),
    VectorField(
        "embedding",
        "HNSW",
        {
            "TYPE": "FLOAT32",
            "DIM": 1536,
            "DISTANCE_METRIC": "COSINE",
            "INITIAL_CAP": 1000,
            "M": 16,
            "EF_CONSTRUCTION": 200,
        }
    )
)

client.ft("product_idx").create_index(
    schema,
    definition=IndexDefinition(prefix=["product:"], index_type=IndexType.HASH)
)

# Semantic search function
def semantic_search(query_embedding: np.ndarray, top_k: int = 10, category: str = None):
    vec_bytes = query_embedding.astype(np.float32).tobytes()
    
    # Build filter
    filter_expr = f"@category:}" if category else "*"
    
    query = (
        Query(f"{filter_expr}=>[KNN {top_k} @embedding $vec AS score]")
        .sort_by("score")
        .return_fields("name", "category", "price", "score")
        .dialect(2)
    )
    
    results = client.ft("product_idx").search(
        query,
        query_params={"vec": vec_bytes}
    )
    
    return [
        {
            "id": doc.id,
            "name": doc.name,
            "category": doc.category,
            "price": float(doc.price),
            "similarity": 1 - float(doc.score)
        }
        for doc in results.docs
    ]

import redis
from redis.commands.json.path import Path

client = redis.Redis(host='localhost', port=6379)

# Store complex nested document
user_doc = {
    "id": "user:12345",
    "profile": {
        "name": "Alice Chen",
        "email": "alice@example.com",
        "preferences": {
            "theme": "dark",
            "language": "en",
            "notifications": {
                "email": True,
                "push": False,
                "sms": True
            }
        }
    },
    "cart": [],
    "orders": [],
    "metadata": {
        "created_at": "2026-01-15T09:00:00Z",
        "last_login": "2026-03-28T08:30:00Z"
    }
}

client.json().set("user:12345", Path.root_path(), user_doc)

# Atomic increment/operations on nested paths
client.json().numincrby("user:12345", "$.cart_count", 1)

# Update specific nested field
client.json().set("user:12345", "$.profile.preferences.theme", "light")

# Get only needed fields (avoid over-fetching)
profile = client.json().get("user:12345", "$.profile.name", "$.metadata.last_login")

# Array operations
client.json().arrappend(
    "user:12345",
    "$.orders",
    {"order_id": "ord:9999", "total": 149.99, "status": "pending"}
)

# Redis 8.0: JSON Path indexing
client.ft("users_idx").create_index(
    [
        TextField("$.profile.name", as_name="name"),
        TagField("$.profile.preferences.language", as_name="lang"),
        NumericField("$.metadata.last_login", as_name="last_login")
    ],
    definition=IndexDefinition(prefix=["user:"], index_type=IndexType.JSON)
)

# Valkey fully supports RESP3 for richer data types
import valkey  # pip install valkey (drop-in redis replacement)

client = valkey.Valkey(
    host='localhost',
    port=6379,
    protocol=3  # RESP3
)

# RESP3 returns typed responses natively
# Hash → dict (not list of pairs)
user_data = client.hgetall("user:12345")
# Returns: {'name': 'Alice', 'age': '28'} instead of ['name', 'Alice', 'age', '28']

# Sets → Python sets
tags = client.smembers("product:tags")
# Returns: {'electronics', 'wireless', 'audio'}

# Push notifications (RESP3 pub/sub)
pubsub = client.pubsub()
pubsub.subscribe("notifications")

def update_leaderboard(user_id: str, score: float, leaderboard: str = "global"):
    """Thread-safe leaderboard update with Redis"""
    pipe = client.pipeline()
    
    # Add/update score
    pipe.zadd(f"leaderboard:{leaderboard}", {user_id: score})
    
    # Trim to top 1000
    pipe.zremrangebyrank(f"leaderboard:{leaderboard}", 0, -1001)
    
    pipe.execute()

def get_leaderboard_with_ranks(leaderboard: str, page: int = 1, per_page: int = 20):
    """Get paginated leaderboard with ranks"""
    start = (page - 1) * per_page
    end = start + per_page - 1
    
    # Get scores with ranks
    entries = client.zrevrange(
        f"leaderboard:{leaderboard}",
        start, end,
        withscores=True
    )
    
    return [
        {
            "rank": start + i + 1,
            "user_id": user_id.decode(),
            "score": score
        }
        for i, (user_id, score) in enumerate(entries)
    ]

def get_user_rank(user_id: str, leaderboard: str = "global"):
    """Get a specific user's rank (0-indexed from top)"""
    rank = client.zrevrank(f"leaderboard:{leaderboard}", user_id)
    score = client.zscore(f"leaderboard:{leaderboard}", user_id)
    return {"rank": rank + 1 if rank is not None else None, "score": score}

# Redis Streams: append-only log with consumer groups
import time

def publish_event(stream: str, event_type: str, data: dict):
    """Publish an event to a Redis Stream"""
    return client.xadd(
        stream,
        {
            "type": event_type,
            "timestamp": str(int(time.time() * 1000)),
            **{k: str(v) for k, v in data.items()}
        },
        maxlen=100_000,  # Keep last 100k events
        approximate=True  # Allow slight overflow for performance
    )

def create_consumer_group(stream: str, group: str):
    """Create a consumer group (idempotent)"""
    try:
        client.xgroup_create(stream, group, id="$", mkstream=True)
    except redis.exceptions.ResponseError as e:
        if "BUSYGROUP" not in str(e):
            raise

def process_events(stream: str, group: str, consumer: str, batch_size: int = 100):
    """Process events with at-least-once delivery guarantee"""
    
    # First, check for unacknowledged messages (recovery)
    pending = client.xreadgroup(
        groupname=group,
        consumername=consumer,
        streams={stream: "0"},  # "0" = read pending messages
        count=batch_size
    )
    
    if not pending:
        # No pending, read new messages
        pending = client.xreadgroup(
            groupname=group,
            consumername=consumer,
            streams={stream: ">"},  # ">" = undelivered messages
            count=batch_size,
            block=1000  # Wait up to 1s for new messages
        )
    
    if not pending:
        return []
    
    processed_ids = []
    for stream_name, messages in pending:
        for msg_id, fields in messages:
            try:
                # Process the event
                handle_event(fields)
                processed_ids.append(msg_id)
            except Exception as e:
                # Log but don't ack - will be retried
                print(f"Failed to process {msg_id}: {e}")
    
    # Acknowledge processed messages
    if processed_ids:
        client.xack(stream, group, *processed_ids)
    
    return processed_ids

# Redis 8 / Valkey: Built-in probabilistic data structures
def setup_bloom_filter(name: str, expected_items: int, error_rate: float = 0.01):
    """Create a Bloom filter for efficient deduplication"""
    # BF.RESERVE name error_rate initial_capacity
    try:
        client.bf().reserve(name, error_rate, expected_items)
    except redis.exceptions.ResponseError:
        pass  # Already exists

def track_seen_event(event_id: str) -> bool:
    """Returns True if this is a NEW event, False if already seen"""
    # BF.ADD returns 1 if newly added, 0 if already existed
    return bool(client.bf().add("seen_events", event_id))

def bulk_dedup(event_ids: list) -> list:
    """Efficiently check/add multiple events at once"""
    results = client.bf().madd("seen_events", *event_ids)
    return [event_id for event_id, is_new in zip(event_ids, results) if is_new]

import asyncio
import aioredis
from typing import Optional, Callable, TypeVar

T = TypeVar('T')

class RedisCache:
    def __init__(self, redis_url: str):
        self.redis = aioredis.from_url(redis_url)
        self._locks: dict = {}
    
    async def get_or_set(
        self,
        key: str,
        fetcher: Callable,
        ttl: int = 3600,
        early_refresh_ttl: int = 60  # Refresh 60s before expiry
    ) -> T:
        """Cache-aside with probabilistic early expiration to prevent stampede"""
        
        # Try to get cached value
        cached = await self.redis.get(key)
        if cached:
            remaining_ttl = await self.redis.ttl(key)
            
            # Probabilistic early refresh (XFetch algorithm)
            # Refresh with increasing probability as TTL decreases
            import math, random
            refresh_probability = math.exp(-remaining_ttl / early_refresh_ttl)
            
            if random.random() > refresh_probability:
                return cached  # Use cached value
        
        # Acquire a distributed lock to prevent stampede
        lock_key = f"lock:{key}"
        lock = await self.redis.set(lock_key, "1", nx=True, ex=10)
        
        if not lock:
            # Another process is fetching - wait and retry
            await asyncio.sleep(0.1)
            return await self.redis.get(key) or await fetcher()
        
        try:
            # Fetch fresh value
            value = await fetcher()
            
            # Store with TTL
            await self.redis.setex(key, ttl, value)
            return value
        finally:
            await self.redis.delete(lock_key)

# Usage
cache = RedisCache("redis://localhost:6379")

async def get_user_profile(user_id: str):
    return await cache.get_or_set(
        f"user:profile:{user_id}",
        lambda: fetch_from_database(user_id),
        ttl=3600
    )