src/client/server.c

Go to the documentation of this file.

 
#include "server.h"
#include "main.h"
#include "crypto.h"
#include "crypto/crypto.h"
#include "crypto/handshake/common.h"
 
#include "platform/abstraction.h"
#include "platform/terminal.h"
#include "platform/system.h"
#include "network/packet.h"
#include "network/network.h"
#include "network/acip/send.h"
#include "network/acip/transport.h"
#include "util/endian.h"
#include "util/ip.h"
#include "common.h"
#include "display.h"
#include "options/options.h"
#include "options/rcu.h" // For RCU-based options access
#include "video/palette.h"
#include "buffer_pool.h"
 
#include <string.h>
#include <stdarg.h>
#include <time.h>
#include <sys/types.h>
#include <stdatomic.h>
#ifndef _WIN32
#include <netinet/tcp.h>
#include <netdb.h>     // For getaddrinfo(), gai_strerror()
#include <arpa/inet.h> // For inet_ntop()
#else
#include <ws2tcpip.h> // For getaddrinfo(), gai_strerror(), inet_ntop()
#endif
 
// Debug flags
#define DEBUG_NETWORK 1
#define DEBUG_THREADS 1
#define DEBUG_MEMORY 1
 
/* ============================================================================
 * Connection State Management
 * ============================================================================ */
 
static socket_t g_sockfd = INVALID_SOCKET_VALUE;
 
static acip_transport_t *g_client_transport = NULL;
 
static atomic_bool g_connection_active = false;
 
static atomic_bool g_connection_lost = false;
 
static atomic_bool g_should_reconnect = false;
 
static uint32_t g_my_client_id = 0;
 
static char g_server_ip[256] = {0};
 
static mutex_t g_send_mutex = {0};
 
/* ============================================================================
 * Crypto State
 * ============================================================================ */
 
crypto_handshake_context_t g_crypto_ctx = {0};
 
static bool g_encryption_enabled = false;
 
/* ============================================================================
 * Reconnection Logic
 * ============================================================================ */
 
#define MAX_RECONNECT_DELAY (5 * 1000 * 1000)
 
static unsigned int get_reconnect_delay(unsigned int reconnect_attempt) {
  // Use integer arithmetic for microsecond calculations
  // Initial delay: 100,000 us (0.1 seconds)
  // Additional delay per attempt: 200,000 us (0.2 seconds)
  unsigned int delay_us = 100000 + (reconnect_attempt - 1) * 200000;
  if (delay_us > MAX_RECONNECT_DELAY)
    delay_us = MAX_RECONNECT_DELAY;
  return delay_us;
}
 
/* ============================================================================
 * Socket Management Functions
 * ============================================================================ */
 
static int close_socket(socket_t socketfd) {
  if (socket_is_valid(socketfd)) {
    log_debug("Closing socket %d", socketfd);
 
    if (socket_close(socketfd) != 0) {
      log_error("Failed to close socket: %s", network_error_string());
      return -1;
    }
 
    // Small delay to ensure socket resources are fully released
    // This prevents WSA error 10038 on subsequent connections
    platform_sleep_usec(50000); // 50ms delay
 
    return 0;
  }
 
  return 0; // Socket already closed or invalid
}
 
/* ============================================================================
 * Public Interface Functions
 * ============================================================================ */
 
int server_connection_init() {
  // Initialize mutex for thread-safe packet sending
  if (mutex_init(&g_send_mutex) != 0) {
    log_error("Failed to initialize send mutex");
    return -1;
  }
 
  // Initialize connection state
  g_sockfd = INVALID_SOCKET_VALUE;
  g_client_transport = NULL;
  atomic_store(&g_connection_active, false);
  atomic_store(&g_connection_lost, false);
  atomic_store(&g_should_reconnect, false);
  g_my_client_id = 0;
 
  return 0;
}
 
