Transactions, Pipelines, and Lua Scripts

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "strconv"
    "time"

    "github.com/redis/go-redis/v9"
)

// Account represents a bank account
type Account struct {
    ID      string  `json:"id"`
    Balance float64 `json:"balance"`
    Version int     `json:"version"` // For optimistic locking
}

// Transaction represents a financial transaction
type Transaction struct {
    ID       string    `json:"id"`
    FromID   string    `json:"from_id"`
    ToID     string    `json:"to_id"`
    Amount   float64   `json:"amount"`
    Status   string    `json:"status"`
    Created  time.Time `json:"created"`
    Updated  time.Time `json:"updated"`
}

// BankingSystem handles account operations with Redis transactions
type BankingSystem struct {
    client *redis.Client
    ctx    context.Context
}

func NewBankingSystem(client *redis.Client) *BankingSystem {
    return &BankingSystem{
        client: client,
        ctx:    context.Background(),
    }
}

// CreateAccount creates a new account with initial balance
func (bs *BankingSystem) CreateAccount(id string, initialBalance float64) error {
    account := &Account{
        ID:      id,
        Balance: initialBalance,
        Version: 1,
    }

    accountJSON, err := json.Marshal(account)
    if err != nil {
        return fmt.Errorf("failed to marshal account: %w", err)
    }

    key := fmt.Sprintf("account:%s", id)
    return bs.client.Set(bs.ctx, key, accountJSON, 0).Err()
}

// GetAccount retrieves account information
func (bs *BankingSystem) GetAccount(id string) (*Account, error) {
    key := fmt.Sprintf("account:%s", id)
    
    data, err := bs.client.Get(bs.ctx, key).Bytes()
    if err != nil {
        return nil, fmt.Errorf("account not found: %w", err)
    }

    var account Account
    if err := json.Unmarshal(data, &account); err != nil {
        return nil, fmt.Errorf("failed to unmarshal account: %w", err)
    }

    return &account, nil
}

// TransferMoney performs atomic money transfer between accounts
func (bs *BankingSystem) TransferMoney(fromID, toID string, amount float64) (*Transaction, error) {
    if amount <= 0 {
        return nil, fmt.Errorf("invalid amount: %f", amount)
    }

    txn := &Transaction{
        ID:      fmt.Sprintf("txn_%d", time.Now().UnixNano()),
        FromID:  fromID,
        ToID:    toID,
        Amount:  amount,
        Status:  "pending",
        Created: time.Now(),
    }

    // Retry loop for optimistic locking
    maxRetries := 5
    for retry := 0; retry < maxRetries; retry++ {
        success, err := bs.attemptTransfer(txn)
        if err != nil {
            return nil, err
        }
        if success {
            txn.Status = "completed"
            txn.Updated = time.Now()
            return txn, nil
        }
        
        // Brief delay before retry
        time.Sleep(time.Millisecond * 10)
    }

    txn.Status = "failed"
    txn.Updated = time.Now()
    return txn, fmt.Errorf("transfer failed after %d retries", maxRetries)
}

