extends Node # Authenticated HTTP boundary for matchmaking. ENet/Steam carries the match # itself; this client only handles queue/proposal control-plane state. signal request_succeeded(operation: String, payload: Dictionary) signal request_failed(operation: String, http_code: int, detail: String) signal session_expired() signal session_changed(player_id: String) signal websocket_event(event: Dictionary) signal websocket_status_changed(status: String) signal assignment_connection_started(assignment: AssignmentState) signal assignment_connection_failed(detail: String) const DEFAULT_BASE_URL := "http://127.0.0.1:8080" const PERSIST_PATH := "user://matchmaking_state.cfg" var base_url := DEFAULT_BASE_URL var access_token := "" var auth_expired := false var player_id := "" var session_expires_at := "" var state: MatchmakingState var ranked_profile: RankedProfileState var assignment: AssignmentState var _request: HTTPRequest var _operation := "" var _last_queue_create: Dictionary = {} var _last_mutation: Dictionary = {} var _last_mutation_retryable := false var _websocket: WebSocketPeer var _websocket_status := "DISCONNECTED" var _websocket_retry_seconds := 0.0 var _websocket_backoff := 1.0 var _pending_assignment_match_id := "" var _pending_resync_resource_id := "" func _ready() -> void: state = MatchmakingState.new() ranked_profile = RankedProfileState.new() assignment = AssignmentState.new() _load_persisted_state() state.changed.connect(_persist_state) _request = HTTPRequest.new() _request.timeout = 10.0 add_child(_request) _request.request_completed.connect(_on_request_completed) state.resync_required.connect(_on_resync_required) _websocket = WebSocketPeer.new() func _process(_delta: float) -> void: if not auth_expired and is_session_expired(session_expires_at): _expire_session() if _websocket == null: return _websocket.poll() var ready_state := _websocket.get_ready_state() if ready_state == WebSocketPeer.STATE_OPEN: _websocket_retry_seconds = 0.0 _websocket_backoff = 1.0 _set_websocket_status("CONNECTED") while _websocket.get_available_packet_count() > 0: _handle_websocket_packet(_websocket.get_packet()) elif ready_state == WebSocketPeer.STATE_CONNECTING: _set_websocket_status("CONNECTING") elif ready_state == WebSocketPeer.STATE_CLOSED: _set_websocket_status("DISCONNECTED") if not auth_expired and is_valid_access_token(access_token): _websocket_retry_seconds -= _delta if _websocket_retry_seconds <= 0.0: _websocket_retry_seconds = _websocket_backoff _websocket_backoff = minf(_websocket_backoff * 2.0, 30.0) connect_event_stream() if not _pending_assignment_match_id.is_empty() and _operation.is_empty() and not player_id.is_empty(): var match_id := _pending_assignment_match_id _pending_assignment_match_id = "" fetch_assignment(match_id) func configure(url: String, token: String) -> bool: var normalized := url.strip_edges().trim_suffix("/") var normalized_token := token.strip_edges() if not is_valid_base_url(normalized) or not is_valid_access_token(normalized_token): return false base_url = normalized access_token = normalized_token session_expires_at = "" auth_expired = false if _websocket != null: connect_event_stream() return true func connect_event_stream() -> Error: if not is_valid_access_token(access_token) or auth_expired or not is_valid_base_url(base_url): return ERR_UNAUTHORIZED var socket_url := websocket_url(base_url) + "/v1/events" _websocket = WebSocketPeer.new() # Godot 4.7 moved handshake headers onto WebSocketPeer; the second # connect_to_url argument is TLSOptions, not an HTTP header array. Keep the # bearer token in the authenticated handshake without putting it in the URL. _websocket.handshake_headers = PackedStringArray(["Authorization: Bearer " + access_token]) var err := _websocket.connect_to_url(socket_url) if err != OK: _set_websocket_status("DISCONNECTED") return err _websocket_retry_seconds = 0.0 _set_websocket_status("CONNECTING") return OK func disconnect_event_stream() -> void: if _websocket != null: _websocket.close() _websocket_retry_seconds = 0.0 _websocket_backoff = 1.0 _set_websocket_status("DISCONNECTED") static func websocket_url(url: String) -> String: if url.begins_with("https://"): return "wss://" + url.trim_prefix("https://") if url.begins_with("http://"): return "ws://" + url.trim_prefix("http://") return "" func queue_create(ticket_id: String, playlist: String, client_build: String, protocol_version: int) -> Error: if not is_valid_resource_id(ticket_id) or (playlist != "casual" and playlist != "ranked") or client_build.is_empty() or protocol_version < 1: return ERR_INVALID_PARAMETER if not state.begin_queue(ticket_id, playlist): return ERR_INVALID_PARAMETER var key := _idempotency_key("queue") _last_queue_create = {"ticket_id": ticket_id, "playlist": playlist, "client_build": client_build, "protocol_version": protocol_version, "key": key} var err := _start_request("queue_create", HTTPClient.METHOD_POST, "/v1/queue", {"ticket_id": ticket_id, "playlist": playlist, "client_build": client_build, "protocol_version": protocol_version}, key) if err != OK: state.fail("Could not start matchmaking: %s" % error_string(err)) return err func login_steam(web_api_ticket: String) -> Error: if not is_valid_web_api_ticket(web_api_ticket): return ERR_INVALID_PARAMETER return _start_request("steam_session", HTTPClient.METHOD_POST, "/v1/session/steam", {"web_api_ticket": web_api_ticket}, "") func retry_queue_create() -> Error: if _last_queue_create.is_empty() or not _last_queue_create.has("ticket_id"): return ERR_INVALID_DATA var ticket_id := String(_last_queue_create["ticket_id"]) var playlist := String(_last_queue_create["playlist"]) if not state.begin_queue(ticket_id, playlist): return ERR_INVALID_PARAMETER var err := _start_request("queue_create", HTTPClient.METHOD_POST, "/v1/queue", {"ticket_id": ticket_id, "playlist": playlist, "client_build": String(_last_queue_create["client_build"]), "protocol_version": int(_last_queue_create["protocol_version"])}, String(_last_queue_create["key"])) if err != OK: state.fail("Could not retry matchmaking: %s" % error_string(err)) return err func can_retry_queue_create() -> bool: return not _last_queue_create.is_empty() and state.phase == MatchmakingState.FAILED and String(_last_queue_create.get("ticket_id", "")) == state.ticket_id func retry_last_mutation() -> Error: if not can_retry_last_mutation(): return ERR_INVALID_DATA var request := _last_mutation.duplicate(true) return _start_request(String(request["operation"]), int(request["method"]), String(request["path"]), request["payload"], String(request["key"]), int(request["expected_revision"])) func can_retry_last_mutation() -> bool: return _last_mutation_retryable and not _last_mutation.is_empty() and _operation.is_empty() and not auth_expired and is_valid_access_token(access_token) func recover_queue(ticket_id: String) -> Error: if not is_valid_resource_id(ticket_id): return ERR_INVALID_PARAMETER return _start_request("queue_recover", HTTPClient.METHOD_GET, "/v1/queue/" + ticket_id, {}, "") func recover_proposal(proposal_id: String) -> Error: if not is_valid_resource_id(proposal_id): return ERR_INVALID_PARAMETER return _start_request("proposal_recover", HTTPClient.METHOD_GET, "/v1/proposals/" + proposal_id, {}, "") static func resync_target(resource_id: String, ticket_id: String, proposal_id: String, proposal_open: bool) -> String: if resource_id == ticket_id and not ticket_id.is_empty(): return ticket_id if resource_id == proposal_id and not proposal_id.is_empty(): return proposal_id if proposal_open else ticket_id return "" func fetch_ranked_profile() -> Error: return _start_request("ranked_profile", HTTPClient.METHOD_GET, "/v1/profile/ranked", {}, "") func fetch_assignment(match_id: String) -> Error: if not is_valid_resource_id(match_id) or player_id.is_empty(): return ERR_INVALID_PARAMETER return _start_request("assignment", HTTPClient.METHOD_GET, "/v1/assignments/" + match_id, {}, "") # Starts the assigned game transport only after AssignmentState has validated # the complete player-scoped manifest. The signed authorisation is passed to # MatchNet's hello RPC, never appended to the endpoint URL or logged. Server # admission remains authoritative; this method only owns the client-side # readiness/transport boundary. func connect_to_assignment() -> Error: if assignment == null or not assignment.available or not _assignment_is_fresh(assignment): var unavailable_detail := "Match assignment is unavailable or expired" assignment_connection_failed.emit(unavailable_detail) return ERR_UNAUTHORIZED var endpoint := _split_assignment_endpoint(assignment.endpoint) if endpoint.is_empty(): var invalid_detail := "Match assignment endpoint is invalid" assignment_connection_failed.emit(invalid_detail) return ERR_INVALID_PARAMETER var transport := NetworkManager.TRANSPORT_ENET if assignment.transport == "enet" else NetworkManager.TRANSPORT_STEAM MatchNet.join_authorisation = assignment.join_authorisation state.mark_connecting() var err := NetworkManager.join(String(endpoint["host"]), int(endpoint["port"]), transport) if err != OK: MatchNet.join_authorisation = "" assignment_connection_failed.emit("Unable to connect to match server") return err assignment_connection_started.emit(assignment) return OK static func _assignment_is_fresh(value: AssignmentState) -> bool: if value == null or not AssignmentState.is_valid_expiry_timestamp(value.expires_at): return false var expiry := Time.get_unix_time_from_datetime_string(value.expires_at) return expiry > Time.get_unix_time_from_system() static func _split_assignment_endpoint(value: String) -> Dictionary: if not AssignmentState._valid_endpoint(value): return {} var separator := value.rfind(":") return {"host": value.substr(0, separator), "port": int(value.substr(separator + 1))} func heartbeat(ticket_id: String, expected_revision: int) -> Error: if not is_valid_resource_id(ticket_id) or expected_revision < 0: return ERR_INVALID_PARAMETER return _start_request("queue_heartbeat", HTTPClient.METHOD_POST, "/v1/queue/%s/heartbeat" % ticket_id, {}, _idempotency_key("heartbeat"), expected_revision) func cancel_queue(ticket_id: String, expected_revision: int) -> Error: if not is_valid_resource_id(ticket_id) or expected_revision < 0 or not state.can_cancel(): return ERR_INVALID_PARAMETER return _start_request("queue_cancel", HTTPClient.METHOD_POST, "/v1/queue/%s/cancel" % ticket_id, {}, _idempotency_key("cancel"), expected_revision) func respond_to_proposal(proposal_id: String, accept: bool, expected_revision: int) -> Error: if not is_valid_resource_id(proposal_id) or expected_revision < 0: return ERR_INVALID_PARAMETER var action := "accept" if accept else "decline" return _start_request("proposal_" + action, HTTPClient.METHOD_POST, "/v1/proposals/%s/%s" % [proposal_id, action], {}, _idempotency_key("proposal"), expected_revision) static func is_valid_base_url(url: String) -> bool: if url.is_empty() or url.contains(" ") or url.contains("\r") or url.contains("\n") or url.contains("?") or url.contains("#") or url.contains("@") or url.ends_with("/"): return false return url.begins_with("http://") or url.begins_with("https://") static func is_valid_web_api_ticket(ticket: String) -> bool: return not ticket.is_empty() and ticket.length() <= 4096 and not ticket.contains("\r") and not ticket.contains("\n") static func is_valid_access_token(token: String) -> bool: var separator := token.find(":") return separator > 0 and separator < token.length() - 1 and token.length() <= 4096 and not token.contains("\r") and not token.contains("\n") static func is_session_expired(expires_at: String, now_unix: int = -1) -> bool: if expires_at.is_empty(): return false if not is_valid_rfc3339_timestamp(expires_at): return true var expiry_unix := Time.get_unix_time_from_datetime_string(expires_at) if expiry_unix < 0: return true var current_unix := now_unix if current_unix < 0: current_unix = int(Time.get_unix_time_from_system()) return expiry_unix <= current_unix static func is_valid_rfc3339_timestamp(value: String) -> bool: if value.is_empty(): return false var timestamp_pattern := RegEx.create_from_string("^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(?:\\.\\d+)?(?:Z|[+-]\\d{2}:\\d{2})$") return timestamp_pattern.search(value) != null static func is_valid_resource_id(value: String) -> bool: if value.length() < 16 or value.length() > 128: return false var resource_pattern := RegEx.create_from_string("^[A-Za-z0-9_-]+$") return resource_pattern.search(value) != null static func is_retryable_mutation_response(response_code: int) -> bool: return response_code == 0 or response_code == HTTPClient.RESPONSE_REQUEST_TIMEOUT or response_code == HTTPClient.RESPONSE_TOO_MANY_REQUESTS or response_code >= 500 static func should_recover_queue_after_conflict(operation: String, response_code: int, ticket_id: String) -> bool: return response_code == HTTPClient.RESPONSE_CONFLICT and operation in ["queue_heartbeat", "queue_cancel"] and not ticket_id.is_empty() static func normalize_ticket(payload: Dictionary) -> Dictionary: var result := payload.duplicate(true) for pair in [["enqueued_at", "enqueued_at_unix"], ["expires_at", "expires_at_unix"]]: var source_key: String = pair[0] var target_key: String = pair[1] if not result.has(source_key): continue if not result[source_key] is String or not is_valid_rfc3339_timestamp(String(result[source_key])): result[target_key] = -1 else: result[target_key] = Time.get_unix_time_from_datetime_string(String(result[source_key])) return result func _start_request(operation: String, method: HTTPClient.Method, path: String, payload: Dictionary, idempotency_key: String, expected_revision: int = -1) -> Error: if _request == null or not _operation.is_empty() or not is_valid_base_url(base_url): return ERR_BUSY if not _operation.is_empty() else ERR_UNAUTHORIZED if operation != "steam_session" and access_token.is_empty(): return ERR_UNAUTHORIZED if operation != "steam_session" and is_session_expired(session_expires_at): _expire_session() return ERR_UNAUTHORIZED var headers := PackedStringArray(["Accept: application/json"]) if operation != "steam_session": headers.append("Authorization: Bearer " + access_token) if not idempotency_key.is_empty(): headers.append("Idempotency-Key: " + idempotency_key) if expected_revision >= 0: headers.append("If-Match-Revision: %d" % expected_revision) var body := "" if payload.is_empty() else JSON.stringify(payload) _operation = operation var err := _request.request(base_url + path, headers, method, body) if err != OK: _operation = "" return err if not idempotency_key.is_empty(): _last_mutation = {"operation": operation, "method": method, "path": path, "payload": payload.duplicate(true), "key": idempotency_key, "expected_revision": expected_revision} _last_mutation_retryable = false return OK func _expire_session() -> void: if auth_expired: return access_token = "" auth_expired = true disconnect_event_stream() state.fail("Session expired; sign in again") ranked_profile.set_error("Session expired; sign in again") session_expired.emit() func _on_request_completed(result: HTTPRequest.Result, response_code: int, _headers: PackedStringArray, body: PackedByteArray) -> void: var operation := _operation _operation = "" if result != HTTPRequest.RESULT_SUCCESS: _last_mutation_retryable = _last_mutation.get("operation", "") == operation if operation == "ranked_profile": ranked_profile.set_error("Ranked profile request failed") elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover": state.fail("Control-plane request failed") else: state.set_notice("Control-plane request failed; retrying is safe") request_failed.emit(operation, response_code, "network error") return var parsed = JSON.parse_string(body.get_string_from_utf8()) if not parsed is Dictionary: _last_mutation_retryable = _last_mutation.get("operation", "") == operation if operation == "ranked_profile": ranked_profile.set_error("Ranked profile returned invalid JSON") elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover": state.fail("Control-plane returned invalid JSON") else: state.set_notice("Control-plane returned invalid JSON; retrying is safe") request_failed.emit(operation, response_code, "invalid JSON") return if response_code < 200 or response_code >= 300: _last_mutation_retryable = _last_mutation.get("operation", "") == operation and is_retryable_mutation_response(response_code) var detail := String(parsed.get("error", "request rejected")) var recover_proposal_after_conflict := response_code == HTTPClient.RESPONSE_CONFLICT and (operation == "proposal_accept" or operation == "proposal_decline") and not state.proposal_id.is_empty() var recover_queue_after_conflict := should_recover_queue_after_conflict(operation, response_code, state.ticket_id) if response_code == HTTPClient.RESPONSE_UNAUTHORIZED: access_token = "" auth_expired = true disconnect_event_stream() state.fail("Session expired; sign in again") ranked_profile.set_error("Session expired; sign in again") session_expired.emit() elif response_code == HTTPClient.RESPONSE_GONE and operation == "queue_recover": state.expire("Queue ticket expired") elif response_code == HTTPClient.RESPONSE_SERVICE_UNAVAILABLE: state.set_notice("Matchmaking is temporarily unavailable; retrying is safe") elif operation == "ranked_profile": ranked_profile.set_error(detail) elif response_code == HTTPClient.RESPONSE_NOT_FOUND and (operation == "queue_recover" or operation == "proposal_recover"): state.fail("Matchmaking record is no longer available") elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover": state.fail(detail) else: state.set_notice(detail) request_failed.emit(operation, response_code, detail) if recover_proposal_after_conflict: _pending_resync_resource_id = state.proposal_id call_deferred("_run_pending_resync") if recover_queue_after_conflict: _pending_resync_resource_id = state.ticket_id call_deferred("_run_pending_resync") return var payload: Dictionary = parsed _last_mutation_retryable = false if operation == "steam_session": var returned_token := String(payload.get("access_token", "")) var returned_player_id := String(payload.get("player_id", "")) if not is_valid_resource_id(returned_player_id) or not is_valid_access_token(returned_token): request_failed.emit(operation, response_code, "invalid session response") return player_id = returned_player_id access_token = returned_token auth_expired = false session_expires_at = String(payload.get("expires_at", "")) connect_event_stream() session_changed.emit(player_id) elif operation == "queue_create" or operation == "queue_recover" or operation == "queue_heartbeat" or operation == "queue_cancel": if not _valid_response_opaque_id(payload, "ticket_id"): state.fail("Queue response contains an invalid ticket identifier") request_failed.emit(operation, response_code, "invalid ticket identifier") return if operation == "queue_create": state.begin_queue(String(payload["ticket_id"]), String(payload.get("playlist", ""))) elif operation.begins_with("proposal_"): if not _valid_response_opaque_id(payload, "proposal_id"): state.fail("Proposal response contains an invalid proposal identifier") request_failed.emit(operation, response_code, "invalid proposal identifier") return if operation.begins_with("queue_"): state.apply_ticket_update(normalize_ticket(payload)) elif operation.begins_with("proposal_"): state.apply_proposal_update(payload) elif operation == "ranked_profile": if not ranked_profile.apply(payload): request_failed.emit(operation, response_code, ranked_profile.error_message) return elif operation == "assignment": if not assignment.apply(payload, player_id): request_failed.emit(operation, response_code, assignment.error_message) return request_succeeded.emit(operation, payload) if not _pending_resync_resource_id.is_empty(): call_deferred("_run_pending_resync") func _handle_websocket_packet(packet: PackedByteArray) -> void: var parsed = JSON.parse_string(packet.get_string_from_utf8()) if not parsed is Dictionary or not _valid_websocket_event(parsed): websocket_status_changed.emit("INVALID_EVENT") return var event: Dictionary = parsed websocket_event.emit(event) var event_name := String(event["event"]) if event_name == "state_changed": var update := event.duplicate(true) update["ticket_id"] = String(event["resource_id"]) if not state.apply_ticket_update(update): return elif event_name == "proposal_changed": var proposal_update := event.duplicate(true) proposal_update["proposal_id"] = String(event["resource_id"]) state.apply_proposal_update(proposal_update) elif event_name == "assignment_changed": state.mark_assignment_ready() _pending_assignment_match_id = String(event["match_id"]) elif event_name == "error": state.set_notice("Control-plane error: %s" % String(event["code"])) _on_resync_required(String(event["resource_id"])) static func _valid_websocket_event(event: Dictionary) -> bool: if not event.has("event") or not event["event"] is String or String(event["event"]).is_empty(): return false if not event.has("revision") or not _valid_revision(event["revision"]): return false if not event.has("resource_id") or not event["resource_id"] is String or not is_valid_resource_id(String(event["resource_id"])): return false if not event.has("occurred_at") or not event["occurred_at"] is String or not is_valid_rfc3339_timestamp(String(event["occurred_at"])): return false var event_name := String(event["event"]) if event_name == "assignment_changed": return event.has("match_id") and event["match_id"] is String and is_valid_resource_id(String(event["match_id"])) and event.has("server_id") and event["server_id"] is String and is_valid_resource_id(String(event["server_id"])) if event_name == "error": return event.has("code") and String(event["code"]) in ["REVISION_GAP", "NOT_AUTHORISED", "INVALID_STATE", "RATE_LIMITED"] if event_name == "state_changed": return event.has("state") and String(event["state"]) in ["QUEUED", "PROPOSED", "ACCEPTED", "ALLOCATING", "PROCESS_READY", "ASSIGNMENT_READY", "ASSIGNED", "CONNECTING", "LIVE", "RESULT_PENDING", "COMPLETED", "CANCELLED", "EXPIRED", "FAILED"] if event_name == "proposal_changed": return event.has("state") and String(event["state"]) in ["OPEN", "ACCEPTED", "DECLINED", "EXPIRED", "CANCELLED"] return false static func _valid_revision(value: Variant) -> bool: if value is int: return int(value) >= 0 if value is float: return is_finite(float(value)) and float(value) >= 0.0 and float(value) == floor(float(value)) return false static func _valid_response_opaque_id(payload: Dictionary, key: String) -> bool: return payload.has(key) and payload[key] is String and is_valid_resource_id(String(payload[key])) func _on_resync_required(resource_id: String) -> void: if not _operation.is_empty(): _pending_resync_resource_id = resource_id return _run_resync(resource_id) func _run_pending_resync() -> void: if not _operation.is_empty() or _pending_resync_resource_id.is_empty(): return var resource_id := _pending_resync_resource_id _pending_resync_resource_id = "" _run_resync(resource_id) func _run_resync(resource_id: String) -> void: var target := resync_target(resource_id, state.ticket_id, state.proposal_id, state.has_open_proposal()) if target == state.ticket_id and not state.ticket_id.is_empty(): recover_queue(state.ticket_id) elif target == state.proposal_id and not state.proposal_id.is_empty(): recover_proposal(state.proposal_id) func _set_websocket_status(status: String) -> void: if _websocket_status == status: return _websocket_status = status websocket_status_changed.emit(status) if status == "CONNECTED": if not state.ticket_id.is_empty() and state.phase not in [MatchmakingState.CANCELLED, MatchmakingState.EXPIRED, MatchmakingState.FAILED, MatchmakingState.LIVE, MatchmakingState.COMPLETED]: var resource_id := state.proposal_id if state.has_open_proposal() else state.ticket_id if _operation.is_empty(): _run_resync(resource_id) else: # A reconnect must not lose its authoritative recovery merely because # the previous mutation has not acknowledged yet. The deferred path # runs after that request completes and avoids an ERR_BUSY drop. _pending_resync_resource_id = resource_id func _idempotency_key(prefix: String) -> String: return "%s-%s-%s" % [prefix, str(Time.get_ticks_usec()), str(randi())] func _persist_state(snapshot: Dictionary) -> void: var config := ConfigFile.new() config.set_value("matchmaking", "snapshot", JSON.stringify(snapshot)) config.save(PERSIST_PATH) func _load_persisted_state() -> void: var config := ConfigFile.new() if config.load(PERSIST_PATH) != OK: return var raw = config.get_value("matchmaking", "snapshot", "") if not raw is String or String(raw).is_empty(): return var parsed = JSON.parse_string(String(raw)) if parsed is Dictionary and not state.restore_snapshot(parsed): state.fail("Saved matchmaking state is invalid")