int server_connection_establish(const char *address, int port, int reconnect_attempt, bool first_connection,
                                bool has_ever_connected) {
  (void)first_connection; // Currently unused
  if (!address || port <= 0) {
    log_error("Invalid address or port parameters");
    return -1;
  }
 
  // Close any existing connection
  if (g_sockfd != INVALID_SOCKET_VALUE) {
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
  }
 
  // Apply reconnection delay if this is a retry
  if (reconnect_attempt > 0) {
    unsigned int delay_us = get_reconnect_delay(reconnect_attempt);
    // Reconnection attempt logged only to file
    platform_sleep_usec(delay_us);
 
    // Check if user requested exit during reconnection delay
    if (should_exit()) {
      log_debug("Exit requested during reconnection delay");
      return -1;
    }
  } else {
    // Initial connection logged only to file
  }
 
  // Resolve server address using getaddrinfo() for IPv4/IPv6 support
  // Special handling for localhost: ensure we try both IPv6 (::1) and IPv4 (127.0.0.1)
  // Many systems only map "localhost" to 127.0.0.1 in /etc/hosts
  bool is_localhost =
      (strcmp(address, "localhost") == 0 || strcmp(address, "127.0.0.1") == 0 || strcmp(address, "::1") == 0);
 
  struct addrinfo hints, *res = NULL, *addr_iter;
  memset(&hints, 0, sizeof(hints));
  hints.ai_family = AF_UNSPEC; // Allow IPv4 or IPv6
  hints.ai_socktype = SOCK_STREAM;
  if (is_localhost) {
    hints.ai_flags = AI_NUMERICSERV; // Optimize for localhost
  }
 
  char port_str[16];
  SAFE_SNPRINTF(port_str, sizeof(port_str), "%d", port);
 
  // For localhost, try IPv6 loopback (::1) first, then fall back to DNS resolution
  if (is_localhost) {
    log_debug("Localhost detected - trying IPv6 loopback [::1]:%s first...", port_str);
    hints.ai_family = AF_INET6;
    hints.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV;
 
    int ipv6_result = getaddrinfo("::1", port_str, &hints, &res);
    if (ipv6_result == 0 && res != NULL) {
      // Try IPv6 loopback connection
      g_sockfd = socket_create(res->ai_family, res->ai_socktype, res->ai_protocol);
      if (g_sockfd != INVALID_SOCKET_VALUE) {
        log_info("Trying IPv6 loopback connection to [::1]:%s...", port_str);
        if (connect_with_timeout(g_sockfd, res->ai_addr, res->ai_addrlen, CONNECT_TIMEOUT)) {
          log_debug("Connection successful using IPv6 loopback");
          SAFE_STRNCPY(g_server_ip, "::1", sizeof(g_server_ip));
          freeaddrinfo(res);
          res = NULL; // Prevent double-free at connection_success label
          goto connection_success;
        }
        if (socket_get_error(g_sockfd) != 0) {
          log_debug("NETWORK_ERROR: %d", (int)socket_get_error(g_sockfd));
        } else {
          // log_debug("IPv6 loopback connection failed: %s", network_error_string());
        }
        close_socket(g_sockfd);
        g_sockfd = INVALID_SOCKET_VALUE;
      }
      freeaddrinfo(res);
      res = NULL;
    }
 
    // Check if user requested exit (Ctrl-C) before trying IPv4
    if (should_exit()) {
      log_debug("Exit requested during connection attempt");
      return -1;
    }
 
    // IPv6 failed, try IPv4 loopback (127.0.0.1)
    log_debug("IPv6 failed, trying IPv4 loopback 127.0.0.1:%s...", port_str);
    hints.ai_family = AF_INET;
 
    int ipv4_result = getaddrinfo("127.0.0.1", port_str, &hints, &res);
    if (ipv4_result == 0 && res != NULL) {
      g_sockfd = socket_create(res->ai_family, res->ai_socktype, res->ai_protocol);
      if (g_sockfd != INVALID_SOCKET_VALUE) {
        log_info("Trying IPv4 loopback connection to 127.0.0.1:%s...", port_str);
        if (connect_with_timeout(g_sockfd, res->ai_addr, res->ai_addrlen, CONNECT_TIMEOUT)) {
          log_debug("Connection successful using IPv4 loopback");
          SAFE_STRNCPY(g_server_ip, "127.0.0.1", sizeof(g_server_ip));
          freeaddrinfo(res);
          res = NULL; // Prevent double-free at connection_success label
          goto connection_success;
        }
        if (socket_get_error(g_sockfd) != 0) {
          log_debug("NETWORK_ERROR: %d", (int)socket_get_error(g_sockfd));
        } else {
          // log_debug("IPv4 loopback connection failed: %s", network_error_string());
        }
        close_socket(g_sockfd);
        g_sockfd = INVALID_SOCKET_VALUE;
      }
      freeaddrinfo(res);
      res = NULL;
    }
 
    // Both IPv6 and IPv4 loopback failed for localhost
    log_warn("Could not connect to localhost using either IPv6 or IPv4 loopback");
    return -1;
  }
 
  // For non-localhost addresses, use standard resolution
  log_debug("Resolving server address '%s' port %s...", address, port_str);
  hints.ai_family = AF_UNSPEC;
  hints.ai_flags = 0;
  int getaddr_result = getaddrinfo(address, port_str, &hints, &res);
  if (getaddr_result != 0) {
    log_error("Failed to resolve server address '%s': %s", address, gai_strerror(getaddr_result));
    return -1;
  }
 
  // Try each address returned by getaddrinfo() until one succeeds
  // Prefer IPv6 over IPv4: try IPv6 addresses first, then fall back to IPv4
  for (int address_family = AF_INET6; address_family >= AF_INET; address_family -= (AF_INET6 - AF_INET)) {
    for (addr_iter = res; addr_iter != NULL; addr_iter = addr_iter->ai_next) {
      // Skip addresses that don't match current pass (IPv6 first, then IPv4)
      if (addr_iter->ai_family != address_family) {
        continue;
      }
 
      // Create socket with appropriate address family
      g_sockfd = socket_create(addr_iter->ai_family, addr_iter->ai_socktype, addr_iter->ai_protocol);
      if (g_sockfd == INVALID_SOCKET_VALUE) {
        log_debug("Could not create socket for address family %d: %s", addr_iter->ai_family, network_error_string());
        continue; // Try next address
      }
 
      // Log which address family we're trying
      if (addr_iter->ai_family == AF_INET) {
        log_debug("Trying IPv4 connection...");
      } else if (addr_iter->ai_family == AF_INET6) {
        log_debug("Trying IPv6 connection...");
      }
 
      // Attempt connection with timeout
      if (connect_with_timeout(g_sockfd, addr_iter->ai_addr, addr_iter->ai_addrlen, CONNECT_TIMEOUT)) {
        // Connection successful!
        log_debug("Connection successful using %s", addr_iter->ai_family == AF_INET    ? "IPv4"
                                                    : addr_iter->ai_family == AF_INET6 ? "IPv6"
                                                                                       : "unknown protocol");
 
        // Extract server IP address for known_hosts
        if (format_ip_address(addr_iter->ai_family, addr_iter->ai_addr, g_server_ip, sizeof(g_server_ip)) ==
            ASCIICHAT_OK) {
          log_debug("Resolved server IP: %s", g_server_ip);
        } else {
          log_warn("Failed to format server IP address");
        }
 
        goto connection_success; // Break out of both loops
      }
 
      // Connection failed - close socket and try next address
      if (socket_get_error(g_sockfd) != 0) {
        log_debug("NETWORK_ERROR: %d", (int)socket_get_error(g_sockfd));
      } else {
        // log_debug("Connection failed: %s", network_error_string());
      }
      close_socket(g_sockfd);
      g_sockfd = INVALID_SOCKET_VALUE;
    }
  }
 
connection_success:
 
  if (res) {
    freeaddrinfo(res);
  }
 
  // If we exhausted all addresses without success, fail
  if (g_sockfd == INVALID_SOCKET_VALUE) {
    log_warn("Could not connect to server %s:%d (tried all addresses)", address, port);
    return -1;
  }
 
  // Connection successful - extract local port for client ID
  struct sockaddr_storage local_addr = {0};
  socklen_t addr_len = sizeof(local_addr);
  if (getsockname(g_sockfd, (struct sockaddr *)&local_addr, &addr_len) == -1) {
    log_error("Failed to get local socket address: %s", network_error_string());
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
    return -1;
  }
 
  // Extract port from either IPv4 or IPv6 address
  int local_port = 0;
  if (((struct sockaddr *)&local_addr)->sa_family == AF_INET) {
    local_port = NET_TO_HOST_U16(((struct sockaddr_in *)&local_addr)->sin_port);
  } else if (((struct sockaddr *)&local_addr)->sa_family == AF_INET6) {
    local_port = NET_TO_HOST_U16(((struct sockaddr_in6 *)&local_addr)->sin6_port);
  }
  g_my_client_id = (uint32_t)local_port;
 
  // Mark connection as active immediately after successful socket connection
  atomic_store(&g_connection_active, true);
  atomic_store(&g_connection_lost, false);
  atomic_store(&g_should_reconnect, false);
 
  // Initialize crypto BEFORE starting protocol handshake
  // Note: server IP is already set above in the connection loop
  log_debug("CLIENT_CONNECT: Calling client_crypto_init()");
  if (client_crypto_init() != 0) {
    log_error("Failed to initialize crypto (password required or incorrect)");
    log_debug("CLIENT_CONNECT: client_crypto_init() failed");
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
    return CONNECTION_ERROR_AUTH_FAILED; // SSH key password was wrong - no retry
  }
 
  // Perform crypto handshake if encryption is enabled
  log_debug("CLIENT_CONNECT: Calling client_crypto_handshake()");
  int handshake_result = client_crypto_handshake(g_sockfd);
  if (handshake_result != 0) {
    log_error("Crypto handshake failed");
    log_debug("CLIENT_CONNECT: client_crypto_handshake() failed with code %d", handshake_result);
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
    FATAL(ERROR_CRYPTO_HANDSHAKE,
          "Crypto handshake failed with server - this usually indicates a protocol mismatch or network issue");
  }
  log_debug("CLIENT_CONNECT: client_crypto_handshake() succeeded");
 
  // Create ACIP transport for protocol-agnostic packet sending
  // The transport wraps the socket with encryption context from the handshake
  const crypto_context_t *crypto_ctx = crypto_client_is_ready() ? crypto_client_get_context() : NULL;
  g_client_transport = acip_tcp_transport_create(g_sockfd, (crypto_context_t *)crypto_ctx);
  if (!g_client_transport) {
    log_error("Failed to create ACIP transport");
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
    return -1;
  }
  log_debug("CLIENT_CONNECT: Created ACIP transport with crypto context");
 
  // Turn OFF terminal logging when successfully connected to server
  // First connection - we'll disable logging after main.c shows the "Connected successfully" message
  if (!GET_OPTION(snapshot_mode)) {
    log_debug("Connected to server - terminal logging will be disabled after initial setup");
  } else {
    log_debug("Connected to server - terminal logging kept enabled for snapshot mode");
  }
 
  // Configure socket options for optimal performance
  if (socket_set_keepalive(g_sockfd, true) < 0) {
    log_warn("Failed to set socket keepalive: %s", network_error_string());
  }
 
  // Configure socket buffers and TCP_NODELAY for optimal performance
  asciichat_error_t sock_config_result = socket_configure_buffers(g_sockfd);
  if (sock_config_result != ASCIICHAT_OK) {
    log_warn("Failed to configure socket: %s", network_error_string());
  }
 
  // Send initial terminal capabilities to server (this may generate debug logs)
  int result = threaded_send_terminal_size_with_auto_detect(GET_OPTION(width), GET_OPTION(height));
  if (result < 0) {
    log_error("Failed to send initial capabilities to server: %s", network_error_string());
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
    return -1;
  }
 
  // Now disable terminal logging after capabilities are sent (for reconnections)
  if (!GET_OPTION(snapshot_mode) && has_ever_connected) {
    log_set_terminal_output(false);
    log_debug("Reconnected to server - terminal logging disabled to prevent interference with ASCII display");
  }
 
  // Send client join packet for multi-user support
  uint32_t my_capabilities = CLIENT_CAP_VIDEO; // Basic video capability
  log_info("GET_OPTION(audio_enabled) = %d (sending CLIENT_JOIN)", GET_OPTION(audio_enabled));
  if (GET_OPTION(audio_enabled)) {
    log_info("Adding CLIENT_CAP_AUDIO to capabilities");
    my_capabilities |= CLIENT_CAP_AUDIO;
  }
  if (GET_OPTION(color_mode) != COLOR_MODE_NONE) {
    my_capabilities |= CLIENT_CAP_COLOR;
  }
  if (GET_OPTION(stretch)) {
    my_capabilities |= CLIENT_CAP_STRETCH;
  }
 
  // Generate display name from username + PID
  const char *display_name = platform_get_username();
 
  char my_display_name[MAX_DISPLAY_NAME_LEN];
  int pid = getpid();
  SAFE_SNPRINTF(my_display_name, sizeof(my_display_name), "%s-%d", display_name, pid);
 
  if (threaded_send_client_join_packet(my_display_name, my_capabilities) < 0) {
    log_error("Failed to send client join packet: %s", network_error_string());
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
    return -1;
  }
 
  // Connection already marked as active after socket creation
 
  return 0;
}
 