// attemptTransfer attempts a single transfer with optimistic locking
func (bs *BankingSystem) attemptTransfer(txn *Transaction) (bool, error) {
    fromKey := fmt.Sprintf("account:%s", txn.FromID)
    toKey := fmt.Sprintf("account:%s", txn.ToID)

    // Start watching the accounts for changes
    err := bs.client.Watch(bs.ctx, func(tx *redis.Tx) error {
        // Get current account states
        fromAccount, err := bs.getAccountInTx(tx, txn.FromID)
        if err != nil {
            return fmt.Errorf("failed to get from account: %w", err)
        }

        toAccount, err := bs.getAccountInTx(tx, txn.ToID)
        if err != nil {
            return fmt.Errorf("failed to get to account: %w", err)
        }

        // Validate transfer
        if fromAccount.Balance < txn.Amount {
            return fmt.Errorf("insufficient funds: balance=%.2f, amount=%.2f", 
                            fromAccount.Balance, txn.Amount)
        }

        // Calculate new balances
        fromAccount.Balance -= txn.Amount
        fromAccount.Version++
        
        toAccount.Balance += txn.Amount
        toAccount.Version++

        // Marshal updated accounts
        fromJSON, err := json.Marshal(fromAccount)
        if err != nil {
            return fmt.Errorf("failed to marshal from account: %w", err)
        }

        toJSON, err := json.Marshal(toAccount)
        if err != nil {
            return fmt.Errorf("failed to marshal to account: %w", err)
        }

        // Execute transaction
        _, err = tx.TxPipelined(bs.ctx, func(pipe redis.Pipeliner) error {
            pipe.Set(bs.ctx, fromKey, fromJSON, 0)
            pipe.Set(bs.ctx, toKey, toJSON, 0)
            
            // Log transaction
            txnKey := fmt.Sprintf("transaction:%s", txn.ID)
            txnJSON, _ := json.Marshal(txn)
            pipe.Set(bs.ctx, txnKey, txnJSON, 24*time.Hour)
            
            return nil
        })

        return err
    }, fromKey, toKey)

    if err == redis.TxFailedErr {
        return false, nil // Retry due to concurrent modification
    }
    
    return err == nil, err
}

// getAccountInTx gets account within a transaction context
func (bs *BankingSystem) getAccountInTx(tx *redis.Tx, accountID string) (*Account, error) {
    key := fmt.Sprintf("account:%s", accountID)
    
    data, err := tx.Get(bs.ctx, key).Bytes()
    if err != nil {
        return nil, err
    }

    var account Account
    if err := json.Unmarshal(data, &account); err != nil {
        return nil, err
    }

    return &account, nil
}

// BatchTransfer performs multiple transfers atomically
func (bs *BankingSystem) BatchTransfer(transfers []struct{ FromID, ToID string; Amount float64 }) error {
    // Collect all keys to watch
    var watchKeys []string
    for _, transfer := range transfers {
        watchKeys = append(watchKeys, 
            fmt.Sprintf("account:%s", transfer.FromID),
            fmt.Sprintf("account:%s", transfer.ToID))
    }

    return bs.client.Watch(bs.ctx, func(tx *redis.Tx) error {
        // Get all accounts
        accounts := make(map[string]*Account)
        
        for _, transfer := range transfers {
            if _, exists := accounts[transfer.FromID]; !exists {
                account, err := bs.getAccountInTx(tx, transfer.FromID)
                if err != nil {
                    return err
                }
                accounts[transfer.FromID] = account
            }
            
            if _, exists := accounts[transfer.ToID]; !exists {
                account, err := bs.getAccountInTx(tx, transfer.ToID)
                if err != nil {
                    return err
                }
                accounts[transfer.ToID] = account
            }
        }

        // Validate all transfers
        for _, transfer := range transfers {
            fromAccount := accounts[transfer.FromID]
            if fromAccount.Balance < transfer.Amount {
                return fmt.Errorf("insufficient funds in account %s", transfer.FromID)
            }
        }

        // Apply all transfers
        for _, transfer := range transfers {
            accounts[transfer.FromID].Balance -= transfer.Amount
            accounts[transfer.FromID].Version++
            
            accounts[transfer.ToID].Balance += transfer.Amount
            accounts[transfer.ToID].Version++
        }

        // Execute all updates
        _, err := tx.TxPipelined(bs.ctx, func(pipe redis.Pipeliner) error {
            for id, account := range accounts {
                accountJSON, _ := json.Marshal(account)
                key := fmt.Sprintf("account:%s", id)
                pipe.Set(bs.ctx, key, accountJSON, 0)
            }
            return nil
        })

        return err
    }, watchKeys...)
}

