mesen_socket_client.cc

Go to the documentation of this file.

#include "app/emu/mesen/mesen_socket_client.h"
 
#include <fcntl.h>
#include <sys/stat.h>
 
#include <filesystem>
 
#ifdef _WIN32
// clang-format off
// winsock2.h must precede afunix.h (defines ADDRESS_FAMILY)
#include <winsock2.h>
#include <ws2tcpip.h>
#include <afunix.h>
#include <io.h>
// clang-format on
#define close closesocket
typedef int ssize_t;
#else
#include <arpa/inet.h>
#include <netdb.h>
#include <netinet/in.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#endif
 
#include <cerrno>
#include <charconv>
#include <chrono>
#include <cstdlib>
#include <cstring>
#include <regex>
#include <sstream>
#include <utility>
 
#include "absl/cleanup/cleanup.h"
#include "absl/strings/numbers.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/str_format.h"
#include "absl/strings/str_split.h"
 
namespace yaze {
namespace emu {
namespace mesen {
 
namespace {
 
// Simple JSON value extraction (avoids pulling in a full JSON library)
std::string ExtractJsonString(const std::string& json, const std::string& key) {
  std::string search = "\"" + key + "\":";
  size_t pos = json.find(search);
  if (pos == std::string::npos)
    return "";
 
  pos += search.length();
  while (pos < json.length() && (json[pos] == ' ' || json[pos] == '\t'))
    pos++;
 
  if (pos >= json.length())
    return "";
 
  if (json[pos] == '"') {
    // String value
    size_t start = pos + 1;
    size_t end = json.find('"', start);
    while (end != std::string::npos && end > 0 && json[end - 1] == '\\') {
      end = json.find('"', end + 1);
    }
    if (end == std::string::npos)
      return "";
    return json.substr(start, end - start);
  } else if (json[pos] == '{') {
    // Object value - find matching brace
    int depth = 1;
    size_t start = pos;
    pos++;
    while (pos < json.length() && depth > 0) {
      if (json[pos] == '{')
        depth++;
      else if (json[pos] == '}')
        depth--;
      pos++;
    }
    return json.substr(start, pos - start);
  } else if (json[pos] == '[') {
    // Array value - find matching bracket
    int depth = 1;
    size_t start = pos;
    pos++;
    while (pos < json.length() && depth > 0) {
      if (json[pos] == '[')
        depth++;
      else if (json[pos] == ']')
        depth--;
      pos++;
    }
    return json.substr(start, pos - start);
  } else {
    // Number or boolean
    size_t start = pos;
    while (pos < json.length() && json[pos] != ',' && json[pos] != '}' &&
           json[pos] != ']') {
      pos++;
    }
    return json.substr(start, pos - start);
  }
}
 
int64_t ExtractJsonInt(const std::string& json, const std::string& key,
                       int64_t default_value = 0) {
  std::string value = ExtractJsonString(json, key);
  if (value.empty())
    return default_value;
 
  // Handle hex strings like "0x7E0000"
  if (value.length() > 2 && value[0] == '0' &&
      (value[1] == 'x' || value[1] == 'X')) {
    int64_t result;
    std::string hex_str = value.substr(2);
    auto [ptr, ec] = std::from_chars(
        hex_str.data(), hex_str.data() + hex_str.size(), result, 16);
    if (ec == std::errc()) {
      return result;
    }
  }
 
  int64_t result;
  if (absl::SimpleAtoi(value, &result)) {
    return result;
  }
  return default_value;
}
 
double ExtractJsonDouble(const std::string& json, const std::string& key,
                         double default_value = 0.0) {
  std::string value = ExtractJsonString(json, key);
  if (value.empty())
    return default_value;
 
  double result;
  if (absl::SimpleAtod(value, &result)) {
    return result;
  }
  return default_value;
}
 
bool ExtractJsonBool(const std::string& json, const std::string& key,
                     bool default_value = false) {
  std::string value = ExtractJsonString(json, key);
  if (value.empty())
    return default_value;
  return value == "true";
}
 
std::string BuildJsonCommand(const std::string& type) {
  return absl::StrFormat("{\"type\":\"%s\"}\n", type);
}
 
std::string EscapeJsonValue(const std::string& value) {
  std::string escaped;
  escaped.reserve(value.size());
  for (char c : value) {
    switch (c) {
      case '\\':
        escaped += "\\\\";
        break;
      case '"':
        escaped += "\\\"";
        break;
      case '\n':
        escaped += "\\n";
        break;
      case '\r':
        escaped += "\\r";
        break;
      case '\t':
        escaped += "\\t";
        break;
      default:
        escaped += c;
        break;
    }
  }
  return escaped;
}
 
std::string BuildJsonCommand(
    const std::string& type,
    const std::vector<std::pair<std::string, std::string>>& params) {
  std::stringstream ss;
  ss << "{\"type\":\"" << type << "\"";
  for (const auto& [key, value] : params) {
    ss << ",\"" << key << "\":\"" << EscapeJsonValue(value) << "\"";
  }
  ss << "}\n";
  return ss.str();
}
 
constexpr size_t kMaxResponseSize = 4 * 1024 * 1024;
constexpr int kConnectTimeoutMs = 2000;
 
int LastSocketError() {
#ifdef _WIN32
  return WSAGetLastError();
#else
  return errno;
#endif
}
 
std::string SocketErrorMessage(int error) {
#ifdef _WIN32
  return absl::StrCat("Winsock error ", error);
#else
  return strerror(error);
#endif
}
 
// Commands must not block a UI thread forever, but the event stream is
// long-lived and idles between emulator events.
constexpr int kDefaultSendTimeoutMs = 5000;
 
// Tests override this to exercise the write deadline without stalling for
// the production timeout.
int SendTimeoutMs() {
  // Read on each connect rather than caching, so a test that sets the
  // override is not affected by whichever test connected first.
  const char* raw = std::getenv("YAZE_MESEN_SEND_TIMEOUT_MS");
  int parsed = 0;
  if (raw != nullptr && absl::SimpleAtoi(raw, &parsed) && parsed > 0) {
    return parsed;
  }
  return kDefaultSendTimeoutMs;
}
 
// The event socket needs a bounded wakeup so Unsubscribe()/Disconnect() can
// join the event thread promptly; a receive timeout there means "no event
// yet", which EventLoop treats as a retry rather than an error.
constexpr int kDefaultEventPollMs = 1000;
 
int EventPollMs() {
  const char* raw = std::getenv("YAZE_MESEN_EVENT_POLL_MS");
  int parsed = 0;
  if (raw != nullptr && absl::SimpleAtoi(raw, &parsed) && parsed > 0) {
    return parsed;
  }
  return kDefaultEventPollMs;
}
 
// A closed peer must surface as a status, never as a process-level signal.
// Linux carries this per send(); BSD/macOS carry it per socket.
#if defined(MSG_NOSIGNAL)
constexpr int kSendFlags = MSG_NOSIGNAL;
#else
constexpr int kSendFlags = 0;
#endif
 
void SuppressSigpipe(SocketHandle fd) {
#if defined(SO_NOSIGPIPE)
  const int enable = 1;
  (void)setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE,
                   reinterpret_cast<const char*>(&enable), sizeof(enable));
#else
  (void)fd;
#endif
}
 
void SetSocketSendTimeout(SocketHandle fd, int timeout_ms) {
#ifdef _WIN32
  const DWORD timeout = static_cast<DWORD>(timeout_ms);
  (void)setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO,
                   reinterpret_cast<const char*>(&timeout), sizeof(timeout));
#else
  timeval timeout{};
  timeout.tv_sec = timeout_ms / 1000;
  timeout.tv_usec = (timeout_ms % 1000) * 1000;
  (void)setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout));
#endif
}
 