bool server_connection_is_active() {
  return atomic_load(&g_connection_active) && (g_sockfd != INVALID_SOCKET_VALUE);
}
 
socket_t server_connection_get_socket() {
  return g_sockfd;
}
 
acip_transport_t *server_connection_get_transport(void) {
  return g_client_transport;
}
 
void server_connection_set_transport(acip_transport_t *transport) {
  log_debug("server_connection_set_transport() called with transport=%p", (void *)transport);
 
  // Clean up any existing transport
  if (g_client_transport) {
    log_warn("Replacing existing transport with new fallback transport");
    acip_transport_destroy(g_client_transport);
  }
 
  log_debug("Setting g_client_transport to %p", (void *)transport);
  g_client_transport = transport;
 
  // Mark connection as active when transport is set
  if (transport) {
    log_debug("Transport is non-NULL, extracting socket...");
    // Extract socket from transport for backward compatibility with socket-based checks
    g_sockfd = acip_transport_get_socket(transport);
    log_debug("Socket extracted: %d", (int)g_sockfd);
 
    atomic_store(&g_connection_active, true);
    log_debug("Server connection transport set and marked active (sockfd=%d)", (int)g_sockfd);
  } else {
    g_sockfd = INVALID_SOCKET_VALUE;
    atomic_store(&g_connection_active, false);
    log_debug("Server connection transport cleared and marked inactive");
  }
 
  log_debug("server_connection_set_transport() completed");
}
 