// Demo usage
func main() {
    client := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer client.Close()

    bs := NewBankingSystem(client)

    // Create test accounts
    bs.CreateAccount("alice", 1000.0)
    bs.CreateAccount("bob", 500.0)
    bs.CreateAccount("charlie", 200.0)

    // Simple transfer
    log.Println("=== Simple Transfer ===")
    txn, err := bs.TransferMoney("alice", "bob", 100.0)
    if err != nil {
        log.Printf("Transfer failed: %v", err)
    } else {
        log.Printf("Transfer successful: %+v", txn)
    }

    // Check balances
    alice, _ := bs.GetAccount("alice")
    bob, _ := bs.GetAccount("bob")
    log.Printf("Alice balance: %.2f", alice.Balance)
    log.Printf("Bob balance: %.2f", bob.Balance)

    // Batch transfer
    log.Println("\n=== Batch Transfer ===")
    batchTransfers := []struct{ FromID, ToID string; Amount float64 }{
        {"alice", "charlie", 50.0},
        {"bob", "alice", 25.0},
    }

    err = bs.BatchTransfer(batchTransfers)
    if err != nil {
        log.Printf("Batch transfer failed: %v", err)
    } else {
        log.Printf("Batch transfer successful")
    }

    // Final balances
    alice, _ = bs.GetAccount("alice")
    bob, _ = bs.GetAccount("bob")
    charlie, _ := bs.GetAccount("charlie")
    log.Printf("Final - Alice: %.2f, Bob: %.2f, Charlie: %.2f", 
               alice.Balance, bob.Balance, charlie.Balance)
}

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
)

// MetricsCollector efficiently collects and stores metrics using pipelines
type MetricsCollector struct {
    client     *redis.Client
    ctx        context.Context
    batchSize  int
    flushTime  time.Duration
    buffer     []MetricPoint
    mutex      sync.Mutex
    flushTimer *time.Timer
}

type MetricPoint struct {
    Name      string
    Value     float64
    Tags      map[string]string
    Timestamp time.Time
}

func NewMetricsCollector(client *redis.Client, batchSize int, flushTime time.Duration) *MetricsCollector {
    mc := &MetricsCollector{
        client:    client,
        ctx:       context.Background(),
        batchSize: batchSize,
        flushTime: flushTime,
        buffer:    make([]MetricPoint, 0, batchSize),
    }

    mc.startFlushTimer()
    return mc
}

// AddMetric adds a metric point to the buffer
func (mc *MetricsCollector) AddMetric(name string, value float64, tags map[string]string) {
    mc.mutex.Lock()
    defer mc.mutex.Unlock()

    metric := MetricPoint{
        Name:      name,
        Value:     value,
        Tags:      tags,
        Timestamp: time.Now(),
    }

    mc.buffer = append(mc.buffer, metric)

    // Flush if buffer is full
    if len(mc.buffer) >= mc.batchSize {
        mc.flushBuffer()
    }
}

// flushBuffer sends all buffered metrics using pipeline
func (mc *MetricsCollector) flushBuffer() {
    if len(mc.buffer) == 0 {
        return
    }

    start := time.Now()
    
    // Create pipeline
    pipe := mc.client.Pipeline()

    // Add all metrics to pipeline
    for _, metric := range mc.buffer {
        mc.addMetricToPipeline(pipe, metric)
    }

    // Execute pipeline
    _, err := pipe.Exec(mc.ctx)
    if err != nil {
        log.Printf("Failed to flush metrics: %v", err)
    } else {
        duration := time.Since(start)
        log.Printf("Flushed %d metrics in %v", len(mc.buffer), duration)
    }

    // Clear buffer
    mc.buffer = mc.buffer[:0]
    
    // Reset timer
    mc.resetFlushTimer()
}