void SetSocketReceiveTimeout(SocketHandle fd, int timeout_ms) {
#ifdef _WIN32
  const DWORD timeout = static_cast<DWORD>(timeout_ms);
  (void)setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO,
                   reinterpret_cast<const char*>(&timeout), sizeof(timeout));
#else
  timeval timeout{};
  timeout.tv_sec = timeout_ms / 1000;
  timeout.tv_usec = (timeout_ms % 1000) * 1000;
  (void)setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));
#endif
}
 
bool LooksLikeTcpEndpoint(const std::string& path) {
  return path.rfind("tcp://", 0) == 0 || path.rfind("tcp:", 0) == 0;
}
 
absl::StatusOr<std::pair<std::string, uint16_t>> ParseTcpEndpoint(
    const std::string& path) {
  std::string raw = path;
  if (raw.rfind("tcp://", 0) == 0) {
    raw = raw.substr(6);
  } else if (raw.rfind("tcp:", 0) == 0) {
    raw = raw.substr(4);
  } else {
    return absl::InvalidArgumentError(
        absl::StrCat("Not a tcp:// endpoint: ", path));
  }
 
  std::string host = "127.0.0.1";
  std::string port_str = raw;
  const auto colon = raw.rfind(':');
  if (colon != std::string::npos) {
    host = raw.substr(0, colon);
    port_str = raw.substr(colon + 1);
    if (host.empty()) {
      host = "127.0.0.1";
    }
  }
  int port = 0;
  if (!absl::SimpleAtoi(port_str, &port) || port <= 0 || port > 65535) {
    return absl::InvalidArgumentError(
        absl::StrCat("Invalid TCP port in ", path));
  }
  return std::make_pair(host, static_cast<uint16_t>(port));
}
 
bool SetSocketBlocking(SocketHandle fd, bool blocking) {
#ifdef _WIN32
  u_long mode = blocking ? 0UL : 1UL;
  return ioctlsocket(fd, FIONBIO, &mode) == 0;
#else
  const int flags = fcntl(fd, F_GETFL, 0);
  if (flags < 0) {
    return false;
  }
  const int next_flags =
      blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
  return fcntl(fd, F_SETFL, next_flags) == 0;
#endif
}
 
// Waits until the socket accepts more data or the budget expires.
// Returns Ok when writable, DeadlineExceeded on timeout.
absl::Status WaitForWritable(SocketHandle fd, int timeout_ms) {
#ifdef _WIN32
  fd_set write_fds;
  FD_ZERO(&write_fds);
  FD_SET(fd, &write_fds);
  timeval timeout;
  timeout.tv_sec = timeout_ms / 1000;
  timeout.tv_usec = (timeout_ms % 1000) * 1000;
  const int ready = select(0, nullptr, &write_fds, nullptr, &timeout);
#else
  pollfd descriptor{};
  descriptor.fd = fd;
  descriptor.events = POLLOUT;
  const int ready = poll(&descriptor, 1, timeout_ms);
#endif
  if (ready == 0) {
    return absl::DeadlineExceededError("Socket did not become writable");
  }
  if (ready < 0) {
    const int error = LastSocketError();
#ifdef _WIN32
    if (error == WSAEINTR) {
      return absl::OkStatus();
    }
#else
    if (error == EINTR) {
      return absl::OkStatus();
    }
#endif
    return absl::InternalError(
        absl::StrCat("Failed waiting to send: ", SocketErrorMessage(error)));
  }
  return absl::OkStatus();
}
 