uint32_t server_connection_get_client_id() {
  return g_my_client_id;
}
 
const char *server_connection_get_ip() {
  return g_server_ip;
}
 
void server_connection_set_ip(const char *ip) {
  if (ip) {
    SAFE_STRNCPY(g_server_ip, ip, sizeof(g_server_ip));
    log_debug("Server IP set to: %s", g_server_ip);
  } else {
    g_server_ip[0] = '\0';
    log_debug("Server IP cleared");
  }
}
 
void server_connection_close() {
  atomic_store(&g_connection_active, false);
 
  // Destroy ACIP transport before closing socket
  if (g_client_transport) {
    acip_transport_destroy(g_client_transport);
    g_client_transport = NULL;
  }
 
  if (g_sockfd != INVALID_SOCKET_VALUE) {
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
  }
 
  g_my_client_id = 0;
 
  // Cleanup crypto context if encryption was enabled
  if (g_encryption_enabled) {
    crypto_handshake_cleanup(&g_crypto_ctx);
    g_encryption_enabled = false;
  }
 
  // Turn ON terminal logging when connection is closed
  log_set_terminal_output(true);
}
 
void server_connection_shutdown() {
  // NOTE: This function may be called from:
  //   - Signal handlers on Unix (async-signal-safe context)
  //   - SetConsoleCtrlHandler callback thread on Windows (separate thread context)
  // Only use atomic operations and simple system calls - NO mutex locks, NO malloc, NO logging.
 
  atomic_store(&g_connection_active, false);
  atomic_store(&g_connection_lost, true);
 
  if (g_sockfd != INVALID_SOCKET_VALUE) {
    // Only shutdown() the socket to interrupt blocking recv()/send() operations.
    // Do NOT close() here - on Windows, closing the socket while another thread
    // is using it is undefined behavior and can cause STATUS_STACK_BUFFER_OVERRUN.
    // The actual socket close happens in server_connection_close() which is called
    // from the main thread after worker threads have been joined.
    socket_shutdown(g_sockfd, SHUT_RDWR);
  }
 
  // DO NOT call log_set_terminal_output() here - it uses mutex which is NOT async-signal-safe.
  // The normal cleanup path in shutdown_client() will handle logging state.
}
 
