go-redis/push_notifications.go

package redis

import (
	"context"
	"fmt"
	"sync"

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

// PushNotificationHandler defines the interface for handling push notifications.
type PushNotificationHandler interface {
	// HandlePushNotification processes a push notification.
	// Returns true if the notification was handled, false otherwise.
	HandlePushNotification(ctx context.Context, notification []interface{}) bool
}

// PushNotificationRegistry manages handlers for different types of push notifications.
type PushNotificationRegistry struct {
	mu        sync.RWMutex
	handlers  map[string]PushNotificationHandler // pushNotificationName -> single handler
	protected map[string]bool                    // pushNotificationName -> protected flag
}

// NewPushNotificationRegistry creates a new push notification registry.
func NewPushNotificationRegistry() *PushNotificationRegistry {
	return &PushNotificationRegistry{
		handlers:  make(map[string]PushNotificationHandler),
		protected: make(map[string]bool),
	}
}

// RegisterHandler registers a handler for a specific push notification name.
// Returns an error if a handler is already registered for this push notification name.
// If protected is true, the handler cannot be unregistered.
func (r *PushNotificationRegistry) RegisterHandler(pushNotificationName string, handler PushNotificationHandler, protected bool) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if _, exists := r.handlers[pushNotificationName]; exists {
		return fmt.Errorf("handler already registered for push notification: %s", pushNotificationName)
	}
	r.handlers[pushNotificationName] = handler
	r.protected[pushNotificationName] = protected
	return nil
}

// UnregisterHandler removes the handler for a specific push notification name.
// Returns an error if the handler is protected.
func (r *PushNotificationRegistry) UnregisterHandler(pushNotificationName string) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.protected[pushNotificationName] {
		return fmt.Errorf("cannot unregister protected handler for push notification: %s", pushNotificationName)
	}

	delete(r.handlers, pushNotificationName)
	delete(r.protected, pushNotificationName)
	return nil
}

// HandleNotification processes a push notification by calling the registered handler.
func (r *PushNotificationRegistry) HandleNotification(ctx context.Context, notification []interface{}) bool {
	if len(notification) == 0 {
		return false
	}

	// Extract push notification name from notification
	pushNotificationName, ok := notification[0].(string)
	if !ok {
		return false
	}

	r.mu.RLock()
	defer r.mu.RUnlock()

	// Call specific handler
	if handler, exists := r.handlers[pushNotificationName]; exists {
		return handler.HandlePushNotification(ctx, notification)
	}

	return false
}

// GetRegisteredPushNotificationNames returns a list of push notification names that have registered handlers.
func (r *PushNotificationRegistry) GetRegisteredPushNotificationNames() []string {
	r.mu.RLock()
	defer r.mu.RUnlock()

	names := make([]string, 0, len(r.handlers))
	for name := range r.handlers {
		names = append(names, name)
	}
	return names
}

// HasHandlers returns true if there are any handlers registered.
func (r *PushNotificationRegistry) HasHandlers() bool {
	r.mu.RLock()
	defer r.mu.RUnlock()

	return len(r.handlers) > 0
}

// PushNotificationProcessor handles the processing of push notifications from Redis.
type PushNotificationProcessor struct {
	registry *PushNotificationRegistry
	enabled  bool
	mu       sync.RWMutex // Protects enabled field
}

// NewPushNotificationProcessor creates a new push notification processor.
func NewPushNotificationProcessor(enabled bool) *PushNotificationProcessor {
	return &PushNotificationProcessor{
		registry: NewPushNotificationRegistry(),
		enabled:  enabled,
	}
}

// IsEnabled returns whether push notification processing is enabled.
func (p *PushNotificationProcessor) IsEnabled() bool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.enabled
}

// SetEnabled enables or disables push notification processing.
func (p *PushNotificationProcessor) SetEnabled(enabled bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.enabled = enabled
}

// GetRegistry returns the push notification registry.
func (p *PushNotificationProcessor) GetRegistry() *PushNotificationRegistry {
	return p.registry
}

// ProcessPendingNotifications checks for and processes any pending push notifications.
func (p *PushNotificationProcessor) ProcessPendingNotifications(ctx context.Context, rd *proto.Reader) error {
	if !p.IsEnabled() || !p.registry.HasHandlers() {
		return nil
	}

	// Check if there are any buffered bytes that might contain push notifications
	if rd.Buffered() == 0 {
		return nil
	}

	// Process any pending push notifications
	for {
		// Peek at the next reply type to see if it's a push notification
		replyType, err := rd.PeekReplyType()
		if err != nil {
			// No more data available or error peeking
			break
		}

		// Check if this is a RESP3 push notification
		if replyType == '>' { // RespPush
			// Read the push notification
			reply, err := rd.ReadReply()
			if err != nil {
				internal.Logger.Printf(ctx, "push: error reading push notification: %v", err)
				break
			}

			// Process the push notification
			if pushSlice, ok := reply.([]interface{}); ok && len(pushSlice) > 0 {
				handled := p.registry.HandleNotification(ctx, pushSlice)
				if handled {
					internal.Logger.Printf(ctx, "push: processed push notification: %v", pushSlice[0])
				} else {
					internal.Logger.Printf(ctx, "push: unhandled push notification: %v", pushSlice[0])
				}
			} else {
				internal.Logger.Printf(ctx, "push: invalid push notification format: %v", reply)
			}
		} else {
			// Not a push notification, stop processing
			break
		}
	}

	return nil
}

// RegisterHandler is a convenience method to register a handler for a specific push notification name.
// Returns an error if a handler is already registered for this push notification name.
// If protected is true, the handler cannot be unregistered.
func (p *PushNotificationProcessor) RegisterHandler(pushNotificationName string, handler PushNotificationHandler, protected bool) error {
	return p.registry.RegisterHandler(pushNotificationName, handler, protected)
}

// Redis Cluster push notification names
const (
	PushNotificationMoving     = "MOVING"
	PushNotificationMigrating  = "MIGRATING"
	PushNotificationMigrated   = "MIGRATED"
	PushNotificationFailingOver = "FAILING_OVER"
	PushNotificationFailedOver  = "FAILED_OVER"
)

// PushNotificationInfo contains metadata about a push notification.
type PushNotificationInfo struct {
	Name string
	Args []interface{}
}

// ParsePushNotificationInfo extracts information from a push notification.
func ParsePushNotificationInfo(notification []interface{}) *PushNotificationInfo {
	if len(notification) == 0 {
		return nil
	}

	name, ok := notification[0].(string)
	if !ok {
		return nil
	}

	return &PushNotificationInfo{
		Name: name,
		Args: notification[1:],
	}
}

// String returns a string representation of the push notification info.
func (info *PushNotificationInfo) String() string {
	if info == nil {
		return "<nil>"
	}
	return info.Name
}