// send() may accept fewer bytes than requested, and a blocking send() can
// park indefinitely when the peer stops reading: SO_SNDTIMEO is not honored
// for Unix-domain sockets on macOS. Send non-blocking and bound the whole
// command with one deadline.
absl::Status SendAll(SocketHandle fd, const std::string& data) {
  const auto deadline = std::chrono::steady_clock::now() +
                        std::chrono::milliseconds(SendTimeoutMs());
  const bool was_blocking = SetSocketBlocking(fd, false);
  absl::Cleanup restore = [fd, was_blocking] {
    if (was_blocking) {
      (void)SetSocketBlocking(fd, true);
    }
  };
 
  size_t offset = 0;
  while (offset < data.size()) {
    const auto now = std::chrono::steady_clock::now();
    const auto remaining =
        std::chrono::duration_cast<std::chrono::milliseconds>(deadline - now)
            .count();
    if (remaining <= 0) {
      return absl::DeadlineExceededError(
          absl::StrCat("Timed out sending command after ", SendTimeoutMs(),
                       "ms (", offset, " of ", data.size(), " bytes)"));
    }
    const ssize_t sent =
        send(fd, data.c_str() + offset, static_cast<int>(data.size() - offset),
             kSendFlags);
    if (sent > 0) {
      offset += static_cast<size_t>(sent);
      continue;
    }
    const int error = LastSocketError();
#ifdef _WIN32
    if (error == WSAEINTR) {
      continue;
    }
    if (error == WSAEWOULDBLOCK || error == WSAETIMEDOUT) {
#else
    if (error == EINTR) {
      continue;
    }
    if (error == EAGAIN || error == EWOULDBLOCK) {
#endif
      const absl::Status writable =
          WaitForWritable(fd, static_cast<int>(remaining));
      if (absl::IsDeadlineExceeded(writable)) {
        return absl::DeadlineExceededError(
            absl::StrCat("Timed out sending command after ", SendTimeoutMs(),
                         "ms (", offset, " of ", data.size(), " bytes)"));
      }
      if (!writable.ok()) {
        return writable;
      }
      continue;
    }
    return absl::InternalError(
        absl::StrCat("Failed to send command: ", SocketErrorMessage(error)));
  }
  return absl::OkStatus();
}
 
absl::Status WaitForConnectComplete(SocketHandle fd,
                                    const std::string& socket_path) {
#ifdef _WIN32
  fd_set write_fds;
  fd_set exception_fds;
  FD_ZERO(&write_fds);
  FD_ZERO(&exception_fds);
  FD_SET(fd, &write_fds);
  FD_SET(fd, &exception_fds);
  timeval timeout;
  timeout.tv_sec = kConnectTimeoutMs / 1000;
  timeout.tv_usec = (kConnectTimeoutMs % 1000) * 1000;
  // Winsock reports successful nonblocking connects through writefds and
  // failed connects through exceptfds. SO_ERROR below determines the result.
  const int ready = select(0, nullptr, &write_fds, &exception_fds, &timeout);
#else
  pollfd descriptor{};
  descriptor.fd = fd;
  descriptor.events = POLLOUT;
  const int ready = poll(&descriptor, 1, kConnectTimeoutMs);
#endif
  if (ready == 0) {
    return absl::DeadlineExceededError(absl::StrCat("Timed out connecting to ",
                                                    socket_path, " after ",
                                                    kConnectTimeoutMs, "ms"));
  }
  if (ready < 0) {
    const int error = LastSocketError();
    return absl::UnavailableError(
        absl::StrCat("Failed while waiting to connect to ", socket_path, ": ",
                     SocketErrorMessage(error)));
  }
 
  int connect_error = 0;
#ifdef _WIN32
  int error_length = sizeof(connect_error);
  const int get_error =
      getsockopt(fd, SOL_SOCKET, SO_ERROR,
                 reinterpret_cast<char*>(&connect_error), &error_length);
#else
  socklen_t error_length = sizeof(connect_error);
  const int get_error =
      getsockopt(fd, SOL_SOCKET, SO_ERROR, &connect_error, &error_length);
#endif
  if (get_error != 0) {
    const int error = LastSocketError();
    return absl::UnavailableError(
        absl::StrCat("Failed to read connect status for ", socket_path, ": ",
                     SocketErrorMessage(error)));
  }
  if (connect_error != 0) {
    return absl::UnavailableError(
        absl::StrCat("Failed to connect to ", socket_path, ": ",
                     SocketErrorMessage(connect_error)));
  }
  return absl::OkStatus();
}
 
absl::Status ConnectTcpEndpoint(const std::string& socket_path,
                                SocketHandle* socket_fd) {
  auto parsed = ParseTcpEndpoint(socket_path);
  if (!parsed.ok()) {
    return parsed.status();
  }
  const std::string& host = parsed->first;
  const uint16_t port = parsed->second;
 
  SocketHandle fd = socket(AF_INET, SOCK_STREAM, 0);
  if (fd == kInvalidSocketHandle) {
    const int error = LastSocketError();
    return absl::InternalError(absl::StrCat("Failed to create TCP socket: ",
                                            SocketErrorMessage(error)));
  }
  SuppressSigpipe(fd);
  SetSocketSendTimeout(fd, SendTimeoutMs());
 
  if (!SetSocketBlocking(fd, false)) {
    const int error = LastSocketError();
    close(fd);
    return absl::InternalError(
        absl::StrCat("Failed to set non-blocking mode on TCP socket: ",
                     SocketErrorMessage(error)));
  }
 
  sockaddr_in addr;
  memset(&addr, 0, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_port = htons(port);
  if (inet_pton(AF_INET, host.c_str(), &addr.sin_addr) != 1) {
    addrinfo hints;
    memset(&hints, 0, sizeof(hints));
    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_STREAM;
    addrinfo* result = nullptr;
    const int gai = getaddrinfo(host.c_str(), nullptr, &hints, &result);
    if (gai != 0 || result == nullptr) {
      close(fd);
      return absl::UnavailableError(
          absl::StrCat("Failed to resolve ", host, ": ",
                       gai != 0 ? gai_strerror(gai) : "no address"));
    }
    addr.sin_addr = reinterpret_cast<sockaddr_in*>(result->ai_addr)->sin_addr;
    freeaddrinfo(result);
  }
 
  const int result =
      connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr));
  const int connect_error = result < 0 ? LastSocketError() : 0;
#ifdef _WIN32
  const bool in_progress = result < 0 && (connect_error == WSAEWOULDBLOCK ||
                                          connect_error == WSAEINPROGRESS ||
                                          connect_error == WSAEALREADY);
#else
  const bool in_progress = result < 0 && connect_error == EINPROGRESS;
#endif
  if (result < 0 && !in_progress) {
    close(fd);
    return absl::UnavailableError(
        absl::StrCat("Failed to connect to ", socket_path, ": ",
                     SocketErrorMessage(connect_error)));
  }
  if (in_progress) {
    auto wait_status = WaitForConnectComplete(fd, socket_path);
    if (!wait_status.ok()) {
      close(fd);
      return wait_status;
    }
  }
 
  if (!SetSocketBlocking(fd, true)) {
    const int error = LastSocketError();
    close(fd);
    return absl::InternalError(
        absl::StrCat("Failed to restore blocking mode on TCP socket: ",
                     SocketErrorMessage(error)));
  }
 
  *socket_fd = fd;
  return absl::OkStatus();
}
 
