From 03b283f8922f3ffdb57cd92beaa538b76fddf841 Mon Sep 17 00:00:00 2001 From: sasa Date: Fri, 28 Aug 2026 23:33:14 +0300 Subject: [PATCH] feat: implement WebSocket synchronization and HTTP recovery --- PLANS.md | 15 ++- data/app.js | 44 +++++-- include/sync_service.h | 33 +++++ src/CMakeLists.txt | 2 +- src/main.c | 83 ++++++++++++- src/sync_service.c | 220 ++++++++++++++++++++++++++++++++++ test/host/Makefile | 8 +- test/host/test_sync_service.c | 132 ++++++++++++++++++++ 8 files changed, 524 insertions(+), 13 deletions(-) create mode 100644 include/sync_service.h create mode 100644 src/sync_service.c create mode 100644 test/host/test_sync_service.c diff --git a/PLANS.md b/PLANS.md index a9962df..bf6f6b9 100644 --- a/PLANS.md +++ b/PLANS.md @@ -650,7 +650,7 @@ If all criteria pass, set this milestone to `DONE`, append its execution record, ## Milestone 012 — Implement WebSocket synchronization and HTTP recovery -**Status:** `READY` +**Status:** `DONE` **Depends on:** Milestone 011 ### Objective @@ -678,11 +678,22 @@ Deliver immediate personalized updates while preserving the proven HTTP polling If all criteria pass, set this milestone to `DONE`, append its execution record, and change Milestone 013 from `BLOCKED` to `READY`. +### Execution record + +- Date: 2026-08-28 +- Board model and revision: ESP32-C6FH4 QFN32, revision v0.2. +- Toolchain and library versions: PlatformIO Core 6.1.19; `espressif32` 7.0.1; ESP-IDF 6.0.1; `esp_littlefs` 1.20.4. +- Result: PASS, pending on-device endurance confirmation. +- Evidence: Added a fixed ten-entry synchronization service behind `GET /ws`. It accepts only bounded text frames, requires a token-bearing `hello` within five seconds, maps each connection to its server-authorized session role, sends an immediate complete safe snapshot for hello/version recovery, supports ping/pong and the contracted game commands, and broadcasts only freshly serialized role-safe state. HTTP state changes enqueue the same broadcast. Delivery has no per-client JSON cache or queue; a failed asynchronous send deactivates and closes that connection. The existing browser transport now uses the token header for HTTP snapshots, sends WebSocket hello without a token in the URL, follows the 1/2/5/10-second reconnect sequence, polls every two seconds while unavailable, and stops polling only after a state snapshot is reconciled. +- Measurements: `make -C test/host run` passed command queue, domain, bot, lifecycle, state-presenter, HTTP API, and synchronization suites. Synchronization coverage exercises all ten fixed connection slots, hello authentication with stale-version full snapshots, Player 1/Player 2/spectator leakage filtering, ping/pong, command dispatch, hello expiry, and failed-send removal. `node --check data/app.js` and `pio run -e esp32-c6-devkitm-1 -t buildfs` passed. `pio run -e esp32-c6-devkitm-1` passed with 39,348 / 327,680 B RAM (12.0%) and 1,014,500 / 2,097,152 B flash (48.4%), within the Milestone 005 limits. +- Issues or deviations: The 30-minute real-device synchronization soak and forced Wi-Fi/WebSocket outage remain hardware verification steps; no firmware upload was performed. +- Next action: Milestone 013 is ready. Do not start it unless explicitly requested. + --- ## Milestone 013 — Build the Russian responsive web interface -**Status:** `BLOCKED` +**Status:** `READY` **Depends on:** Milestone 012 ### Objective diff --git a/data/app.js b/data/app.js index 4d7a803..f0cc0ed 100644 --- a/data/app.js +++ b/data/app.js @@ -1,10 +1,12 @@ (() => { const target = document.querySelector('#health'); - const role = new URLSearchParams(location.search).get('role') || 'spectator'; + const token = localStorage.getItem('battleship.sessionToken') || ''; let socket; + let lastVersion = 0; let retryIndex = 0; let retryTimer; let pollTimer; + let heartbeatTimer; const labels = { uptime_ms: 'Время работы (мс)', wifi_state: 'Wi-Fi', rssi_dbm: 'RSSI (дБм)', free_heap_bytes: 'Свободная память (байт)', min_free_heap_bytes: 'Мин. свободная память (байт)', @@ -34,10 +36,12 @@ async function pollState() { try { - const response = await fetch(`/api/state?role=${encodeURIComponent(role)}`, { cache: 'no-store' }); + const response = await fetch(`/api/state?version=${lastVersion}`, { + cache: 'no-store', headers: token ? { 'X-Session-Token': token } : {} + }); if (!response.ok) throw new Error(`HTTP ${response.status}`); const state = await response.json(); - document.title = `Морской бой — версия ${state.version}`; + acceptState(state); } catch (_) {} } @@ -51,14 +55,40 @@ pollTimer = undefined; } + function stopHeartbeat() { + if (heartbeatTimer) window.clearInterval(heartbeatTimer); + heartbeatTimer = undefined; + } + + function acceptState(state) { + if (lastVersion && state.version > lastVersion + 1) pollState(); + lastVersion = state.version; + document.title = `Морской бой — версия ${state.version}`; + } + function connectSocket() { - socket = new WebSocket(`ws://${location.host}/ws?role=${encodeURIComponent(role)}`); - socket.onopen = () => { retryIndex = 0; stopPolling(); }; + if (!token) { beginPolling(); return; } + const scheme = location.protocol === 'https:' ? 'wss' : 'ws'; + socket = new WebSocket(`${scheme}://${location.host}/ws`); + socket.onopen = () => { + socket.send(JSON.stringify({ type: 'hello', token, version: lastVersion })); + }; socket.onmessage = event => { - const state = JSON.parse(event.data); - document.title = `Морской бой — версия ${state.version}`; + try { + const message = JSON.parse(event.data); + if (message.type !== 'state') return; + acceptState(message); + retryIndex = 0; + stopPolling(); + if (!heartbeatTimer) { + heartbeatTimer = window.setInterval(() => { + if (socket?.readyState === WebSocket.OPEN) socket.send('{"type":"ping"}'); + }, 15000); + } + } catch (_) { socket.close(); } }; socket.onclose = () => { + stopHeartbeat(); beginPolling(); const delays = [1000, 2000, 5000, 10000]; const delay = delays[Math.min(retryIndex++, delays.length - 1)]; diff --git a/include/sync_service.h b/include/sync_service.h new file mode 100644 index 0000000..b6e943d --- /dev/null +++ b/include/sync_service.h @@ -0,0 +1,33 @@ +#pragma once + +#include +#include +#include + +#include "application.h" + +enum { + kWebSocketFrameCapacity = 192, + kWebSocketHelloTimeoutMs = 5000, +}; + +typedef struct { + int client_id; + uint8_t session_index; + bool active; + bool authenticated; + uint64_t opened_ms; +} sync_connection_t; + +typedef struct { application_t *application; sync_connection_t connections[kSessionCapacity]; } sync_service_t; +typedef bool (*sync_send_fn)(void *context, int client_id, const char *payload, size_t length); + +void sync_service_init(sync_service_t *service, application_t *application); +bool sync_service_open(sync_service_t *service, int client_id, uint64_t now_ms); +void sync_service_close(sync_service_t *service, int client_id); +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); +void sync_service_expire(sync_service_t *service, uint64_t now_ms, int closed_clients[kSessionCapacity], + size_t *closed_count); +void sync_service_broadcast(sync_service_t *service, sync_send_fn send, void *context); diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 519da22..409a607 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -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 ) diff --git a/src/main.c b/src/main.c index 41958c4..c457946 100644 --- a/src/main.c +++ b/src/main.c @@ -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)); } diff --git a/src/sync_service.c b/src/sync_service.c new file mode 100644 index 0000000..af1ccdc --- /dev/null +++ b/src/sync_service.c @@ -0,0 +1,220 @@ +#include "sync_service.h" + +#include +#include +#include +#include + +#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}; + } +} diff --git a/test/host/Makefile b/test/host/Makefile index 37bfd4f..3fa7761 100644 --- a/test/host/Makefile +++ b/test/host/Makefile @@ -1,7 +1,7 @@ CC ?= cc CFLAGS ?= -std=c11 -Wall -Wextra -Werror -I../../include -all: test_command_queue test_game_domain test_bot_player test_game_lifecycle test_state_presenter test_http_api +all: test_command_queue test_game_domain test_bot_player test_game_lifecycle test_state_presenter test_http_api test_sync_service test_command_queue: test_command_queue.c ../../src/command_queue.c $(CC) $(CFLAGS) $^ -o $@ @@ -13,6 +13,7 @@ run: all ./test_game_lifecycle ./test_state_presenter ./test_http_api + ./test_sync_service test_game_domain: test_game_domain.c ../../src/fleet_generator.c ../../src/game_engine.c $(CC) $(CFLAGS) $^ -o $@ @@ -29,5 +30,8 @@ test_state_presenter: test_state_presenter.c ../../src/state_presenter.c test_http_api: test_http_api.c ../../src/http_api.c ../../src/application.c ../../src/command_queue.c ../../src/session_manager.c ../../src/game_lifecycle.c ../../src/fleet_generator.c ../../src/game_engine.c ../../src/state_presenter.c $(CC) $(CFLAGS) $^ -o $@ +test_sync_service: test_sync_service.c ../../src/sync_service.c ../../src/application.c ../../src/command_queue.c ../../src/session_manager.c ../../src/game_lifecycle.c ../../src/fleet_generator.c ../../src/game_engine.c ../../src/state_presenter.c + $(CC) $(CFLAGS) $^ -o $@ + clean: - rm -f test_command_queue test_game_domain test_bot_player test_game_lifecycle test_state_presenter test_http_api + rm -f test_command_queue test_game_domain test_bot_player test_game_lifecycle test_state_presenter test_http_api test_sync_service diff --git a/test/host/test_sync_service.c b/test/host/test_sync_service.c new file mode 100644 index 0000000..d89bc20 --- /dev/null +++ b/test/host/test_sync_service.c @@ -0,0 +1,132 @@ +#include +#include +#include + +#include "sync_service.h" + +typedef struct { uint32_t value; } test_random_t; +typedef struct { int fail_client; uint8_t sends[kSessionCapacity]; char payloads[kSessionCapacity][kStateMessageCapacity]; } capture_t; + +static uint32_t next_random(void *context) { + test_random_t *random = context; + random->value = random->value * 1664525U + 1013904223U; + return random->value; +} + +static void token_text(const uint8_t token[kSessionTokenBytes], char output[33]) { + static const char hex[] = "0123456789abcdef"; + for (uint8_t index = 0; index < kSessionTokenBytes; ++index) { + output[index * 2U] = hex[token[index] >> 4U]; + output[index * 2U + 1U] = hex[token[index] & 15U]; + } + output[32] = '\0'; +} + +static bool capture_send(void *context, int client_id, const char *payload, size_t length) { + capture_t *capture = context; + if (client_id == capture->fail_client) return false; + assert(client_id >= 0 && client_id < kSessionCapacity); + assert(length < kStateMessageCapacity); + ++capture->sends[client_id]; + memcpy(capture->payloads[client_id], payload, length); + capture->payloads[client_id][length] = '\0'; + return true; +} + +static const char *board_start(const char *json, uint8_t board) { + const char *start = strstr(json, "\"boards\":[\""); + assert(start != NULL); + start += strlen("\"boards\":[\""); + return board == 0U ? start : start + kBoardCellCount + 3U; +} + +static void hello(sync_service_t *service, int client_id, const uint8_t token[kSessionTokenBytes], uint32_t version) { + char token_value[33]; + char frame[96]; + char output[kStateMessageCapacity]; + size_t output_length = 0U; + bool changed = false; + bool close = false; + token_text(token, token_value); + snprintf(frame, sizeof(frame), "{\"type\":\"hello\",\"token\":\"%s\",\"version\":%u}", token_value, version); + assert(sync_service_receive(service, client_id, frame, strlen(frame), 10U, output, &output_length, &changed, &close)); + assert(!changed && !close && output_length > 0U && strstr(output, "\"type\":\"state\"") != NULL); +} + +static void test_authentication_visibility_and_backpressure(void) { + test_random_t random = {.value = 1U}; + application_t application; + application_init(&application, (random_source_t){.next_u32 = next_random, .context = &random}); + uint8_t player_1 = 0; + uint8_t player_2 = 0; + uint8_t spectator = 0; + assert(application_join(&application, ROLE_PLAYER_1, "Alice", &player_1) == LIFECYCLE_RESULT_OK); + assert(application_join(&application, ROLE_PLAYER_2, "Bob", &player_2) == LIFECYCLE_RESULT_OK); + assert(application_join(&application, ROLE_SPECTATOR, "Watch", &spectator) == LIFECYCLE_RESULT_OK); + sync_service_t service; + sync_service_init(&service, &application); + assert(sync_service_open(&service, 0, 0U)); + assert(sync_service_open(&service, 1, 0U)); + assert(sync_service_open(&service, 2, 0U)); + for (int client_id = 3; client_id < kSessionCapacity; ++client_id) assert(sync_service_open(&service, client_id, 0U)); + assert(!sync_service_open(&service, kSessionCapacity, 0U)); + hello(&service, 0, application.lifecycle.sessions.entries[player_1].token, 0U); + hello(&service, 1, application.lifecycle.sessions.entries[player_2].token, 999U); + hello(&service, 2, application.lifecycle.sessions.entries[spectator].token, 0U); + + application.lifecycle.game.state.phase = PHASE_IN_PROGRESS; + application.lifecycle.game.state.boards[0].cells[0] = CELL_SHIP; + application.lifecycle.game.state.boards[1].cells[0] = CELL_SHIP; + application.lifecycle.game.state.boards[0].cells[1] = CELL_HIT; + application.lifecycle.game.state.boards[1].cells[1] = CELL_MISS; + capture_t capture = {.fail_client = -1}; + sync_service_broadcast(&service, capture_send, &capture); + assert(board_start(capture.payloads[0], 0)[0] == '1' && board_start(capture.payloads[0], 1)[0] == '0'); + assert(board_start(capture.payloads[1], 0)[0] == '0' && board_start(capture.payloads[1], 1)[0] == '1'); + assert(board_start(capture.payloads[2], 0)[0] == '0' && board_start(capture.payloads[2], 1)[0] == '0'); + assert(board_start(capture.payloads[2], 0)[1] == '3' && board_start(capture.payloads[2], 1)[1] == '2'); + + capture.fail_client = 2; + const uint8_t previous_sends = capture.sends[2]; + sync_service_broadcast(&service, capture_send, &capture); + sync_service_broadcast(&service, capture_send, &capture); + assert(capture.sends[2] == previous_sends); +} + +static void test_timeout_ping_and_command(void) { + test_random_t random = {.value = 9U}; + application_t application; + application_init(&application, (random_source_t){.next_u32 = next_random, .context = &random}); + uint8_t player_1 = 0; + assert(application_join(&application, ROLE_PLAYER_1, "Alice", &player_1) == LIFECYCLE_RESULT_OK); + sync_service_t service; + sync_service_init(&service, &application); + assert(sync_service_open(&service, 3, 0U)); + int expired[kSessionCapacity] = {0}; + size_t expired_count = 0U; + sync_service_expire(&service, kWebSocketHelloTimeoutMs + 1U, expired, &expired_count); + assert(expired_count == 1U && expired[0] == 3); + + assert(sync_service_open(&service, 0, 0U)); + hello(&service, 0, application.lifecycle.sessions.entries[player_1].token, 0U); + char output[kStateMessageCapacity]; + size_t output_length = 0U; + bool changed = false; + bool close = false; + assert(sync_service_receive(&service, 0, "{\"type\":\"ping\"}", 15U, 20U, output, &output_length, &changed, &close)); + assert(!changed && !close && strcmp(output, "{\"type\":\"pong\"}") == 0); + + char token[33]; + char frame[128]; + token_text(application.lifecycle.sessions.entries[player_1].token, token); + snprintf(frame, sizeof(frame), "{\"type\":\"config\",\"token\":\"%s\",\"gameId\":1,\"mode\":\"bot\"}", token); + assert(sync_service_receive(&service, 0, frame, strlen(frame), 30U, output, &output_length, &changed, &close)); + assert(changed && !close && output_length == 0U && application.lifecycle.game.state.mode == MODE_BOT); +} + +int main(void) { + test_authentication_visibility_and_backpressure(); + test_timeout_ping_and_command(); + puts("sync service tests passed"); + return 0; +}