package websocket import ( "context" "fmt" "strings" "time" ) // DatabaseHandler menangani operasi database type DatabaseHandler struct { hub *Hub } func NewDatabaseHandler(hub *Hub) *DatabaseHandler { return &DatabaseHandler{hub: hub} } func (h *DatabaseHandler) HandleMessage(client *Client, message WebSocketMessage) error { switch message.Type { case DatabaseInsertMessage: return h.handleDatabaseInsert(client, message) case DatabaseQueryMessage: return h.handleDatabaseQuery(client, message) case DatabaseCustomQueryMessage: return h.handleDatabaseCustomQuery(client, message) default: return fmt.Errorf("unsupported database message type: %s", message.Type) } } func (h *DatabaseHandler) MessageType() MessageType { return DatabaseInsertMessage // Primary type for registration } func (h *DatabaseHandler) handleDatabaseInsert(client *Client, message WebSocketMessage) error { data, ok := message.Data.(map[string]interface{}) if !ok { client.sendErrorResponse("Invalid database insert format", "Data must be an object") return nil } table, ok := data["m_deviceqris"].(string) if !ok || table == "" { client.sendErrorResponse("Invalid table name", "table is required") return nil } insertData, ok := data["data"].(map[string]interface{}) if !ok { client.sendErrorResponse("Invalid insert data", "data must be an object") return nil } // Perform actual database insert if h.hub.dbService != nil { // Get database connection db, err := h.hub.GetDatabaseConnection("simrs_prod") if err != nil { client.sendErrorResponse("Database connection error", err.Error()) return nil } // Build insert query columns := make([]string, 0, len(insertData)) values := make([]interface{}, 0, len(insertData)) placeholders := make([]string, 0, len(insertData)) i := 1 for col, val := range insertData { columns = append(columns, col) values = append(values, val) placeholders = append(placeholders, fmt.Sprintf("$%d", i)) i++ } query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", table, strings.Join(columns, ", "), strings.Join(placeholders, ", ")) // Execute insert ctx, cancel := context.WithTimeout(client.ctx, 30*time.Second) defer cancel() result, err := db.ExecContext(ctx, query, values...) if err != nil { client.sendErrorResponse("Database insert error", err.Error()) return nil } rowsAffected, _ := result.RowsAffected() client.sendDirectResponse("db_insert_result", map[string]interface{}{ "table": table, "status": "success", "rows_affected": rowsAffected, "message": "Data inserted successfully", }) } else { client.sendErrorResponse("Database service not available", "Database service is not initialized") } return nil } func (h *DatabaseHandler) handleDatabaseQuery(client *Client, message WebSocketMessage) error { data, ok := message.Data.(map[string]interface{}) if !ok { client.sendErrorResponse("Invalid database query format", "Data must be an object") return nil } table, ok := data["m_deviceqris"].(string) if !ok || table == "" { client.sendErrorResponse("Invalid table name", "table is required") return nil } // Execute query results, err := h.hub.ExecuteDatabaseQuery("simrs_prod", fmt.Sprintf("SELECT * FROM %s LIMIT 100", table)) if err != nil { client.sendErrorResponse("Database query error", err.Error()) return nil } client.sendDirectResponse("db_query_result", map[string]interface{}{ "table": table, "status": "success", "results": results, "count": len(results), }) return nil } func (h *DatabaseHandler) handleDatabaseCustomQuery(client *Client, message WebSocketMessage) error { data, ok := message.Data.(map[string]interface{}) if !ok { client.sendErrorResponse("Invalid database query format", "Data must be an object") return nil } database, ok := data["database"].(string) if !ok || database == "" { database = "simrs_prod" } query, ok := data["query"].(string) if !ok || query == "" { client.sendErrorResponse("Invalid query", "query is required") return nil } // Execute custom query results, err := h.hub.ExecuteDatabaseQuery(database, query) if err != nil { client.sendErrorResponse("Database query error", err.Error()) return nil } client.sendDirectResponse("db_query_result", map[string]interface{}{ "database": database, "query": query, "status": "success", "results": results, "count": len(results), }) return nil } // AdminHandler menangani operasi admin type AdminHandler struct { hub *Hub } func NewAdminHandler(hub *Hub) *AdminHandler { return &AdminHandler{hub: hub} } func (h *AdminHandler) HandleMessage(client *Client, message WebSocketMessage) error { switch message.Type { case AdminKickClientMessage: return h.handleAdminKickClient(client, message) case AdminKillServerMessage: return h.handleAdminKillServer(client, message) case AdminClearLogsMessage: return h.handleAdminClearLogs(client, message) default: return fmt.Errorf("unsupported admin message type: %s", message.Type) } } func (h *AdminHandler) MessageType() MessageType { return AdminKickClientMessage // Primary type for registration } func (h *AdminHandler) handleAdminKickClient(client *Client, message WebSocketMessage) error { data, ok := message.Data.(map[string]interface{}) if !ok { client.sendErrorResponse("Invalid admin command format", "Data must be an object") return nil } targetClientID, ok := data["client_id"].(string) if !ok || targetClientID == "" { client.sendErrorResponse("Invalid target client ID", "client_id is required") return nil } // Check if target client exists h.hub.mu.RLock() targetClient, exists := h.hub.clientsByID[targetClientID] h.hub.mu.RUnlock() if !exists { client.sendErrorResponse("Client not found", fmt.Sprintf("Client %s not found", targetClientID)) return nil } // Log activity h.hub.logActivity("admin_kick_client", client.ID, fmt.Sprintf("Kicked client: %s", targetClientID)) // Disconnect the target client targetClient.cancel() targetClient.Conn.Close() client.sendDirectResponse("admin_command_result", map[string]interface{}{ "command": "kick_client", "target_client_id": targetClientID, "status": "success", "message": "Client kicked successfully", }) return nil } func (h *AdminHandler) handleAdminKillServer(client *Client, message WebSocketMessage) error { // Log activity h.hub.logActivity("admin_kill_server", client.ID, "Server kill signal received") // For testing purposes, just acknowledge (don't actually kill server) client.sendDirectResponse("admin_command_result", map[string]interface{}{ "command": "kill_server", "status": "acknowledged", "message": "Kill server signal received (test mode - server not actually killed)", }) return nil } func (h *AdminHandler) handleAdminClearLogs(client *Client, message WebSocketMessage) error { // Clear activity logs h.hub.activityMu.Lock() h.hub.activityLog = make([]ActivityLog, 0, h.hub.config.ActivityLogSize) h.hub.activityMu.Unlock() // Log the clear action h.hub.logActivity("admin_clear_logs", client.ID, "Activity logs cleared") client.sendDirectResponse("admin_command_result", map[string]interface{}{ "command": "clear_logs", "status": "success", "message": "Server logs cleared successfully", }) return nil } // RoomHandler menangani operasi ruangan type RoomHandler struct { hub *Hub } func NewRoomHandler(hub *Hub) *RoomHandler { return &RoomHandler{hub: hub} } func (h *RoomHandler) HandleMessage(client *Client, message WebSocketMessage) error { switch message.Type { case JoinRoomMessage: return h.handleJoinRoom(client, message) case LeaveRoomMessage: return h.handleLeaveRoom(client, message) case GetRoomInfoMessage: return h.handleGetRoomInfo(client, message) default: return fmt.Errorf("unsupported room message type: %s", message.Type) } } func (h *RoomHandler) MessageType() MessageType { return JoinRoomMessage // Primary type for registration } func (h *RoomHandler) handleJoinRoom(client *Client, message WebSocketMessage) error { data, ok := message.Data.(map[string]interface{}) if !ok { client.sendErrorResponse("Invalid join room format", "Data must be an object") return nil } roomName, ok := data["room"].(string) if !ok || roomName == "" { client.sendErrorResponse("Invalid room name", "room is required") return nil } // Update client room oldRoom := client.Room client.Room = roomName // Update hub room mapping h.hub.mu.Lock() if oldRoom != "" { if roomClients, exists := h.hub.rooms[oldRoom]; exists { delete(roomClients, client) if len(roomClients) == 0 { delete(h.hub.rooms, oldRoom) } } } if h.hub.rooms[roomName] == nil { h.hub.rooms[roomName] = make(map[*Client]bool) } h.hub.rooms[roomName][client] = true h.hub.mu.Unlock() // Log activity h.hub.logActivity("client_join_room", client.ID, fmt.Sprintf("Room: %s", roomName)) // Notify other clients in the room h.hub.broadcaster.BroadcastServerToRoom(roomName, "user_joined_room", map[string]interface{}{ "client_id": client.ID, "user_id": client.UserID, "room": roomName, "joined_at": time.Now().Unix(), }) // Send response client.sendDirectResponse("room_joined", map[string]interface{}{ "room": roomName, "previous_room": oldRoom, }) return nil } func (h *RoomHandler) handleLeaveRoom(client *Client, message WebSocketMessage) error { oldRoom := client.Room if oldRoom == "" { client.sendErrorResponse("Not in any room", "Client is not currently in a room") return nil } // Update hub room mapping h.hub.mu.Lock() if roomClients, exists := h.hub.rooms[oldRoom]; exists { delete(roomClients, client) if len(roomClients) == 0 { delete(h.hub.rooms, oldRoom) } } h.hub.mu.Unlock() // Clear client room client.Room = "" // Log activity h.hub.logActivity("client_leave_room", client.ID, fmt.Sprintf("Room: %s", oldRoom)) // Notify other clients in the room h.hub.broadcaster.BroadcastServerToRoom(oldRoom, "user_left_room", map[string]interface{}{ "client_id": client.ID, "user_id": client.UserID, "room": oldRoom, "left_at": time.Now().Unix(), }) // Send response client.sendDirectResponse("room_left", map[string]interface{}{ "room": oldRoom, }) return nil } func (h *RoomHandler) handleGetRoomInfo(client *Client, message WebSocketMessage) error { h.hub.mu.RLock() defer h.hub.mu.RUnlock() roomInfo := make(map[string]interface{}) // Info ruangan saat ini if client.Room != "" { if roomClients, exists := h.hub.rooms[client.Room]; exists { clientIDs := make([]string, 0, len(roomClients)) for client := range roomClients { clientIDs = append(clientIDs, client.ID) } roomInfo["current_room"] = map[string]interface{}{ "name": client.Room, "client_count": len(roomClients), "clients": clientIDs, } } } // Info semua ruangan allRooms := make(map[string]int) for roomName, clients := range h.hub.rooms { allRooms[roomName] = len(clients) } roomInfo["all_rooms"] = allRooms roomInfo["total_rooms"] = len(h.hub.rooms) client.sendDirectResponse("room_info", roomInfo) return nil } // MonitoringHandler menangani operasi monitoring type MonitoringHandler struct { hub *Hub } func NewMonitoringHandler(hub *Hub) *MonitoringHandler { return &MonitoringHandler{hub: hub} } func (h *MonitoringHandler) HandleMessage(client *Client, message WebSocketMessage) error { switch message.Type { case GetStatsMessage: return h.handleGetStats(client, message) case GetServerStatsMessage: return h.handleGetServerStats(client, message) case GetSystemHealthMessage: return h.handleGetSystemHealth(client, message) default: return fmt.Errorf("unsupported monitoring message type: %s", message.Type) } } func (h *MonitoringHandler) MessageType() MessageType { return GetStatsMessage // Primary type for registration } func (h *MonitoringHandler) handleGetStats(client *Client, message WebSocketMessage) error { stats := h.hub.GetStats() client.sendDirectResponse("stats", stats) return nil } func (h *MonitoringHandler) handleGetServerStats(client *Client, message WebSocketMessage) error { // Create monitoring manager instance monitoringManager := NewMonitoringManager(h.hub) detailedStats := monitoringManager.GetDetailedStats() client.sendDirectResponse("server_stats", detailedStats) return nil } func (h *MonitoringHandler) handleGetSystemHealth(client *Client, message WebSocketMessage) error { systemHealth := make(map[string]interface{}) if h.hub.dbService != nil { systemHealth["databases"] = h.hub.dbService.Health() systemHealth["available_dbs"] = h.hub.dbService.ListDBs() } systemHealth["websocket_status"] = "healthy" systemHealth["uptime_seconds"] = time.Since(h.hub.startTime).Seconds() client.sendDirectResponse("system_health", systemHealth) return nil } // ConnectionHandler menangani operasi koneksi type ConnectionHandler struct { hub *Hub } func NewConnectionHandler(hub *Hub) *ConnectionHandler { return &ConnectionHandler{hub: hub} } func (h *ConnectionHandler) HandleMessage(client *Client, message WebSocketMessage) error { switch message.Type { case PingMessage: return h.handlePing(client, message) case PongMessage: return h.handlePong(client, message) case HeartbeatMessage: return h.handleHeartbeat(client, message) case ConnectionTestMessage: return h.handleConnectionTest(client, message) case OnlineUsersMessage: return h.handleGetOnlineUsers(client, message) default: return fmt.Errorf("unsupported connection message type: %s", message.Type) } } func (h *ConnectionHandler) MessageType() MessageType { return PingMessage // Primary type for registration } func (h *ConnectionHandler) handlePing(client *Client, message WebSocketMessage) error { // Send pong response pongMsg := NewWebSocketMessage(PongMessage, map[string]interface{}{ "timestamp": message.Timestamp.Unix(), "client_id": client.ID, }, client.ID, "") select { case client.Send <- pongMsg: default: // Channel penuh, abaikan } return nil } func (h *ConnectionHandler) handlePong(client *Client, message WebSocketMessage) error { // Pong sudah ditangani di level koneksi, tapi kita bisa log aktivitas h.hub.logActivity("pong_received", client.ID, "Pong message received") return nil } func (h *ConnectionHandler) handleHeartbeat(client *Client, message WebSocketMessage) error { // Send heartbeat acknowledgment heartbeatAck := NewWebSocketMessage(MessageType("heartbeat_ack"), map[string]interface{}{ "timestamp": time.Now().Unix(), "client_uptime": time.Since(client.connectedAt).Seconds(), "server_uptime": time.Since(h.hub.startTime).Seconds(), "client_id": client.ID, }, client.ID, "") select { case client.Send <- heartbeatAck: default: // Channel penuh, abaikan } return nil } func (h *ConnectionHandler) handleConnectionTest(client *Client, message WebSocketMessage) error { // Send connection test result testResult := NewWebSocketMessage(MessageType("connection_test_result"), map[string]interface{}{ "timestamp": time.Now().Unix(), "client_id": client.ID, "connection_status": "healthy", "latency_ms": 0, // Could be calculated if ping timestamp is provided "uptime_seconds": time.Since(client.connectedAt).Seconds(), }, client.ID, "") select { case client.Send <- testResult: default: // Channel penuh, abaikan } return nil } func (h *ConnectionHandler) handleGetOnlineUsers(client *Client, message WebSocketMessage) error { h.hub.mu.RLock() defer h.hub.mu.RUnlock() users := make([]map[string]interface{}, 0, len(h.hub.clients)) for client := range h.hub.clients { user := map[string]interface{}{ "id": client.ID, "static_id": client.StaticID, "user_id": client.UserID, "room": client.Room, "ip_address": client.IPAddress, "connected_at": client.connectedAt, "last_ping": client.lastPing, "last_pong": client.lastPong, "is_active": client.isClientActive(), } users = append(users, user) } response := NewWebSocketMessage(OnlineUsersMessage, map[string]interface{}{ "users": users, "count": len(users), }, client.ID, "") select { case client.Send <- response: default: // Channel penuh, abaikan } return nil } // SetupDefaultHandlers sets up all default message handlers func SetupDefaultHandlers(registry *MessageRegistry, hub *Hub) { // Register database handler dbHandler := NewDatabaseHandler(hub) registry.RegisterHandler(dbHandler) // Register admin handler adminHandler := NewAdminHandler(hub) registry.RegisterHandler(adminHandler) // Register room handler roomHandler := NewRoomHandler(hub) registry.RegisterHandler(roomHandler) // Register monitoring handler monitoringHandler := NewMonitoringHandler(hub) registry.RegisterHandler(monitoringHandler) // Register connection handler connectionHandler := NewConnectionHandler(hub) registry.RegisterHandler(connectionHandler) }