absl::Status ConnectUnixEndpoint(const std::string& socket_path,
                                 SocketHandle* socket_fd) {
  SocketHandle fd = socket(AF_UNIX, SOCK_STREAM, 0);
  if (fd == kInvalidSocketHandle) {
    const int error = LastSocketError();
    return absl::InternalError(
        absl::StrCat("Failed to create socket: ", SocketErrorMessage(error)));
  }
  SuppressSigpipe(fd);
  SetSocketSendTimeout(fd, SendTimeoutMs());
 
  struct sockaddr_un addr;
  memset(&addr, 0, sizeof(addr));
  addr.sun_family = AF_UNIX;
  strncpy(addr.sun_path, socket_path.c_str(), sizeof(addr.sun_path) - 1);
 
  if (connect(fd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) <
      0) {
    const int error = LastSocketError();
    close(fd);
    return absl::UnavailableError(absl::StrCat(
        "Failed to connect to ", socket_path, ": ", SocketErrorMessage(error)));
  }
 
  *socket_fd = fd;
  return absl::OkStatus();
}
 
absl::Status ConnectSocketToPath(const std::string& socket_path,
                                 SocketHandle* socket_fd) {
  if (socket_fd == nullptr) {
    return absl::InvalidArgumentError("socket_fd output pointer is null");
  }
  if (LooksLikeTcpEndpoint(socket_path)) {
    return ConnectTcpEndpoint(socket_path, socket_fd);
  }
  return ConnectUnixEndpoint(socket_path, socket_fd);
}
 
void ShutdownSocketFd(SocketHandle fd) {
  if (fd == kInvalidSocketHandle) {
    return;
  }
#ifdef _WIN32
  shutdown(fd, SD_BOTH);
#else
  shutdown(fd, SHUT_RDWR);
#endif
}
 
}  // namespace
 
MesenSocketClient::MesenSocketClient() {
#ifdef _WIN32
  WSADATA winsock_data{};
  winsock_startup_error_ = WSAStartup(MAKEWORD(2, 2), &winsock_data);
  winsock_started_ = winsock_startup_error_ == 0;
#endif
}
 
MesenSocketClient::~MesenSocketClient() {
  Disconnect();
#ifdef _WIN32
  if (winsock_started_) {
    WSACleanup();
  }
#endif
}
 
absl::Status MesenSocketClient::Connect() {
  auto paths = FindSocketPaths();
  if (paths.empty()) {
    return absl::NotFoundError(
        "No Mesen2 socket found. Is Mesen2-OoS running?");
  }
  return Connect(paths[0]);
}
 
absl::Status MesenSocketClient::Connect(const std::string& socket_path) {
#ifdef _WIN32
  if (!winsock_started_) {
    return absl::InternalError(
        absl::StrCat("Failed to initialize Winsock: ",
                     SocketErrorMessage(winsock_startup_error_)));
  }
#endif
 
  // A mid-command failure clears connected_ but leaves socket_fd_ open, so
  // IsConnected() is not a reliable cleanup gate. Disconnect() is idempotent.
  Disconnect();
 
  SocketHandle fd = kInvalidSocketHandle;
  auto connect_status = ConnectSocketToPath(socket_path, &fd);
  if (!connect_status.ok()) {
    return connect_status;
  }
 
  socket_fd_ = fd;
  socket_path_ = socket_path;
  connected_ = true;
 
  // Verify connection with a ping
  auto status = Ping();
  if (!status.ok()) {
    Disconnect();
    return status;
  }
 
  return absl::OkStatus();
}
 
void MesenSocketClient::Disconnect() {
  // Best-effort: stop event stream first so background thread exits cleanly.
  (void)Unsubscribe();
 
  if (socket_fd_ != kInvalidSocketHandle) {
    close(socket_fd_);
    socket_fd_ = kInvalidSocketHandle;
  }
  socket_path_.clear();
  connected_ = false;
}
 
bool MesenSocketClient::IsConnected() const {
  return connected_;
}
 
std::vector<std::string> MesenSocketClient::FindSocketPaths() {
  const char* env_path = std::getenv("MESEN2_SOCKET_PATH");
  if (env_path && env_path[0] != '\0') {
    std::string env(env_path);
    if (LooksLikeTcpEndpoint(env)) {
      // An explicit TCP target is authoritative. Preserve malformed values so
      // Connect() reports the configuration error instead of attaching to an
      // unrelated local emulator discovered in the temporary directory.
      return {env};
    }
#ifdef _WIN32
    // Windows AF_UNIX sockets don't report S_IFSOCK via stat; trust the env var
    if (std::filesystem::exists(env_path)) {
      return {env};
    }
#else
    struct stat st;
    if (stat(env_path, &st) == 0 && (st.st_mode & S_IFMT) == S_IFSOCK) {
      return {env};
    }
#endif
  }
 
  std::vector<std::string> paths;
  namespace fs = std::filesystem;
  std::vector<fs::path> search_paths;
 
#ifdef _WIN32
  search_paths.push_back(fs::temp_directory_path());
#else
  search_paths.push_back("/tmp");
#endif
 
  std::regex socket_pattern("mesen2-\\d+\\.sock");
 
  for (const auto& search_path : search_paths) {
    std::error_code ec;
    if (!fs::exists(search_path, ec))
      continue;
 
    for (const auto& entry : fs::directory_iterator(search_path, ec)) {
      if (ec)
        break;
      // On Windows, checking is_socket might be unreliable or not supported for
      // AF_UNIX files, so we mainly rely on the filename pattern.
      std::string filename = entry.path().filename().string();
      if (std::regex_match(filename, socket_pattern)) {
        paths.push_back(entry.path().string());
      }
    }
  }
 
  return paths;
}
 
std::vector<std::string> MesenSocketClient::ListAvailableSockets() {
  return FindSocketPaths();
}
 