// addMetricToPipeline adds a single metric to the pipeline
func (mc *MetricsCollector) addMetricToPipeline(pipe redis.Pipeliner, metric MetricPoint) {
    timestamp := metric.Timestamp.Unix()
    
    // Time series data (sorted set with timestamp as score)
    timeSeriesKey := fmt.Sprintf("ts:%s", metric.Name)
    pipe.ZAdd(mc.ctx, timeSeriesKey, redis.Z{
        Score:  float64(timestamp),
        Member: fmt.Sprintf("%.2f:%d", metric.Value, timestamp),
    })
    pipe.Expire(mc.ctx, timeSeriesKey, 7*24*time.Hour) // 7 days retention

    // Latest value (for quick access)
    latestKey := fmt.Sprintf("latest:%s", metric.Name)
    pipe.Set(mc.ctx, latestKey, metric.Value, time.Hour)

    // Counter for total occurrences
    countKey := fmt.Sprintf("count:%s", metric.Name)
    pipe.Incr(mc.ctx, countKey)
    pipe.Expire(mc.ctx, countKey, 24*time.Hour)

    // Hourly aggregation
    hour := metric.Timestamp.Truncate(time.Hour).Unix()
    hourlyKey := fmt.Sprintf("hourly:%s:%d", metric.Name, hour)
    pipe.IncrByFloat(mc.ctx, hourlyKey, metric.Value)
    pipe.Expire(mc.ctx, hourlyKey, 48*time.Hour)

    // Add to tags index if tags exist
    if len(metric.Tags) > 0 {
        for tagKey, tagVal := range metric.Tags {
            tagIndexKey := fmt.Sprintf("tag:%s:%s:%s", metric.Name, tagKey, tagVal)
            pipe.SAdd(mc.ctx, tagIndexKey, timestamp)
            pipe.Expire(mc.ctx, tagIndexKey, 7*24*time.Hour)
        }
    }
}

func (mc *MetricsCollector) startFlushTimer() {
    mc.flushTimer = time.AfterFunc(mc.flushTime, func() {
        mc.mutex.Lock()
        mc.flushBuffer()
        mc.mutex.Unlock()
    })
}

func (mc *MetricsCollector) resetFlushTimer() {
    if mc.flushTimer != nil {
        mc.flushTimer.Stop()
    }
    mc.startFlushTimer()
}

// Flush manually flushes all buffered metrics
func (mc *MetricsCollector) Flush() {
    mc.mutex.Lock()
    defer mc.mutex.Unlock()
    mc.flushBuffer()
}

// Close flushes remaining metrics and stops the collector
func (mc *MetricsCollector) Close() {
    mc.mutex.Lock()
    defer mc.mutex.Unlock()
    
    if mc.flushTimer != nil {
        mc.flushTimer.Stop()
    }
    
    mc.flushBuffer()
}

// BatchDataLoader efficiently loads large datasets using pipelines
type BatchDataLoader struct {
    client    *redis.Client
    ctx       context.Context
    batchSize int
}

func NewBatchDataLoader(client *redis.Client, batchSize int) *BatchDataLoader {
    return &BatchDataLoader{
        client:    client,
        ctx:       context.Background(),
        batchSize: batchSize,
    }
}

// LoadUsers loads multiple users efficiently
func (bdl *BatchDataLoader) LoadUsers(userIDs []string) (map[string]interface{}, error) {
    results := make(map[string]interface{})
    
    // Process in batches
    for i := 0; i < len(userIDs); i += bdl.batchSize {
        end := i + bdl.batchSize
        if end > len(userIDs) {
            end = len(userIDs)
        }
        
        batch := userIDs[i:end]
        batchResults, err := bdl.loadUserBatch(batch)
        if err != nil {
            return nil, err
        }
        
        // Merge results
        for k, v := range batchResults {
            results[k] = v
        }
    }
    
    return results, nil
}

func (bdl *BatchDataLoader) loadUserBatch(userIDs []string) (map[string]interface{}, error) {
    pipe := bdl.client.Pipeline()
    
    // Queue all operations
    commands := make(map[string]*redis.StringCmd)
    for _, userID := range userIDs {
        key := fmt.Sprintf("user:%s", userID)
        commands[userID] = pipe.Get(bdl.ctx, key)
    }
    
    // Execute pipeline
    _, err := pipe.Exec(bdl.ctx)
    if err != nil && err != redis.Nil {
        return nil, fmt.Errorf("pipeline execution failed: %w", err)
    }
    
    // Collect results
    results := make(map[string]interface{})
    for userID, cmd := range commands {
        if cmd.Err() == nil {
            results[userID] = cmd.Val()
        } else if cmd.Err() != redis.Nil {
            log.Printf("Failed to load user %s: %v", userID, cmd.Err())
        }
    }
    
    return results, nil
}

