Add lobby history retrieval protocol: probe/request/data items, chunked transfer, local dedup & merge

Author: jolavillette

src/chat/distributedchat.cc

@@ -443,6 +443,10 @@ bool DistributedChatService::handleRecvItem(RsChatItem *item)
 		case RS_PKT_SUBTYPE_CHAT_LOBBY_UNSUBSCRIBE:       handleFriendUnsubscribeLobby     (dynamic_cast<RsChatLobbyUnsubscribeItem      *>(item)) ; break ;
 		case RS_PKT_SUBTYPE_CHAT_LOBBY_LIST_REQUEST:      handleRecvChatLobbyListRequest   (dynamic_cast<RsChatLobbyListRequestItem      *>(item)) ; break ;
 		case RS_PKT_SUBTYPE_CHAT_LOBBY_LIST:              handleRecvChatLobbyList          (dynamic_cast<RsChatLobbyListItem             *>(item)) ; break ;
+		case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_PROBE:      handleRecvLobbyHistoryProbe        (dynamic_cast<RsChatLobbyHistoryProbeItem         *>(item)) ; break ;
+		case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_PROBE_RESP: handleRecvLobbyHistoryProbeResponse(dynamic_cast<RsChatLobbyHistoryProbeResponseItem *>(item)) ; break ;
+		case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_REQUEST:    handleRecvLobbyHistoryRequest      (dynamic_cast<RsChatLobbyHistoryRequestItem       *>(item)) ; break ;
+		case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_DATA:       handleRecvLobbyHistoryData         (dynamic_cast<RsChatLobbyHistoryDataItem          *>(item)) ; break ;
 		default:                                          return false ;
 	}
 	return true ;
@@ -2263,3 +2267,244 @@ bool DistributedChatService::processLoadListItem(const RsItem *item)
 	return false ;
 }
 