absl::StatusOr<std::string> MesenSocketClient::SendCommandOnSocket(
    SocketHandle fd, const std::string& json, bool update_connection_state) {
  if (fd == kInvalidSocketHandle) {
    return absl::FailedPreconditionError("Socket is not connected");
  }
 
  // Send command
  if (const absl::Status sent = SendAll(fd, json); !sent.ok()) {
    if (update_connection_state) {
      connected_ = false;
    }
    return sent;
  }
 
  // Receive response (with timeout)
  SetSocketReceiveTimeout(fd, 5000);
 
  std::string response;
  char buffer[4096];
  while (true) {
    ssize_t received = recv(fd, buffer, sizeof(buffer), 0);
    if (received < 0) {
      const int error = LastSocketError();
      // SO_RCVTIMEO makes recv() non-restartable, so a caught signal arrives
      // here as EINTR and must be retried rather than failing the command.
#ifdef _WIN32
      if (error == WSAEINTR) {
        continue;
      }
      if (error == WSAEWOULDBLOCK || error == WSAETIMEDOUT) {
#else
      if (error == EINTR) {
        continue;
      }
      if (error == EAGAIN || error == EWOULDBLOCK) {
#endif
        if (response.empty()) {
          return absl::DeadlineExceededError("Timeout waiting for response");
        }
        break;
      }
      if (update_connection_state) {
        connected_ = false;
      }
      return absl::InternalError(absl::StrCat("Failed to receive response: ",
                                              SocketErrorMessage(error)));
    }
    if (received == 0) {
      break;
    }
    response.append(buffer, static_cast<size_t>(received));
    if (response.size() > kMaxResponseSize) {
      return absl::ResourceExhaustedError("Mesen2 response too large");
    }
    if (response.find('\n') != std::string::npos) {
      break;
    }
  }
  if (response.empty()) {
    return absl::DeadlineExceededError("Empty response from Mesen2");
  }
  auto newline_pos = response.find('\n');
  if (newline_pos != std::string::npos) {
    response = response.substr(0, newline_pos);
  }
 
  return ParseResponse(response);
}
 
absl::StatusOr<std::string> MesenSocketClient::SendCommand(
    const std::string& json) {
  if (!IsConnected()) {
    return absl::FailedPreconditionError("Not connected to Mesen2");
  }
 
  std::lock_guard<std::mutex> lock(command_mutex_);
  return SendCommandOnSocket(socket_fd_, json,
                             /*update_connection_state=*/true);
}
 
absl::StatusOr<std::string> MesenSocketClient::ParseResponse(
    const std::string& response) {
  bool success = ExtractJsonBool(response, "success");
  if (!success) {
    std::string error = ExtractJsonString(response, "error");
    if (error.empty())
      error = "Unknown Mesen2 error";
    return absl::InternalError(error);
  }
 
  std::string data = ExtractJsonString(response, "data");
  return data.empty() ? response : data;
}
 
// ─────────────────────────────────────────────────────────────────────────────
// Control Commands
// ─────────────────────────────────────────────────────────────────────────────
 
absl::Status MesenSocketClient::Ping() {
  auto result = SendCommand(BuildJsonCommand("PING"));
  if (!result.ok())
    return result.status();
  return absl::OkStatus();
}
 
absl::StatusOr<MesenState> MesenSocketClient::GetState() {
  auto result = SendCommand(BuildJsonCommand("STATE"));
  if (!result.ok())
    return result.status();
 
  MesenState state;
  state.running = ExtractJsonBool(*result, "running");
  state.paused = ExtractJsonBool(*result, "paused");
  state.debugging = ExtractJsonBool(*result, "debugging");
  state.frame = ExtractJsonInt(*result, "frame");
  state.fps = ExtractJsonDouble(*result, "fps");
  state.console_type = ExtractJsonInt(*result, "consoleType");
  return state;
}
 
absl::Status MesenSocketClient::Pause() {
  auto result = SendCommand(BuildJsonCommand("PAUSE"));
  return result.status();
}
 
absl::Status MesenSocketClient::Resume() {
  auto result = SendCommand(BuildJsonCommand("RESUME"));
  return result.status();
}
 
absl::Status MesenSocketClient::Reset() {
  auto result = SendCommand(BuildJsonCommand("RESET"));
  return result.status();
}
 
absl::Status MesenSocketClient::Frame() {
  auto result = SendCommand(BuildJsonCommand("FRAME"));
  return result.status();
}
 
absl::Status MesenSocketClient::Step(int count, const std::string& mode) {
  auto result = SendCommand(BuildJsonCommand(
      "STEP", {{"count", std::to_string(count)}, {"mode", mode}}));
  return result.status();
}
 
// ─────────────────────────────────────────────────────────────────────────────
// Input Commands
// ─────────────────────────────────────────────────────────────────────────────
 
absl::Status MesenSocketClient::SetButton(emu::input::SnesButton button,
                                          bool pressed) {
  current_input_.SetButton(button, pressed);
  return SetButtons(current_input_);
}
 
absl::Status MesenSocketClient::SetButtons(
    const emu::input::ControllerState& state) {
  current_input_ = state;
 
  std::vector<std::string> buttons;
  if (state.IsPressed(emu::input::SnesButton::A))
    buttons.push_back("a");
  if (state.IsPressed(emu::input::SnesButton::B))
    buttons.push_back("b");
  if (state.IsPressed(emu::input::SnesButton::X))
    buttons.push_back("x");
  if (state.IsPressed(emu::input::SnesButton::Y))
    buttons.push_back("y");
  if (state.IsPressed(emu::input::SnesButton::L))
    buttons.push_back("l");
  if (state.IsPressed(emu::input::SnesButton::R))
    buttons.push_back("r");
  if (state.IsPressed(emu::input::SnesButton::SELECT))
    buttons.push_back("select");
  if (state.IsPressed(emu::input::SnesButton::START))
    buttons.push_back("start");
  if (state.IsPressed(emu::input::SnesButton::UP))
    buttons.push_back("up");
  if (state.IsPressed(emu::input::SnesButton::DOWN))
    buttons.push_back("down");
  if (state.IsPressed(emu::input::SnesButton::LEFT))
    buttons.push_back("left");
  if (state.IsPressed(emu::input::SnesButton::RIGHT))
    buttons.push_back("right");
 
  std::string buttons_str;
  for (size_t i = 0; i < buttons.size(); ++i) {
    if (i > 0)
      buttons_str += ",";
    buttons_str += buttons[i];
  }
 
  auto result =
      SendCommand(BuildJsonCommand("INPUT", {{"buttons", buttons_str}}));
  return result.status();
}
 
// ─────────────────────────────────────────────────────────────────────────────
// Memory Commands
// ─────────────────────────────────────────────────────────────────────────────
 
absl::StatusOr<uint8_t> MesenSocketClient::ReadByte(uint32_t addr) {
  auto result = SendCommand(
      BuildJsonCommand("READ", {{"addr", absl::StrFormat("0x%06X", addr)}}));
  if (!result.ok())
    return result.status();
  return static_cast<uint8_t>(ExtractJsonInt(*result, "data", 0));
}
 
absl::StatusOr<uint16_t> MesenSocketClient::ReadWord(uint32_t addr) {
  auto result = SendCommand(
      BuildJsonCommand("READ16", {{"addr", absl::StrFormat("0x%06X", addr)}}));
  if (!result.ok())
    return result.status();
  return static_cast<uint16_t>(ExtractJsonInt(*result, "data", 0));
}
 
absl::StatusOr<std::vector<uint8_t>> MesenSocketClient::ReadBlock(uint32_t addr,
                                                                  size_t len) {
  auto result = SendCommand(
      BuildJsonCommand("READBLOCK", {{"addr", absl::StrFormat("0x%06X", addr)},
                                     {"len", std::to_string(len)}}));
  if (!result.ok())
    return result.status();
 
  // Response is hex string
  std::string hex = ExtractJsonString(*result, "data");
  if (hex.empty()) {
    // Try raw data field
    hex = *result;
  }
 
  std::vector<uint8_t> data;
  data.reserve(len);
 
  for (size_t i = 0; i + 1 < hex.length(); i += 2) {
    int byte;
    std::string byte_hex = hex.substr(i, 2);
    auto [ptr, ec] = std::from_chars(
        byte_hex.data(), byte_hex.data() + byte_hex.size(), byte, 16);
    if (ec == std::errc()) {
      data.push_back(static_cast<uint8_t>(byte));
    }
  }
  return data;
}
 
absl::Status MesenSocketClient::WriteByte(uint32_t addr, uint8_t value) {
  auto result = SendCommand(
      BuildJsonCommand("WRITE", {{"addr", absl::StrFormat("0x%06X", addr)},
                                 {"value", absl::StrFormat("0x%02X", value)}}));
  return result.status();
}
 
absl::Status MesenSocketClient::WriteWord(uint32_t addr, uint16_t value) {
  auto result = SendCommand(BuildJsonCommand(
      "WRITE16", {{"addr", absl::StrFormat("0x%06X", addr)},
                  {"value", absl::StrFormat("0x%04X", value)}}));
  return result.status();
}
 
absl::Status MesenSocketClient::WriteBlock(uint32_t addr,
                                           const std::vector<uint8_t>& data) {
  std::stringstream hex;
  for (uint8_t byte : data) {
    hex << absl::StrFormat("%02X", byte);
  }
  auto result = SendCommand(BuildJsonCommand(
      "WRITEBLOCK",
      {{"addr", absl::StrFormat("0x%06X", addr)}, {"hex", hex.str()}}));
  return result.status();
}
 
// ─────────────────────────────────────────────────────────────────────────────
// Debugging Commands
// ─────────────────────────────────────────────────────────────────────────────
 
absl::StatusOr<CpuState> MesenSocketClient::GetCpuState() {
  auto result = SendCommand(BuildJsonCommand("CPU"));
  if (!result.ok())
    return result.status();
 
  CpuState state{};
  state.A = static_cast<uint16_t>(ExtractJsonInt(*result, "a"));
  state.X = static_cast<uint16_t>(ExtractJsonInt(*result, "x"));
  state.Y = static_cast<uint16_t>(ExtractJsonInt(*result, "y"));
  state.SP = static_cast<uint16_t>(ExtractJsonInt(*result, "sp"));
  state.D = static_cast<uint16_t>(ExtractJsonInt(*result, "d"));
  state.PC = static_cast<uint32_t>(ExtractJsonInt(*result, "pc"));
  state.K = static_cast<uint8_t>(ExtractJsonInt(*result, "k"));
  state.DBR = static_cast<uint8_t>(ExtractJsonInt(*result, "dbr"));
  state.P = static_cast<uint8_t>(ExtractJsonInt(*result, "p"));
  state.emulation_mode = ExtractJsonBool(*result, "emulationMode");
  return state;
}
 
absl::StatusOr<std::string> MesenSocketClient::Disassemble(uint32_t addr,
                                                           int count) {
  auto result = SendCommand(
      BuildJsonCommand("DISASM", {{"addr", absl::StrFormat("0x%06X", addr)},
                                  {"count", std::to_string(count)}}));
  if (!result.ok())
    return result.status();
  return *result;
}
 
absl::StatusOr<int> MesenSocketClient::AddBreakpoint(
    uint32_t addr, BreakpointType type, const std::string& condition) {
  std::string type_str;
  switch (type) {
    case BreakpointType::kExecute:
      type_str = "exec";
      break;
    case BreakpointType::kRead:
      type_str = "read";
      break;
    case BreakpointType::kWrite:
      type_str = "write";
      break;
    case BreakpointType::kReadWrite:
      type_str = "rw";
      break;
  }
 
  std::vector<std::pair<std::string, std::string>> params = {
      {"action", "add"},
      {"addr", absl::StrFormat("0x%06X", addr)},
      {"bptype", type_str}};
 
  if (!condition.empty()) {
    params.push_back({"condition", condition});
  }
 
  auto result = SendCommand(BuildJsonCommand("BREAKPOINT", params));
  if (!result.ok())
    return result.status();
 
  return static_cast<int>(ExtractJsonInt(*result, "id", -1));
}
 
absl::Status MesenSocketClient::RemoveBreakpoint(int id) {
  auto result = SendCommand(BuildJsonCommand(
      "BREAKPOINT", {{"action", "remove"}, {"id", std::to_string(id)}}));
  return result.status();
}
 
absl::Status MesenSocketClient::ClearBreakpoints() {
  auto result =
      SendCommand(BuildJsonCommand("BREAKPOINT", {{"action", "clear"}}));
  return result.status();
}
 
absl::StatusOr<std::string> MesenSocketClient::GetTrace(int count) {
  auto result = SendCommand(
      BuildJsonCommand("TRACE", {{"count", std::to_string(count)}}));
  if (!result.ok())
    return result.status();
  return *result;
}
 
// ─────────────────────────────────────────────────────────────────────────────
// ALTTP Commands
// ─────────────────────────────────────────────────────────────────────────────
 
absl::StatusOr<GameState> MesenSocketClient::GetGameState() {
  auto result = SendCommand(BuildJsonCommand("GAMESTATE"));
  if (!result.ok())
    return result.status();
 
  GameState state;
 
  // Parse link state
  std::string link_json = ExtractJsonString(*result, "link");
  state.link.x = static_cast<uint16_t>(ExtractJsonInt(link_json, "x"));
  state.link.y = static_cast<uint16_t>(ExtractJsonInt(link_json, "y"));
  state.link.layer = static_cast<uint8_t>(ExtractJsonInt(link_json, "layer"));
  state.link.direction =
      static_cast<uint8_t>(ExtractJsonInt(link_json, "direction"));
  state.link.state = static_cast<uint8_t>(ExtractJsonInt(link_json, "state"));
  state.link.pose = static_cast<uint8_t>(ExtractJsonInt(link_json, "pose"));
 
  // Parse health
  std::string health_json = ExtractJsonString(*result, "health");
  state.items.current_health =
      static_cast<uint8_t>(ExtractJsonInt(health_json, "current"));
  state.items.max_health =
      static_cast<uint8_t>(ExtractJsonInt(health_json, "max"));
 
  // Parse items
  std::string items_json = ExtractJsonString(*result, "items");
  state.items.magic = static_cast<uint8_t>(ExtractJsonInt(items_json, "magic"));
  state.items.rupees =
      static_cast<uint16_t>(ExtractJsonInt(items_json, "rupees"));
  state.items.bombs = static_cast<uint8_t>(ExtractJsonInt(items_json, "bombs"));
  state.items.arrows =
      static_cast<uint8_t>(ExtractJsonInt(items_json, "arrows"));
 
  // Parse game mode
  std::string game_json = ExtractJsonString(*result, "game");
  state.game.mode = static_cast<uint8_t>(ExtractJsonInt(game_json, "mode"));
  state.game.submode =
      static_cast<uint8_t>(ExtractJsonInt(game_json, "submode"));
  state.game.indoors = ExtractJsonBool(game_json, "indoors");
  state.game.room_id =
      static_cast<uint16_t>(ExtractJsonInt(game_json, "room_id"));
  state.game.overworld_area =
      static_cast<uint8_t>(ExtractJsonInt(game_json, "overworld_area"));
 
  return state;
}
 
absl::StatusOr<std::vector<SpriteInfo>> MesenSocketClient::GetSprites(
    bool all) {
  std::vector<std::pair<std::string, std::string>> params;
  if (all) {
    params.push_back({"all", "true"});
  }
 
  auto result =
      SendCommand(params.empty() ? BuildJsonCommand("SPRITES")
                                 : BuildJsonCommand("SPRITES", params));
  if (!result.ok())
    return result.status();
 
  std::vector<SpriteInfo> sprites;
 
  // Find sprites array in response
  size_t array_start = result->find("[");
  size_t array_end = result->rfind("]");
  if (array_start == std::string::npos || array_end == std::string::npos) {
    return sprites;
  }
 
  std::string array_content =
      result->substr(array_start + 1, array_end - array_start - 1);
 
  // Parse each sprite object
  size_t pos = 0;
  while ((pos = array_content.find("{", pos)) != std::string::npos) {
    size_t end = array_content.find("}", pos);
    if (end == std::string::npos)
      break;
 
    std::string sprite_json = array_content.substr(pos, end - pos + 1);
 
    SpriteInfo sprite;
    sprite.slot = static_cast<int>(ExtractJsonInt(sprite_json, "slot"));
    sprite.type = static_cast<uint8_t>(ExtractJsonInt(sprite_json, "type"));
    sprite.state = static_cast<uint8_t>(ExtractJsonInt(sprite_json, "state"));
    sprite.x = static_cast<uint16_t>(ExtractJsonInt(sprite_json, "x"));
    sprite.y = static_cast<uint16_t>(ExtractJsonInt(sprite_json, "y"));
    sprite.health = static_cast<uint8_t>(ExtractJsonInt(sprite_json, "health"));
    sprite.subtype =
        static_cast<uint8_t>(ExtractJsonInt(sprite_json, "subtype"));
 
    sprites.push_back(sprite);
    pos = end + 1;
  }
 
  return sprites;
}
 
absl::Status MesenSocketClient::SetCollisionOverlay(bool enable,
                                                    const std::string& colmap) {
  auto result = SendCommand(BuildJsonCommand(
      "COLLISION_OVERLAY",
      {{"action", enable ? "enable" : "disable"}, {"colmap", colmap}}));
  return result.status();
}
 
// ─────────────────────────────────────────────────────────────────────────────
// Save State Commands
// ─────────────────────────────────────────────────────────────────────────────
 
absl::Status MesenSocketClient::SaveState(int slot) {
  auto result = SendCommand(
      BuildJsonCommand("SAVESTATE", {{"slot", std::to_string(slot)}}));
  return result.status();
}
 
absl::Status MesenSocketClient::LoadState(int slot) {
  auto result = SendCommand(
      BuildJsonCommand("LOADSTATE", {{"slot", std::to_string(slot)}}));
  return result.status();
}
 
absl::StatusOr<std::string> MesenSocketClient::Screenshot() {
  auto result = SendCommand(BuildJsonCommand("SCREENSHOT"));
  if (!result.ok())
    return result.status();
  return *result;  // Base64 PNG data
}
 
// ─────────────────────────────────────────────────────────────────────────────
// Event Subscription
// ─────────────────────────────────────────────────────────────────────────────
 
absl::Status MesenSocketClient::Subscribe(
    const std::vector<std::string>& events) {
  if (!IsConnected()) {
    return absl::FailedPreconditionError("Not connected to Mesen2");
  }
  if (events.empty()) {
    return absl::InvalidArgumentError("At least one event must be requested");
  }
 
  // Ensure we don't leak a previous subscription socket/thread.
  (void)Unsubscribe();
 
  std::stringstream ss;
  for (size_t i = 0; i < events.size(); ++i) {
    if (i > 0)
      ss << ",";
    ss << events[i];
  }
 
  SocketHandle event_fd = kInvalidSocketHandle;
  auto connect_status = ConnectSocketToPath(socket_path_, &event_fd);
  if (!connect_status.ok()) {
    return connect_status;
  }
 
  const std::string subscribe_command =
      BuildJsonCommand("SUBSCRIBE", {{"events", ss.str()}});
  if (const absl::Status sent = SendAll(event_fd, subscribe_command);
      !sent.ok()) {
    close(event_fd);
    return sent;
  }
 
  SetSocketReceiveTimeout(event_fd, 5000);
 
  std::string response;
  char buffer[4096];
  while (true) {
    const ssize_t received = recv(event_fd, buffer, sizeof(buffer), 0);
    if (received < 0) {
      const int error = LastSocketError();
#ifdef _WIN32
      if (error == WSAEINTR) {
        continue;
      }
      if (error == WSAEWOULDBLOCK || error == WSAETIMEDOUT) {
#else
      if (error == EINTR) {
        continue;
      }
      if (error == EAGAIN || error == EWOULDBLOCK) {
#endif
        close(event_fd);
        return absl::DeadlineExceededError(
            "Timeout waiting for subscribe response");
      }
      close(event_fd);
      return absl::InternalError(absl::StrCat("Failed to receive response: ",
                                              SocketErrorMessage(error)));
    }
    if (received == 0) {
      break;
    }
    response.append(buffer, static_cast<size_t>(received));
    if (response.size() > kMaxResponseSize) {
      close(event_fd);
      return absl::ResourceExhaustedError("Mesen2 response too large");
    }
    if (response.find('\n') != std::string::npos) {
      break;
    }
  }
 
  const size_t newline_pos = response.find('\n');
  if (newline_pos == std::string::npos) {
    close(event_fd);
    return absl::DeadlineExceededError("Malformed subscribe response");
  }
 
  auto parse_status = ParseResponse(response.substr(0, newline_pos));
  if (!parse_status.ok()) {
    close(event_fd);
    return parse_status.status();
  }
 
  // The handshake deadline must not outlive the handshake: the event stream
  // idles between emulator events. Swap it for a short poll interval so a
  // quiet stream never looks like a failure, while Unsubscribe() still has a
  // bounded wakeup to join the event thread (shutdown() alone does not
  // interrupt a blocking recv() on Windows).
  SetSocketReceiveTimeout(event_fd, EventPollMs());
 
  pending_event_payload_ = response.substr(newline_pos + 1);
  event_socket_fd_ = event_fd;
  event_thread_running_ = true;
  event_thread_ = std::thread(&MesenSocketClient::EventLoop, this);
  return absl::OkStatus();
}
 
absl::Status MesenSocketClient::Unsubscribe() {
  event_thread_running_ = false;
  ShutdownSocketFd(event_socket_fd_);
  if (event_thread_.joinable()) {
    event_thread_.join();
  }
 
  if (event_socket_fd_ != kInvalidSocketHandle) {
    close(event_socket_fd_);
    event_socket_fd_ = kInvalidSocketHandle;
  }
  pending_event_payload_.clear();
 
  return absl::OkStatus();
}
 
void MesenSocketClient::SetEventCallback(EventCallback callback) {
  std::lock_guard<std::mutex> lock(event_callback_mutex_);
  event_callback_ = std::move(callback);
}
 
EventListenerId MesenSocketClient::AddEventListener(EventCallback callback) {
  if (!callback) {
    return 0;
  }
  std::lock_guard<std::mutex> lock(event_callback_mutex_);
  const EventListenerId id = next_event_listener_id_++;
  event_listeners_[id] = std::move(callback);
  return id;
}
 
void MesenSocketClient::RemoveEventListener(EventListenerId id) {
  if (id == 0) {
    return;
  }
  std::lock_guard<std::mutex> lock(event_callback_mutex_);
  event_listeners_.erase(id);
}
 
void MesenSocketClient::EventLoop() {
  const SocketHandle event_fd = event_socket_fd_;
  if (event_fd == kInvalidSocketHandle) {
    return;
  }
 
  std::string pending = pending_event_payload_;
  pending_event_payload_.clear();
  char buffer[4096];
 
  auto dispatch_pending_lines = [&]() {
    size_t newline_pos = pending.find('\n');
    while (newline_pos != std::string::npos) {
      std::string line = pending.substr(0, newline_pos);
      pending.erase(0, newline_pos + 1);
 
      if (!line.empty() && line.find("\"success\"") == std::string::npos) {
        MesenEvent event;
        event.raw_json = line;
        event.type = ExtractJsonString(line, "event");
        if (event.type.empty()) {
          event.type = ExtractJsonString(line, "type");
        }
        event.address = static_cast<uint32_t>(
            ExtractJsonInt(line, "address", ExtractJsonInt(line, "addr", 0)));
        event.frame = static_cast<uint64_t>(ExtractJsonInt(line, "frame", 0));
 
        EventCallback callback;
        std::vector<EventCallback> listeners;
        {
          std::lock_guard<std::mutex> lock(event_callback_mutex_);
          callback = event_callback_;
          listeners.reserve(event_listeners_.size());
          for (const auto& [id, listener] : event_listeners_) {
            if (listener) {
              listeners.push_back(listener);
            }
          }
        }
        if (callback && !event.type.empty()) {
          callback(event);
        }
        if (!event.type.empty()) {
          for (const auto& listener : listeners) {
            listener(event);
          }
        }
      }
 
      newline_pos = pending.find('\n');
    }
  };
 
  while (event_thread_running_) {
    dispatch_pending_lines();
 
    const ssize_t received = recv(event_fd, buffer, sizeof(buffer), 0);
    if (received < 0) {
      // A receive timeout means "no event yet", not "stream over"; only a
      // real socket error or a closed peer ends the loop.
#ifdef _WIN32
      const int err = WSAGetLastError();
      if (err == WSAEINTR || err == WSAEWOULDBLOCK || err == WSAETIMEDOUT) {
        continue;
      }
#else
      if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK ||
          errno == ETIMEDOUT) {
        continue;
      }
#endif
      break;
    }
 
    if (received == 0) {
      break;
    }
 
    pending.append(buffer, static_cast<size_t>(received));
    dispatch_pending_lines();
  }
}
 
}  // namespace mesen
}  // namespace emu
}  // namespace yaze