Pub/Sub and Real-time Features

package main

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

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

// Message represents a pub/sub message
type Message struct {
    Type      string                 `json:"type"`
    Channel   string                 `json:"channel"`
    Payload   map[string]interface{} `json:"payload"`
    Timestamp time.Time              `json:"timestamp"`
    UserID    int                    `json:"user_id,omitempty"`
}

// PubSubManager handles Redis pub/sub operations
type PubSubManager struct {
    client *redis.Client
    ctx    context.Context
}

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

// Publisher methods
func (psm *PubSubManager) PublishMessage(channel string, message *Message) error {
    message.Channel = channel
    message.Timestamp = time.Now()

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

    subscribers, err := psm.client.Publish(psm.ctx, channel, messageJSON).Result()
    if err != nil {
        return fmt.Errorf("failed to publish message: %w", err)
    }

    log.Printf("Published message to %s, reached %d subscribers", channel, subscribers)
    return nil
}

func (psm *PubSubManager) PublishChatMessage(roomID string, userID int, username, text string) error {
    message := &Message{
        Type:   "chat_message",
        UserID: userID,
        Payload: map[string]interface{}{
            "username": username,
            "text":     text,
            "room_id":  roomID,
        },
    }

    channel := fmt.Sprintf("chat:room:%s", roomID)
    return psm.PublishMessage(channel, message)
}

func (psm *PubSubManager) PublishNotification(userID int, title, body string) error {
    message := &Message{
        Type:   "notification",
        UserID: userID,
        Payload: map[string]interface{}{
            "title": title,
            "body":  body,
        },
    }

    // Send to user-specific channel
    userChannel := fmt.Sprintf("notifications:user:%d", userID)
    if err := psm.PublishMessage(userChannel, message); err != nil {
        return err
    }

    // Also send to global notifications channel
    globalChannel := "notifications:global"
    return psm.PublishMessage(globalChannel, message)
}

// Subscriber methods
func (psm *PubSubManager) Subscribe(channels []string, messageHandler func(*Message)) error {
    pubsub := psm.client.Subscribe(psm.ctx, channels...)
    defer pubsub.Close()

    log.Printf("Subscribed to channels: %v", channels)

    // Wait for subscription confirmation
    _, err := pubsub.Receive(psm.ctx)
    if err != nil {
        return fmt.Errorf("failed to confirm subscription: %w", err)
    }

    // Listen for messages
    ch := pubsub.Channel()
    for msg := range ch {
        var message Message
        if err := json.Unmarshal([]byte(msg.Payload), &message); err != nil {
            log.Printf("Failed to unmarshal message: %v", err)
            continue
        }

        // Handle message in goroutine to avoid blocking
        go messageHandler(&message)
    }

    return nil
}

func (psm *PubSubManager) SubscribeWithPattern(patterns []string, messageHandler func(*Message)) error {
    pubsub := psm.client.PSubscribe(psm.ctx, patterns...)
    defer pubsub.Close()

    log.Printf("Subscribed to patterns: %v", patterns)

    // Wait for subscription confirmation
    _, err := pubsub.Receive(psm.ctx)
    if err != nil {
        return fmt.Errorf("failed to confirm pattern subscription: %w", err)
    }

    // Listen for messages
    ch := pubsub.Channel()
    for msg := range ch {
        var message Message
        if err := json.Unmarshal([]byte(msg.Payload), &message); err != nil {
            log.Printf("Failed to unmarshal message: %v", err)
            continue
        }

        // Set the actual channel from Redis message
        message.Channel = msg.Channel

        // Handle message in goroutine
        go messageHandler(&message)
    }

    return nil
}

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

    psm := NewPubSubManager(client)

    // Start subscriber in goroutine
    go func() {
        channels := []string{"chat:room:general", "notifications:global"}
        
        err := psm.Subscribe(channels, func(msg *Message) {
            log.Printf("Received message on %s: %s", msg.Channel, msg.Type)
            log.Printf("Payload: %+v", msg.Payload)
        })
        
        if err != nil {
            log.Printf("Subscription error: %v", err)
        }
    }()

    // Start pattern subscriber in another goroutine
    go func() {
        patterns := []string{"chat:room:*", "notifications:user:*"}
        
        err := psm.SubscribeWithPattern(patterns, func(msg *Message) {
            log.Printf("Pattern match on %s: %s", msg.Channel, msg.Type)
            
            if msg.Type == "chat_message" {
                username := msg.Payload["username"].(string)
                text := msg.Payload["text"].(string)
                log.Printf("Chat from %s: %s", username, text)
            }
        })
        
        if err != nil {
            log.Printf("Pattern subscription error: %v", err)
        }
    }()

    // Wait for subscribers to start
    time.Sleep(1 * time.Second)

    // Publish some messages
    psm.PublishChatMessage("general", 123, "alice", "Hello everyone!")
    psm.PublishChatMessage("dev", 456, "bob", "Anyone working on the new feature?")
    psm.PublishNotification(123, "Welcome", "Welcome to our platform!")

    // Keep running
    time.Sleep(5 * time.Second)
    log.Println("Demo completed")
}

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "strconv"
    "sync"

    "github.com/gin-gonic/gin"
    "github.com/gorilla/websocket"
    "github.com/redis/go-redis/v9"
)

