Advanced Patterns and Use Cases

calendar_today January 12, 2024 schedule 19 min read

Learning Objectives: By the end of this lesson, you will be able to:

Implement advanced Redis patterns for complex business requirements

Build sophisticated applications using Redis as a primary data store

Design event-driven architectures with Redis Streams

Create geospatial applications with Redis geolocation features

Implement time-series data storage and analytics with Redis

1. Welcome and Introduction (5 minutes)

Icebreaker Activity

Congratulations! You’ve mastered Redis from zero to hero. Now it’s time to explore the cutting-edge patterns that power today’s most innovative applications - from real-time location services like Uber, to time-series analytics at Tesla, to event streaming at Netflix. Let’s discover how Redis can be the foundation for revolutionary applications!

Lesson Overview

Duration: 120 minutes
Prerequisites: All previous lessons completed
Materials Needed: Go environment, Redis server with modules, advanced Redis knowledge

2. Learning Objectives and Session Outline (5 minutes)

What We’ll Cover Today:

Redis Streams - Event streaming and processing pipelines
Geospatial Applications - Location-based services and proximity searches
Time-Series Data - High-frequency data storage and analytics
Graph Algorithms - Social networks and recommendation engines
Machine Learning Integration - Feature stores and model serving
Advanced Architectural Patterns - Microservices, CQRS, and event sourcing

Success Criteria:

By the end of this session, you should be able to architect and implement advanced Redis-based solutions for enterprise-scale applications.

3. Foundation Knowledge (15-20 minutes)

Advanced Redis Patterns Overview

1. Event Streaming with Redis Streams

Producer → [Redis Stream] → Consumer Group 1 → Worker 1, 2, 3
                         → Consumer Group 2 → Worker A, B
                         → Consumer Group 3 → Analytics Pipeline

2. Geospatial Computing

User Location → [Redis Geo] → Proximity Search → Nearby Points of Interest
                           → Distance Calculation → Route Optimization
                           → Geofencing → Alert System

3. Time-Series Architecture

Sensors → [Time-Series Ingestion] → [Redis TimeSeries] → [Aggregation] → Dashboards
       → [Real-time Alerting]     → [Historical Analysis] → [Predictive Models]

4. Live Coding Demonstration (30-35 minutes)

Example 1: Advanced Event Streaming with Redis Streams

package main

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

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

// Event represents a business event in the stream
type Event struct {
    ID        string                 `json:"id"`
    Type      string                 `json:"type"`
    Source    string                 `json:"source"`
    Timestamp time.Time              `json:"timestamp"`
    Data      map[string]interface{} `json:"data"`
    Version   string                 `json:"version"`
}

// EventStreamManager handles Redis Streams operations
type EventStreamManager struct {
    client *redis.Client
    ctx    context.Context
}

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

// PublishEvent publishes an event to a stream
func (esm *EventStreamManager) PublishEvent(streamKey string, event *Event) (string, error) {
    eventData := map[string]interface{}{
        "type":      event.Type,
        "source":    event.Source,
        "timestamp": event.Timestamp.Unix(),
        "data":      mustMarshal(event.Data),
        "version":   event.Version,
    }

    result, err := esm.client.XAdd(esm.ctx, &redis.XAddArgs{
        Stream: streamKey,
        MaxLen: 10000, // Keep last 10k events
        Approx: true,  // Approximate trimming for performance
        Values: eventData,
    }).Result()

    if err != nil {
        return "", fmt.Errorf("failed to publish event: %w", err)
    }

    event.ID = result
    log.Printf("Published event %s to stream %s", event.ID, streamKey)
    return result, nil
}

// CreateConsumerGroup creates a consumer group for parallel processing
func (esm *EventStreamManager) CreateConsumerGroup(streamKey, groupName string, startID string) error {
    err := esm.client.XGroupCreateMkStream(esm.ctx, streamKey, groupName, startID).Err()
    if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
        return fmt.Errorf("failed to create consumer group: %w", err)
    }
    return nil
}