// PerformanceTest demonstrates pipeline performance benefits
func PerformanceTest(client *redis.Client) {
    ctx := context.Background()
    numOperations := 1000

    // Test 1: Individual operations
    start := time.Now()
    for i := 0; i < numOperations; i++ {
        key := fmt.Sprintf("test:%d", i)
        client.Set(ctx, key, fmt.Sprintf("value_%d", i), time.Minute)
    }
    individualTime := time.Since(start)

    // Clear test data
    client.FlushDB(ctx)

    // Test 2: Pipeline operations
    start = time.Now()
    pipe := client.Pipeline()
    
    for i := 0; i < numOperations; i++ {
        key := fmt.Sprintf("test:%d", i)
        pipe.Set(ctx, key, fmt.Sprintf("value_%d", i), time.Minute)
    }
    
    _, err := pipe.Exec(ctx)
    if err != nil {
        log.Printf("Pipeline failed: %v", err)
        return
    }
    
    pipelineTime := time.Since(start)

    // Results
    speedup := float64(individualTime) / float64(pipelineTime)
    log.Printf("Performance Test Results:")
    log.Printf("Individual operations: %v", individualTime)
    log.Printf("Pipeline operations: %v", pipelineTime)
    log.Printf("Speedup: %.2fx faster", speedup)
}

func main() {
    client := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer client.Close()

    // Metrics collection demo
    log.Println("=== Metrics Collection Demo ===")
    collector := NewMetricsCollector(client, 100, 5*time.Second)
    defer collector.Close()

    // Simulate metric collection
    for i := 0; i < 250; i++ {
        collector.AddMetric("cpu_usage", float64(i%100), map[string]string{
            "host":   "server1",
            "region": "us-east-1",
        })
        
        collector.AddMetric("memory_usage", float64((i*7)%100), map[string]string{
            "host": "server1",
        })
        
        // Add some delay to simulate real metrics
        if i%50 == 0 {
            time.Sleep(100 * time.Millisecond)
        }
    }

    collector.Flush() // Force flush remaining metrics

    // Batch loading demo
    log.Println("\n=== Batch Loading Demo ===")
    loader := NewBatchDataLoader(client, 10)

    // Create test users
    pipe := client.Pipeline()
    userIDs := make([]string, 25)
    for i := 0; i < 25; i++ {
        userID := fmt.Sprintf("user_%d", i)
        userIDs[i] = userID
        pipe.Set(context.Background(), fmt.Sprintf("user:%s", userID), 
                 fmt.Sprintf(`{"id":"%s","name":"User %d"}`, userID, i), 0)
    }
    pipe.Exec(context.Background())

    // Load users in batches
    start := time.Now()
    users, err := loader.LoadUsers(userIDs)
    loadTime := time.Since(start)
    
    if err != nil {
        log.Printf("Failed to load users: %v", err)
    } else {
        log.Printf("Loaded %d users in %v", len(users), loadTime)
    }

    // Performance comparison
    log.Println("\n=== Performance Test ===")
    PerformanceTest(client)
}

package main

import (
    "context"
    "crypto/sha1"
    "encoding/hex"
    "fmt"
    "log"
    "time"

    "github.com/redis/go-redis/v9"
)

// LuaScriptManager handles Redis Lua script execution
type LuaScriptManager struct {
    client *redis.Client
    ctx    context.Context
    cache  map[string]string // Script SHA cache
}

func NewLuaScriptManager(client *redis.Client) *LuaScriptManager {
    return &LuaScriptManager{
        client: client,
        ctx:    context.Background(),
        cache:  make(map[string]string),
    }
}

// registerScript registers a Lua script and returns its SHA
func (lsm *LuaScriptManager) registerScript(script string) string {
    // Calculate SHA1 of script
    h := sha1.New()
    h.Write([]byte(script))
    sha := hex.EncodeToString(h.Sum(nil))
    
    // Cache the SHA
    lsm.cache[script] = sha
    
    return sha
}