var upgrader = websocket.Upgrader{
    ReadBufferSize:  1024,
    WriteBufferSize: 1024,
    CheckOrigin: func(r *http.Request) bool {
        return true // Allow all origins in demo
    },
}

// ChatMessage represents a chat message
type ChatMessage struct {
    Type     string `json:"type"`
    RoomID   string `json:"room_id"`
    UserID   int    `json:"user_id"`
    Username string `json:"username"`
    Message  string `json:"message"`
    Time     string `json:"time"`
}

// Client represents a WebSocket client
type Client struct {
    ID       string
    UserID   int
    Username string
    RoomID   string
    Conn     *websocket.Conn
    Send     chan []byte
    Hub      *ChatHub
}

// ChatHub manages WebSocket connections and Redis pub/sub
type ChatHub struct {
    clients    map[*Client]bool
    broadcast  chan []byte
    register   chan *Client
    unregister chan *Client
    redis      *redis.Client
    ctx        context.Context
    mutex      sync.RWMutex
}

func NewChatHub(redisClient *redis.Client) *ChatHub {
    return &ChatHub{
        clients:    make(map[*Client]bool),
        broadcast:  make(chan []byte),
        register:   make(chan *Client),
        unregister: make(chan *Client),
        redis:      redisClient,
        ctx:        context.Background(),
    }
}

func (h *ChatHub) Run() {
    // Start Redis subscriber
    go h.subscribeToRedis()

    for {
        select {
        case client := <-h.register:
            h.registerClient(client)

        case client := <-h.unregister:
            h.unregisterClient(client)

        case message := <-h.broadcast:
            h.broadcastToClients(message)
        }
    }
}

func (h *ChatHub) registerClient(client *Client) {
    h.mutex.Lock()
    defer h.mutex.Unlock()

    h.clients[client] = true
    log.Printf("Client %s joined room %s", client.Username, client.RoomID)

    // Send welcome message
    welcomeMsg := ChatMessage{
        Type:    "system",
        RoomID:  client.RoomID,
        Message: fmt.Sprintf("%s joined the room", client.Username),
        Time:    getCurrentTime(),
    }

    if msgJSON, err := json.Marshal(welcomeMsg); err == nil {
        h.publishToRedis(fmt.Sprintf("chat:room:%s", client.RoomID), msgJSON)
    }
}

func (h *ChatHub) unregisterClient(client *Client) {
    h.mutex.Lock()
    defer h.mutex.Unlock()

    if _, ok := h.clients[client]; ok {
        delete(h.clients, client)
        close(client.Send)
        
        log.Printf("Client %s left room %s", client.Username, client.RoomID)

        // Send goodbye message
        goodbyeMsg := ChatMessage{
            Type:    "system",
            RoomID:  client.RoomID,
            Message: fmt.Sprintf("%s left the room", client.Username),
            Time:    getCurrentTime(),
        }

        if msgJSON, err := json.Marshal(goodbyeMsg); err == nil {
            h.publishToRedis(fmt.Sprintf("chat:room:%s", client.RoomID), msgJSON)
        }
    }
}

func (h *ChatHub) broadcastToClients(message []byte) {
    var chatMsg ChatMessage
    if err := json.Unmarshal(message, &chatMsg); err != nil {
        log.Printf("Failed to unmarshal broadcast message: %v", err)
        return
    }

    h.mutex.RLock()
    defer h.mutex.RUnlock()

    for client := range h.clients {
        // Only send to clients in the same room
        if client.RoomID == chatMsg.RoomID {
            select {
            case client.Send <- message:
            default:
                // Client's send channel is full, close it
                delete(h.clients, client)
                close(client.Send)
            }
        }
    }
}

