Add timestamp to protobuf and only send unseen messages to clients

2025-04-16 11:49:49 +02:00
parent f9ec811b10
commit c720b66818
4 changed files with 260 additions and 179 deletions
@@ -24,9 +24,10 @@ const (

 type Player struct {
 	sync.Mutex
-	ID       int
-	X, Y     int
-	Username string
+	ID                   int
+	X, Y                 int
+	Username             string
+	LastSeenMsgTimestamp int64 // Track the last message timestamp this player has seen
 }

 var (
@@ -82,7 +83,10 @@ func main() {
 }

 func handleConnection(conn net.Conn) {
-	defer conn.Close()
+	defer func() {
+		conn.Close()
+		log.Printf("Connection closed and cleanup complete")
+	}()

 	// Get client IP
 	remoteAddr := conn.RemoteAddr().String()
@@ -188,29 +192,57 @@ func handleConnection(conn net.Conn) {
 		return
 	}

-	player := &Player{
-		ID:       playerID,
-		X:        x,
-		Y:        y,
-		Username: username,
+	log.Printf("Player %d (%s) authenticated successfully, checking for existing session", playerID, username)
+
+	// Check for existing session and force disconnect if needed
+	mu.Lock()
+	existingPlayer, alreadyLoggedIn := players[playerID]
+	if alreadyLoggedIn {
+		log.Printf("Player %d (%s) is already logged in, forcing disconnect of old session", playerID, username)
+		// An existing session is found - clean it up
+		if oldConn, exists := playerConns[playerID]; exists {
+			// Try to close the old connection
+			oldConn.Close()
+			delete(playerConns, playerID)
+		}
+		// Keep the player object but update its connection
+		existingPlayer.X = x
+		existingPlayer.Y = y
+		playerConns[playerID] = conn
+		mu.Unlock()
+	} else {
+		// Create a new player
+		player := &Player{
+			ID:                   playerID,
+			X:                    x,
+			Y:                    y,
+			Username:             username,
+			LastSeenMsgTimestamp: 0, // Initialize to 0 to receive all messages initially
+		}
+		players[playerID] = player
+		playerConns[playerID] = conn
+		mu.Unlock()
+		existingPlayer = player
+
+		// Announce connection
+		addSystemMessage(fmt.Sprintf("%s connected", username))
 	}

-	// Prevent multiple logins
-	mu.Lock()
-	for _, p := range players {
-		if p.Username == username {
-			mu.Unlock()
-			response := &pb.ServerMessage{
-				AuthSuccess:  false,
-				ErrorMessage: "Account already logged in",
+	// Ensure player state is saved on any kind of disconnect
+	defer func() {
+		if p, exists := players[playerID]; exists {
+			if err := db.SavePlayerState(playerID, p.X, p.Y); err != nil {
+				log.Printf("Error saving state for player %d: %v", playerID, err)
 			}
-			writeMessage(conn, response)
-			return
 		}
-	}
-	players[playerID] = player
-	playerConns[playerID] = conn
-	mu.Unlock()
+		addSystemMessage(fmt.Sprintf("%s disconnected", username))
+		mu.Lock()
+		delete(players, playerID)
+		delete(playerConns, playerID)
+		delete(actionQueue, playerID)
+		mu.Unlock()
+		log.Printf("Player %d (%s) disconnected", playerID, username)
+	}()

 	// Send initial state with correct position
 	response = &pb.ServerMessage{
@@ -225,29 +257,13 @@ func handleConnection(conn net.Conn) {
 		ProtocolVersion: protoVersion,
 	}

-	addSystemMessage(fmt.Sprintf("%s connected", username))
-
-	// Ensure player state is saved on any kind of disconnect
-	defer func() {
-		if err := db.SavePlayerState(playerID, player.X, player.Y); err != nil {
-			log.Printf("Error saving state for player %d: %v", playerID, err)
-		}
-		addSystemMessage(fmt.Sprintf("%s disconnected", player.Username))
-		mu.Lock()
-		delete(players, playerID)
-		delete(playerConns, playerID)
-		delete(actionQueue, playerID)
-		mu.Unlock()
-		log.Printf("Player %d disconnected", playerID)
-	}()
-
 	// Send player ID to client
 	if err := writeMessage(conn, response); err != nil {
 		log.Printf("Failed to send player ID: %v", err)
 		return
 	}

-	fmt.Printf("Player %d connected\n", playerID)
+	log.Printf("Player %d (%s) connected successfully", playerID, username)

 	// Listen for incoming actions from this player
 	for {
@@ -276,6 +292,11 @@ func handleConnection(conn net.Conn) {
 			continue
 		}

+		// Update the last seen message timestamp
+		if batch.LastSeenMessageTimestamp > 0 {
+			existingPlayer.LastSeenMsgTimestamp = batch.LastSeenMessageTimestamp
+		}
+
 		// Queue the actions for processing
 		if batch.PlayerId == int32(playerID) {
 			for _, action := range batch.Actions {
@@ -284,7 +305,9 @@ func handleConnection(conn net.Conn) {
 					return
 				}
 			}
+			mu.Lock()
 			actionQueue[playerID] = append(actionQueue[playerID], batch.Actions...)
+			mu.Unlock()
 		}
 	}
 }
@@ -330,11 +353,30 @@ func addSystemMessage(content string) {

 func processActions() {
 	mu.Lock()
-	defer mu.Unlock()
+	// Make a list of players to process first, to avoid lock contention
+	activePlayers := make(map[int]*Player)
+	for id, p := range players {
+		activePlayers[id] = p
+	}
+	activeConns := make(map[int]net.Conn)
+	for id, conn := range playerConns {
+		activeConns[id] = conn
+	}
+	activeQueues := make(map[int][]*pb.Action)
+	for id, actions := range actionQueue {
+		if len(actions) > 0 {
+			activeQueues[id] = actions
+			actionQueue[id] = nil // Clear the queue early to avoid double processing
+		}
+	}
+	mu.Unlock()

-	// Update players based on queued actions
-	for playerID, actions := range actionQueue {
-		player := players[playerID]
+	// Process actions without holding the global lock
+	for playerID, actions := range activeQueues {
+		player, exists := activePlayers[playerID]
+		if !exists {
+			continue
+		}
 		player.Lock()
 		for _, action := range actions {
 			switch action.Type {
@@ -348,20 +390,21 @@ func processActions() {
 			}
 		}
 		player.Unlock()
-		actionQueue[playerID] = nil // Clear the action queue after processing
 	}

-	// Prepare and broadcast the current game state
+	// Prepare current game state
 	currentTick := time.Now().UnixNano() / int64(tickRate)
-	state := &pb.ServerMessage{
-		CurrentTick: currentTick,
-		Players:     make([]*pb.PlayerState, 0, len(players)),
-	}

-	// Convert players to PlayerState
-	for id, p := range players {
+	// Get recent messages for new connections
+	chatMutex.RLock()
+	recentMessages := chatHistory[max(0, len(chatHistory)-5):] // Get last 5 for new connections
+	chatMutex.RUnlock()
+
+	// To avoid holding locks too long, prepare player states first
+	playerStates := make([]*pb.PlayerState, 0, len(activePlayers))
+	for id, p := range activePlayers {
 		p.Lock()
-		state.Players = append(state.Players, &pb.PlayerState{
+		playerStates = append(playerStates, &pb.PlayerState{
 			PlayerId: int32(id),
 			X:        int32(p.X),
 			Y:        int32(p.Y),
@@ -370,15 +413,69 @@ func processActions() {
 		p.Unlock()
 	}

-	// Add chat messages to the state
-	chatMutex.RLock()
-	state.ChatMessages = chatHistory[max(0, len(chatHistory)-5):] // Only send last 5 messages
-	chatMutex.RUnlock()
+	// Now send updates to each player
+	for playerID, conn := range activeConns {
+		player, exists := activePlayers[playerID]
+		if !exists {
+			continue
+		}

-	// Send to each connected player
-	for _, conn := range playerConns {
+		state := &pb.ServerMessage{
+			CurrentTick: currentTick,
+			Players:     playerStates,
+		}
+
+		// Add chat messages - only send those the player hasn't seen
+		player.Lock()
+		lastSeen := player.LastSeenMsgTimestamp
+		player.Unlock()
+
+		chatMutex.RLock()
+		var newMessages []*pb.ChatMessage
+
+		// For new connections, send the 5 most recent messages
+		if lastSeen == 0 && len(recentMessages) > 0 {
+			newMessages = recentMessages
+			if len(newMessages) > 0 {
+				// Update the player's timestamp to the latest message
+				player.Lock()
+				player.LastSeenMsgTimestamp = newMessages[len(newMessages)-1].Timestamp
+				player.Unlock()
+			}
+		} else {
+			// For existing connections, only send new messages
+			for _, msg := range chatHistory {
+				if msg.Timestamp > lastSeen {
+					newMessages = append(newMessages, msg)
+				}
+			}
+
+			// Update the player's timestamp if we sent them new messages
+			if len(newMessages) > 0 {
+				player.Lock()
+				player.LastSeenMsgTimestamp = newMessages[len(newMessages)-1].Timestamp
+				player.Unlock()
+			}
+		}
+
+		state.ChatMessages = newMessages
+		chatMutex.RUnlock()
+
+		// Log the number of messages we're sending
+		if len(newMessages) > 0 {
+			log.Printf("Sending %d new messages to player %d", len(newMessages), playerID)
+		}
+
+		// Send the state to the player - do this without holding any locks
 		if err := writeMessage(conn, state); err != nil {
-			log.Printf("Failed to send update: %v", err)
+			log.Printf("Failed to send update to player %d: %v", playerID, err)
+
+			// Handle connection errors by removing the player
+			mu.Lock()
+			delete(players, playerID)
+			delete(playerConns, playerID)
+			delete(actionQueue, playerID)
+			mu.Unlock()
 		}
 	}
 }