// executeScript executes a Lua script with fallback to EVAL if not loaded
func (lsm *LuaScriptManager) executeScript(script string, keys []string, args ...interface{}) (interface{}, error) {
    sha, exists := lsm.cache[script]
    if !exists {
        sha = lsm.registerScript(script)
    }

    // Try EVALSHA first (more efficient)
    result, err := lsm.client.EvalSha(lsm.ctx, sha, keys, args...).Result()
    if err != nil && err.Error() == "NOSCRIPT No matching script. Please use EVAL." {
        // Script not loaded, use EVAL
        result, err = lsm.client.Eval(lsm.ctx, script, keys, args...).Result()
    }

    return result, err
}

// Distributed Lock Implementation with Lua
func (lsm *LuaScriptManager) AcquireDistributedLock(lockKey, clientID string, ttlSeconds int) (bool, error) {
    script := `
        local lock_key = KEYS[1]
        local client_id = ARGV[1]
        local ttl = tonumber(ARGV[2])
        
        -- Try to acquire lock
        local result = redis.call('SET', lock_key, client_id, 'NX', 'EX', ttl)
        
        if result then
            return 1
        else
            -- Check if we already own this lock
            local current_owner = redis.call('GET', lock_key)
            if current_owner == client_id then
                -- Extend the lock
                redis.call('EXPIRE', lock_key, ttl)
                return 1
            end
            return 0
        end
    `

    result, err := lsm.executeScript(script, []string{lockKey}, clientID, ttlSeconds)
    if err != nil {
        return false, err
    }

    return result.(int64) == 1, nil
}

func (lsm *LuaScriptManager) ReleaseLock(lockKey, clientID string) (bool, error) {
    script := `
        local lock_key = KEYS[1]
        local client_id = ARGV[1]
        
        -- Check if we own the lock
        local current_owner = redis.call('GET', lock_key)
        if current_owner == client_id then
            redis.call('DEL', lock_key)
            return 1
        else
            return 0
        end
    `

    result, err := lsm.executeScript(script, []string{lockKey}, clientID)
    if err != nil {
        return false, err
    }

    return result.(int64) == 1, nil
}

// Complex Rate Limiting with Multiple Windows
func (lsm *LuaScriptManager) ComplexRateLimit(key string, limits map[string]int) (bool, map[string]int, error) {
    script := `
        local key_prefix = KEYS[1]
        local now = tonumber(ARGV[1])
        local limits = {}
        local windows = {}
        
        -- Parse limits and windows from arguments
        for i = 2, #ARGV, 2 do
            local window = ARGV[i]
            local limit = tonumber(ARGV[i + 1])
            table.insert(windows, window)
            limits[window] = limit
        end
        
        local results = {}
        local allowed = true
        
        for _, window in ipairs(windows) do
            local window_seconds = tonumber(window)
            local limit = limits[window]
            local bucket_key = key_prefix .. ':' .. window
            local window_start = now - window_seconds
            
            -- Clean old entries
            redis.call('ZREMRANGEBYSCORE', bucket_key, 0, window_start)
            
            -- Count current requests
            local current_count = redis.call('ZCARD', bucket_key)
            
            if current_count >= limit then
                allowed = false
            end
            
            results[window] = limit - current_count
            
            -- Add current request if allowed
            if allowed then
                redis.call('ZADD', bucket_key, now, now)
                redis.call('EXPIRE', bucket_key, window_seconds)
                results[window] = results[window] - 1
            end
        end
        
        return {allowed and 1 or 0, cjson.encode(results)}
    `

    // Build arguments
    args := []interface{}{time.Now().Unix()}
    for window, limit := range limits {
        args = append(args, window, limit)
    }

    result, err := lsm.executeScript(script, []string{key}, args...)
    if err != nil {
        return false, nil, err
    }

    res := result.([]interface{})
    allowed := res[0].(int64) == 1
    
    // Parse remaining counts (would need JSON parsing in real implementation)
    remaining := make(map[string]int)
    // Simplified - in real implementation, parse JSON from res[1]
    
    return allowed, remaining, nil
}

