src/client/server.c

Go to the documentation of this file.

 
#include "server.h"
#include "main.h"
#include "../main.h" // Global exit API
#include "crypto.h"
#include <ascii-chat/crypto/crypto.h>
#include <ascii-chat/crypto/handshake/common.h>
#include <ascii-chat/video/h265/encoder.h>
 
#include <ascii-chat/platform/abstraction.h>
#include <ascii-chat/platform/terminal.h>
#include <ascii-chat/platform/system.h>
#include <ascii-chat/network/packet/packet.h>
#include <ascii-chat/network/network.h>
#include <ascii-chat/network/acip/send.h>
#include <ascii-chat/network/acip/client.h>
#include <ascii-chat/network/acip/transport.h>
#include <ascii-chat/network/connection_endpoint.h>
#include <ascii-chat/network/connection_factory.h>
#include <ascii-chat/network/websocket/internal.h>
#include <ascii-chat/util/endian.h>
#include <ascii-chat/util/ip.h>
#include <ascii-chat/util/time.h>
#include <ascii-chat/util/url.h> // For WebSocket URL detection
#include <ascii-chat/common.h>
#include "display.h"
#include <ascii-chat/options/options.h>
#include <ascii-chat/options/rcu.h> // For RCU-based options access
#include <ascii-chat/video/ascii/palette.h>
#include <ascii-chat/media/codecs.h>
#include <ascii-chat/buffer_pool.h>
 
#include <string.h>
#include <stdarg.h>
#include <time.h>
#include <sys/types.h>
#include <ascii-chat/atomic.h>
#include <ascii-chat/debug/named.h> // For NAMED_REGISTER_ATOMIC macro
 
#include <ascii-chat/platform/network.h> // Consolidates platform-specific network headers (includes TCP options)
 
// 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_t g_connection_active = {0};
 
static atomic_t g_connection_lost = {0};
 
static atomic_t g_should_reconnect = {0};
 
static uint32_t g_my_client_id = 0;
 
static char g_server_ip[256] = {0};
 
static int g_server_port = 0;
 
static mutex_t g_send_mutex = {0};
 
/* ============================================================================
 * Crypto State
 * ============================================================================ */
 
crypto_handshake_context_t g_crypto_ctx = {0};
 