// ConsumeEvents consumes events from a stream using consumer groups
func (esm *EventStreamManager) ConsumeEvents(streamKey, groupName, consumerName string, handler func(*Event) error) error {
    for {
        // Read from stream with consumer group
        streams, err := esm.client.XReadGroup(esm.ctx, &redis.XReadGroupArgs{
            Group:    groupName,
            Consumer: consumerName,
            Streams:  []string{streamKey, ">"},
            Count:    10,
            Block:    time.Second,
        }).Result()

        if err != nil {
            if err == redis.Nil {
                continue // No new messages
            }
            log.Printf("Error reading from stream: %v", err)
            time.Sleep(time.Second)
            continue
        }

        // Process each stream
        for _, stream := range streams {
            for _, message := range stream.Messages {
                event, err := esm.parseStreamMessage(message)
                if err != nil {
                    log.Printf("Failed to parse message %s: %v", message.ID, err)
                    continue
                }

                // Handle the event
                if err := handler(event); err != nil {
                    log.Printf("Failed to handle event %s: %v", event.ID, err)
                    // In production, you might want to retry or send to dead letter queue
                    continue
                }

                // Acknowledge the message
                esm.client.XAck(esm.ctx, streamKey, groupName, message.ID)
                log.Printf("Processed event %s", event.ID)
            }
        }
    }
}

// parseStreamMessage converts Redis stream message to Event
func (esm *EventStreamManager) parseStreamMessage(msg redis.XMessage) (*Event, error) {
    event := &Event{
        ID: msg.ID,
    }

    if eventType, ok := msg.Values["type"].(string); ok {
        event.Type = eventType
    }

    if source, ok := msg.Values["source"].(string); ok {
        event.Source = source
    }

    if timestampStr, ok := msg.Values["timestamp"].(string); ok {
        if timestamp, err := time.Parse("2006-01-02T15:04:05Z07:00", timestampStr); err == nil {
            event.Timestamp = timestamp
        }
    }

    if version, ok := msg.Values["version"].(string); ok {
        event.Version = version
    }

    if dataStr, ok := msg.Values["data"].(string); ok {
        if err := json.Unmarshal([]byte(dataStr), &event.Data); err != nil {
            return nil, fmt.Errorf("failed to unmarshal event data: %w", err)
        }
    }

    return event, nil
}

// E-commerce Order Processing Pipeline
type OrderProcessor struct {
    eventManager *EventStreamManager
    streamKey    string
}

func NewOrderProcessor(eventManager *EventStreamManager) *OrderProcessor {
    return &OrderProcessor{
        eventManager: eventManager,
        streamKey:    "orders:stream",
    }
}

// ProcessOrderPlaced handles order placement events
func (op *OrderProcessor) ProcessOrderPlaced(event *Event) error {
    log.Printf("Processing order placement: %v", event.Data)

    orderID := event.Data["order_id"].(string)
    customerID := event.Data["customer_id"].(string)
    
    // Validate inventory
    if err := op.validateInventory(orderID); err != nil {
        return op.publishEvent("order_failed", map[string]interface{}{
            "order_id":    orderID,
            "customer_id": customerID,
            "reason":      "insufficient_inventory",
        })
    }

    // Reserve inventory
    if err := op.reserveInventory(orderID); err != nil {
        return err
    }

    // Publish inventory reserved event
    return op.publishEvent("inventory_reserved", map[string]interface{}{
        "order_id":    orderID,
        "customer_id": customerID,
    })
}

// PublishEvent helper method
func (op *OrderProcessor) publishEvent(eventType string, data map[string]interface{}) error {
    event := &Event{
        Type:      eventType,
        Source:    "order_processor",
        Timestamp: time.Now(),
        Data:      data,
        Version:   "1.0",
    }

    _, err := op.eventManager.PublishEvent(op.streamKey, event)
    return err
}

func (op *OrderProcessor) validateInventory(orderID string) error {
    // Simulate inventory validation
    log.Printf("Validating inventory for order %s", orderID)
    return nil
}

func (op *OrderProcessor) reserveInventory(orderID string) error {
    // Simulate inventory reservation
    log.Printf("Reserving inventory for order %s", orderID)
    return nil
}

// Helper function
func mustMarshal(v interface{}) string {
    data, _ := json.Marshal(v)
    return string(data)
}

Example 2: Geospatial Location Services

package main

