feat: implement WebSocket synchronization and HTTP recovery

2026-08-28 23:33:14 +03:00
parent 95b3450317
commit 03b283f892
8 changed files with 524 additions and 13 deletions
@@ -2,7 +2,7 @@
 # without default 'CMakeLists.txt' file.

 idf_component_register(
-  SRCS "main.c" "application.c" "command_queue.c" "fleet_generator.c" "game_engine.c" "bot_player.c" "session_manager.c" "game_lifecycle.c" "state_presenter.c" "http_api.c"
+  SRCS "main.c" "application.c" "command_queue.c" "fleet_generator.c" "game_engine.c" "bot_player.c" "session_manager.c" "game_lifecycle.c" "state_presenter.c" "http_api.c" "sync_service.c"
  INCLUDE_DIRS "../include"
  REQUIRES esp_event esp_http_server esp_netif esp_wifi esp_littlefs nvs_flash
 )
@@ -19,6 +19,7 @@
 #include "freertos/portmacro.h"
 #include "http_api.h"
 #include "nvs_flash.h"
+#include "sync_service.h"

 #if __has_include("wifi_config.h")
 #include "wifi_config.h"
@@ -40,7 +41,9 @@ static bool s_littlefs_mounted;
 static httpd_handle_t s_server;
 static application_t s_application;
 static http_api_t s_api;
+static sync_service_t s_sync;
 static uint16_t s_rejected_input;
+static esp_timer_handle_t s_sync_timer;
 #if WIFI_CONFIG_AVAILABLE
 static esp_timer_handle_t s_reconnect_timer;
 #endif
@@ -90,6 +93,36 @@ static esp_err_t send_api_response(httpd_req_t *request, const http_api_response
  return httpd_resp_send(request, response->body, response->body_length);
 }