h265_encoder_t *volatile g_h265_encoder = NULL;
 
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 = 100 * US_PER_MS_INT + (reconnect_attempt - 1) * 200 * US_PER_MS_INT;
  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_us(50 * US_PER_MS_INT); // 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, "send") != 0) {
    log_error("Failed to initialize send mutex");
    return -1;
  }
 
  // Register connection atomics for debug sync state monitoring (one-time registration)
  static bool connection_atomics_registered = false;
  if (!connection_atomics_registered) {
    NAMED_REGISTER_ATOMIC(&g_connection_active, "server_connection_active_flag", NULL);
    NAMED_REGISTER_ATOMIC(&g_connection_lost, "server_connection_lost_flag", NULL);
    NAMED_REGISTER_ATOMIC(&g_should_reconnect, "should_attempt_reconnection_flag", NULL);
    connection_atomics_registered = true;
  }
 
  // Initialize connection state
  g_sockfd = INVALID_SOCKET_VALUE;
  g_client_transport = NULL;
  atomic_store_bool(&g_connection_active, false);
  atomic_store_bool(&g_connection_lost, false);
  atomic_store_bool(&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_us(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
  }
 
  connection_endpoint_t endpoint = {0};
  asciichat_error_t endpoint_result = connection_endpoint_resolve(address, port, &endpoint);
  if (endpoint_result != ASCIICHAT_OK) {
    log_error("Failed to resolve connection endpoint: %s", asciichat_error_string(endpoint_result));
    return -1;
  }
 
  // Check for WebSocket URL - handle separately from TCP
  if (endpoint.protocol == CONNECTION_ENDPOINT_WEBSOCKET) {
    // WebSocket connection - bypass TCP socket creation
    // Use the original address (URL already contains port if specified)
    const char *ws_url = endpoint.input;
    log_info("Connecting via WebSocket: %s (host=%s, port=%u)", ws_url, endpoint.host, endpoint.port);
 
    // Set server IP and port for crypto handshake
    // This ensures the crypto context has proper server address info for known_hosts verification
    server_connection_set_ip(endpoint.host);
    server_connection_set_port(endpoint.port);
    log_debug("CLIENT_CONNECT: Set server IP=%s, port=%u from WebSocket endpoint", endpoint.host, endpoint.port);
 
    // Initialize crypto if encryption is enabled
    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");
      return CONNECTION_ERROR_AUTH_FAILED;
    }
 
    // Get crypto context for transport
    const crypto_context_t *crypto_ctx = crypto_client_is_ready() ? crypto_client_get_context() : NULL;
 
    // Create WebSocket transport through the shared factory.
    asciichat_error_t ws_result =
        connection_factory_open("client", ws_url, 0, (crypto_context_t *)crypto_ctx, &g_client_transport, &endpoint);
    if (ws_result != ASCIICHAT_OK) {
      log_error("Failed to create WebSocket ACIP transport");
      return -1;
    }
    if (!g_client_transport) {
      log_error("Failed to create WebSocket ACIP transport");
      return -1;
    }
    log_debug("CLIENT_CONNECT: Created WebSocket ACIP transport with crypto context");
 
    // Set connection as active
    atomic_store_bool(&g_connection_active, true);
    atomic_store_bool(&g_connection_lost, false);
 
    // CRITICAL: Wait for WebSocket connection to be established before sending
    // The service thread is running but may not have completed TLS handshake yet
    // Don't block indefinitely - use timeout loops to allow responsiveness
    log_debug("CLIENT_CONNECT: Waiting for WebSocket connection to be established...");
    const int CONNECT_CHECK_ATTEMPTS = 300; // 300 * 50ms = 15 seconds timeout
    int connect_attempts = 0;
    while (!acip_transport_is_connected(g_client_transport) && connect_attempts < CONNECT_CHECK_ATTEMPTS) {
      platform_sleep_us(50000); // Sleep 50ms between checks
      connect_attempts++;
      if (connect_attempts % 20 == 0) { // Log every 1 second (20 * 50ms)
        log_info("CLIENT_CONNECT: Waiting for connection... (%d seconds elapsed)", connect_attempts / 20);
      }
    }
 
    if (!acip_transport_is_connected(g_client_transport)) {
      log_error("Failed to establish WebSocket connection within 15 seconds");
      acip_transport_destroy(g_client_transport);
      g_client_transport = NULL;
      atomic_store_bool(&g_connection_active, false);
      return -1;
    }
 
    log_info("CLIENT_CONNECT: WebSocket connection established, proceeding with handshake");
 
    // Send initial terminal capabilities to server
    int result = threaded_send_terminal_size_with_auto_detect((int)terminal_get_effective_width(),
                                                              (int)terminal_get_effective_height());
    if (result < 0) {
      log_error("Failed to send initial capabilities to server: %s", network_error_string());
      acip_transport_destroy(g_client_transport);
      g_client_transport = NULL;
      return -1;
    }
 
    // Disable terminal logging after initial setup (for non-snapshot mode)
    if (!GET_OPTION(snapshot_mode) && has_ever_connected) {
      log_set_terminal_output(false);
    }
 
    // Build capabilities flags
    uint32_t my_capabilities = CLIENT_CAP_VIDEO;
    log_debug("GET_OPTION(audio_enabled) = %d (sending CLIENT_JOIN)", GET_OPTION(audio_enabled));
    if (GET_OPTION(audio_enabled)) {
      log_debug("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());
      acip_transport_destroy(g_client_transport);
      g_client_transport = NULL;
      return -1;
    }
 
    log_info("WebSocket connection established successfully");
    return 0;
  }
 
  asciichat_error_t connect_result =
      connection_factory_open("client", address, (uint16_t)port, NULL, &g_client_transport, &endpoint);
  if (connect_result != ASCIICHAT_OK || !g_client_transport) {
    log_error("Failed to establish TCP transport: %s", asciichat_error_string(connect_result));
    return -1;
  }
 
  g_sockfd = g_client_transport->methods->get_socket(g_client_transport);
  if (g_sockfd == INVALID_SOCKET_VALUE) {
    log_error("TCP transport did not expose a valid socket");
    acip_transport_destroy(g_client_transport);
    g_client_transport = NULL;
    return -1;
  }
 
  SAFE_STRNCPY(g_server_ip, endpoint.host, sizeof(g_server_ip));
  log_debug("CLIENT_CONNECT: Resolved TCP endpoint host=%s port=%u", endpoint.host, endpoint.port);
 
  // 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_bool(&g_connection_active, true);
  atomic_store_bool(&g_connection_lost, false);
  atomic_store_bool(&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
  }
 
  // Create the persistent TCP transport before handshake.
  // It starts with NULL crypto; client_crypto_handshake() sets crypto_ctx on success.
  g_client_transport = acip_tcp_transport_create("client", g_sockfd, NULL);
  if (!g_client_transport) {
    log_error("Failed to create TCP ACIP transport");
    close_socket(g_sockfd);
    g_sockfd = INVALID_SOCKET_VALUE;
    return -1;
  }
  log_debug("CLIENT_CONNECT: Created TCP ACIP transport (crypto pending handshake)");
 
  // Perform crypto handshake if encryption is enabled
  log_debug("CLIENT_CONNECT: Calling client_crypto_handshake()");
  int handshake_result = client_crypto_handshake(g_client_transport);
  if (handshake_result != 0) {
    log_error("Crypto handshake failed");
    log_debug("CLIENT_CONNECT: client_crypto_handshake() failed with code %d", handshake_result);
    acip_transport_destroy(g_client_transport);
    g_client_transport = NULL;
    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");
 
  // 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((int)terminal_get_effective_width(),
                                                            (int)terminal_get_effective_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_debug("GET_OPTION(audio_enabled) = %d (sending CLIENT_JOIN)", GET_OPTION(audio_enabled));
  if (GET_OPTION(audio_enabled)) {
    log_debug("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() {
  // For TCP: check socket validity
  // For WebRTC: socket is INVALID_SOCKET_VALUE but transport exists
  return atomic_load_bool(&g_connection_active) && (g_sockfd != INVALID_SOCKET_VALUE || g_client_transport != NULL);
}
 
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("[TRANSPORT_LIFECYCLE] server_connection_set_transport() called with transport=%p", (void *)transport);
 
  // Clean up any existing transport
  if (g_client_transport) {
    log_warn("[TRANSPORT_LIFECYCLE] Replacing existing transport=%p with new transport=%p (potential premature "
             "destruction?)",
             (void *)g_client_transport, (void *)transport);
    log_debug("[TRANSPORT_LIFECYCLE] OLD transport is_connected=%s before destruction",
              acip_transport_is_connected(g_client_transport) ? "true" : "false");
    acip_transport_destroy(g_client_transport);
    log_debug("[TRANSPORT_LIFECYCLE] OLD transport destroyed");
  }
 
  log_debug("[TRANSPORT_LIFECYCLE] 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_LIFECYCLE] Transport is non-NULL, extracting socket...");
    g_sockfd = acip_transport_get_socket(transport);
    log_debug("[TRANSPORT_LIFECYCLE] Socket extracted: %d from transport=%p", (int)g_sockfd, (void *)transport);
 
    atomic_store_bool(&g_connection_active, true);
    atomic_store_bool(&g_connection_lost, false); // Reset lost flag for new connection
    log_debug("[TRANSPORT_LIFECYCLE] Server connection transport set and marked active (transport=%p, sockfd=%d, "
              "is_connected=%s)",
              (void *)transport, (int)g_sockfd, acip_transport_is_connected(transport) ? "true" : "false");
  } else {
    g_sockfd = INVALID_SOCKET_VALUE;
    atomic_store_bool(&g_connection_active, false);
    log_debug("[TRANSPORT_LIFECYCLE] Server connection transport cleared and marked inactive");
  }
 
  log_debug("[TRANSPORT_LIFECYCLE] 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_set_port(int port) {
  g_server_port = port;
  log_debug("Server port set to: %d", g_server_port);
}
 
int server_connection_get_port(void) {
  return g_server_port;
}
 
void server_connection_close() {
  log_debug("[TRANSPORT_LIFECYCLE] server_connection_close() called");
 
  // Acquire send mutex to prevent race with threaded_send_image_frame()
  // This ensures any in-flight sends complete before we destroy the transport
  mutex_lock(&g_send_mutex);
  atomic_store_bool(&g_connection_active, false);
 
  // Destroy ACIP transport before closing socket
  if (g_client_transport) {
    log_debug("[TRANSPORT_LIFECYCLE] Destroying transport=%p (is_connected=%s) before closing socket",
              (void *)g_client_transport, acip_transport_is_connected(g_client_transport) ? "true" : "false");
    acip_transport_destroy(g_client_transport);
    g_client_transport = NULL;
    log_debug("[TRANSPORT_LIFECYCLE] Transport destroyed, g_client_transport set to NULL");
  }
 
  mutex_unlock(&g_send_mutex);
 
  // The socket belongs to the connection handle's TCP/WebSocket owner. The
  // caller cleans up that owner after protocol workers have stopped; closing
  // g_sockfd here would close the same descriptor a second time.
  g_sockfd = INVALID_SOCKET_VALUE;
 
  g_my_client_id = 0;
 
  // Cleanup crypto context if encryption was enabled
  if (g_encryption_enabled) {
    log_debug("[TRANSPORT_LIFECYCLE] Cleaning up crypto context");
    crypto_handshake_destroy(&g_crypto_ctx);
    g_encryption_enabled = false;
  }
 
  // Turn ON terminal logging when connection is closed (unless it was disabled with --quiet)
  if (!GET_OPTION(quiet)) {
    log_set_terminal_output(true);
  }
  log_debug("[TRANSPORT_LIFECYCLE] server_connection_close() completed");
}
static void server_connection_interrupt_websocket_recv(void) {
  // Access the transport
  acip_transport_t *transport = g_client_transport;
  if (!transport || !transport->impl_data) {
    return; // No transport
  }
 
  // Check if this is a WebSocket transport
  if (transport->methods && transport->methods->get_type) {
    acip_transport_type_t type = transport->methods->get_type(transport);
    if (type == ACIP_TRANSPORT_WEBSOCKET) {
      // For WebSocket, directly signal the recv condition variable to wake
      // any blocked threads. This is async-signal-safe (doesn't require mutex).
      // The waiting threads will re-acquire their mutexes and check is_connected.
      websocket_transport_data_t *ws_data = (websocket_transport_data_t *)transport->impl_data;
 
      // Signal recv condition - wakes threads waiting for fragments
      // This is safe to call without holding the recv_mutex
      cond_broadcast(&ws_data->recv_cond);
      cond_broadcast(&ws_data->state_cond);
    }
  }
}
 
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_bool(&g_connection_active, false);
  atomic_store_bool(&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);
  }
 
  // For WebSocket transport: interrupt blocking recv() calls
  // The recv() checks is_connected before waiting for data. Setting this to false
  // will cause any blocked recv() to return immediately with CONNECTION_CLOSED.
  // We use a function that's async-signal-safe (no mutexes) to do this.
  server_connection_interrupt_websocket_recv();
 
  // 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_bool(&g_connection_lost, true);
  atomic_store_bool(&g_connection_active, false);
 
  // Don't re-enable terminal logging here - let the splash screen handle it
  // The reconnection splash will capture and display logs properly
  display_full_reset();
}
 