+/***************** Lobby History Retrieval Protocol *****************/
+
+#include "chat/rschatitems.h"
+#include "pqi/p3historymgr.h"
+
+bool DistributedChatService::requestLobbyHistory(const ChatLobbyId& lobby_id)
+{
+	RsStackMutex stack(mDistributedChatMtx); /********** STACK LOCKED MTX ******/
+
+	std::map<ChatLobbyId,ChatLobbyEntry>::iterator it = _chat_lobbys.find(lobby_id) ;
+
+	if(it == _chat_lobbys.end())
+	{
+		std::cerr << "(EE) requestLobbyHistory(): lobby " << std::hex << lobby_id << std::dec << " not found." << std::endl;
+		return false ;
+	}
+
+	// Send a probe to every direct friend participating in this lobby
+
+	for(std::set<RsPeerId>::const_iterator fit(it->second.participating_friends.begin()); fit != it->second.participating_friends.end(); ++fit)
+	{
+		if(mServControl->isPeerConnected(mServType, *fit))
+		{
+			RsChatLobbyHistoryProbeItem *item = new RsChatLobbyHistoryProbeItem ;
+			item->lobby_id = lobby_id ;
+			item->PeerId(*fit) ;
+
+			sendChatItem(item) ;
+
+			std::cerr << "requestLobbyHistory(): sent probe to peer " << *fit << " for lobby " << std::hex << lobby_id << std::dec << std::endl;
+		}
+	}
+
+	return true ;
+}
+
+void DistributedChatService::handleRecvLobbyHistoryProbe(RsChatLobbyHistoryProbeItem *item)
+{
+	if(!item) return ;
+
+	// Check we are subscribed to this lobby
+	{
+		RsStackMutex stack(mDistributedChatMtx); /********** STACK LOCKED MTX ******/
+
+		if(_chat_lobbys.find(item->lobby_id) == _chat_lobbys.end())
+		{
+			std::cerr << "(WW) handleRecvLobbyHistoryProbe(): lobby " << std::hex << item->lobby_id << std::dec << " not found. Ignoring." << std::endl;
+			return ;
+		}
+	}
+
+	// Retrieve our local history for this lobby
+	std::list<HistoryMsg> msgs ;
+	mHistMgr->getMessages(ChatId(item->lobby_id), msgs, 0) ; // 0 = get all available
+
+	uint32_t available_count = (uint32_t)msgs.size() ;
+	uint32_t oldest_ts = 0 ;
+
+	if(!msgs.empty())
+		oldest_ts = msgs.front().sendTime ;
+
+	// Send response back to the requesting peer
+	RsChatLobbyHistoryProbeResponseItem *response = new RsChatLobbyHistoryProbeResponseItem ;
+	response->lobby_id = item->lobby_id ;
+	response->available_count = available_count ;
+	response->oldest_timestamp = oldest_ts ;
+	response->PeerId(item->PeerId()) ;
+
+	sendChatItem(response) ;
+
+	std::cerr << "handleRecvLobbyHistoryProbe(): responded to " << item->PeerId() << " — " << available_count << " msgs available, oldest TS=" << oldest_ts << std::endl;
+}
+
+void DistributedChatService::handleRecvLobbyHistoryProbeResponse(RsChatLobbyHistoryProbeResponseItem *item)
+{
+	if(!item) return ;
+
+	std::cerr << "handleRecvLobbyHistoryProbeResponse(): peer " << item->PeerId()
+	          << " has " << item->available_count << " messages for lobby " << std::hex << item->lobby_id << std::dec
+	          << ", oldest TS=" << item->oldest_timestamp << std::endl;
+
+	auto ev = std::make_shared<RsChatLobbyEvent>();
+	ev->mEventCode = RsChatLobbyEventCode::CHAT_LOBBY_EVENT_HISTORY_PROBE_RESPONSE;
+	ev->mPeerId       = item->PeerId() ;
+	ev->mLobbyId      = item->lobby_id ;
+	ev->mGenericCount = item->available_count ;
+	ev->mTimeShift    = item->oldest_timestamp ; // using mTimeShift to store the timestamp
+	rsEvents->postEvent(ev);
+}
+
+bool DistributedChatService::requestLobbyHistoryFromPeer(const ChatLobbyId& lobby_id, const RsPeerId& peer_id, uint32_t max_count, uint32_t oldest_ts)
+{
+	RsStackMutex stack(mDistributedChatMtx); /********** STACK LOCKED MTX ******/
+
+	std::map<ChatLobbyId,ChatLobbyEntry>::iterator it = _chat_lobbys.find(lobby_id) ;
+
+	if(it == _chat_lobbys.end())
+	{
+		std::cerr << "(EE) requestLobbyHistoryFromPeer(): lobby " << std::hex << lobby_id << std::dec << " not found." << std::endl;
+		return false ;
+	}
+
+	if(!mServControl->isPeerConnected(mServType, peer_id))
+	{
+		std::cerr << "(EE) requestLobbyHistoryFromPeer(): peer " << peer_id << " is not connected." << std::endl;
+		return false ;
+	}
+
+	RsChatLobbyHistoryRequestItem *request = new RsChatLobbyHistoryRequestItem ;
+	request->lobby_id = lobby_id ;
+	request->max_count = max_count ;
+	request->oldest_timestamp = oldest_ts ;
+	request->PeerId(peer_id) ;
+
+	sendChatItem(request) ;
+
+	std::cerr << "requestLobbyHistoryFromPeer(): sent request to peer " << peer_id << " for lobby " << std::hex << lobby_id << std::dec << " (max " << max_count << " msgs, oldest TS=" << oldest_ts << ")" << std::endl;
+	return true ;
+}
+
+void DistributedChatService::handleRecvLobbyHistoryRequest(RsChatLobbyHistoryRequestItem *item)
+{
+	if(!item) return ;
+
+	// Check we are subscribed to this lobby
+	{
+		RsStackMutex stack(mDistributedChatMtx); /********** STACK LOCKED MTX ******/
+
+		if(_chat_lobbys.find(item->lobby_id) == _chat_lobbys.end())
+		{
+			std::cerr << "(WW) handleRecvLobbyHistoryRequest(): lobby " << std::hex << item->lobby_id << std::dec << " not found. Ignoring." << std::endl;
+			return ;
+		}
+	}
+
+	// Retrieve local history
+	std::list<HistoryMsg> msgs ;
+	mHistMgr->getMessages(ChatId(item->lobby_id), msgs, item->max_count) ;
+
+	RsChatLobbyHistoryDataItem *data_item = new RsChatLobbyHistoryDataItem ;
+	data_item->lobby_id = item->lobby_id ;
+	data_item->PeerId(item->PeerId()) ;
+	size_t count = 0;
+	size_t current_chunk_size = 0;
+	const size_t MAX_CHUNK_SIZE = 120 * 1024; // ~120 KB to stay safely under 128KB limit
+
+	for(std::list<HistoryMsg>::const_iterator it(msgs.begin()); it != msgs.end(); ++it)
+	{
+		if(it->sendTime >= item->oldest_timestamp)
+		{
+			LobbyHistoryMsgEntry entry ;
+			// HistoryMsg only stores the author's nick (peerName) and peerId, not the GXS ID.
+			entry.author_id = RsGxsId() ;
+			entry.nick      = it->peerName ;
+			entry.send_time = it->sendTime ;
+			entry.message   = it->message ;
+			entry.incoming  = it->incoming ;
+
+			// Calculate approximate size of this entry
+			// author_id (std::string approx 32 bytes) + nick + timestamp (4 bytes) + message + incoming (1 byte)
+			size_t entry_size = 32 + entry.nick.size() + 4 + entry.message.size() + 1 + 50; // +50 for serialization overhead
+
+			// If adding this message exceeds the chunk size (and the chunk isn't empty), send the current chunk first
+			if (current_chunk_size + entry_size > MAX_CHUNK_SIZE && !data_item->msgs.empty()) {
+				std::cerr << "handleRecvLobbyHistoryRequest(): sending chunk of " << data_item->msgs.size() << " msgs (" << current_chunk_size << " bytes) to " << item->PeerId() << std::endl;
+				sendChatItem(data_item) ;
+				
+				// Create a new data item for the next chunk
+				data_item = new RsChatLobbyHistoryDataItem ;
+				data_item->lobby_id = item->lobby_id ;
+				data_item->PeerId(item->PeerId()) ;
+				current_chunk_size = 0;
+			}
+
+			data_item->msgs.push_back(entry) ;
+			current_chunk_size += entry_size;
+			count++;
+		}
+	}
+
+	if (!data_item->msgs.empty()) {
+		std::cerr << "handleRecvLobbyHistoryRequest(): sending final chunk of " << data_item->msgs.size() << " msgs to " << item->PeerId() << std::endl;
+		sendChatItem(data_item) ;
+	} else {
+		// Clean up the unused item if we sent everything cleanly in batches, or if there were 0 messages.
+		delete data_item;
+	}
+
+	std::cerr << "handleRecvLobbyHistoryRequest(): finished. Sent a total of " << count << " messages." << std::endl;
+}
+
+void DistributedChatService::handleRecvLobbyHistoryData(RsChatLobbyHistoryDataItem *item)
+{
+	if(!item) return ;
+
+	std::cerr << "handleRecvLobbyHistoryData(): received " << item->msgs.size() << " messages from peer " << item->PeerId()
+	          << " for lobby " << std::hex << item->lobby_id << std::dec << std::endl;
+
+	// Deduplicate and save to local history database
+	std::list<HistoryMsg> existingMsgs;
+	mHistMgr->getMessages(ChatId(item->lobby_id), existingMsgs, 0);
+	
+	std::set<std::pair<uint32_t, std::string> > existingSet;
+	for (std::list<HistoryMsg>::const_iterator it = existingMsgs.begin(); it != existingMsgs.end(); ++it) {
+		existingSet.insert(std::make_pair(it->sendTime, it->message));
+	}
+
+	int addedCount = 0;
+	for (std::vector<LobbyHistoryMsgEntry>::const_iterator it = item->msgs.begin(); it != item->msgs.end(); ++it) {
+		if (existingSet.find(std::make_pair(it->send_time, it->message)) == existingSet.end()) {
+			// Not a duplicate
+			ChatMessage cm;
+			cm.chat_id = ChatId(item->lobby_id);
+			cm.lobby_peer_gxs_id = it->author_id;
+			cm.peer_alternate_nickname = it->nick;
+			cm.chatflags = 0; 
+			cm.sendTime = it->send_time;
+			cm.recvTime = it->send_time; // Keep the original send_time for chronological sorting
+			cm.msg = it->message;
+			cm.incoming = it->incoming;
+			cm.online = true;
+
+			mHistMgr->addMessage(cm);
+			addedCount++;
+			
+			// Add to our set just in case the received chunk contains duplicates within itself
+			existingSet.insert(std::make_pair(it->send_time, it->message));
+		}
+	}
+
+	std::cerr << "handleRecvLobbyHistoryData(): merged " << addedCount << " new messages into local history." << std::endl;
+
+	auto ev = std::make_shared<RsChatLobbyEvent>();
+	ev->mEventCode = RsChatLobbyEventCode::CHAT_LOBBY_EVENT_HISTORY_DATA;
+	ev->mPeerId    = item->PeerId() ;
+	ev->mLobbyId   = item->lobby_id ;
+	ev->mHistoryMsgs = item->msgs ; // Copy the received messages to the event
+	rsEvents->postEvent(ev);
+}
+
+