void server_connection_lost() {
  atomic_store(&g_connection_lost, true);
  atomic_store(&g_connection_active, false);
 
  // Turn ON terminal logging when connection is lost
  log_set_terminal_output(true);
  display_full_reset();
}
 
bool server_connection_is_lost() {
  return atomic_load(&g_connection_lost);
}
 
void server_connection_cleanup() {
  log_set_terminal_output(true);
  server_connection_close();
  mutex_destroy(&g_send_mutex);
}
 
/* ============================================================================
 * Thread Safety Interface
 * ============================================================================ */
 
/* ============================================================================
 * Thread-Safe Wrapper Functions
 * ============================================================================ */
 
asciichat_error_t threaded_send_packet(packet_type_t type, const void *data, size_t len) {
  mutex_lock(&g_send_mutex);
 
  // Recheck connection status and transport INSIDE the mutex to prevent TOCTOU race
  if (!atomic_load(&g_connection_active) || !g_client_transport) {
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_NETWORK, "Connection not active or transport unavailable");
  }
 
  // Use ACIP transport which handles encryption and compression automatically
  asciichat_error_t result = packet_send_via_transport(g_client_transport, type, data, len);
 
  mutex_unlock(&g_send_mutex);
 
  // If send failed due to network error, signal connection loss
  if (result != ASCIICHAT_OK) {
    server_connection_lost();
    return result;
  }
 
  return ASCIICHAT_OK;
}
 