import (
    "context"
    "fmt"
    "log"
    "math"

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

// Location represents a geographic location
type Location struct {
    ID        string  `json:"id"`
    Name      string  `json:"name"`
    Latitude  float64 `json:"latitude"`
    Longitude float64 `json:"longitude"`
    Type      string  `json:"type"`
    Metadata  map[string]interface{} `json:"metadata"`
}

// GeospatialService provides location-based operations
type GeospatialService struct {
    client *redis.Client
    ctx    context.Context
}

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

// AddLocation adds a location to the geospatial index
func (gs *GeospatialService) AddLocation(setKey string, location *Location) error {
    // Add to geospatial index
    err := gs.client.GeoAdd(gs.ctx, setKey, &redis.GeoLocation{
        Name:      location.ID,
        Longitude: location.Longitude,
        Latitude:  location.Latitude,
    }).Err()

    if err != nil {
        return fmt.Errorf("failed to add location to geo index: %w", err)
    }

    // Store location metadata
    metadataKey := fmt.Sprintf("location:%s:metadata", location.ID)
    metadataJSON := mustMarshal(location)
    err = gs.client.Set(gs.ctx, metadataKey, metadataJSON, 0).Err()
    if err != nil {
        return fmt.Errorf("failed to store location metadata: %w", err)
    }

    log.Printf("Added location %s (%s) at %.6f,%.6f", 
               location.ID, location.Name, location.Latitude, location.Longitude)
    return nil
}

// FindNearbyLocations finds locations within a radius
func (gs *GeospatialService) FindNearbyLocations(setKey string, lat, lon, radiusKm float64, limit int) ([]*Location, error) {
    // Query nearby locations
    geoResults, err := gs.client.GeoRadius(gs.ctx, setKey, lon, lat, &redis.GeoRadiusQuery{
        Radius:      radiusKm,
        Unit:        "km",
        WithCoord:   true,
        WithDist:    true,
        Count:       limit,
        Sort:        "ASC", // Sort by distance
    }).Result()

    if err != nil {
        return nil, fmt.Errorf("failed to query nearby locations: %w", err)
    }

    var locations []*Location
    for _, result := range geoResults {
        location, err := gs.getLocationMetadata(result.Name)
        if err != nil {
            log.Printf("Failed to get metadata for location %s: %v", result.Name, err)
            continue
        }

        // Add distance information
        location.Metadata["distance_km"] = result.Dist
        location.Metadata["coordinates"] = map[string]float64{
            "latitude":  result.Latitude,
            "longitude": result.Longitude,
        }

        locations = append(locations, location)
    }

    return locations, nil
}

// FindLocationsInArea finds locations within a geographic area
func (gs *GeospatialService) FindLocationsInArea(setKey string, minLat, minLon, maxLat, maxLon float64) ([]*Location, error) {
    // Use GEORADIUS with multiple queries or implement bounding box logic
    centerLat := (minLat + maxLat) / 2
    centerLon := (minLon + maxLon) / 2
    
    // Calculate radius to cover the bounding box
    radius := gs.calculateBoundingBoxRadius(minLat, minLon, maxLat, maxLon)
    
    allLocations, err := gs.FindNearbyLocations(setKey, centerLat, centerLon, radius, 1000)
    if err != nil {
        return nil, err
    }

    // Filter locations within the actual bounding box
    var filteredLocations []*Location
    for _, location := range allLocations {
        if location.Latitude >= minLat && location.Latitude <= maxLat &&
           location.Longitude >= minLon && location.Longitude <= maxLon {
            filteredLocations = append(filteredLocations, location)
        }
    }

    return filteredLocations, nil
}

// CalculateDistance calculates distance between two locations
func (gs *GeospatialService) CalculateDistance(setKey, location1, location2 string) (float64, error) {
    result, err := gs.client.GeoDist(gs.ctx, setKey, location1, location2, "km").Result()
    if err != nil {
        return 0, fmt.Errorf("failed to calculate distance: %w", err)
    }
    return result, nil
}

// RideMatching service using geospatial features
type RideMatchingService struct {
    geo *GeospatialService
}

func NewRideMatchingService(geo *GeospatialService) *RideMatchingService {
    return &RideMatchingService{geo: geo}
}

// AddDriver adds a driver to the available drivers pool
func (rms *RideMatchingService) AddDriver(driverID string, lat, lon float64, metadata map[string]interface{}) error {
    driver := &Location{
        ID:        driverID,
        Name:      fmt.Sprintf("Driver %s", driverID),
        Latitude:  lat,
        Longitude: lon,
        Type:      "driver",
        Metadata:  metadata,
    }

    return rms.geo.AddLocation("available_drivers", driver)
}

// FindNearestDrivers finds the closest available drivers
func (rms *RideMatchingService) FindNearestDrivers(passengerLat, passengerLon float64, maxDistance float64, limit int) ([]*Location, error) {
    return rms.geo.FindNearbyLocations("available_drivers", passengerLat, passengerLon, maxDistance, limit)
}

// MatchRide matches a passenger with the best available driver
func (rms *RideMatchingService) MatchRide(passengerID string, passengerLat, passengerLon float64) (*Location, error) {
    // Find nearby drivers
    drivers, err := rms.FindNearestDrivers(passengerLat, passengerLon, 5.0, 10) // 5km radius, max 10 drivers
    if err != nil {
        return nil, err
    }

    if len(drivers) == 0 {
        return nil, fmt.Errorf("no available drivers found")
    }

    // Select best driver based on criteria (distance, rating, etc.)
    bestDriver := drivers[0] // Closest driver for simplicity
    
    // Remove driver from available pool
    err = rms.geo.client.ZRem(rms.geo.ctx, "available_drivers", bestDriver.ID).Err()
    if err != nil {
        return nil, fmt.Errorf("failed to remove driver from available pool: %w", err)
    }

    log.Printf("Matched passenger %s with driver %s (%.2f km away)", 
               passengerID, bestDriver.ID, bestDriver.Metadata["distance_km"].(float64))

    return bestDriver, nil
}

// Helper methods
func (gs *GeospatialService) getLocationMetadata(locationID string) (*Location, error) {
    metadataKey := fmt.Sprintf("location:%s:metadata", locationID)
    metadataJSON, err := gs.client.Get(gs.ctx, metadataKey).Result()
    if err != nil {
        return nil, err
    }

    var location Location
    if err := json.Unmarshal([]byte(metadataJSON), &location); err != nil {
        return nil, err
    }

    return &location, nil
}

func (gs *GeospatialService) calculateBoundingBoxRadius(minLat, minLon, maxLat, maxLon float64) float64 {
    // Calculate diagonal distance of bounding box
    const earthRadiusKm = 6371
    
    dLat := (maxLat - minLat) * math.Pi / 180
    dLon := (maxLon - minLon) * math.Pi / 180
    
    a := math.Sin(dLat/2)*math.Sin(dLat/2) +
        math.Cos(minLat*math.Pi/180)*math.Cos(maxLat*math.Pi/180)*
        math.Sin(dLon/2)*math.Sin(dLon/2)
    
    c := 2 * math.Atan2(math.Sqrt(a), math.Sqrt(1-a))
    
    return earthRadiusKm * c
}

Example 3: Time-Series Data Management

package main

import (
    "context"
    "fmt"
    "log"
    "sort"
    "strconv"
    "time"

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

// TimeSeriesPoint represents a single data point
type TimeSeriesPoint struct {
    Timestamp time.Time `json:"timestamp"`
    Value     float64   `json:"value"`
    Tags      map[string]string `json:"tags,omitempty"`
}

// TimeSeriesManager handles time-series operations
type TimeSeriesManager struct {
    client *redis.Client
    ctx    context.Context
}

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

// AddPoint adds a data point to a time series
func (tsm *TimeSeriesManager) AddPoint(seriesKey string, point *TimeSeriesPoint) error {
    // Use sorted set with timestamp as score
    score := float64(point.Timestamp.Unix())
    member := fmt.Sprintf("%.6f:%d", point.Value, point.Timestamp.UnixNano())
    
    err := tsm.client.ZAdd(tsm.ctx, seriesKey, redis.Z{
        Score:  score,
        Member: member,
    }).Err()
    
    if err != nil {
        return fmt.Errorf("failed to add time series point: %w", err)
    }

    // Set expiration for automatic cleanup (e.g., 30 days)
    tsm.client.Expire(tsm.ctx, seriesKey, 30*24*time.Hour)
    
    return nil
}

// AddPoints adds multiple points efficiently
func (tsm *TimeSeriesManager) AddPoints(seriesKey string, points []*TimeSeriesPoint) error {
    if len(points) == 0 {
        return nil
    }

    pipe := tsm.client.Pipeline()
    
    for _, point := range points {
        score := float64(point.Timestamp.Unix())
        member := fmt.Sprintf("%.6f:%d", point.Value, point.Timestamp.UnixNano())
        
        pipe.ZAdd(tsm.ctx, seriesKey, redis.Z{
            Score:  score,
            Member: member,
        })
    }
    
    pipe.Expire(tsm.ctx, seriesKey, 30*24*time.Hour)
    
    _, err := pipe.Exec(tsm.ctx)
    if err != nil {
        return fmt.Errorf("failed to add time series points: %w", err)
    }
    
    log.Printf("Added %d points to series %s", len(points), seriesKey)
    return nil
}

// QueryRange retrieves points within a time range
func (tsm *TimeSeriesManager) QueryRange(seriesKey string, start, end time.Time) ([]*TimeSeriesPoint, error) {
    startScore := float64(start.Unix())
    endScore := float64(end.Unix())
    
    results, err := tsm.client.ZRangeByScore(tsm.ctx, seriesKey, &redis.ZRangeBy{
        Min:    fmt.Sprintf("%.6f", startScore),
        Max:    fmt.Sprintf("%.6f", endScore),
        Offset: 0,
        Count:  10000, // Limit to prevent memory issues
    }).Result()
    
    if err != nil {
        return nil, fmt.Errorf("failed to query time series range: %w", err)
    }
    
    var points []*TimeSeriesPoint
    for _, result := range results {
        point, err := tsm.parseTimeSeriesPoint(result)
        if err != nil {
            log.Printf("Failed to parse time series point %s: %v", result, err)
            continue
        }
        points = append(points, point)
    }
    
    return points, nil
}

// Aggregate computes aggregations over time windows
func (tsm *TimeSeriesManager) Aggregate(seriesKey string, start, end time.Time, windowSize time.Duration, aggregationType string) (map[time.Time]float64, error) {
    points, err := tsm.QueryRange(seriesKey, start, end)
    if err != nil {
        return nil, err
    }
    
    aggregates := make(map[time.Time]float64)
    windowData := make(map[time.Time][]float64)
    
    // Group points by time window
    for _, point := range points {
        windowStart := point.Timestamp.Truncate(windowSize)
        windowData[windowStart] = append(windowData[windowStart], point.Value)
    }
    
    // Calculate aggregates for each window
    for windowStart, values := range windowData {
        switch aggregationType {
        case "avg":
            aggregates[windowStart] = tsm.calculateAverage(values)
        case "sum":
            aggregates[windowStart] = tsm.calculateSum(values)
        case "min":
            aggregates[windowStart] = tsm.calculateMin(values)
        case "max":
            aggregates[windowStart] = tsm.calculateMax(values)
        case "count":
            aggregates[windowStart] = float64(len(values))
        default:
            return nil, fmt.Errorf("unsupported aggregation type: %s", aggregationType)
        }
    }
    
    return aggregates, nil
}

// Real-time IoT sensor monitoring system
type IoTMonitoringSystem struct {
    timeSeries *TimeSeriesManager
    client     *redis.Client
    ctx        context.Context
}

func NewIoTMonitoringSystem(client *redis.Client) *IoTMonitoringSystem {
    return &IoTMonitoringSystem{
        timeSeries: NewTimeSeriesManager(client),
        client:     client,
        ctx:        context.Background(),
    }
}

// RecordSensorData records data from IoT sensors
func (iot *IoTMonitoringSystem) RecordSensorData(sensorID string, sensorType string, value float64) error {
    point := &TimeSeriesPoint{
        Timestamp: time.Now(),
        Value:     value,
        Tags: map[string]string{
            "sensor_id":   sensorID,
            "sensor_type": sensorType,
        },
    }
    
    seriesKey := fmt.Sprintf("sensor:%s:%s", sensorType, sensorID)
    
    if err := iot.timeSeries.AddPoint(seriesKey, point); err != nil {
        return err
    }
    
    // Update latest value for quick access
    latestKey := fmt.Sprintf("sensor:latest:%s", sensorID)
    iot.client.Set(iot.ctx, latestKey, value, time.Hour)
    
    // Check for alerts
    return iot.checkAlerts(sensorID, sensorType, value)
}

// checkAlerts checks if sensor values trigger any alerts
func (iot *IoTMonitoringSystem) checkAlerts(sensorID, sensorType string, value float64) error {
    alertsKey := fmt.Sprintf("alerts:config:%s", sensorType)
    
    // Get alert thresholds (in production, this would be more sophisticated)
    config, err := iot.client.HGetAll(iot.ctx, alertsKey).Result()
    if err != nil {
        return nil // No alerts configured
    }
    
    if maxThresholdStr, exists := config["max_threshold"]; exists {
        if maxThreshold, err := strconv.ParseFloat(maxThresholdStr, 64); err == nil {
            if value > maxThreshold {
                return iot.triggerAlert(sensorID, "HIGH_VALUE", value, maxThreshold)
            }
        }
    }
    
    if minThresholdStr, exists := config["min_threshold"]; exists {
        if minThreshold, err := strconv.ParseFloat(minThresholdStr, 64); err == nil {
            if value < minThreshold {
                return iot.triggerAlert(sensorID, "LOW_VALUE", value, minThreshold)
            }
        }
    }
    
    return nil
}

// triggerAlert triggers an alert for abnormal sensor readings
func (iot *IoTMonitoringSystem) triggerAlert(sensorID, alertType string, currentValue, threshold float64) error {
    alert := map[string]interface{}{
        "sensor_id":     sensorID,
        "alert_type":    alertType,
        "current_value": currentValue,
        "threshold":     threshold,
        "timestamp":     time.Now().Unix(),
    }
    
    alertJSON := mustMarshal(alert)
    
    // Store alert
    alertKey := fmt.Sprintf("alerts:active:%s", sensorID)
    iot.client.Set(iot.ctx, alertKey, alertJSON, 24*time.Hour)
    
    // Publish alert to subscribers
    iot.client.Publish(iot.ctx, "alerts:channel", alertJSON)
    
    log.Printf("ALERT: Sensor %s triggered %s alert (value: %.2f, threshold: %.2f)", 
               sensorID, alertType, currentValue, threshold)
    
    return nil
}

// GetSensorHistory retrieves historical data for a sensor
func (iot *IoTMonitoringSystem) GetSensorHistory(sensorID, sensorType string, hours int) ([]*TimeSeriesPoint, error) {
    seriesKey := fmt.Sprintf("sensor:%s:%s", sensorType, sensorID)
    end := time.Now()
    start := end.Add(-time.Duration(hours) * time.Hour)
    
    return iot.timeSeries.QueryRange(seriesKey, start, end)
}

// GetSensorStats calculates statistics for a sensor over a time period
func (iot *IoTMonitoringSystem) GetSensorStats(sensorID, sensorType string, hours int) (map[string]float64, error) {
    points, err := iot.GetSensorHistory(sensorID, sensorType, hours)
    if err != nil {
        return nil, err
    }
    
    if len(points) == 0 {
        return nil, fmt.Errorf("no data available for sensor %s", sensorID)
    }
    
    values := make([]float64, len(points))
    for i, point := range points {
        values[i] = point.Value
    }
    
    stats := map[string]float64{
        "count":   float64(len(values)),
        "average": iot.timeSeries.calculateAverage(values),
        "min":     iot.timeSeries.calculateMin(values),
        "max":     iot.timeSeries.calculateMax(values),
        "sum":     iot.timeSeries.calculateSum(values),
    }
    
    return stats, nil
}

// Helper methods for time series calculations
func (tsm *TimeSeriesManager) calculateAverage(values []float64) float64 {
    if len(values) == 0 {
        return 0
    }
    sum := tsm.calculateSum(values)
    return sum / float64(len(values))
}

func (tsm *TimeSeriesManager) calculateSum(values []float64) float64 {
    sum := 0.0
    for _, v := range values {
        sum += v
    }
    return sum
}

func (tsm *TimeSeriesManager) calculateMin(values []float64) float64 {
    if len(values) == 0 {
        return 0
    }
    min := values[0]
    for _, v := range values[1:] {
        if v < min {
            min = v
        }
    }
    return min
}

func (tsm *TimeSeriesManager) calculateMax(values []float64) float64 {
    if len(values) == 0 {
        return 0
    }
    max := values[0]
    for _, v := range values[1:] {
        if v > max {
            max = v
        }
    }
    return max
}

func (tsm *TimeSeriesManager) parseTimeSeriesPoint(member string) (*TimeSeriesPoint, error) {
    // Parse member format: "value:nanoseconds"
    parts := strings.Split(member, ":")
    if len(parts) != 2 {
        return nil, fmt.Errorf("invalid time series member format: %s", member)
    }
    
    value, err := strconv.ParseFloat(parts[0], 64)
    if err != nil {
        return nil, fmt.Errorf("invalid value in time series member: %s", parts[0])
    }
    
    nanos, err := strconv.ParseInt(parts[1], 10, 64)
    if err != nil {
        return nil, fmt.Errorf("invalid timestamp in time series member: %s", parts[1])
    }
    
    return &TimeSeriesPoint{
        Timestamp: time.Unix(0, nanos),
        Value:     value,
    }, nil
}

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

    // Event Streaming Demo
    log.Println("=== Event Streaming Demo ===")
    eventManager := NewEventStreamManager(client)
    
    // Create consumer group
    eventManager.CreateConsumerGroup("orders:stream", "order_processors", "0")
    
    // Start order processor in background
    processor := NewOrderProcessor(eventManager)
    go func() {
        eventManager.ConsumeEvents("orders:stream", "order_processors", "worker1", processor.ProcessOrderPlaced)
    }()
    
    // Publish some test events
    for i := 0; i < 5; i++ {
        event := &Event{
            Type:      "order_placed",
            Source:    "web_app",
            Timestamp: time.Now(),
            Data: map[string]interface{}{
                "order_id":    fmt.Sprintf("order_%d", i),
                "customer_id": fmt.Sprintf("customer_%d", i%3),
                "amount":      100.0 + float64(i*50),
            },
            Version: "1.0",
        }
        
        eventManager.PublishEvent("orders:stream", event)
        time.Sleep(500 * time.Millisecond)
    }

    // Geospatial Demo
    log.Println("\n=== Geospatial Demo ===")
    geoService := NewGeospatialService(client)
    rideService := NewRideMatchingService(geoService)
    
    // Add some drivers
    drivers := []struct {
        id  string
        lat float64
        lon float64
    }{
        {"driver1", 37.7749, -122.4194}, // San Francisco
        {"driver2", 37.7849, -122.4094}, // Nearby
        {"driver3", 37.7649, -122.4294}, // Another location
    }
    
    for _, driver := range drivers {
        rideService.AddDriver(driver.id, driver.lat, driver.lon, map[string]interface{}{
            "rating": 4.5,
            "car_type": "sedan",
        })
    }
    
    // Find a ride
    passengerLat, passengerLon := 37.7749, -122.4194
    matchedDriver, err := rideService.MatchRide("passenger1", passengerLat, passengerLon)
    if err != nil {
        log.Printf("Failed to match ride: %v", err)
    } else {
        log.Printf("Matched with driver: %s", matchedDriver.ID)
    }

    // IoT Time Series Demo
    log.Println("\n=== IoT Time Series Demo ===")
    iotSystem := NewIoTMonitoringSystem(client)
    
    // Configure alert thresholds
    client.HSet(context.Background(), "alerts:config:temperature", map[string]interface{}{
        "max_threshold": "30.0",
        "min_threshold": "10.0",
    })
    
    // Simulate sensor data
    for i := 0; i < 20; i++ {
        temperature := 20.0 + float64(i)*0.5 // Gradually increasing temperature
        err := iotSystem.RecordSensorData("temp001", "temperature", temperature)
        if err != nil {
            log.Printf("Failed to record sensor data: %v", err)
        }
        time.Sleep(100 * time.Millisecond)
    }
    
    // Get sensor statistics
    stats, err := iotSystem.GetSensorStats("temp001", "temperature", 1)
    if err != nil {
        log.Printf("Failed to get sensor stats: %v", err)
    } else {
        log.Printf("Sensor stats: %+v", stats)
    }

    time.Sleep(2 * time.Second)
    log.Println("Advanced patterns demo completed!")
}

5. Hands-On Practice (40-45 minutes)

Exercise 1: Build a Complete Event-Driven Microservice (20 minutes)

Objective: Create a microservice architecture using Redis Streams

Requirements:

Order service publishes events
Inventory service processes reservations
Payment service handles transactions
Notification service sends alerts
Implement error handling and dead letter queues

Exercise 2: Create a Location-Based Recommendation Engine (15 minutes)

Objective: Build recommendations based on location and user behavior

Requirements:

Store user preferences and location history
Find nearby points of interest
Recommend based on similarity and proximity
Implement geofencing for location-based triggers

Exercise 3: Real-Time Analytics Dashboard (10 minutes)

Objective: Build real-time analytics using time-series data

Requirements:

Collect multiple metrics from different sources
Create real-time aggregations
Implement anomaly detection
Build alerting system for threshold breaches

6. Architectural Patterns and Best Practices (20 minutes)

Advanced Architectural Patterns

1. Event Sourcing with Redis Streams

// Event store implementation
type EventStore struct {
    streams map[string]*EventStreamManager
}

func (es *EventStore) AppendEvent(aggregateID string, event *Event) error {
    streamKey := fmt.Sprintf("events:%s", aggregateID)
    return es.streams[streamKey].PublishEvent(streamKey, event)
}

func (es *EventStore) ReplayEvents(aggregateID string, fromVersion int) ([]*Event, error) {
    // Replay events from specific version to rebuild aggregate state
}

2. CQRS (Command Query Responsibility Segregation)

type CQRSSystem struct {
    commandStore *redis.Client  // Write side
    readStore    *redis.Client  // Read side (potentially different instance)
    eventBus     *EventStreamManager
}

func (cqrs *CQRSSystem) ExecuteCommand(cmd Command) error {
    // Process command, generate events
    // Update command store
    // Publish events for read side updates
}

func (cqrs *CQRSSystem) Query(query Query) (interface{}, error) {
    // Query optimized read models
    // Return projected data
}

3. Saga Pattern for Distributed Transactions

type SagaOrchestrator struct {
    eventManager *EventStreamManager
    sagaStore    *redis.Client
}

func (so *SagaOrchestrator) StartSaga(sagaID string, steps []SagaStep) error {
    // Store saga state
    // Execute first step
    // Handle compensating actions on failure
}

Performance and Scaling Considerations

1. Data Partitioning Strategies

// Partition by hash of key
func (ps *PartitioningService) GetPartition(key string) int {
    hash := crc32.ChecksumIEEE([]byte(key))
    return int(hash % uint32(ps.numPartitions))
}

// Partition by time range for time-series data
func (ps *PartitioningService) GetTimePartition(timestamp time.Time) string {
    return fmt.Sprintf("partition:%s", timestamp.Format("2006-01-02"))
}

2. Caching Layers for Geospatial Data

type GeoCache struct {
    hotspots map[string]*GeoIndex  // Frequently accessed areas
    client   *redis.Client
}

func (gc *GeoCache) GetNearbyWithCache(lat, lon float64) ([]*Location, error) {
    // Check hot cache first
    // Fall back to Redis geo queries
    // Update cache with results
}

7. Summary and Review (10 minutes)

Key Takeaways

Redis Streams enable sophisticated event-driven architectures with guaranteed delivery
Geospatial features power location-based applications with high performance
Time-series patterns handle high-frequency data efficiently
Advanced patterns like CQRS and Event Sourcing scale to enterprise requirements
Proper data modeling and partitioning are crucial for performance

Pattern Selection Matrix

Use Case	Pattern	Benefits	Considerations
Event Processing	Redis Streams	Guaranteed delivery, scaling	Memory usage
Location Services	Geospatial	Built-in algorithms	Limited to Earth coordinates
Time-Series	Sorted Sets	Simple, fast	Manual aggregation
Real-time Analytics	Multiple patterns	Flexibility	Complexity

Architecture Evolution Path

Simple Cache → Session Store → Pub/Sub → Streams → Advanced Patterns
     ↓              ↓            ↓         ↓           ↓
  Basic Ops    Stateful Apps  Real-time  Event-Driven  Enterprise

Next Steps for Mastery

Practice building complete systems using multiple patterns
Study production case studies from major companies
Contribute to Redis ecosystem with modules and tools
Mentor others in Redis best practices
Stay updated with Redis roadmap and new features

Congratulations! 🎉

You have completed the Redis Zero to Hero journey! You now have the knowledge and skills to:

Build high-performance applications with Redis
Design scalable architectures for enterprise workloads
Implement advanced patterns for complex business requirements
Operate Redis systems in production environments
Mentor and teach others in Redis mastery

You are now a Redis Expert! Use your knowledge to build amazing applications and contribute to the Redis community. The journey of learning never ends - keep exploring, keep building, and keep sharing your knowledge with others.

For Instructors

Teaching Tips

Use real-world examples throughout the lesson
Encourage students to think about their own use cases
Demonstrate the progression from simple to complex patterns
Show actual production architectures and their evolution
Emphasize the importance of monitoring and observability

Assessment Ideas

Capstone Project: Build a complete application using multiple advanced patterns
Architecture Review: Design a system architecture for a given business requirement
Pattern Matching: Identify appropriate Redis patterns for various scenarios
Performance Analysis: Optimize an existing Redis-based system
Teaching Exercise: Prepare and deliver a lesson on a specific Redis topic

This completes your comprehensive Redis learning journey. Well done! 🚀