src/chat/distributedchat.h

@@ -46,6 +46,11 @@ class RsChatItem ;
 class RsChatMsgItem ;
 class RsGixs ;
 
+class RsChatLobbyHistoryProbeItem ;
+class RsChatLobbyHistoryProbeResponseItem ;
+class RsChatLobbyHistoryRequestItem ;
+class RsChatLobbyHistoryDataItem ;
+
 class DistributedChatService
 {
 	public:
@@ -79,6 +84,10 @@ class DistributedChatService
 		void getListOfNearbyChatLobbies(std::vector<VisibleChatLobbyRecord>& public_lobbies) ;
 		bool joinVisibleChatLobby(const ChatLobbyId& id, const RsGxsId &gxs_id) ;
 
+		// Lobby history retrieval protocol
+		bool requestLobbyHistory(const ChatLobbyId& lobby_id) ;
+		bool requestLobbyHistoryFromPeer(const ChatLobbyId& lobby_id, const RsPeerId& peer_id, uint32_t max_count, uint32_t oldest_ts) ;
+
 	protected:
 		bool handleRecvItem(RsChatItem *) ;
 
@@ -125,6 +134,12 @@ class DistributedChatService
 		void sendLobbyStatusNewPeer(const ChatLobbyId& lobby_id) ;
 		void sendLobbyStatusKeepAlive(const ChatLobbyId&) ;
 
+		// Lobby history retrieval handlers
+		void handleRecvLobbyHistoryProbe(RsChatLobbyHistoryProbeItem *item) ;
+		void handleRecvLobbyHistoryProbeResponse(RsChatLobbyHistoryProbeResponseItem *item) ;
+		void handleRecvLobbyHistoryRequest(RsChatLobbyHistoryRequestItem *item) ;
+		void handleRecvLobbyHistoryData(RsChatLobbyHistoryDataItem *item) ;
+
 		bool locked_initLobbyBouncableObject(const ChatLobbyId& id,RsChatLobbyBouncingObject&) ;
 		void locked_printDebugInfo() const ;
 		RsGxsId locked_getDefaultIdentity();

src/chat/p3chatservice.cc

@@ -554,6 +554,16 @@ ChatLobbyId p3ChatService::createChatLobby(const std::string& lobby_name,const R
     return DistributedChatService::createChatLobby(lobby_name,lobby_identity,lobby_topic,invited_friends,privacy_type) ;
 }
 
+bool p3ChatService::requestLobbyHistory(const ChatLobbyId& lobby_id)
+{
+    return DistributedChatService::requestLobbyHistory(lobby_id) ;
+}
+
+bool p3ChatService::requestLobbyHistoryFromPeer(const ChatLobbyId& lobby_id, const RsPeerId& peer_id, uint32_t max_count, uint32_t oldest_ts)
+{
+    return DistributedChatService::requestLobbyHistoryFromPeer(lobby_id, peer_id, max_count, oldest_ts) ;
+}
+
 
 void p3ChatService::sendChatItem(RsChatItem *item)
 {

src/chat/p3chatservice.h

@@ -138,6 +138,8 @@ class p3ChatService :
     virtual void getDefaultIdentityForChatLobby(RsGxsId& nick_name) override;
     virtual void setLobbyAutoSubscribe(const ChatLobbyId& lobby_id, const bool autoSubscribe) override;
     virtual bool getLobbyAutoSubscribe(const ChatLobbyId& lobby_id) override;
+    virtual bool requestLobbyHistory(const ChatLobbyId& lobby_id) override;
+    virtual bool requestLobbyHistoryFromPeer(const ChatLobbyId& lobby_id, const RsPeerId& peer_id, uint32_t max_count, uint32_t oldest_ts) override;
 
     /** methods that will call the DistantChatService parent
      */

src/chat/rschatitems.cc

@@ -56,6 +56,10 @@ RsItem *RsChatSerialiser::create_item(uint16_t service_id,uint8_t item_sub_id) c
 	case RS_PKT_SUBTYPE_CHAT_AVATAR_INFO: return new RsChatAvatarInfoItem();
     case RS_PKT_SUBTYPE_CHAT_AVATAR_CONFIG: return new RsChatAvatarConfigItem();
 	case RS_PKT_SUBTYPE_OUTGOING_MAP: return new PrivateOugoingMapItem();
+	case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_PROBE:      return new RsChatLobbyHistoryProbeItem();
+	case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_PROBE_RESP: return new RsChatLobbyHistoryProbeResponseItem();
+	case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_REQUEST:    return new RsChatLobbyHistoryRequestItem();
+	case RS_PKT_SUBTYPE_CHAT_LOBBY_HISTORY_DATA:       return new RsChatLobbyHistoryDataItem();
 	default:
 		std::cerr << "Unknown packet type in chat!" << std::endl;
 		return NULL;
@@ -220,3 +224,32 @@ void PrivateOugoingMapItem::serial_process(
         RsGenericSerializer::SerializeJob j,
         RsGenericSerializer::SerializeContext& ctx )
 { RS_SERIAL_PROCESS(store); }
+
+/***************** Lobby History Retrieval Protocol *****************/
+
+
+
+void RsChatLobbyHistoryProbeItem::serial_process(RsGenericSerializer::SerializeJob j,RsGenericSerializer::SerializeContext& ctx)
+{
+	RsTypeSerializer::serial_process<uint64_t>(j,ctx,lobby_id,"lobby_id") ;
+}
+
+void RsChatLobbyHistoryProbeResponseItem::serial_process(RsGenericSerializer::SerializeJob j,RsGenericSerializer::SerializeContext& ctx)
+{
+	RsTypeSerializer::serial_process<uint64_t>(j,ctx,lobby_id,         "lobby_id") ;
+	RsTypeSerializer::serial_process<uint32_t>(j,ctx,available_count,  "available_count") ;
+	RsTypeSerializer::serial_process<uint32_t>(j,ctx,oldest_timestamp, "oldest_timestamp") ;
+}
+
+void RsChatLobbyHistoryRequestItem::serial_process(RsGenericSerializer::SerializeJob j,RsGenericSerializer::SerializeContext& ctx)
+{
+	RsTypeSerializer::serial_process<uint64_t>(j,ctx,lobby_id,         "lobby_id") ;
+	RsTypeSerializer::serial_process<uint32_t>(j,ctx,max_count,        "max_count") ;
+	RsTypeSerializer::serial_process<uint32_t>(j,ctx,oldest_timestamp, "oldest_timestamp") ;
+}
+
+void RsChatLobbyHistoryDataItem::serial_process(RsGenericSerializer::SerializeJob j,RsGenericSerializer::SerializeContext& ctx)
+{
+	RsTypeSerializer::serial_process<uint64_t>(j,ctx,lobby_id,"lobby_id") ;
+	RsTypeSerializer::serial_process          (j,ctx,msgs,    "msgs") ;
+}