int threaded_send_audio_batch_packet(const float *samples, int num_samples, int batch_count) {
  mutex_lock(&g_send_mutex);
 
  // Recheck connection status INSIDE the mutex to prevent TOCTOU race
  socket_t sockfd = server_connection_get_socket();
  if (!atomic_load(&g_connection_active) || sockfd == INVALID_SOCKET_VALUE) {
    mutex_unlock(&g_send_mutex);
    return -1;
  }
 
  // Get crypto context if encryption is enabled
  const crypto_context_t *crypto_ctx = crypto_client_is_ready() ? crypto_client_get_context() : NULL;
  int result = send_audio_batch_packet(sockfd, samples, num_samples, batch_count, (crypto_context_t *)crypto_ctx);
 
  mutex_unlock(&g_send_mutex);
 
  // If send failed due to network error, signal connection loss
  if (result < 0) {
    server_connection_lost();
  }
 
  return result;
}
 
asciichat_error_t threaded_send_audio_opus(const uint8_t *opus_data, size_t opus_size, int sample_rate,
                                           int frame_duration) {
  mutex_lock(&g_send_mutex);
 
  // Recheck connection status and transport INSIDE the mutex to prevent TOCTOU race
  if (!atomic_load(&g_connection_active) || !g_client_transport) {
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_NETWORK, "Connection not active or transport unavailable");
  }
 
  // Build Opus packet with header
  size_t header_size = 16; // sample_rate (4), frame_duration (4), reserved (8)
  size_t total_size = header_size + opus_size;
  void *packet_data = buffer_pool_alloc(NULL, total_size);
  if (!packet_data) {
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_MEMORY, "Failed to allocate buffer for Opus packet: %zu bytes", total_size);
  }
 
  // Write header in network byte order
  uint8_t *buf = (uint8_t *)packet_data;
  uint32_t sr = HOST_TO_NET_U32((uint32_t)sample_rate);
  uint32_t fd = HOST_TO_NET_U32((uint32_t)frame_duration);
  memcpy(buf, &sr, 4);
  memcpy(buf + 4, &fd, 4);
  memset(buf + 8, 0, 8); // Reserved
 
  // Copy Opus data
  memcpy(buf + header_size, opus_data, opus_size);
 
  // Send packet via ACIP transport (handles encryption automatically)
  asciichat_error_t result =
      packet_send_via_transport(g_client_transport, PACKET_TYPE_AUDIO_OPUS, packet_data, total_size);
 
  // Clean up
  buffer_pool_free(NULL, packet_data, total_size);
  mutex_unlock(&g_send_mutex);
 
  // If send failed due to network error, signal connection loss
  if (result != ASCIICHAT_OK) {
    server_connection_lost();
    return result;
  }
 
  return ASCIICHAT_OK;
}
 
