jcreek/CosmicClash
1
0
Fork 0
mirror of https://github.com/jcreek/CosmicClash.git synced 2026-09-10 16:04:04 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
889e30a434673e2e7383bd18dff305316d7153df
CosmicClash/Game/scripts/control_plane_client.gd
T
Josh Creek 889e30a434 fix(multiplayer): reset expiry on reauthentication
2026-09-01 22:26:39 +01:00

569 lines
24 KiB
GDScript
Raw Blame History

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 ticket_id.is_empty() 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 ticket_id.is_empty():
return ERR_INVALID_PARAMETER
return _start_request("queue_recover", HTTPClient.METHOD_GET, "/v1/queue/" + ticket_id, {}, "")
func recover_proposal(proposal_id: String) -> Error:
if proposal_id.is_empty():
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 match_id.is_empty() 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 value.expires_at.is_empty():
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 ticket_id.is_empty() 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 ticket_id.is_empty() 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 proposal_id.is_empty() 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)
if result.has("enqueued_at") and result["enqueued_at"] is String:
result["enqueued_at_unix"] = Time.get_unix_time_from_datetime_string(String(result["enqueued_at"]))
if result.has("expires_at") and result["expires_at"] is String:
result["expires_at_unix"] = Time.get_unix_time_from_datetime_string(String(result["expires_at"]))
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 not String(event["match_id"]).is_empty() and event.has("server_id") and event["server_id"] is String and not String(event["server_id"]).is_empty()
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")
Reference in New Issue View Git Blame Copy Permalink