func (h *ChatHub) subscribeToRedis() {
    pubsub := h.redis.PSubscribe(h.ctx, "chat:room:*")
    defer pubsub.Close()

    ch := pubsub.Channel()
    for msg := range ch {
        // Forward Redis message to WebSocket clients
        h.broadcast <- []byte(msg.Payload)
    }
}

func (h *ChatHub) publishToRedis(channel string, message []byte) {
    if err := h.redis.Publish(h.ctx, channel, message).Err(); err != nil {
        log.Printf("Failed to publish to Redis: %v", err)
    }
}

// Client handling
func (c *Client) ReadPump() {
    defer func() {
        c.Hub.unregister <- c
        c.Conn.Close()
    }()

    for {
        var chatMsg ChatMessage
        if err := c.Conn.ReadJSON(&chatMsg); err != nil {
            if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
                log.Printf("WebSocket error: %v", err)
            }
            break
        }

        // Set client info and timestamp
        chatMsg.UserID = c.UserID
        chatMsg.Username = c.Username
        chatMsg.RoomID = c.RoomID
        chatMsg.Time = getCurrentTime()
        chatMsg.Type = "message"

        // Publish to Redis (will be broadcast to all clients)
        if msgJSON, err := json.Marshal(chatMsg); err == nil {
            channel := fmt.Sprintf("chat:room:%s", c.RoomID)
            c.Hub.publishToRedis(channel, msgJSON)
        }
    }
}

func (c *Client) WritePump() {
    defer c.Conn.Close()

    for {
        select {
        case message, ok := <-c.Send:
            if !ok {
                c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }

            if err := c.Conn.WriteMessage(websocket.TextMessage, message); err != nil {
                log.Printf("WebSocket write error: %v", err)
                return
            }
        }
    }
}

func getCurrentTime() string {
    return fmt.Sprintf("%02d:%02d", 
        time.Now().Hour(), time.Now().Minute())
}

// HTTP handlers
func setupRoutes(hub *ChatHub) *gin.Engine {
    r := gin.Default()

    // Serve static files
    r.Static("/static", "./static")
    r.LoadHTMLGlob("templates/*")

    r.GET("/", func(c *gin.Context) {
        c.HTML(http.StatusOK, "chat.html", nil)
    })

    r.GET("/ws", func(c *gin.Context) {
        handleWebSocket(hub, c.Writer, c.Request)
    })

    r.POST("/api/rooms/:roomId/messages", func(c *gin.Context) {
        // REST API for sending messages (alternative to WebSocket)
        roomID := c.Param("roomId")
        
        var req struct {
            UserID   int    `json:"user_id"`
            Username string `json:"username"`
            Message  string `json:"message"`
        }

        if err := c.ShouldBindJSON(&req); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
            return
        }

        chatMsg := ChatMessage{
            Type:     "message",
            RoomID:   roomID,
            UserID:   req.UserID,
            Username: req.Username,
            Message:  req.Message,
            Time:     getCurrentTime(),
        }

        if msgJSON, err := json.Marshal(chatMsg); err == nil {
            channel := fmt.Sprintf("chat:room:%s", roomID)
            hub.publishToRedis(channel, msgJSON)
            c.JSON(http.StatusOK, gin.H{"status": "sent"})
        } else {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to send message"})
        }
    })

    return r
}

func handleWebSocket(hub *ChatHub, w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Printf("WebSocket upgrade error: %v", err)
        return
    }

    // Get client info from query parameters
    userIDStr := r.URL.Query().Get("user_id")
    username := r.URL.Query().Get("username")
    roomID := r.URL.Query().Get("room_id")

    if userIDStr == "" || username == "" || roomID == "" {
        conn.Close()
        return
    }

    userID, _ := strconv.Atoi(userIDStr)

    client := &Client{
        ID:       fmt.Sprintf("%s_%d", username, userID),
        UserID:   userID,
        Username: username,
        RoomID:   roomID,
        Conn:     conn,
        Send:     make(chan []byte, 256),
        Hub:      hub,
    }

    hub.register <- client

    // Start goroutines for reading and writing
    go client.WritePump()
    go client.ReadPump()
}

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

    // Create and start chat hub
    hub := NewChatHub(redisClient)
    go hub.Run()

    // Setup HTTP routes
    r := setupRoutes(hub)

    log.Println("Chat server starting on :8080...")
    log.Fatal(r.Run(":8080"))
}