asciichat_error_t threaded_send_audio_opus_batch(const uint8_t *opus_data, size_t opus_size,
                                                 const uint16_t *frame_sizes, int frame_count) {
  mutex_lock(&g_send_mutex);
 
  // Recheck connection status and transport INSIDE the mutex to prevent TOCTOU race
  if (!atomic_load(&g_connection_active) || !g_client_transport) {
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_NETWORK, "Connection not active or transport unavailable");
  }
 
  // Opus uses 20ms frames at 48kHz (960 samples = 20ms)
  asciichat_error_t result =
      acip_send_audio_opus_batch(g_client_transport, opus_data, opus_size, frame_sizes, frame_count, 48000, 20);
 
  mutex_unlock(&g_send_mutex);
 
  // If send failed due to network error, signal connection loss
  if (result != ASCIICHAT_OK) {
    server_connection_lost();
    return result;
  }
 
  return ASCIICHAT_OK;
}
 
int threaded_send_ping_packet(void) {
  // Use threaded_send_packet which handles encryption, mutex locking, and connection state
  return threaded_send_packet(PACKET_TYPE_PING, NULL, 0);
}
 
int threaded_send_pong_packet(void) {
  // Use threaded_send_packet which handles encryption, mutex locking, and connection state
  return threaded_send_packet(PACKET_TYPE_PONG, NULL, 0);
}
 
int threaded_send_stream_start_packet(uint32_t stream_type) {
  socket_t sockfd = server_connection_get_socket();
  if (!atomic_load(&g_connection_active) || sockfd == INVALID_SOCKET_VALUE) {
    return -1;
  }
 
  // Build STREAM_START packet locally
  uint32_t type_data = HOST_TO_NET_U32(stream_type);
 
  // Use threaded_send_packet() which handles encryption
  return threaded_send_packet(PACKET_TYPE_STREAM_START, &type_data, sizeof(type_data));
}
 
