package websocket

import (
	"primesys_command_producer/src/logger"
)

type Hub struct {
	// Registered clients, a user can have more than one open connection
	clients map[*Client]bool

	register chan *Client

	unregister chan *Client

	// messages to be delivered to every connection of a user
	toUser chan userMessage

	// kafka topic in which command requests are pushed
	topic string
}

type userMessage struct {
	userId string
	data   []byte
}

// socketHub is shared so queue processing can reach connected users
var socketHub = newHub()

func newHub() *Hub {
	return &Hub{
		clients:    make(map[*Client]bool),
		register:   make(chan *Client),
		unregister: make(chan *Client),
		toUser:     make(chan userMessage),
	}
}

// sendToUser delivers data to every open connection of the user
func (h *Hub) sendToUser(userId string, data []byte) {
	h.toUser <- userMessage{userId: userId, data: data}
}

func (h *Hub) run() {
	for {
		select {
		case client := <-h.register:
			h.clients[client] = true
			logger.Info("Client registered userId=%s total=%d", client.UserID, len(h.clients))
		case client := <-h.unregister:
			if _, ok := h.clients[client]; ok {
				delete(h.clients, client)
				close(client.send)
				logger.Info("Client unregistered userId=%s total=%d", client.UserID, len(h.clients))
			}
		case msg := <-h.toUser:
			delivered := false
			for client := range h.clients {
				if client.UserID != msg.userId {
					continue
				}
				select {
				case client.send <- msg.data:
					delivered = true
				default:
					logger.Warn("userId=%s send buffer full, response dropped", client.UserID)
				}
			}
			if !delivered {
				logger.Warn("userId=%s not connected, response not delivered", msg.userId)
			}
		}
	}
}