// Atomic Counter with Expiration and Callback
func (lsm *LuaScriptManager) AtomicCounterWithCallback(counterKey string, increment int, maxValue int, ttlSeconds int) (int, bool, error) {
    script := `
        local counter_key = KEYS[1]
        local increment = tonumber(ARGV[1])
        local max_value = tonumber(ARGV[2])
        local ttl = tonumber(ARGV[3])
        
        -- Get current value
        local current = tonumber(redis.call('GET', counter_key) or 0)
        local new_value = current + increment
        
        -- Check if we exceed maximum
        if new_value > max_value then
            return {current, 0} -- {current_value, exceeded_flag}
        end
        
        -- Set new value with TTL
        redis.call('SET', counter_key, new_value, 'EX', ttl)
        
        -- Check if we've reached a milestone
        local milestone_reached = 0
        if new_value >= max_value then
            milestone_reached = 1
        end
        
        return {new_value, milestone_reached}
    `

    result, err := lsm.executeScript(script, []string{counterKey}, increment, maxValue, ttlSeconds)
    if err != nil {
        return 0, false, err
    }

    res := result.([]interface{})
    newValue := int(res[0].(int64))
    milestoneReached := res[1].(int64) == 1

    return newValue, milestoneReached, nil
}

// Leaderboard with Score Updates
func (lsm *LuaScriptManager) UpdateLeaderboard(leaderboardKey, playerID string, scoreChange int, maxEntries int) (int, int, error) {
    script := `
        local lb_key = KEYS[1]
        local player_id = ARGV[1]
        local score_change = tonumber(ARGV[2])
        local max_entries = tonumber(ARGV[3])
        
        -- Update player score
        local new_score = redis.call('ZINCRBY', lb_key, score_change, player_id)
        
        -- Trim leaderboard to max entries (keep top scores)
        local total_entries = redis.call('ZCARD', lb_key)
        if total_entries > max_entries then
            local to_remove = total_entries - max_entries
            redis.call('ZREMRANGEBYRANK', lb_key, 0, to_remove - 1)
        end
        
        -- Get player rank (1-based, from highest score)
        local rank = redis.call('ZREVRANK', lb_key, player_id)
        if rank then
            rank = rank + 1 -- Convert to 1-based
        else
            rank = -1 -- Player not in leaderboard
        end
        
        return {tonumber(new_score), rank}
    `

    result, err := lsm.executeScript(script, []string{leaderboardKey}, playerID, scoreChange, maxEntries)
    if err != nil {
        return 0, 0, err
    }

    res := result.([]interface{})
    newScore := int(res[0].(int64))
    rank := int(res[1].(int64))

    return newScore, rank, nil
}

// Demo usage of Lua scripts
func main() {
    client := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer client.Close()

    lsm := NewLuaScriptManager(client)

    // Distributed Lock Demo
    log.Println("=== Distributed Lock Demo ===")
    clientID := "client_123"
    lockKey := "resource_lock"

    acquired, err := lsm.AcquireDistributedLock(lockKey, clientID, 30)
    if err != nil {
        log.Printf("Failed to acquire lock: %v", err)
    } else if acquired {
        log.Printf("Lock acquired successfully")
        
        // Simulate work
        time.Sleep(1 * time.Second)
        
        // Release lock
        released, err := lsm.ReleaseLock(lockKey, clientID)
        if err != nil {
            log.Printf("Failed to release lock: %v", err)
        } else if released {
            log.Printf("Lock released successfully")
        }
    } else {
        log.Printf("Failed to acquire lock - already held")
    }

    // Atomic Counter Demo
    log.Println("\n=== Atomic Counter Demo ===")
    counterKey := "game_score"
    
    for i := 0; i < 15; i++ {
        value, milestone, err := lsm.AtomicCounterWithCallback(counterKey, 1, 10, 60)
        if err != nil {
            log.Printf("Counter error: %v", err)
            continue
        }
        
        log.Printf("Counter: %d", value)
        if milestone {
            log.Printf("Milestone reached! Counter hit maximum value")
        }
        
        time.Sleep(100 * time.Millisecond)
    }

    // Leaderboard Demo
    log.Println("\n=== Leaderboard Demo ===")
    leaderboardKey := "game_leaderboard"
    
    players := []string{"alice", "bob", "charlie", "diana", "eve"}
    
    for _, player := range players {
        scoreChange := (len(player) * 10) + int(time.Now().Unix()%50)
        score, rank, err := lsm.UpdateLeaderboard(leaderboardKey, player, scoreChange, 3)
        if err != nil {
            log.Printf("Leaderboard error: %v", err)
            continue
        }
        
        if rank > 0 {
            log.Printf("Player %s: Score=%d, Rank=%d", player, score, rank)
        } else {
            log.Printf("Player %s: Score=%d, Not in top 3", player, score)
        }
    }

    // Show final leaderboard
    log.Println("\nFinal Leaderboard:")
    leaderboard, err := client.ZRevRangeWithScores(context.Background(), leaderboardKey, 0, 2).Result()
    if err != nil {
        log.Printf("Failed to get leaderboard: %v", err)
    } else {
        for i, entry := range leaderboard {
            log.Printf("%d. %s: %.0f points", i+1, entry.Member, entry.Score)
        }
    }
}