package main

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

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

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

// EventBus handles event publishing and subscribing
type EventBus struct {
    client *redis.Client
    ctx    context.Context
}

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

func (eb *EventBus) PublishEvent(event *Event) error {
    event.Timestamp = time.Now()
    
    eventJSON, err := json.Marshal(event)
    if err != nil {
        return fmt.Errorf("failed to marshal event: %w", err)
    }

    // Publish to specific event type channel
    channel := fmt.Sprintf("events:%s", event.Type)
    if err := eb.client.Publish(eb.ctx, channel, eventJSON).Err(); err != nil {
        return fmt.Errorf("failed to publish event: %w", err)
    }

    // Also publish to service-specific channel
    serviceChannel := fmt.Sprintf("events:%s:%s", event.Source, event.Type)
    if err := eb.client.Publish(eb.ctx, serviceChannel, eventJSON).Err(); err != nil {
        log.Printf("Failed to publish to service channel: %v", err)
    }

    log.Printf("Published event %s of type %s", event.ID, event.Type)
    return nil
}

func (eb *EventBus) Subscribe(eventTypes []string, handler func(*Event)) error {
    channels := make([]string, len(eventTypes))
    for i, eventType := range eventTypes {
        channels[i] = fmt.Sprintf("events:%s", eventType)
    }

    pubsub := eb.client.Subscribe(eb.ctx, channels...)
    defer pubsub.Close()

    log.Printf("Subscribed to event types: %v", eventTypes)

    ch := pubsub.Channel()
    for msg := range ch {
        var event Event
        if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil {
            log.Printf("Failed to unmarshal event: %v", err)
            continue
        }

        // Handle event in goroutine
        go handler(&event)
    }

    return nil
}

// Service implementations
type OrderService struct {
    eventBus *EventBus
}

func NewOrderService(eventBus *EventBus) *OrderService {
    return &OrderService{eventBus: eventBus}
}

func (os *OrderService) CreateOrder(customerID, productID int, amount float64) error {
    // Business logic for creating order
    orderID := fmt.Sprintf("order_%d", time.Now().Unix())
    
    log.Printf("Creating order %s for customer %d", orderID, customerID)
    
    // Publish order created event
    event := &Event{
        ID:     fmt.Sprintf("evt_%d", time.Now().UnixNano()),
        Type:   "order_created",
        Source: "order_service",
        Data: map[string]interface{}{
            "order_id":    orderID,
            "customer_id": customerID,
            "product_id":  productID,
            "amount":      amount,
        },
        Version: "1.0",
    }

    return os.eventBus.PublishEvent(event)
}

type InventoryService struct {
    eventBus *EventBus
}

func NewInventoryService(eventBus *EventBus) *InventoryService {
    is := &InventoryService{eventBus: eventBus}
    
    // Subscribe to order events
    go is.subscribeToEvents()
    
    return is
}

func (is *InventoryService) subscribeToEvents() {
    is.eventBus.Subscribe([]string{"order_created"}, func(event *Event) {
        if event.Type == "order_created" {
            productID := int(event.Data["product_id"].(float64))
            orderID := event.Data["order_id"].(string)
            
            log.Printf("Inventory: Processing order %s for product %d", orderID, productID)
            
            // Simulate inventory check
            time.Sleep(100 * time.Millisecond)
            
            // Publish inventory reserved event
            reservedEvent := &Event{
                ID:     fmt.Sprintf("evt_%d", time.Now().UnixNano()),
                Type:   "inventory_reserved",
                Source: "inventory_service",
                Data: map[string]interface{}{
                    "order_id":   orderID,
                    "product_id": productID,
                    "reserved":   true,
                },
                Version: "1.0",
            }
            
            is.eventBus.PublishEvent(reservedEvent)
        }
    })
}

type PaymentService struct {
    eventBus *EventBus
}

func NewPaymentService(eventBus *EventBus) *PaymentService {
    ps := &PaymentService{eventBus: eventBus}
    
    // Subscribe to inventory events
    go ps.subscribeToEvents()
    
    return ps
}

