mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
574 lines
24 KiB
GDScript
574 lines
24 KiB
GDScript
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 returned_player_id.is_empty() 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":
|
|
state.begin_queue(String(payload.get("ticket_id", "")), String(payload.get("playlist", "")))
|
|
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
|
|
|
|
|
|
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")
|