bool server_connection_is_lost() {
  return atomic_load_bool(&g_connection_lost);
}
 
void server_connection_cleanup() {
  if (!GET_OPTION(quiet)) {
    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) {
  // Lock mutex for entire send operation to prevent concurrent socket writes
  mutex_lock(&g_send_mutex);
 
  // Check connection status and get transport reference
  if (!atomic_load_bool(&g_connection_active) || !g_client_transport) {
    log_debug("[TRANSPORT_LIFECYCLE] threaded_send_packet() check failed: active=%s, transport=%p",
              atomic_load_bool(&g_connection_active) ? "true" : "false", (void *)g_client_transport);
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_NETWORK, "Connection not active or transport unavailable");
  }
 
  // Get transport reference - transport has its own internal synchronization
  acip_transport_t *transport = g_client_transport;
  log_debug_every(LOG_RATE_SLOW, "[TRANSPORT_LIFECYCLE] threaded_send_packet() using transport=%p, is_connected=%s",
                  (void *)transport, acip_transport_is_connected(transport) ? "true" : "false");
 
  // Network I/O happens while holding mutex to prevent concurrent socket writes
  asciichat_error_t result = packet_send_via_transport(transport, type, data, len, 0);
 
  // Unlock after send completes
  mutex_unlock(&g_send_mutex);
 
  // If send failed due to network error, signal connection loss
  if (result != ASCIICHAT_OK) {
    log_debug("[TRANSPORT_LIFECYCLE] threaded_send_packet() send failed, calling server_connection_lost()");
    server_connection_lost();
    return result;
  }
 
  return ASCIICHAT_OK;
}
 