+static bool websocket_send(void *unused, int client_id, const char *payload, size_t length) {
+  (void)unused;
+  httpd_ws_frame_t frame = {.final = true, .type = HTTPD_WS_TYPE_TEXT, .payload = (uint8_t *)payload, .len = length};
+  const esp_err_t result = httpd_ws_send_frame_async(s_server, client_id, &frame);
+  if (result != ESP_OK) httpd_sess_trigger_close(s_server, client_id);
+  return result == ESP_OK;
+}
+
+static void sync_broadcast_work(void *unused) {
+  (void)unused;
+  sync_service_broadcast(&s_sync, websocket_send, NULL);
+}
+
+static void queue_state_broadcast(void) {
+  if (s_server != NULL && httpd_queue_work(s_server, sync_broadcast_work, NULL) != ESP_OK) ++s_rejected_input;
+}
+
+static void sync_expire_work(void *unused) {
+  (void)unused;
+  int closed[kSessionCapacity] = {0};
+  size_t closed_count = 0U;
+  sync_service_expire(&s_sync, (uint64_t)(esp_timer_get_time() / 1000U), closed, &closed_count);
+  for (size_t index = 0; index < closed_count; ++index) httpd_sess_trigger_close(s_server, closed[index]);
+}
+
+static void sync_timer_callback(void *unused) {
+  (void)unused;
+  if (s_server != NULL) httpd_queue_work(s_server, sync_expire_work, NULL);
+}
+
 static bool request_has_json_content_type(httpd_req_t *request) {
  const size_t length = httpd_req_get_hdr_value_len(request, "Content-Type");
  if (length == 0U || length >= 64U) return false;
@@ -139,10 +172,53 @@ static esp_err_t api_handler(httpd_req_t *request) {
  const http_api_request_t api_request = {.method = request->method == HTTP_GET ? HTTP_API_GET : HTTP_API_POST, .route = route,
      .content_type_json = request_has_json_content_type(request), .body = body, .body_length = body_length,
      .session_token = token[0] == '\0' ? NULL : token, .target_too_large = target_too_large};
+  const uint32_t version_before = s_application.lifecycle.game.state.version;
  if (!http_api_handle(&s_api, &api_request, &response)) return ESP_FAIL;
+  if (s_application.lifecycle.game.state.version != version_before) queue_state_broadcast();
  return send_api_response(request, &response);
 }

+static esp_err_t websocket_handler(httpd_req_t *request) {
+  const int client_id = httpd_req_to_sockfd(request);
+  if (request->method == HTTP_GET) {
+    if (sync_service_open(&s_sync, client_id, (uint64_t)(esp_timer_get_time() / 1000U))) return ESP_OK;
+    httpd_sess_trigger_close(s_server, client_id);
+    return ESP_FAIL;
+  }
+  httpd_ws_frame_t frame = {0};
+  if (httpd_ws_recv_frame(request, &frame, 0U) != ESP_OK || frame.type != HTTPD_WS_TYPE_TEXT ||
+      frame.len > kWebSocketFrameCapacity) {
+    ++s_rejected_input;
+    sync_service_close(&s_sync, client_id);
+    httpd_sess_trigger_close(s_server, client_id);
+    return ESP_OK;
+  }
+  char input[kWebSocketFrameCapacity + 1U] = {0};
+  frame.payload = (uint8_t *)input;
+  if (httpd_ws_recv_frame(request, &frame, sizeof(input) - 1U) != ESP_OK) {
+    sync_service_close(&s_sync, client_id);
+    httpd_sess_trigger_close(s_server, client_id);
+    return ESP_OK;
+  }
+  char output[kStateMessageCapacity] = {0};
+  size_t output_length = 0U;
+  bool state_changed = false;
+  bool close_client = false;
+  if (!sync_service_receive(&s_sync, client_id, input, frame.len, (uint64_t)(esp_timer_get_time() / 1000U),
+                            output, &output_length, &state_changed, &close_client)) return ESP_FAIL;
+  if (output_length > 0U) {
+    httpd_ws_frame_t response = {.final = true, .type = HTTPD_WS_TYPE_TEXT,
+        .payload = (uint8_t *)output, .len = output_length};
+    if (httpd_ws_send_frame(request, &response) != ESP_OK) close_client = true;
+  }
+  if (state_changed) queue_state_broadcast();
+  if (close_client) {
+    sync_service_close(&s_sync, client_id);
+    httpd_sess_trigger_close(s_server, client_id);
+  }
+  return ESP_OK;
+}
+
 #if WIFI_CONFIG_AVAILABLE
 static void reconnect_timer_callback(void *unused) {
  (void)unused;
@@ -252,7 +328,7 @@ static esp_err_t static_file_handler(httpd_req_t *request) {

 static esp_err_t start_http_server(void) {
  httpd_config_t config = HTTPD_DEFAULT_CONFIG();
-  config.max_uri_handlers = 14U; config.max_open_sockets = kHttpMaxOpenSockets; config.uri_match_fn = httpd_uri_match_wildcard; config.lru_purge_enable = true;
+  config.max_uri_handlers = 15U; config.max_open_sockets = kHttpMaxOpenSockets; config.uri_match_fn = httpd_uri_match_wildcard; config.lru_purge_enable = true;
  ESP_RETURN_ON_ERROR(httpd_start(&s_server, &config), kLogTag, "http server start failed");
  const httpd_uri_t routes[] = {
      {.uri = "/", .method = HTTP_GET, .handler = root_handler},
@@ -266,6 +342,7 @@ static esp_err_t start_http_server(void) {
      {.uri = "/api/game/rematch", .method = HTTP_POST, .handler = api_handler, .user_ctx = (void *)(uintptr_t)HTTP_API_ROUTE_REMATCH},
      {.uri = "/api/game/abort", .method = HTTP_POST, .handler = api_handler, .user_ctx = (void *)(uintptr_t)HTTP_API_ROUTE_ABORT},
      {.uri = "/api/state", .method = HTTP_GET, .handler = api_handler, .user_ctx = (void *)(uintptr_t)HTTP_API_ROUTE_STATE},
+      {.uri = "/ws", .method = HTTP_GET, .handler = websocket_handler, .is_websocket = true},
      {.uri = "/*", .method = HTTP_GET, .handler = static_file_handler},
  };
  for (size_t index = 0; index < sizeof(routes) / sizeof(routes[0]); ++index) ESP_RETURN_ON_ERROR(httpd_register_uri_handler(s_server, &routes[index]), kLogTag, "route registration failed");
@@ -284,8 +361,12 @@ void app_main(void) {
  ESP_ERROR_CHECK(result); ESP_ERROR_CHECK(esp_netif_init()); ESP_ERROR_CHECK(esp_event_loop_create_default());
  application_init(&s_application, (random_source_t){.next_u32 = platform_random, .context = NULL});
  http_api_init(&s_api, &s_application);
+  sync_service_init(&s_sync, &s_application);
  mount_littlefs();
  ESP_ERROR_CHECK(start_http_server());
+  const esp_timer_create_args_t sync_timer_args = {.callback = sync_timer_callback, .name = "sync_expire"};
+  ESP_ERROR_CHECK(esp_timer_create(&sync_timer_args, &s_sync_timer));
+  ESP_ERROR_CHECK(esp_timer_start_periodic(s_sync_timer, 1000000U));
  result = start_wifi();
  if (result != ESP_OK) ESP_LOGW(kLogTag, "wifi unavailable: %s", esp_err_to_name(result));
 }
@@ -0,0 +1,220 @@
+#include "sync_service.h"
+
+#include <inttypes.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <string.h>
+
+#include "state_presenter.h"
+
+static void response_write(char output[kStateMessageCapacity], size_t *output_length, const char *format, ...) {
+  va_list arguments;
+  va_start(arguments, format);
+  const int length = vsnprintf(output, kStateMessageCapacity, format, arguments);
+  va_end(arguments);
+  *output_length = length >= 0 && (size_t)length < kStateMessageCapacity ? (size_t)length : 0U;
+}
+
+static void response_error(char output[kStateMessageCapacity], size_t *output_length, const char *code,
+                           const char *message, uint32_t version) {
+  response_write(output, output_length, "{\"ok\":false,\"code\":\"%s\",\"message\":\"%s\",\"version\":%" PRIu32 "}",
+                 code, message, version);
+}
+
+static bool token_bytes(const char *text, uint8_t token[kSessionTokenBytes]) {
+  if (text == NULL || strlen(text) != kSessionTokenBytes * 2U) return false;
+  for (uint8_t index = 0; index < kSessionTokenBytes; ++index) {
+    const char high = text[index * 2U];
+    const char low = text[index * 2U + 1U];
+    if (high < '0' || (high > '9' && high < 'a') || high > 'f' ||
+        low < '0' || (low > '9' && low < 'a') || low > 'f') return false;
+    const uint8_t high_value = (uint8_t)(high <= '9' ? high - '0' : high - 'a' + 10);
+    const uint8_t low_value = (uint8_t)(low <= '9' ? low - '0' : low - 'a' + 10);
+    token[index] = (uint8_t)((high_value << 4U) | low_value);
+  }
+  return true;
+}
+
+static bool compact_json(const char *frame, size_t frame_length, char output[kWebSocketFrameCapacity + 1U]) {
+  if (frame == NULL || frame_length == 0U || frame_length > kWebSocketFrameCapacity) return false;
+  bool in_string = false;
+  bool escaped = false;
+  size_t written = 0;
+  for (size_t index = 0; index < frame_length; ++index) {
+    const unsigned char character = (unsigned char)frame[index];
+    if (character < 0x20U && character != ' ' && character != '\n' && character != '\r' && character != '\t') return false;
+    if (!in_string && (character == ' ' || character == '\n' || character == '\r' || character == '\t')) continue;
+    if (written >= kWebSocketFrameCapacity) return false;
+    output[written++] = (char)character;
+    if (in_string && escaped) escaped = false;
+    else if (in_string && character == '\\') escaped = true;
+    else if (character == '"') in_string = !in_string;
+  }
+  output[written] = '\0';
+  return !in_string && !escaped;
+}
+
+static sync_connection_t *connection_for(sync_service_t *service, int client_id) {
+  if (service == NULL) return NULL;
+  for (uint8_t index = 0; index < kSessionCapacity; ++index) {
+    if (service->connections[index].active && service->connections[index].client_id == client_id) return &service->connections[index];
+  }
+  return NULL;
+}
+
+static void lifecycle_error(char output[kStateMessageCapacity], size_t *output_length, lifecycle_result_t result,
+                            uint32_t version) {
+  switch (result) {
+    case LIFECYCLE_RESULT_INVALID_MODE: response_error(output, output_length, "INVALID_MODE", "Некорректный режим", version); break;
+    case LIFECYCLE_RESULT_INVALID_COORDINATE: response_error(output, output_length, "INVALID_COORDINATE", "Некорректные координаты", version); break;
+    case LIFECYCLE_RESULT_FORBIDDEN_ROLE: response_error(output, output_length, "FORBIDDEN_ROLE", "Роль не может выполнить действие", version); break;
+    case LIFECYCLE_RESULT_WRONG_PHASE: response_error(output, output_length, "WRONG_PHASE", "Действие недоступно сейчас", version); break;
+    case LIFECYCLE_RESULT_NOT_YOUR_TURN: response_error(output, output_length, "NOT_YOUR_TURN", "Сейчас ход соперника", version); break;
+    case LIFECYCLE_RESULT_CELL_ALREADY_SHOT: response_error(output, output_length, "CELL_ALREADY_SHOT", "Клетка уже обстреляна", version); break;
+    case LIFECYCLE_RESULT_STALE_GAME: response_error(output, output_length, "STALE_GAME", "Партия уже изменилась", version); break;
+    default: response_error(output, output_length, "SERVER_BUSY", "Сервер занят", version); break;
+  }
+}
+
+static bool parse_command(const char *json, app_command_t *command, uint8_t token_bytes_out[kSessionTokenBytes]) {
+  char token[kSessionTokenBytes * 2U + 1U] = {0};
+  char mode[8] = {0};
+  uint32_t game_id = 0;
+  uint32_t x = 0;
+  uint32_t y = 0;
+  int consumed = 0;
+  if (sscanf(json, "{\"type\":\"config\",\"token\":\"%32[0-9a-f]\",\"gameId\":%" SCNu32 ",\"mode\":\"%7[a-z]\"}%n", token, &game_id, mode, &consumed) == 3 && json[consumed] == '\0') {
+    command->type = COMMAND_CONFIG;
+    command->mode = strcmp(mode, "human") == 0 ? MODE_HUMAN : strcmp(mode, "bot") == 0 ? MODE_BOT : (game_mode_t)UINT8_MAX;
+  } else if (sscanf(json, "{\"type\":\"shot\",\"token\":\"%32[0-9a-f]\",\"gameId\":%" SCNu32 ",\"x\":%" SCNu32 ",\"y\":%" SCNu32 "}%n", token, &game_id, &x, &y, &consumed) == 4 && json[consumed] == '\0') {
+    command->type = COMMAND_SHOT;
+    command->coordinate = x < kBoardWidth && y < kBoardHeight ?
+        (coordinate_t){.x = (uint8_t)x, .y = (uint8_t)y} : (coordinate_t){.x = kBoardWidth, .y = 0U};
+  } else if (sscanf(json, "{\"type\":\"start\",\"token\":\"%32[0-9a-f]\",\"gameId\":%" SCNu32 "}%n", token, &game_id, &consumed) == 2 && json[consumed] == '\0') {
+    command->type = COMMAND_START;
+  } else if (sscanf(json, "{\"type\":\"rematch\",\"token\":\"%32[0-9a-f]\",\"gameId\":%" SCNu32 "}%n", token, &game_id, &consumed) == 2 && json[consumed] == '\0') {
+    command->type = COMMAND_REMATCH;
+  } else if (sscanf(json, "{\"type\":\"abort\",\"token\":\"%32[0-9a-f]\",\"gameId\":%" SCNu32 "}%n", token, &game_id, &consumed) == 2 && json[consumed] == '\0') {
+    command->type = COMMAND_ABORT;
+  } else return false;
+  if (!token_bytes(token, token_bytes_out)) return false;
+  command->game_id = game_id;
+  return true;
+}
+
+void sync_service_init(sync_service_t *service, application_t *application) {
+  if (service != NULL) *service = (sync_service_t){.application = application};
+}
+
+bool sync_service_open(sync_service_t *service, int client_id, uint64_t now_ms) {
+  if (service == NULL || service->application == NULL || connection_for(service, client_id) != NULL) return false;
+  for (uint8_t index = 0; index < kSessionCapacity; ++index) {
+    if (!service->connections[index].active) {
+      service->connections[index] = (sync_connection_t){.client_id = client_id, .active = true, .opened_ms = now_ms};
+      return true;
+    }
+  }
+  return false;
+}
+
+void sync_service_close(sync_service_t *service, int client_id) {
+  sync_connection_t *connection = connection_for(service, client_id);
+  if (connection != NULL) *connection = (sync_connection_t){0};
+}
+
+bool sync_service_receive(sync_service_t *service, int client_id, const char *frame, size_t frame_length,
+                          uint64_t now_ms, char output[kStateMessageCapacity], size_t *output_length,
+                          bool *state_changed, bool *close_client) {
+  if (output_length != NULL) *output_length = 0U;
+  if (state_changed != NULL) *state_changed = false;
+  if (close_client != NULL) *close_client = false;
+  sync_connection_t *connection = connection_for(service, client_id);
+  if (service == NULL || service->application == NULL || connection == NULL || output == NULL || output_length == NULL) return false;
+  const uint32_t version = service->application->lifecycle.game.state.version;
+  if (now_ms - connection->opened_ms > kWebSocketHelloTimeoutMs && !connection->authenticated) {
+    response_error(output, output_length, "UNAUTHORIZED", "Сессия не найдена", version);
+    if (close_client != NULL) *close_client = true;
+    return true;
+  }
+  char json[kWebSocketFrameCapacity + 1U];
+  if (!compact_json(frame, frame_length, json)) {
+    response_error(output, output_length, "MALFORMED_JSON", "Некорректный JSON", version);
+    return true;
+  }
+  if (!connection->authenticated) {
+    char token_text[kSessionTokenBytes * 2U + 1U] = {0};
+    uint32_t client_version = 0;
+    int consumed = 0;
+    if (sscanf(json, "{\"type\":\"hello\",\"token\":\"%32[0-9a-f]\",\"version\":%" SCNu32 "}%n", token_text, &client_version, &consumed) != 2 || json[consumed] != '\0') {
+      response_error(output, output_length, "UNAUTHORIZED", "Сессия не найдена", version);
+      if (close_client != NULL) *close_client = true;
+      return true;
+    }
+    uint8_t token[kSessionTokenBytes];
+    uint8_t session_index = 0;
+    if (!token_bytes(token_text, token) || !application_session_for_token(service->application, token, &session_index)) {
+      response_error(output, output_length, "UNAUTHORIZED", "Сессия не найдена", version);
+      if (close_client != NULL) *close_client = true;
+      return true;
+    }
+    for (uint8_t index = 0; index < kSessionCapacity; ++index) {
+      if (service->connections[index].active && service->connections[index].authenticated &&
+          service->connections[index].session_index == session_index) service->connections[index].active = false;
+    }
+    connection->authenticated = true;
+    connection->session_index = session_index;
+    (void)client_version;
+    if (!state_presenter_write_lifecycle(&service->application->lifecycle,
+                                         service->application->lifecycle.sessions.entries[session_index].role,
+                                         output, kStateMessageCapacity, output_length)) {
+      response_error(output, output_length, "SERVER_BUSY", "Сервер занят", version);
+    }
+    return true;
+  }
+  if (strcmp(json, "{\"type\":\"ping\"}") == 0) {
+    response_write(output, output_length, "{\"type\":\"pong\"}");
+    return true;
+  }
+  app_command_t command = {0};
+  uint8_t token[kSessionTokenBytes];
+  if (!parse_command(json, &command, token) ||
+      !application_session_for_token(service->application, token, &command.session_index) ||
+      command.session_index != connection->session_index) {
+    response_error(output, output_length, "UNAUTHORIZED", "Сессия не найдена", version);
+    return true;
+  }
+  command.version = version;
+  const lifecycle_result_t result = application_submit(service->application, &command);
+  if (result != LIFECYCLE_RESULT_OK) {
+    lifecycle_error(output, output_length, result, service->application->lifecycle.game.state.version);
+  } else if (state_changed != NULL) {
+    *state_changed = true;
+  }
+  return true;
+}
+
+void sync_service_expire(sync_service_t *service, uint64_t now_ms, int closed_clients[kSessionCapacity],
+                         size_t *closed_count) {
+  if (closed_count != NULL) *closed_count = 0U;
+  if (service == NULL || closed_clients == NULL || closed_count == NULL) return;
+  for (uint8_t index = 0; index < kSessionCapacity; ++index) {
+    sync_connection_t *connection = &service->connections[index];
+    if (connection->active && !connection->authenticated && now_ms - connection->opened_ms > kWebSocketHelloTimeoutMs) {
+      closed_clients[(*closed_count)++] = connection->client_id;
+      *connection = (sync_connection_t){0};
+    }
+  }
+}
+
+void sync_service_broadcast(sync_service_t *service, sync_send_fn send, void *context) {
+  if (service == NULL || service->application == NULL || send == NULL) return;
+  for (uint8_t index = 0; index < kSessionCapacity; ++index) {
+    sync_connection_t *connection = &service->connections[index];
+    if (!connection->active || !connection->authenticated) continue;
+    char payload[kStateMessageCapacity];
+    size_t length = 0U;
+    const role_t role = service->application->lifecycle.sessions.entries[connection->session_index].role;
+    if (!state_presenter_write_lifecycle(&service->application->lifecycle, role, payload, sizeof(payload), &length) ||
+        !send(context, connection->client_id, payload, length)) *connection = (sync_connection_t){0};
+  }
+}