func (ps *PaymentService) subscribeToEvents() {
    ps.eventBus.Subscribe([]string{"inventory_reserved"}, func(event *Event) {
        if event.Type == "inventory_reserved" && event.Data["reserved"].(bool) {
            orderID := event.Data["order_id"].(string)
            
            log.Printf("Payment: Processing payment for order %s", orderID)
            
            // Simulate payment processing
            time.Sleep(200 * time.Millisecond)
            
            // Publish payment processed event
            paymentEvent := &Event{
                ID:     fmt.Sprintf("evt_%d", time.Now().UnixNano()),
                Type:   "payment_processed",
                Source: "payment_service",
                Data: map[string]interface{}{
                    "order_id": orderID,
                    "status":   "completed",
                    "amount":   100.00,
                },
                Version: "1.0",
            }
            
            ps.eventBus.PublishEvent(paymentEvent)
        }
    })
}

type NotificationService struct {
    eventBus *EventBus
}

func NewNotificationService(eventBus *EventBus) *NotificationService {
    ns := &NotificationService{eventBus: eventBus}
    
    // Subscribe to multiple event types
    go ns.subscribeToEvents()
    
    return ns
}

func (ns *NotificationService) subscribeToEvents() {
    ns.eventBus.Subscribe([]string{"order_created", "payment_processed"}, func(event *Event) {
        switch event.Type {
        case "order_created":
            customerID := int(event.Data["customer_id"].(float64))
            orderID := event.Data["order_id"].(string)
            log.Printf("Notification: Sending order confirmation to customer %d for order %s", customerID, orderID)
            
        case "payment_processed":
            if event.Data["status"].(string) == "completed" {
                orderID := event.Data["order_id"].(string)
                log.Printf("Notification: Sending payment confirmation for order %s", orderID)
            }
        }
    })
}

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

    // Create event bus
    eventBus := NewEventBus(client)

    // Initialize services
    orderService := NewOrderService(eventBus)
    inventoryService := NewInventoryService(eventBus)
    paymentService := NewPaymentService(eventBus)
    notificationService := NewNotificationService(eventBus)

    // Wait for subscribers to start
    time.Sleep(1 * time.Second)

    // Simulate some orders
    log.Println("=== Simulating Order Processing ===")
    
    orderService.CreateOrder(123, 456, 99.99)
    orderService.CreateOrder(789, 123, 149.99)
    orderService.CreateOrder(456, 789, 79.99)

    // Keep services running
    time.Sleep(5 * time.Second)
    
    log.Println("=== Event Processing Demo Completed ===")
    
    // Keep references to avoid unused variable warnings
    _ = inventoryService
    _ = paymentService
    _ = notificationService
}

type SystemMetric struct {
    Type      string    `json:"type"`
    Value     float64   `json:"value"`
    Unit      string    `json:"unit"`
    Timestamp time.Time `json:"timestamp"`
    Host      string    `json:"host"`
}

func publishMetrics(psm *PubSubManager) {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        // TODO: Collect system metrics
        // TODO: Publish to metrics:system:* channels
        // TODO: Publish CPU, memory, disk usage
    }
}

// Connection cleanup example
func (c *Client) cleanup() {
    c.Hub.unregister <- c
    c.Conn.Close()
    log.Printf("Client %s disconnected", c.ID)
}

// Monitor pub/sub channels
func monitorChannels(client *redis.Client) {
    pubsub := client.Subscribe(context.Background(), "__keyspace@0__:*")
    defer pubsub.Close()

    ch := pubsub.Channel()
    for msg := range ch {
        log.Printf("Channel activity: %s -> %s", msg.Channel, msg.Payload)
    }
}

// Check active channels
info, err := client.PubSubChannels(ctx, "*").Result()
log.Printf("Active channels: %v", info)

// Connection management
type ConnectionManager struct {
    connections map[string]*Client
    mutex       sync.RWMutex
    maxConn     int
}

func (cm *ConnectionManager) AddConnection(client *Client) error {
    cm.mutex.Lock()
    defer cm.mutex.Unlock()
    
    if len(cm.connections) >= cm.maxConn {
        return fmt.Errorf("maximum connections reached")
    }
    
    cm.connections[client.ID] = client
    return nil
}

// Message rate limiting
type RateLimiter struct {
    limits map[string]int // messages per second per channel
    counts map[string]int
    mutex  sync.RWMutex
}

func (rl *RateLimiter) Allow(channel string) bool {
    rl.mutex.Lock()
    defer rl.mutex.Unlock()
    
    limit, exists := rl.limits[channel]
    if !exists {
        return true
    }
    
    count := rl.counts[channel]
    if count >= limit {
        return false
    }
    
    rl.counts[channel] = count + 1
    return true
}