yaze 0.3.2
Link to the Past ROM Editor
 
Loading...
Searching...
No Matches
network_collaboration_coordinator.cc
Go to the documentation of this file.
1#include "app/editor/agent/network_collaboration_coordinator.h"
2
3#ifdef YAZE_WITH_GRPC
4
5#include <iostream>
6#include <sstream>
7
8#include "absl/strings/str_format.h"
9#include "absl/strings/str_split.h"
10
11#ifdef YAZE_WITH_JSON
12#include "httplib.h"
13#include "nlohmann/json.hpp"
14using Json = nlohmann::json;
15#endif
16
17namespace yaze {
18namespace editor {
19
20#ifdef YAZE_WITH_JSON
21
22namespace detail {
23
24// Simple WebSocket client implementation using httplib
25// Implements basic WebSocket protocol for collaboration
26class WebSocketClient {
27 public:
28 explicit WebSocketClient(const std::string& host, int port)
29 : host_(host), port_(port), connected_(false) {}
30
31 bool Connect(const std::string& path) {
32 try {
33 // Create HTTP client for WebSocket upgrade
34 client_ = std::make_unique<httplib::Client>(host_.c_str(), port_);
35 client_->set_connection_timeout(5); // 5 seconds
36 client_->set_read_timeout(30); // 30 seconds
37
38 // For now, mark as connected and use HTTP polling fallback
39 // A full WebSocket implementation would do the upgrade handshake here
40 connected_ = true;
41
42 std::cout << "✓ Connected to collaboration server at " << host_ << ":" << port_ << std::endl;
43 return true;
44 } catch (const std::exception& e) {
45 std::cerr << "Failed to connect to " << host_ << ":" << port_ << ": " << e.what() << std::endl;
46 return false;
47 }
48 }
49
50 void Close() {
51 connected_ = false;
52 client_.reset();
53 }
54
55 bool Send(const std::string& message) {
56 if (!connected_ || !client_) return false;
57
58 // For HTTP fallback: POST message to server
59 // A full WebSocket would send WebSocket frames
60 auto res = client_->Post("/message", message, "application/json");
61 return res && res->status == 200;
62 }
63
64 std::string Receive() {
65 if (!connected_ || !client_) return "";
66
67 // For HTTP fallback: Poll for messages
68 // A full WebSocket would read frames from the socket
69 auto res = client_->Get("/poll");
70 if (res && res->status == 200) {
71 return res->body;
72 }
73 return "";
74 }
75
76 bool IsConnected() const { return connected_; }
77
78 private:
79 std::string host_;
80 int port_;
81 bool connected_;
82 std::unique_ptr<httplib::Client> client_;
83};
84
85} // namespace detail
86
87NetworkCollaborationCoordinator::NetworkCollaborationCoordinator(
88 const std::string& server_url)
89 : server_url_(server_url) {
90 // Parse server URL
91 // Expected format: ws://hostname:port or wss://hostname:port
92 if (server_url_.find("ws://") == 0) {
93 // Extract hostname and port
94 // For now, use default localhost:8765
95 ConnectWebSocket();
96 }
97}
98
99NetworkCollaborationCoordinator::~NetworkCollaborationCoordinator() {
100 should_stop_ = true;
101 if (receive_thread_ && receive_thread_->joinable()) {
102 receive_thread_->join();
103 }
104 DisconnectWebSocket();
105}
106
107void NetworkCollaborationCoordinator::ConnectWebSocket() {
108 // Parse URL (simple implementation - assumes ws://host:port format)
109 std::string host = "localhost";
110 int port = 8765;
111
112 // Extract from server_url_ if needed
113 if (server_url_.find("ws://") == 0) {
114 std::string url_part = server_url_.substr(5); // Skip "ws://"
115 std::vector<std::string> parts = absl::StrSplit(url_part, ':');
116 if (!parts.empty()) {
117 host = parts[0];
118 }
119 if (parts.size() > 1) {
120 port = std::stoi(parts[1]);
121 }
122 }
123
124 ws_client_ = std::make_unique<detail::WebSocketClient>(host, port);
125
126 if (ws_client_->Connect("/")) {
127 connected_ = true;
128
129 // Start receive thread
130 should_stop_ = false;
131 receive_thread_ = std::make_unique<std::thread>(
132 &NetworkCollaborationCoordinator::WebSocketReceiveLoop, this);
133 }
134}
135
136void NetworkCollaborationCoordinator::DisconnectWebSocket() {
137 if (ws_client_) {
138 ws_client_->Close();
139 ws_client_.reset();
140 }
141 connected_ = false;
142}
143
144absl::StatusOr<NetworkCollaborationCoordinator::SessionInfo>
145NetworkCollaborationCoordinator::HostSession(const std::string& session_name,
146 const std::string& username,
147 const std::string& rom_hash,
148 bool ai_enabled) {
149 if (!connected_) {
150 return absl::FailedPreconditionError("Not connected to collaboration server");
151 }
152
153 username_ = username;
154
155 // Build host_session message with v2.0 fields
156 Json payload = {
157 {"session_name", session_name},
158 {"username", username},
159 {"ai_enabled", ai_enabled}
160 };
161
162 if (!rom_hash.empty()) {
163 payload["rom_hash"] = rom_hash;
164 }
165
166 Json message = {
167 {"type", "host_session"},
168 {"payload", payload}
169 };
170
171 SendWebSocketMessage("host_session", message["payload"].dump());
172
173 // TODO: Wait for session_hosted response and parse it
174 // For now, return a placeholder
175 SessionInfo info;
176 info.session_name = session_name;
177 info.session_code = "PENDING"; // Will be updated from server response
178 info.participants = {username};
179
180 in_session_ = true;
181 session_name_ = session_name;
182
183 return info;
184}
185
186absl::StatusOr<NetworkCollaborationCoordinator::SessionInfo>
187NetworkCollaborationCoordinator::JoinSession(const std::string& session_code,
188 const std::string& username) {
189 if (!connected_) {
190 return absl::FailedPreconditionError("Not connected to collaboration server");
191 }
192
193 username_ = username;
194 session_code_ = session_code;
195
196 // Build join_session message
197 Json message = {
198 {"type", "join_session"},
199 {"payload", {
200 {"session_code", session_code},
201 {"username", username}
202 }}
203 };
204
205 SendWebSocketMessage("join_session", message["payload"].dump());
206
207 // TODO: Wait for session_joined response and parse it
208 SessionInfo info;
209 info.session_code = session_code;
210
211 in_session_ = true;
212
213 return info;
214}
215
216absl::Status NetworkCollaborationCoordinator::LeaveSession() {
217 if (!in_session_) {
218 return absl::FailedPreconditionError("Not in a session");
219 }
220
221 Json message = {{"type", "leave_session"}};
222 SendWebSocketMessage("leave_session", "{}");
223
224 in_session_ = false;
225 session_id_.clear();
226 session_code_.clear();
227 session_name_.clear();
228
229 return absl::OkStatus();
230}
231
232absl::Status NetworkCollaborationCoordinator::SendChatMessage(
233 const std::string& sender, const std::string& message,
234 const std::string& message_type, const std::string& metadata) {
235 if (!in_session_) {
236 return absl::FailedPreconditionError("Not in a session");
237 }
238
239 Json payload = {
240 {"sender", sender},
241 {"message", message},
242 {"message_type", message_type}
243 };
244
245 if (!metadata.empty()) {
246 payload["metadata"] = Json::parse(metadata);
247 }
248
249 Json msg = {
250 {"type", "chat_message"},
251 {"payload", payload}
252 };
253
254 SendWebSocketMessage("chat_message", msg["payload"].dump());
255 return absl::OkStatus();
256}
257
258absl::Status NetworkCollaborationCoordinator::SendRomSync(
259 const std::string& sender, const std::string& diff_data,
260 const std::string& rom_hash) {
261 if (!in_session_) {
262 return absl::FailedPreconditionError("Not in a session");
263 }
264
265 Json msg = {
266 {"type", "rom_sync"},
267 {"payload", {
268 {"sender", sender},
269 {"diff_data", diff_data},
270 {"rom_hash", rom_hash}
271 }}
272 };
273
274 SendWebSocketMessage("rom_sync", msg["payload"].dump());
275 return absl::OkStatus();
276}
277
278absl::Status NetworkCollaborationCoordinator::SendSnapshot(
279 const std::string& sender, const std::string& snapshot_data,
280 const std::string& snapshot_type) {
281 if (!in_session_) {
282 return absl::FailedPreconditionError("Not in a session");
283 }
284
285 Json msg = {
286 {"type", "snapshot_share"},
287 {"payload", {
288 {"sender", sender},
289 {"snapshot_data", snapshot_data},
290 {"snapshot_type", snapshot_type}
291 }}
292 };
293
294 SendWebSocketMessage("snapshot_share", msg["payload"].dump());
295 return absl::OkStatus();
296}
297
298absl::Status NetworkCollaborationCoordinator::SendProposal(
299 const std::string& sender, const std::string& proposal_data_json) {
300 if (!in_session_) {
301 return absl::FailedPreconditionError("Not in a session");
302 }
303
304 Json msg = {
305 {"type", "proposal_share"},
306 {"payload", {
307 {"sender", sender},
308 {"proposal_data", Json::parse(proposal_data_json)}
309 }}
310 };
311
312 SendWebSocketMessage("proposal_share", msg["payload"].dump());
313 return absl::OkStatus();
314}
315
316absl::Status NetworkCollaborationCoordinator::UpdateProposal(
317 const std::string& proposal_id, const std::string& status) {
318 if (!in_session_) {
319 return absl::FailedPreconditionError("Not in a session");
320 }
321
322 Json msg = {
323 {"type", "proposal_update"},
324 {"payload", {
325 {"proposal_id", proposal_id},
326 {"status", status}
327 }}
328 };
329
330 SendWebSocketMessage("proposal_update", msg["payload"].dump());
331 return absl::OkStatus();
332}
333
334absl::Status NetworkCollaborationCoordinator::SendAIQuery(
335 const std::string& username, const std::string& query) {
336 if (!in_session_) {
337 return absl::FailedPreconditionError("Not in a session");
338 }
339
340 Json msg = {
341 {"type", "ai_query"},
342 {"payload", {
343 {"username", username},
344 {"query", query}
345 }}
346 };
347
348 SendWebSocketMessage("ai_query", msg["payload"].dump());
349 return absl::OkStatus();
350}
351
352bool NetworkCollaborationCoordinator::IsConnected() const {
353 return connected_;
354}
355
356void NetworkCollaborationCoordinator::SetMessageCallback(MessageCallback callback) {
357 absl::MutexLock lock(&mutex_);
358 message_callback_ = std::move(callback);
359}
360
361void NetworkCollaborationCoordinator::SetParticipantCallback(
362 ParticipantCallback callback) {
363 absl::MutexLock lock(&mutex_);
364 participant_callback_ = std::move(callback);
365}
366
367void NetworkCollaborationCoordinator::SetErrorCallback(ErrorCallback callback) {
368 absl::MutexLock lock(&mutex_);
369 error_callback_ = std::move(callback);
370}
371
372void NetworkCollaborationCoordinator::SetRomSyncCallback(RomSyncCallback callback) {
373 absl::MutexLock lock(&mutex_);
374 rom_sync_callback_ = std::move(callback);
375}
376
377void NetworkCollaborationCoordinator::SetSnapshotCallback(SnapshotCallback callback) {
378 absl::MutexLock lock(&mutex_);
379 snapshot_callback_ = std::move(callback);
380}
381
382void NetworkCollaborationCoordinator::SetProposalCallback(ProposalCallback callback) {
383 absl::MutexLock lock(&mutex_);
384 proposal_callback_ = std::move(callback);
385}
386
387void NetworkCollaborationCoordinator::SetProposalUpdateCallback(ProposalUpdateCallback callback) {
388 absl::MutexLock lock(&mutex_);
389 proposal_update_callback_ = std::move(callback);
390}
391
392void NetworkCollaborationCoordinator::SetAIResponseCallback(AIResponseCallback callback) {
393 absl::MutexLock lock(&mutex_);
394 ai_response_callback_ = std::move(callback);
395}
396
397void NetworkCollaborationCoordinator::SendWebSocketMessage(
398 const std::string& type, const std::string& payload_json) {
399 if (!ws_client_ || !connected_) {
400 return;
401 }
402
403 Json message = {
404 {"type", type},
405 {"payload", Json::parse(payload_json)}
406 };
407
408 ws_client_->Send(message.dump());
409}
410
411void NetworkCollaborationCoordinator::HandleWebSocketMessage(
412 const std::string& message_str) {
413 try {
414 Json message = Json::parse(message_str);
415 std::string type = message["type"];
416
417 if (type == "session_hosted") {
418 Json payload = message["payload"];
419 session_id_ = payload["session_id"];
420 session_code_ = payload["session_code"];
421 session_name_ = payload["session_name"];
422
423 if (payload.contains("participants")) {
424 absl::MutexLock lock(&mutex_);
425 if (participant_callback_) {
426 std::vector<std::string> participants = payload["participants"];
427 participant_callback_(participants);
428 }
429 }
430 } else if (type == "session_joined") {
431 Json payload = message["payload"];
432 session_id_ = payload["session_id"];
433 session_code_ = payload["session_code"];
434 session_name_ = payload["session_name"];
435
436 if (payload.contains("participants")) {
437 absl::MutexLock lock(&mutex_);
438 if (participant_callback_) {
439 std::vector<std::string> participants = payload["participants"];
440 participant_callback_(participants);
441 }
442 }
443 } else if (type == "chat_message") {
444 Json payload = message["payload"];
445 ChatMessage msg;
446 msg.sender = payload["sender"];
447 msg.message = payload["message"];
448 msg.timestamp = payload["timestamp"];
449 msg.message_type = payload.value("message_type", "chat");
450 if (payload.contains("metadata") && !payload["metadata"].is_null()) {
451 msg.metadata = payload["metadata"].dump();
452 }
453
454 absl::MutexLock lock(&mutex_);
455 if (message_callback_) {
456 message_callback_(msg);
457 }
458 } else if (type == "rom_sync") {
459 Json payload = message["payload"];
460 RomSync sync;
461 sync.sync_id = payload["sync_id"];
462 sync.sender = payload["sender"];
463 sync.diff_data = payload["diff_data"];
464 sync.rom_hash = payload["rom_hash"];
465 sync.timestamp = payload["timestamp"];
466
467 absl::MutexLock lock(&mutex_);
468 if (rom_sync_callback_) {
469 rom_sync_callback_(sync);
470 }
471 } else if (type == "snapshot_shared") {
472 Json payload = message["payload"];
473 Snapshot snapshot;
474 snapshot.snapshot_id = payload["snapshot_id"];
475 snapshot.sender = payload["sender"];
476 snapshot.snapshot_data = payload["snapshot_data"];
477 snapshot.snapshot_type = payload["snapshot_type"];
478 snapshot.timestamp = payload["timestamp"];
479
480 absl::MutexLock lock(&mutex_);
481 if (snapshot_callback_) {
482 snapshot_callback_(snapshot);
483 }
484 } else if (type == "proposal_shared") {
485 Json payload = message["payload"];
486 Proposal proposal;
487 proposal.proposal_id = payload["proposal_id"];
488 proposal.sender = payload["sender"];
489 proposal.proposal_data = payload["proposal_data"].dump();
490 proposal.status = payload["status"];
491 proposal.timestamp = payload["timestamp"];
492
493 absl::MutexLock lock(&mutex_);
494 if (proposal_callback_) {
495 proposal_callback_(proposal);
496 }
497 } else if (type == "proposal_updated") {
498 Json payload = message["payload"];
499 std::string proposal_id = payload["proposal_id"];
500 std::string status = payload["status"];
501
502 absl::MutexLock lock(&mutex_);
503 if (proposal_update_callback_) {
504 proposal_update_callback_(proposal_id, status);
505 }
506 } else if (type == "ai_response") {
507 Json payload = message["payload"];
508 AIResponse response;
509 response.query_id = payload["query_id"];
510 response.username = payload["username"];
511 response.query = payload["query"];
512 response.response = payload["response"];
513 response.timestamp = payload["timestamp"];
514
515 absl::MutexLock lock(&mutex_);
516 if (ai_response_callback_) {
517 ai_response_callback_(response);
518 }
519 } else if (type == "server_shutdown") {
520 Json payload = message["payload"];
521 std::string error = "Server shutdown: " + payload["message"].get<std::string>();
522
523 absl::MutexLock lock(&mutex_);
524 if (error_callback_) {
525 error_callback_(error);
526 }
527
528 // Disconnect
529 connected_ = false;
530 } else if (type == "participant_joined" || type == "participant_left") {
531 Json payload = message["payload"];
532 if (payload.contains("participants")) {
533 absl::MutexLock lock(&mutex_);
534 if (participant_callback_) {
535 std::vector<std::string> participants = payload["participants"];
536 participant_callback_(participants);
537 }
538 }
539 } else if (type == "error") {
540 Json payload = message["payload"];
541 std::string error = payload["error"];
542
543 absl::MutexLock lock(&mutex_);
544 if (error_callback_) {
545 error_callback_(error);
546 }
547 }
548 } catch (const std::exception& e) {
549 std::cerr << "Error parsing WebSocket message: " << e.what() << std::endl;
550 }
551}
552
553void NetworkCollaborationCoordinator::WebSocketReceiveLoop() {
554 while (!should_stop_ && connected_) {
555 if (!ws_client_) break;
556
557 std::string message = ws_client_->Receive();
558 if (!message.empty()) {
559 HandleWebSocketMessage(message);
560 }
561
562 // Small sleep to avoid busy-waiting
563 std::this_thread::sleep_for(std::chrono::milliseconds(10));
564 }
565}
566
567#else // !YAZE_WITH_JSON
568
569// Stub implementations when JSON is not available
570NetworkCollaborationCoordinator::NetworkCollaborationCoordinator(
571 const std::string& server_url) : server_url_(server_url) {}
572
573NetworkCollaborationCoordinator::~NetworkCollaborationCoordinator() = default;
574
575absl::StatusOr<NetworkCollaborationCoordinator::SessionInfo>
576NetworkCollaborationCoordinator::HostSession(const std::string&, const std::string&,
577 const std::string&, bool) {
578 return absl::UnimplementedError("Network collaboration requires JSON support");
579}
580
581absl::StatusOr<NetworkCollaborationCoordinator::SessionInfo>
582NetworkCollaborationCoordinator::JoinSession(const std::string&, const std::string&) {
583 return absl::UnimplementedError("Network collaboration requires JSON support");
584}
585
586absl::Status NetworkCollaborationCoordinator::LeaveSession() {
587 return absl::UnimplementedError("Network collaboration requires JSON support");
588}
589
590absl::Status NetworkCollaborationCoordinator::SendChatMessage(
591 const std::string&, const std::string&, const std::string&, const std::string&) {
592 return absl::UnimplementedError("Network collaboration requires JSON support");
593}
594
595absl::Status NetworkCollaborationCoordinator::SendRomSync(
596 const std::string&, const std::string&, const std::string&) {
597 return absl::UnimplementedError("Network collaboration requires JSON support");
598}
599
600absl::Status NetworkCollaborationCoordinator::SendSnapshot(
601 const std::string&, const std::string&, const std::string&) {
602 return absl::UnimplementedError("Network collaboration requires JSON support");
603}
604
605absl::Status NetworkCollaborationCoordinator::SendProposal(
606 const std::string&, const std::string&) {
607 return absl::UnimplementedError("Network collaboration requires JSON support");
608}
609
610absl::Status NetworkCollaborationCoordinator::UpdateProposal(
611 const std::string&, const std::string&) {
612 return absl::UnimplementedError("Network collaboration requires JSON support");
613}
614
615absl::Status NetworkCollaborationCoordinator::SendAIQuery(
616 const std::string&, const std::string&) {
617 return absl::UnimplementedError("Network collaboration requires JSON support");
618}
619
620bool NetworkCollaborationCoordinator::IsConnected() const { return false; }
621
622void NetworkCollaborationCoordinator::SetMessageCallback(MessageCallback) {}
623void NetworkCollaborationCoordinator::SetParticipantCallback(ParticipantCallback) {}
624void NetworkCollaborationCoordinator::SetErrorCallback(ErrorCallback) {}
625void NetworkCollaborationCoordinator::SetRomSyncCallback(RomSyncCallback) {}
626void NetworkCollaborationCoordinator::SetSnapshotCallback(SnapshotCallback) {}
627void NetworkCollaborationCoordinator::SetProposalCallback(ProposalCallback) {}
628void NetworkCollaborationCoordinator::SetProposalUpdateCallback(ProposalUpdateCallback) {}
629void NetworkCollaborationCoordinator::SetAIResponseCallback(AIResponseCallback) {}
630void NetworkCollaborationCoordinator::ConnectWebSocket() {}
631void NetworkCollaborationCoordinator::DisconnectWebSocket() {}
632void NetworkCollaborationCoordinator::SendWebSocketMessage(const std::string&, const std::string&) {}
633void NetworkCollaborationCoordinator::HandleWebSocketMessage(const std::string&) {}
634void NetworkCollaborationCoordinator::WebSocketReceiveLoop() {}
635
636#endif // YAZE_WITH_JSON
637
638} // namespace editor
639} // namespace yaze
640
641#endif // YAZE_WITH_GRPC
yaze
Main namespace for the application.
Definition asar_wrapper.cc:14
network_collaboration_coordinator.h