int threaded_send_audio_batch_packet(const float *samples, int num_samples, int batch_count) {
  // Lock mutex for entire send operation to prevent concurrent socket writes
  mutex_lock(&g_send_mutex);
 
  // Check connection status and get transport reference
  if (!atomic_load_bool(&g_connection_active) || !g_client_transport) {
    mutex_unlock(&g_send_mutex);
    return -1;
  }
 
  // Get transport reference - transport has its own internal synchronization
  acip_transport_t *transport = g_client_transport;
 
  // Network I/O happens while holding mutex to prevent concurrent socket writes
  asciichat_error_t result = acip_send_audio_batch(transport, samples, (uint32_t)num_samples, (uint32_t)batch_count);
 
  // Unlock after send completes
  mutex_unlock(&g_send_mutex);
 
  // If send failed due to network error, signal connection loss
  if (result != ASCIICHAT_OK) {
    server_connection_lost();
    return -1;
  }
 
  return 0;
}
 
asciichat_error_t threaded_send_audio_opus(const uint8_t *opus_data, size_t opus_size, int sample_rate,
                                           int frame_duration) {
  // Get transport reference while holding mutex (brief lock)
  mutex_lock(&g_send_mutex);
 
  // Check connection status and get transport reference
  if (!atomic_load_bool(&g_connection_active) || !g_client_transport) {
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_NETWORK, "Connection not active or transport unavailable");
  }
 
  // Get transport reference - transport has its own internal synchronization
  acip_transport_t *transport = g_client_transport;
  mutex_unlock(&g_send_mutex);
 
  // Build Opus packet with header (outside mutex - no blocking I/O yet)
  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) {
    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);
 
  // Network I/O happens OUTSIDE the mutex to prevent deadlock on TCP buffer full
  asciichat_error_t result =
      packet_send_via_transport(transport, PACKET_TYPE_AUDIO_OPUS_BATCH, packet_data, total_size, 0);
 
  // Clean up
  buffer_pool_free(NULL, packet_data, total_size);
 
  // 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) {
  // Lock mutex for entire send operation to prevent concurrent socket writes
  mutex_lock(&g_send_mutex);
 
  // Check connection status and get transport reference
  if (!atomic_load_bool(&g_connection_active) || !g_client_transport) {
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_NETWORK, "Connection not active or transport unavailable");
  }
 
  // Get transport reference - transport has its own internal synchronization
  acip_transport_t *transport = g_client_transport;
 
  // Network I/O happens while holding mutex to prevent concurrent socket writes
  // Opus uses 20ms frames at 48kHz (960 samples = 20ms)
  asciichat_error_t result =
      acip_send_audio_opus_batch(transport, opus_data, opus_size, frame_sizes, frame_count, 48000, 20);
 
  // Unlock after send completes
  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_image_frame(const void *pixel_data, uint32_t width, uint32_t height,
                                            uint32_t pixel_format) {
  if (!pixel_data) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Pixel data is NULL");
  }
 
  mutex_lock(&g_send_mutex);
 
  // Get transport reference - may be NULL if connection was lost
  acip_transport_t *transport = server_connection_get_transport();
  if (!transport || !server_connection_is_active()) {
    mutex_unlock(&g_send_mutex);
    return SET_ERRNO(ERROR_NETWORK, "Transport unavailable");
  }
 
  // Send image frame with socket lock held
  asciichat_error_t result = acip_send_image_frame(transport, pixel_data, width, height, pixel_format);
 
  // Unlock after send completes
  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_image_frame_h265(const void *pixel_data, uint32_t width, uint32_t height) {
  if (!pixel_data) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Pixel data is NULL");
  }
 
  // Get encoder - may be NULL if not initialized
  h265_encoder_t *encoder = g_h265_encoder;
  if (!encoder) {
    return SET_ERRNO(ERROR_INVALID_STATE, "H.265 encoder not initialized");
  }
 
  // Allocate buffer and encode frame WITHOUT holding the send mutex.
  // This avoids long critical section that can cause deadlocks with named registry operations.
  log_info("[H265_SEND_FRAME_1] Allocating buffer and encoding frame (no mutex held)");
  size_t output_buffer_size = 5 + (width * height * 2);
  uint8_t *output_buffer = buffer_pool_alloc(NULL, output_buffer_size);
  if (!output_buffer) {
    log_error("Failed to allocate output buffer for frame");
    return SET_ERRNO(ERROR_MEMORY, "Failed to allocate H.265 output buffer: %zu bytes", output_buffer_size);
  }
 
  // Encode frame
  size_t encoded_size = output_buffer_size - 5;
  asciichat_error_t encode_result = h265_encode(encoder, (uint16_t)width, (uint16_t)height, (const uint8_t *)pixel_data,
                                                output_buffer + 5, &encoded_size);
  log_info("[H265_SEND_FRAME_2] ✅ Frame encoded, encoded_size=%zu, result=%d", encoded_size, encode_result);
 
  if (encode_result != ASCIICHAT_OK) {
    log_error("H.265 encoding failed: %s", asciichat_error_string(encode_result));
    buffer_pool_free(NULL, output_buffer, output_buffer_size);
    return encode_result;
  }
 
  // Prepare packet header: [flags:u8][width:u16][height:u16][h265_data...]
  output_buffer[0] = 0; // flags: no flags set by default
  output_buffer[1] = (width >> 8) & 0xFF;
  output_buffer[2] = width & 0xFF;
  output_buffer[3] = (height >> 8) & 0xFF;
  output_buffer[4] = height & 0xFF;
 
  size_t total_size = 5 + encoded_size;
 
  // NOW acquire the send mutex just for the actual network transmission
  log_info("[H265_SEND_FRAME_3] About to acquire g_send_mutex for network send");
  mutex_lock(&g_send_mutex);
  log_info("[H265_SEND_FRAME_4] ✅ Acquired g_send_mutex, checking transport");
 
  // Get transport reference - may be NULL if connection was lost
  acip_transport_t *transport = server_connection_get_transport();
  if (!transport || !server_connection_is_active()) {
    mutex_unlock(&g_send_mutex);
    log_info("[H265_SEND_FRAME_5] Transport unavailable, releasing mutex");
    buffer_pool_free(NULL, output_buffer, output_buffer_size);
    return SET_ERRNO(ERROR_NETWORK, "Transport unavailable");
  }
 
  // Send H.265 frame with socket lock held (short critical section)
  log_info("[H265_SEND_FRAME_6] Sending packet via transport (%zu bytes)", total_size);
  asciichat_error_t result =
      packet_send_via_transport(transport, PACKET_TYPE_IMAGE_FRAME_H265, output_buffer, total_size, 0);
  log_info("[H265_SEND_FRAME_7] packet_send_via_transport returned: %d", result);
 
  // Unlock after send completes
  mutex_unlock(&g_send_mutex);
  log_info("[H265_SEND_FRAME_8] ✅ Released g_send_mutex");
 
  buffer_pool_free(NULL, output_buffer, output_buffer_size);
  log_info("[H265_SEND_FRAME_9] ✅ Freed output buffer");
 
  // 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);
}
 
