feat: connect Godot matchmaking event stream

Game/scripts/control_plane_client.gd

@@ -8,6 +8,8 @@ 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)
 
 const DEFAULT_BASE_URL := "http://127.0.0.1:8080"
 const PERSIST_PATH := "user://matchmaking_state.cfg"
@@ -24,6 +26,8 @@ var assignment: AssignmentState
 var _request: HTTPRequest
 var _operation := ""
 var _last_queue_create: Dictionary = {}
+var _websocket: WebSocketPeer
+var _websocket_status := "DISCONNECTED"
 
 
 func _ready() -> void:
@@ -36,6 +40,23 @@ func _ready() -> void:
 	_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 _websocket == null:
+		return
+	_websocket.poll()
+	var ready_state := _websocket.get_ready_state()
+	if ready_state == WebSocketPeer.STATE_OPEN:
+		_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")
 
 
 func configure(url: String, token: String) -> bool:
@@ -49,6 +70,33 @@ func configure(url: String, token: String) -> bool:
 	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()
+	var err := _websocket.connect_to_url(socket_url, PackedStringArray(["Authorization: Bearer " + access_token]))
+	if err != OK:
+		_set_websocket_status("DISCONNECTED")
+		return err
+	_set_websocket_status("CONNECTING")
+	return OK
+
+
+func disconnect_event_stream() -> void:
+	if _websocket != null:
+		_websocket.close()
+	_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
@@ -242,6 +290,52 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head
 	request_succeeded.emit(operation, payload)
 
 
+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)
+
+
+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 (event["revision"] is int or event["revision"] is float):
+		return false
+	if int(event["revision"]) < 0 or not event.has("resource_id") or not event["resource_id"] is String or String(event["resource_id"]).is_empty():
+		return false
+	if not event.has("occurred_at") or not event["occurred_at"] is String or String(event["occurred_at"]).is_empty():
+		return false
+	var event_name := String(event["event"])
+	return event_name in ["state_changed", "proposal_changed", "assignment_changed", "error"]
+
+
+func _on_resync_required(resource_id: String) -> void:
+	if resource_id == state.ticket_id and not state.ticket_id.is_empty():
+		recover_queue(state.ticket_id)
+	elif resource_id == 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)
+
+
 func _idempotency_key(prefix: String) -> String:
 	return "%s-%s-%s" % [prefix, str(Time.get_ticks_usec()), str(randi())]