type InventoryManager struct {
    client *redis.Client
    ctx    context.Context
}

func (im *InventoryManager) ReserveItems(orderID string, items map[string]int) error {
    // TODO: Use Redis transaction to atomically:
    // 1. Check available inventory
    // 2. Reserve items with timeout
    // 3. Update inventory counts
    // 4. Log reservation
}

func (im *InventoryManager) ConfirmReservation(orderID string) error {
    // TODO: Convert reservation to permanent allocation
}

func (im *InventoryManager) CancelReservation(orderID string) error {
    // TODO: Return reserved items to available inventory
}

// Transaction debugging
func debugTransaction(client *redis.Client) {
    client.Watch(context.Background(), func(tx *redis.Tx) error {
        log.Printf("Transaction starting...")
        
        _, err := tx.TxPipelined(context.Background(), func(pipe redis.Pipeliner) error {
            log.Printf("Executing commands in transaction...")
            pipe.Set(context.Background(), "key1", "value1", 0)
            pipe.Incr(context.Background(), "counter")
            return nil
        })
        
        if err != nil {
            log.Printf("Transaction failed: %v", err)
        } else {
            log.Printf("Transaction completed successfully")
        }
        
        return err
    }, "watched_key")
}

// Pipeline performance monitoring  
func monitorPipeline(pipe redis.Pipeliner) {
    start := time.Now()
    results, err := pipe.Exec(context.Background())
    duration := time.Since(start)
    
    log.Printf("Pipeline: %d commands, %v duration, error: %v", 
               len(results), duration, err)
}

// Lua script error handling
func safeLuaExecution(client *redis.Client, script string, keys []string, args ...interface{}) {
    result, err := client.Eval(context.Background(), script, keys, args...).Result()
    if err != nil {
        log.Printf("Lua script error: %v", err)
        log.Printf("Script: %s", script)
        log.Printf("Keys: %v", keys)
        log.Printf("Args: %v", args)
    } else {
        log.Printf("Lua script result: %v", result)
    }
}

// Optimal pipeline sizes
const (
    OptimalPipelineSize = 100  // Balance memory vs performance
    MaxPipelineSize     = 1000 // Never exceed this
)

// Pipeline with size management
func optimizedPipeline(client *redis.Client, operations []func(redis.Pipeliner)) error {
    for i := 0; i < len(operations); i += OptimalPipelineSize {
        end := i + OptimalPipelineSize
        if end > len(operations) {
            end = len(operations)
        }
        
        pipe := client.Pipeline()
        for j := i; j < end; j++ {
            operations[j](pipe)
        }
        
        _, err := pipe.Exec(context.Background())
        if err != nil {
            return err
        }
    }
    return nil
}

// Lua script best practices
func efficientLuaScript() string {
    return `
        -- Use local variables for better performance
        local key = KEYS[1]
        local value = ARGV[1]
        
        -- Minimize Redis calls in loops
        local results = {}
        for i = 1, #KEYS do
            local result = redis.call('GET', KEYS[i])
            table.insert(results, result)
        end
        
        -- Return structured data
        return cjson.encode(results)
    `
}