int threaded_send_terminal_size_with_auto_detect(unsigned short width, unsigned short height) {
 
  socket_t sockfd = server_connection_get_socket();
  if (!atomic_load(&g_connection_active) || sockfd == INVALID_SOCKET_VALUE) {
    return -1;
  }
 
  // Build terminal capabilities packet locally
  // Detect terminal capabilities automatically
  terminal_capabilities_t caps = detect_terminal_capabilities();
 
  // Apply user's color mode override
  caps = apply_color_mode_override(caps);
 
  // Check if detection was reliable, use fallback only for auto-detection
  if (!caps.detection_reliable && GET_OPTION(color_mode) == COLOR_MODE_AUTO) {
    log_warn("Terminal capability detection not reliable, using fallback");
    SAFE_MEMSET(&caps, sizeof(caps), 0, sizeof(caps));
    caps.color_level = TERM_COLOR_NONE;
    caps.color_count = 2;
    caps.capabilities = 0;
    SAFE_STRNCPY(caps.term_type, "unknown", sizeof(caps.term_type));
    SAFE_STRNCPY(caps.colorterm, "", sizeof(caps.colorterm));
    caps.detection_reliable = 0;
  }
 
  // Convert to network packet format with proper byte order
  terminal_capabilities_packet_t net_packet;
  net_packet.capabilities = HOST_TO_NET_U32(caps.capabilities);
  net_packet.color_level = HOST_TO_NET_U32(caps.color_level);
  net_packet.color_count = HOST_TO_NET_U32(caps.color_count);
  net_packet.render_mode = HOST_TO_NET_U32(caps.render_mode);
  net_packet.width = HOST_TO_NET_U16(width);
  net_packet.height = HOST_TO_NET_U16(height);
  net_packet.palette_type = HOST_TO_NET_U32(GET_OPTION(palette_type));
  net_packet.utf8_support = HOST_TO_NET_U32(caps.utf8_support ? 1 : 0);
 
  const options_t *opts = options_get();
  if (GET_OPTION(palette_type) == PALETTE_CUSTOM && GET_OPTION(palette_custom_set)) {
    const char *palette_custom = opts && opts->palette_custom_set ? opts->palette_custom : "";
    SAFE_STRNCPY(net_packet.palette_custom, palette_custom, sizeof(net_packet.palette_custom));
    net_packet.palette_custom[sizeof(net_packet.palette_custom) - 1] = '\0';
  } else {
    SAFE_MEMSET(net_packet.palette_custom, sizeof(net_packet.palette_custom), 0, sizeof(net_packet.palette_custom));
  }
 
  // Set desired FPS
  if (g_max_fps > 0) {
    net_packet.desired_fps = (uint8_t)(g_max_fps > 144 ? 144 : g_max_fps);
  } else {
    net_packet.desired_fps = caps.desired_fps;
  }
 
  if (net_packet.desired_fps == 0) {
    net_packet.desired_fps = DEFAULT_MAX_FPS;
  }
 
  SAFE_STRNCPY(net_packet.term_type, caps.term_type, sizeof(net_packet.term_type));
  net_packet.term_type[sizeof(net_packet.term_type) - 1] = '\0';
 
  SAFE_STRNCPY(net_packet.colorterm, caps.colorterm, sizeof(net_packet.colorterm));
  net_packet.colorterm[sizeof(net_packet.colorterm) - 1] = '\0';
 
  net_packet.detection_reliable = caps.detection_reliable;
  net_packet.utf8_support = GET_OPTION(force_utf8) ? 1 : 0;
 
  SAFE_MEMSET(net_packet.reserved, sizeof(net_packet.reserved), 0, sizeof(net_packet.reserved));
 
  // Use threaded_send_packet() which handles encryption
  return threaded_send_packet(PACKET_TYPE_CLIENT_CAPABILITIES, &net_packet, sizeof(net_packet));
}
 
int threaded_send_client_join_packet(const char *display_name, uint32_t capabilities) {
  socket_t sockfd = server_connection_get_socket();
  if (!atomic_load(&g_connection_active) || sockfd == INVALID_SOCKET_VALUE) {
    return -1;
  }
 
  // Build CLIENT_JOIN packet locally
  client_info_packet_t join_packet;
  SAFE_MEMSET(&join_packet, sizeof(join_packet), 0, sizeof(join_packet));
  join_packet.client_id = HOST_TO_NET_U32(0); // Will be assigned by server
  SAFE_SNPRINTF(join_packet.display_name, MAX_DISPLAY_NAME_LEN, "%s", display_name ? display_name : "Unknown");
  join_packet.capabilities = HOST_TO_NET_U32(capabilities);
 
  // Use threaded_send_packet() which handles encryption
  int send_result = threaded_send_packet(PACKET_TYPE_CLIENT_JOIN, &join_packet, sizeof(join_packet));
  if (send_result == 0) {
    mutex_lock(&g_send_mutex);
    bool active = atomic_load(&g_connection_active);
    socket_t socket_snapshot = g_sockfd;
    const crypto_context_t *crypto_ctx = crypto_client_is_ready() ? crypto_client_get_context() : NULL;
    if (active && socket_snapshot != INVALID_SOCKET_VALUE) {
      (void)log_network_message(
          socket_snapshot, (const struct crypto_context_t *)crypto_ctx, LOG_INFO, REMOTE_LOG_DIRECTION_CLIENT_TO_SERVER,
          "CLIENT_JOIN sent (display=\"%s\", capabilities=0x%x)", join_packet.display_name, capabilities);
    }
    mutex_unlock(&g_send_mutex);
  }
  return send_result;
}