asciichat_error_t threaded_send_stream_start_packet(uint32_t stream_type) {
  // Connection and transport availability is checked by threaded_send_packet()
 
  // 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));
}
 
asciichat_error_t threaded_send_terminal_size_with_auto_detect(unsigned short width, unsigned short height) {
  // Log the dimensions being sent to server (helps debug dimension mismatch issues)
  log_debug("Sending terminal size to server: %ux%u (auto_width=%d, auto_height=%d)", width, height,
            GET_OPTION(auto_width), GET_OPTION(auto_height));
 
  // Connection and transport availability is checked by threaded_send_packet()
 
  // Build terminal capabilities packet locally
  // Detect terminal capabilities automatically
  terminal_capabilities_t caps = detect_terminal_capabilities();
 
  // Set wants_padding based on snapshot mode and TTY status
  // Disable padding when:
  // - In snapshot mode (one frame and exit)
  // - When stdout is not a TTY (piped/redirected output)
  // Enable padding for interactive terminal sessions
  bool is_snapshot_mode = GET_OPTION(snapshot_mode);
  bool is_interactive = terminal_is_interactive();
  caps.wants_padding = is_interactive && !is_snapshot_mode;
 
  log_debug("Client capabilities: wants_padding=%d (snapshot=%d, interactive=%d, stdin_tty=%d, stdout_tty=%d)",
            caps.wants_padding, is_snapshot_mode, is_interactive, terminal_is_stdin_tty(), terminal_is_stdout_tty());
 
  // 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;
    // Preserve wants_padding even in fallback mode
    caps.wants_padding = is_interactive && !is_snapshot_mode;
  }
 
  // 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
  int fps = GET_OPTION(fps);
  if (fps > 0) {
    net_packet.desired_fps = (uint8_t)(fps > 144 ? 144 : 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;
  // Send UTF-8 support flag: true for AUTO (default) and TRUE settings, false for FALSE setting
  net_packet.utf8_support = (GET_OPTION(force_utf8) != UTF8_SETTING_FALSE) ? 1 : 0;
 
  // Set wants_padding flag (1=padding enabled, 0=no padding for snapshot/piped modes)
  net_packet.wants_padding = caps.wants_padding ? 1 : 0;
 
  // Set codec capabilities (client supports all codecs by default)
  net_packet.codec_capabilities_video = HOST_TO_NET_U32(VIDEO_CODEC_CAP_ALL);
  net_packet.codec_capabilities_audio = HOST_TO_NET_U32(AUDIO_CODEC_CAP_ALL);
 
  // 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) {
  // Connection and transport availability is checked by threaded_send_packet()
 
  // 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_u64(&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;
}