Add multiplayer item use and session chat

network/direct_enet_transport.gd

@@ -11,7 +11,9 @@ func start_host(
 	var enet_peer := ENetMultiplayerPeer.new()
 	if not bind_address.is_empty() and bind_address != "*":
 		enet_peer.set_bind_ip(bind_address)
-	var error: Error = enet_peer.create_server(port, max_clients, 5)
+	var error: Error = enet_peer.create_server(
+		port, max_clients, NetworkProtocol.ENET_CHANNEL_COUNT
+	)
 	if error != OK:
 		transport_error.emit("Unable to host UDP port %d." % port)
 		return error
@@ -27,7 +29,9 @@ func connect_to_route(route: ConnectionRoute) -> Error:
 		return ERR_INVALID_PARAMETER
 	var endpoint: ConnectionEndpoint = route.direct_endpoint
 	var enet_peer := ENetMultiplayerPeer.new()
-	var error: Error = enet_peer.create_client(endpoint.host, endpoint.port, 5)
+	var error: Error = enet_peer.create_client(
+		endpoint.host, endpoint.port, NetworkProtocol.ENET_CHANNEL_COUNT
+	)
 	if error != OK:
 		transport_error.emit(
 			"Unable to connect to %s." % endpoint.normalized_display

network/network_chat_protocol.gd

@@ -0,0 +1,61 @@
+class_name NetworkChatProtocol
+extends RefCounted
+
+const CAPABILITY: StringName = &"chat_v1"
+const RELIABLE_CHANNEL: int = NetworkProtocol.CHAT_RELIABLE_CHANNEL
+const MAX_VISIBLE_CHARACTERS: int = 300
+const MAX_UTF8_BYTES: int = 1200
+const MAX_HISTORY: int = 100
+const LATE_JOIN_HISTORY: int = 50
+const MAX_ID_LENGTH: int = 96
+
+enum Kind { PLAYER, SYSTEM }
+
+
+static func sanitize_body(value: Variant) -> String:
+	if typeof(value) != TYPE_STRING:
+		return ""
+	var result: String = str(value).replace("\r", " ").replace("\n", " ")
+	result = result.replace("\t", " ").strip_edges()
+	if result.is_empty() or result.length() > MAX_VISIBLE_CHARACTERS:
+		return ""
+	if result.to_utf8_buffer().size() > MAX_UTF8_BYTES:
+		return ""
+	for index: int in result.length():
+		var codepoint: int = result.unicode_at(index)
+		if codepoint < 32 or codepoint == 127:
+			return ""
+	return result
+
+
+static func validate_request(data: Variant) -> bool:
+	return (
+		typeof(data) == TYPE_DICTIONARY
+		and _valid_id(data.get("request_id"))
+		and _valid_id(data.get("session_id"))
+		and not sanitize_body(data.get("body")).is_empty()
+	)
+
+
+static func validate_message(data: Variant) -> bool:
+	return (
+		typeof(data) == TYPE_DICTIONARY
+		and _valid_id(data.get("message_id"))
+		and _valid_id(data.get("session_id"))
+		and typeof(data.get("sequence")) == TYPE_INT
+		and int(data["sequence"]) >= 0
+		and typeof(data.get("kind")) == TYPE_INT
+		and int(data["kind"]) in [Kind.PLAYER, Kind.SYSTEM]
+		and typeof(data.get("sender_peer_id")) == TYPE_INT
+		and typeof(data.get("sender_display_name")) == TYPE_STRING
+		and str(data["sender_display_name"]).length() <= 24
+		and not sanitize_body(data.get("body")).is_empty()
+	)
+
+
+static func _valid_id(value: Variant) -> bool:
+	return (
+		typeof(value) == TYPE_STRING
+		and not str(value).is_empty()
+		and str(value).length() <= MAX_ID_LENGTH
+	)

network/network_chat_protocol.gd.uid

@@ -0,0 +1 @@
+uid://c3hg65v27hi2g

network/network_chat_service.gd

@@ -0,0 +1,235 @@
+class_name NetworkChatService
+extends Node
+
+const BURST_COUNT: int = 3
+const WINDOW_COUNT: int = 5
+const WINDOW_SECONDS: float = 10.0
+
+signal message_received(message: Dictionary)
+signal send_rejected(message: String)
+signal history_replaced(messages: Array[Dictionary])
+
+var _session: NetworkSession
+var _history: Array[Dictionary] = []
+var _seen_messages: Dictionary[String, bool] = {}
+var _request_ledgers: Dictionary[int, Dictionary] = {}
+var _rate_times: Dictionary[int, Array] = {}
+var _sequence: int = 0
+var _peer_names: Dictionary[int, String] = {}
+
+
+func setup(session: NetworkSession) -> void:
+	_session = session
+	_session.peer_authenticated.connect(_on_peer_authenticated)
+	_session.peer_removed.connect(_on_peer_removed)
+	_session.state_changed.connect(_on_session_state_changed)
+	_session.peer_display_name_changed.connect(
+		func(peer_id: int, display_name: String) -> void:
+			_peer_names[peer_id] = display_name
+	)
+
+
+func send_local_message(body: String) -> bool:
+	if (
+		_session == null
+		or not _session.is_gameplay_session_active()
+		or (
+			not _session.is_host()
+			and not _session.supports_server_capability(
+				NetworkChatProtocol.CAPABILITY
+			)
+		)
+	):
+		send_rejected.emit("Chat is unavailable.")
+		return false
+	var clean := NetworkChatProtocol.sanitize_body(body)
+	if clean.is_empty():
+		send_rejected.emit("Message is empty or too long.")
+		return false
+	var request := {
+		"request_id": _new_id("chat_request"),
+		"session_id": _session.get_session_id(),
+		"body": clean,
+	}
+	if _session.is_host():
+		_handle_request(_session.get_local_peer_id(), request)
+	else:
+		submit_chat_message.rpc_id(1, request)
+	return true
+
+
+func get_history() -> Array[Dictionary]:
+	return _history.duplicate(true)
+
+
+@rpc("any_peer", "call_remote", "reliable", NetworkChatProtocol.RELIABLE_CHANNEL)
+func submit_chat_message(data: Dictionary) -> void:
+	var sender_id := multiplayer.get_remote_sender_id()
+	if _session.is_host() and _session.is_authenticated_peer(sender_id):
+		_handle_request(sender_id, data)
+
+
+func _handle_request(peer_id: int, data: Dictionary) -> void:
+	if (
+		not NetworkChatProtocol.validate_request(data)
+		or str(data["session_id"]) != _session.get_session_id()
+	):
+		_send_rejection(peer_id, "Message could not be sent.")
+		return
+	var request_id: String = data["request_id"]
+	var ledger: Dictionary = _request_ledgers.get(peer_id, {})
+	if ledger.has(request_id):
+		_send_message(peer_id, ledger[request_id])
+		return
+	if not _consume_rate(peer_id):
+		_send_rejection(peer_id, "Slow down.")
+		return
+	var record := _session.get_peer_record(peer_id)
+	if record == null:
+		return
+	var message := _make_message(
+		NetworkChatProtocol.Kind.PLAYER,
+		peer_id,
+		record.display_name,
+		NetworkChatProtocol.sanitize_body(data["body"])
+	)
+	ledger[request_id] = message.duplicate(true)
+	while ledger.size() > 64:
+		ledger.erase(ledger.keys().front())
+	_request_ledgers[peer_id] = ledger
+	_broadcast(message)
+
+
+func _make_message(
+	kind: int,
+	peer_id: int,
+	display_name: String,
+	body: String,
+) -> Dictionary:
+	_sequence += 1
+	return {
+		"message_id": _new_id("chat"),
+		"session_id": _session.get_session_id(),
+		"sequence": _sequence,
+		"kind": kind,
+		"sender_peer_id": peer_id,
+		"sender_display_name": display_name.left(24),
+		"body": body,
+	}
+
+
+func _broadcast(message: Dictionary) -> void:
+	_apply_message(message)
+	receive_chat_message.rpc(message)
+
+
+func _send_message(peer_id: int, message: Dictionary) -> void:
+	if peer_id == _session.get_local_peer_id():
+		_apply_message(message)
+	else:
+		receive_chat_message.rpc_id(peer_id, message)
+
+
+@rpc("authority", "call_remote", "reliable", NetworkChatProtocol.RELIABLE_CHANNEL)
+func receive_chat_message(data: Dictionary) -> void:
+	_apply_message(data)
+
+
+func _apply_message(data: Dictionary) -> void:
+	if (
+		not NetworkChatProtocol.validate_message(data)
+		or str(data["session_id"]) != _session.get_session_id()
+		or _seen_messages.has(str(data["message_id"]))
+	):
+		return
+	_seen_messages[data["message_id"]] = true
+	_history.append(data.duplicate(true))
+	while _history.size() > NetworkChatProtocol.MAX_HISTORY:
+		_history.pop_front()
+	message_received.emit(data.duplicate(true))
+
+
+func _send_rejection(peer_id: int, message: String) -> void:
+	if peer_id == _session.get_local_peer_id():
+		send_rejected.emit(message)
+	else:
+		receive_chat_rejection.rpc_id(peer_id, message.left(80))
+
+
+@rpc("authority", "call_remote", "reliable", NetworkChatProtocol.RELIABLE_CHANNEL)
+func receive_chat_rejection(message: String) -> void:
+	send_rejected.emit(message.left(80))
+
+
+func _on_peer_authenticated(peer_id: int, display_name: String) -> void:
+	if not _session.is_host():
+		return
+	_peer_names[peer_id] = display_name
+	var start := maxi(0, _history.size() - NetworkChatProtocol.LATE_JOIN_HISTORY)
+	receive_chat_history.rpc_id(peer_id, _history.slice(start))
+	_broadcast(_make_message(
+		NetworkChatProtocol.Kind.SYSTEM, 0, "", "%s joined." % display_name
+	))
+
+
+func _on_peer_removed(peer_id: int) -> void:
+	_request_ledgers.erase(peer_id)
+	_rate_times.erase(peer_id)
+	if not _session.is_host():
+		return
+	var name: String = _peer_names.get(peer_id, "Player")
+	_peer_names.erase(peer_id)
+	_broadcast(_make_message(
+		NetworkChatProtocol.Kind.SYSTEM, 0, "", "%s left." % name
+	))
+
+
+@rpc("authority", "call_remote", "reliable", NetworkChatProtocol.RELIABLE_CHANNEL)
+func receive_chat_history(values: Array) -> void:
+	if values.size() > NetworkChatProtocol.LATE_JOIN_HISTORY:
+		return
+	_history.clear()
+	_seen_messages.clear()
+	for value: Variant in values:
+		if NetworkChatProtocol.validate_message(value):
+			_apply_message(value)
+	history_replaced.emit(_history.duplicate(true))
+
+
+func _consume_rate(peer_id: int) -> bool:
+	var now := Time.get_ticks_msec() / 1000.0
+	var times: Array = _rate_times.get(peer_id, [])
+	while not times.is_empty() and now - float(times.front()) > WINDOW_SECONDS:
+		times.pop_front()
+	var recent_burst := 0
+	for value: Variant in times:
+		if now - float(value) <= 1.0:
+			recent_burst += 1
+	if recent_burst >= BURST_COUNT or times.size() >= WINDOW_COUNT:
+		_rate_times[peer_id] = times
+		return false
+	times.append(now)
+	_rate_times[peer_id] = times
+	return true
+
+
+func _on_session_state_changed(state: NetworkSession.State) -> void:
+	if state in [
+		NetworkSession.State.INACTIVE,
+		NetworkSession.State.DISCONNECTING,
+		NetworkSession.State.CONNECTION_FAILED,
+		NetworkSession.State.SERVER_LOST,
+	]:
+		_history.clear()
+		_seen_messages.clear()
+		_request_ledgers.clear()
+		_rate_times.clear()
+		_peer_names.clear()
+		_sequence = 0
+		history_replaced.emit([])
+
+
+func _new_id(prefix: String) -> String:
+	return "%s:%s" % [
+		prefix, Crypto.new().generate_random_bytes(16).hex_encode(),
+	]

network/network_chat_service.gd.uid

@@ -0,0 +1 @@
+uid://b10jayoog5dc2

network/network_fishing_service.gd

@@ -32,6 +32,7 @@ var _local_capacity: PlayerCoolerCapacity
 var _save_manager: PlayerSaveManager
 var _item_catalog: ItemCatalog
 var _fish_catalog: FishPoolType
+var _item_use: NetworkItemUseService
 var _attempts: Dictionary[int, NetworkFishingAttempt] = {}
 var _request_ledgers: Dictionary[int, Dictionary] = {}
 var _result_ledgers: Dictionary[String, bool] = {}
@@ -53,6 +54,7 @@ func setup(
 	save_manager: PlayerSaveManager,
 	item_catalog: ItemCatalog,
 	fish_catalog: FishPoolType,
+	item_use: NetworkItemUseService,
 ) -> void:
 	_session = session
 	_spawn_service = spawn_service
@@ -63,6 +65,7 @@ func setup(
 	_save_manager = save_manager
 	_item_catalog = item_catalog
 	_fish_catalog = fish_catalog
+	_item_use = item_use
 	if not _session.peer_removed.is_connected(_on_peer_removed):
 		_session.peer_removed.connect(_on_peer_removed)
 	if not _session.state_changed.is_connected(_on_session_state_changed):
@@ -259,8 +262,12 @@ func _handle_cast_request(peer_id: int, data: Dictionary) -> void:
 	if region == null or region.fish_pool == null:
 		_record_and_reject(peer_id, request_id, "Cannot fish here.")
 		return
+	var effects: PlayerItemEffects = (
+		_item_use.get_effects_for_peer(peer_id)
+		if _item_use != null else null
+	)
 	var selected_fish: FishDataType = _select_authoritative_fish(
-		region, data
+		region, data, effects
 	)
 	if selected_fish == null:
 		_record_and_reject(peer_id, request_id, "Nothing is biting here.")
@@ -275,10 +282,14 @@ func _handle_cast_request(peer_id: int, data: Dictionary) -> void:
 	attempt.target = target
 	attempt.bobber_position = target
 	attempt.fish_id = selected_fish.id
-	attempt.reel_speed = float(data["reel_speed"])
-	attempt.barrier_damage = int(data["barrier_damage"])
+	attempt.reel_speed = float(data["reel_speed"]) * (
+		effects.get_reel_multiplier() if effects != null else 1.0
+	)
+	attempt.barrier_damage = int(data["barrier_damage"]) + (
+		effects.get_barrier_bonus() if effects != null else 0
+	)
 	attempt.bite_time_remaining = _fishing_spot.wait_time * float(
-		data["bite_multiplier"]
+		effects.get_bite_time_multiplier() if effects != null else 1.0
 	)
 	attempt.controller = CatchController.new()
 	add_child(attempt.controller)
@@ -359,6 +370,7 @@ func _make_waiting_snapshot(
 func _select_authoritative_fish(
 	region: FishableWaterRegion,
 	data: Dictionary,
+	effects: PlayerItemEffects,
 ) -> FishDataType:
 	var evidence_log := CollectionLogType.new()
 	for value: Variant in data["discovered_fish_ids"]:
@@ -370,8 +382,11 @@ func _select_authoritative_fish(
 	selector.undiscovered_weight_multiplier = (
 		_fishing_spot.undiscovered_weight_multiplier
 	)
-	for value: Variant in data["rarity_multipliers"]:
-		selector.rarity_weight_multipliers.append(float(value))
+	for rarity: int in range(FishDataType.Rarity.size()):
+		selector.rarity_weight_multipliers.append(
+			effects.get_rarity_weight_multiplier(rarity)
+			if effects != null else 1.0
+		)
 	selector.begin_roll()
 	var context: FishingContextType = _fishing_spot.build_network_context(region)
 	return selector.select_fish(region.fish_pool, context, evidence_log)

network/network_item_protocol.gd

@@ -0,0 +1,83 @@
+class_name NetworkItemProtocol
+extends RefCounted
+
+const ITEM_USE_CAPABILITY: StringName = &"item_use_v1"
+const EQUIPMENT_CAPABILITY: StringName = &"equipment_v1"
+const RELIABLE_CHANNEL: int = NetworkProtocol.ITEM_RELIABLE_CHANNEL
+const MAX_ID_LENGTH: int = 96
+const MAX_ITEM_ID_LENGTH: int = 96
+const MAX_MESSAGE_LENGTH: int = 160
+
+
+static func validate_use_request(data: Variant) -> String:
+	if typeof(data) != TYPE_DICTIONARY:
+		return "Item use could not be completed."
+	var value: Dictionary = data
+	for key: String in [
+		"request_id", "session_id", "item_id", "quantity",
+	]:
+		if not value.has(key):
+			return "Item use could not be completed."
+	if (
+		not _valid_id(value["request_id"])
+		or not _valid_id(value["session_id"])
+		or typeof(value["item_id"]) not in [TYPE_STRING, TYPE_STRING_NAME]
+		or str(value["item_id"]).is_empty()
+		or str(value["item_id"]).length() > MAX_ITEM_ID_LENGTH
+		or typeof(value["quantity"]) != TYPE_INT
+		or int(value["quantity"]) < 1
+		or int(value["quantity"]) > 99
+	):
+		return "Item use could not be completed."
+	return ""
+
+
+static func validate_use_result(data: Variant) -> bool:
+	if typeof(data) != TYPE_DICTIONARY:
+		return false
+	var value: Dictionary = data
+	return (
+		_valid_id(value.get("result_id"))
+		and _valid_id(value.get("request_id"))
+		and _valid_id(value.get("session_id"))
+		and typeof(value.get("target_peer_id")) == TYPE_INT
+		and typeof(value.get("accepted")) == TYPE_BOOL
+		and typeof(value.get("item_id")) in [TYPE_STRING, TYPE_STRING_NAME]
+		and not str(value["item_id"]).is_empty()
+		and str(value["item_id"]).length() <= MAX_ITEM_ID_LENGTH
+		and typeof(value.get("quantity")) == TYPE_INT
+		and int(value["quantity"]) >= 0
+		and int(value["quantity"]) <= 1
+		and typeof(value.get("duration")) in [TYPE_FLOAT, TYPE_INT]
+		and is_finite(float(value["duration"]))
+		and float(value["duration"]) >= 0.0
+		and typeof(value.get("message")) == TYPE_STRING
+		and str(value["message"]).length() <= MAX_MESSAGE_LENGTH
+	)
+
+
+static func validate_equipped_state(data: Variant) -> bool:
+	if typeof(data) != TYPE_DICTIONARY:
+		return false
+	var value: Dictionary = data
+	return (
+		_valid_id(value.get("session_id"))
+		and typeof(value.get("owner_peer_id")) == TYPE_INT
+		and int(value["owner_peer_id"]) > 0
+		and typeof(value.get("item_id")) in [TYPE_STRING, TYPE_STRING_NAME]
+		and str(value["item_id"]).length() <= MAX_ITEM_ID_LENGTH
+		and typeof(value.get("category")) == TYPE_INT
+		and int(value["category"]) >= -1
+		and int(value["category"]) < ItemData.Category.size()
+		and typeof(value.get("revision")) == TYPE_INT
+		and int(value["revision"]) >= 0
+		and typeof(value.get("owns_item")) == TYPE_BOOL
+	)
+
+
+static func _valid_id(value: Variant) -> bool:
+	return (
+		typeof(value) == TYPE_STRING
+		and not str(value).is_empty()
+		and str(value).length() <= MAX_ID_LENGTH
+	)

network/network_item_protocol.gd.uid

@@ -0,0 +1 @@
+uid://bphibe6g81mhm

network/network_item_use_service.gd

@@ -0,0 +1,388 @@
+class_name NetworkItemUseService
+extends Node
+
+const MAX_LEDGER_ENTRIES: int = 64
+
+signal local_item_use_pending(request_id: String)
+signal local_item_use_finished(accepted: bool, message: String)
+signal equipped_state_changed(peer_id: int, item_id: StringName, category: int)
+
+var _session: NetworkSession
+var _spawn_service: PlayerSpawnService
+var _catalog: ItemCatalog
+var _local_bag: PlayerBag
+var _local_effects: PlayerItemEffects
+var _save_manager: PlayerSaveManager
+var _requests: Dictionary[int, Dictionary] = {}
+var _pending_by_peer: Dictionary[int, String] = {}
+var _result_owners: Dictionary[String, int] = {}
+var _received_results: Dictionary[String, bool] = {}
+var _pending_local: Dictionary = {}
+var _equipped_states: Dictionary[int, Dictionary] = {}
+var _local_equipped_revision: int = 0
+
+
+func setup(
+	session: NetworkSession,
+	spawn_service: PlayerSpawnService,
+	catalog: ItemCatalog,
+	local_bag: PlayerBag,
+	local_effects: PlayerItemEffects,
+	save_manager: PlayerSaveManager,
+) -> void:
+	_session = session
+	_spawn_service = spawn_service
+	_catalog = catalog
+	_local_bag = local_bag
+	_local_effects = local_effects
+	_save_manager = save_manager
+	_session.peer_removed.connect(_on_peer_removed)
+	_session.state_changed.connect(_on_session_state_changed)
+	_session.peer_authenticated.connect(_on_peer_authenticated)
+
+
+func request_use(item_id: StringName) -> String:
+	if not _pending_local.is_empty():
+		local_item_use_finished.emit(false, "An item use is already pending.")
+		return ""
+	if (
+		_session == null
+		or not _session.is_gameplay_session_active()
+		or (
+			not _session.is_host()
+			and not _session.supports_server_capability(
+				NetworkItemProtocol.ITEM_USE_CAPABILITY
+			)
+		)
+	):
+		local_item_use_finished.emit(false, "Item use is unavailable.")
+		return ""
+	var request_id := _new_id("item")
+	var data := {
+		"request_id": request_id,
+		"session_id": _session.get_session_id(),
+		"item_id": str(item_id),
+		"quantity": _local_bag.get_quantity(item_id),
+	}
+	_pending_local = data.duplicate(true)
+	local_item_use_pending.emit(request_id)
+	if _session.is_host():
+		_handle_use_request(_session.get_local_peer_id(), data)
+	else:
+		submit_item_use.rpc_id(1, data)
+	return request_id
+
+
+@rpc("any_peer", "call_remote", "reliable", NetworkItemProtocol.RELIABLE_CHANNEL)
+func submit_item_use(data: Dictionary) -> void:
+	var sender_id := multiplayer.get_remote_sender_id()
+	if _session.is_host() and _session.is_authenticated_peer(sender_id):
+		_handle_use_request(sender_id, data)
+
+
+func _handle_use_request(peer_id: int, data: Dictionary) -> void:
+	var request_id := str(data.get("request_id", ""))
+	var ledger: Dictionary = _requests.get(peer_id, {})
+	if ledger.has(request_id):
+		_send_result(peer_id, ledger[request_id])
+		return
+	var error := NetworkItemProtocol.validate_use_request(data)
+	if (
+		error.is_empty()
+		and str(data["session_id"]) != _session.get_session_id()
+	):
+		error = "Item use could not be completed."
+	if error.is_empty() and _pending_by_peer.has(peer_id):
+		error = "An item use is already pending."
+	var item_id := StringName(str(data.get("item_id", "")))
+	var item: ItemData = _catalog.get_item_by_id(item_id)
+	var avatar: Player = _spawn_service.get_avatar(peer_id)
+	var duration: float = 0.0
+	if error.is_empty() and (
+		item == null
+		or item.category != ItemData.Category.CONSUMABLE
+		or not item.usable
+	):
+		error = "That item cannot be used now."
+	if error.is_empty() and int(data["quantity"]) < 1:
+		error = "You do not have that item."
+	if error.is_empty() and (
+		avatar == null
+		or avatar.is_water_recovery_active()
+	):
+		error = "That item cannot be used now."
+	if error.is_empty():
+		duration = avatar.item_effects.get_effect_duration(item_id)
+		if duration <= 0.0:
+			error = "That item cannot be used now."
+	var result := {
+		"result_id": _new_id("item_result"),
+		"request_id": request_id if not request_id.is_empty() else "invalid",
+		"session_id": _session.get_session_id(),
+		"target_peer_id": peer_id,
+		"accepted": error.is_empty(),
+		"item_id": str(item_id) if not item_id.is_empty() else "invalid",
+		"quantity": 1 if error.is_empty() else 0,
+		"duration": duration if error.is_empty() else 0.0,
+		"message": "Used %s." % item.display_name if error.is_empty() else error,
+	}
+	ledger[request_id] = result.duplicate(true)
+	_bound(ledger)
+	_requests[peer_id] = ledger
+	_result_owners[result["result_id"]] = peer_id
+	_pending_by_peer[peer_id] = request_id
+	_send_result(peer_id, result)
+
+
+func _send_result(peer_id: int, result: Dictionary) -> void:
+	if peer_id == _session.get_local_peer_id():
+		_apply_result(result)
+	else:
+		receive_item_use_result.rpc_id(peer_id, result)
+
+
+@rpc("authority", "call_remote", "reliable", NetworkItemProtocol.RELIABLE_CHANNEL)
+func receive_item_use_result(data: Dictionary) -> void:
+	_apply_result(data)
+
+
+func _apply_result(data: Dictionary) -> void:
+	if (
+		not NetworkItemProtocol.validate_use_result(data)
+		or str(data["session_id"]) != _session.get_session_id()
+		or int(data["target_peer_id"]) != _session.get_local_peer_id()
+	):
+		return
+	var result_id: String = data["result_id"]
+	if _received_results.has(result_id):
+		_acknowledge(data, true, "")
+		return
+	if (
+		_pending_local.is_empty()
+		or str(data["request_id"]) != str(_pending_local.get("request_id", ""))
+	):
+		return
+	if not bool(data["accepted"]):
+		_received_results[result_id] = true
+		_pending_local.clear()
+		local_item_use_finished.emit(false, str(data["message"]))
+		_acknowledge(data, false, str(data["message"]))
+		return
+	var item_id := StringName(str(data["item_id"]))
+	var bag_snapshot := _local_bag.get_all_items()
+	if (
+		_local_bag.get_quantity(item_id) < 1
+		or not _local_bag.remove_item(item_id, 1)
+		or not _save_manager.save_if_dirty()
+	):
+		_local_bag.replace_all_items(bag_snapshot)
+		_save_manager.save_if_dirty()
+		_pending_local.clear()
+		local_item_use_finished.emit(false, "Item use could not be completed.")
+		_acknowledge(data, false, "Item use could not be completed.")
+		return
+	_received_results[result_id] = true
+	_bound(_received_results)
+	_pending_local.clear()
+	local_item_use_finished.emit(true, str(data["message"]))
+	_acknowledge(data, true, "")
+
+
+func _acknowledge(data: Dictionary, applied: bool, message: String) -> void:
+	if _session.is_host():
+		_handle_ack(
+			_session.get_local_peer_id(),
+			str(data["result_id"]),
+			str(data["request_id"]),
+			applied,
+			message
+		)
+	else:
+		acknowledge_item_use.rpc_id(
+			1, str(data["result_id"]), str(data["request_id"]), applied,
+			message.left(NetworkItemProtocol.MAX_MESSAGE_LENGTH)
+		)
+
+
+@rpc("any_peer", "call_remote", "reliable", NetworkItemProtocol.RELIABLE_CHANNEL)
+func acknowledge_item_use(
+	result_id: String,
+	request_id: String,
+	applied: bool,
+	message: String,
+) -> void:
+	var sender_id := multiplayer.get_remote_sender_id()
+	if _session.is_host() and _session.is_authenticated_peer(sender_id):
+		_handle_ack(sender_id, result_id, request_id, applied, message)
+
+
+func _handle_ack(
+	peer_id: int,
+	result_id: String,
+	request_id: String,
+	applied: bool,
+	_message: String,
+) -> void:
+	if (
+		_result_owners.get(result_id, 0) != peer_id
+		or _pending_by_peer.get(peer_id, "") != request_id
+	):
+		return
+	_pending_by_peer.erase(peer_id)
+	if not applied:
+		return
+	var result: Dictionary = _requests.get(peer_id, {}).get(request_id, {})
+	var avatar := _spawn_service.get_avatar(peer_id)
+	if avatar == null or result.is_empty():
+		return
+	var item_id := StringName(str(result["item_id"]))
+	avatar.item_effects.activate_authoritative(item_id, float(result["duration"]))
+	_broadcast_effect(peer_id, item_id, float(result["duration"]))
+
+
+func get_effects_for_peer(peer_id: int) -> PlayerItemEffects:
+	var avatar := _spawn_service.get_avatar(peer_id)
+	return avatar.item_effects if avatar != null else null
+
+
+func submit_local_equipped(item_id: StringName, owns_item: bool) -> void:
+	if _session == null or not _session.is_gameplay_session_active():
+		return
+	_local_equipped_revision += 1
+	var item: ItemData = _catalog.get_item_by_id(item_id)
+	var data := {
+		"session_id": _session.get_session_id(),
+		"owner_peer_id": _session.get_local_peer_id(),
+		"item_id": str(item_id) if item != null and owns_item else "",
+		"category": int(item.category) if item != null and owns_item else -1,
+		"revision": _local_equipped_revision,
+		"owns_item": owns_item and item != null,
+	}
+	if _session.is_host():
+		_handle_equipped_state(_session.get_local_peer_id(), data)
+	elif _session.supports_server_capability(
+		NetworkItemProtocol.EQUIPMENT_CAPABILITY
+	):
+		submit_equipped_state.rpc_id(1, data)
+
+
+@rpc("any_peer", "call_remote", "reliable", NetworkItemProtocol.RELIABLE_CHANNEL)
+func submit_equipped_state(data: Dictionary) -> void:
+	var sender_id := multiplayer.get_remote_sender_id()
+	if _session.is_host() and _session.is_authenticated_peer(sender_id):
+		_handle_equipped_state(sender_id, data)
+
+
+func _handle_equipped_state(peer_id: int, data: Dictionary) -> void:
+	if (
+		not NetworkItemProtocol.validate_equipped_state(data)
+		or str(data["session_id"]) != _session.get_session_id()
+		or int(data["owner_peer_id"]) != peer_id
+	):
+		return
+	var previous: Dictionary = _equipped_states.get(peer_id, {})
+	if int(data["revision"]) <= int(previous.get("revision", -1)):
+		return
+	var item_id := StringName(str(data["item_id"]))
+	var item: ItemData = _catalog.get_item_by_id(item_id)
+	if not item_id.is_empty() and (
+		item == null
+		or not bool(data["owns_item"])
+		or int(item.category) != int(data["category"])
+	):
+		return
+	_equipped_states[peer_id] = data.duplicate(true)
+	_apply_equipped(data)
+	receive_equipped_state.rpc(data)
+
+
+@rpc("authority", "call_remote", "reliable", NetworkItemProtocol.RELIABLE_CHANNEL)
+func receive_equipped_state(data: Dictionary) -> void:
+	if NetworkItemProtocol.validate_equipped_state(data):
+		_equipped_states[int(data["owner_peer_id"])] = data.duplicate(true)
+		_apply_equipped(data)
+
+
+func _apply_equipped(data: Dictionary) -> void:
+	var peer_id: int = data["owner_peer_id"]
+	var avatar := _spawn_service.get_avatar(peer_id)
+	if avatar != null:
+		avatar.set_active_item_is_rod(
+			int(data["category"]) == ItemData.Category.ROD
+		)
+	equipped_state_changed.emit(
+		peer_id, StringName(str(data["item_id"])), int(data["category"])
+	)
+
+
+func _broadcast_effect(peer_id: int, item_id: StringName, duration: float) -> void:
+	var data := {
+		"session_id": _session.get_session_id(),
+		"owner_peer_id": peer_id,
+		"item_id": str(item_id),
+		"remaining": duration,
+	}
+	_apply_effect_snapshot(data)
+	receive_effect_snapshot.rpc(data)
+
+
+@rpc("authority", "call_remote", "reliable", NetworkItemProtocol.RELIABLE_CHANNEL)
+func receive_effect_snapshot(data: Dictionary) -> void:
+	_apply_effect_snapshot(data)
+
+
+func _apply_effect_snapshot(data: Dictionary) -> void:
+	if (
+		typeof(data.get("owner_peer_id")) != TYPE_INT
+		or typeof(data.get("item_id")) != TYPE_STRING
+		or typeof(data.get("remaining")) not in [TYPE_FLOAT, TYPE_INT]
+		or str(data.get("session_id", "")) != _session.get_session_id()
+	):
+		return
+	var avatar := _spawn_service.get_avatar(int(data["owner_peer_id"]))
+	if avatar != null:
+		avatar.item_effects.activate_authoritative(
+			StringName(data["item_id"]), float(data["remaining"])
+		)
+
+
+func _on_peer_authenticated(peer_id: int, _display_name: String) -> void:
+	if not _session.is_host():
+		return
+	for state: Dictionary in _equipped_states.values():
+		receive_equipped_state.rpc_id(peer_id, state)
+
+
+func _on_peer_removed(peer_id: int) -> void:
+	_requests.erase(peer_id)
+	_pending_by_peer.erase(peer_id)
+	_equipped_states.erase(peer_id)
+
+
+func _on_session_state_changed(state: NetworkSession.State) -> void:
+	if state in [
+		NetworkSession.State.INACTIVE,
+		NetworkSession.State.DISCONNECTING,
+		NetworkSession.State.CONNECTION_FAILED,
+		NetworkSession.State.SERVER_LOST,
+	]:
+		var pending := not _pending_local.is_empty()
+		_requests.clear()
+		_pending_by_peer.clear()
+		_result_owners.clear()
+		_received_results.clear()
+		_equipped_states.clear()
+		_pending_local.clear()
+		if pending:
+			local_item_use_finished.emit(false, "Connection lost.")
+
+
+func _bound(values: Dictionary) -> void:
+	while values.size() > MAX_LEDGER_ENTRIES:
+		values.erase(values.keys().front())
+
+
+func _new_id(prefix: String) -> String:
+	return "%s:%s" % [
+		prefix, Crypto.new().generate_random_bytes(16).hex_encode(),
+	]

network/network_item_use_service.gd.uid

@@ -0,0 +1 @@
+uid://dx21uu253nmpw

network/network_profile_preferences.gd

@@ -34,6 +34,29 @@ func load_or_create() -> bool:
 	return _save_atomic()
 
 
+func set_display_name(value: String) -> bool:
+	var clean_name: String = value.strip_edges()
+	if not is_valid_display_name(clean_name):
+		return false
+	var previous: String = display_name
+	display_name = clean_name
+	if _save_atomic():
+		return true
+	display_name = previous
+	return false
+
+
+static func is_valid_display_name(value: String) -> bool:
+	var clean_name := value.strip_edges()
+	if clean_name.is_empty() or clean_name.length() > 24:
+		return false
+	for index: int in clean_name.length():
+		var codepoint: int = clean_name.unicode_at(index)
+		if codepoint < 32 or codepoint == 127:
+			return false
+	return true
+
+
 func _load_existing() -> bool:
 	var file := FileAccess.open(PROFILE_PATH, FileAccess.READ)
 	if file == null:

network/network_protocol.gd

@@ -3,14 +3,18 @@ extends RefCounted
 
 const PROTOCOL_VERSION: int = 2
 const GAME_BUILD: String = "prealpha"
-const MAX_DISPLAY_NAME_LENGTH: int = 48
+const MAX_DISPLAY_NAME_LENGTH: int = 24
 const MAX_PROFILE_ID_LENGTH: int = 96
 const MAX_NONCE_LENGTH: int = 96
 # ENet channels: 0 reliable lifecycle, 1 movement input, 2 movement
 # snapshots, 3 fishing input, 4 fishing snapshots, 5 reliable sales,
-# 6 reliable shop transactions.
+# 6 reliable shop transactions, 7 reliable item/equipment lifecycle,
+# 8 reliable ordered session chat.
 const SALE_RELIABLE_CHANNEL: int = 5
 const SHOP_RELIABLE_CHANNEL: int = 6
+const ITEM_RELIABLE_CHANNEL: int = 7
+const CHAT_RELIABLE_CHANNEL: int = 8
+const ENET_CHANNEL_COUNT: int = 9
 
 enum RejectionCode {
 	NONE,
@@ -109,6 +113,9 @@ static func make_server_hello(
 			"fishing_v1",
 			"sale_v1",
 			"shop_v1",
+			"item_use_v1",
+			"equipment_v1",
+			"chat_v1",
 		]),
 	}
 

network/network_session.gd

@@ -16,6 +16,7 @@ signal peer_authenticated(peer_id: int, display_name: String)
 signal peer_removed(peer_id: int)
 signal host_openness_changed(is_open: bool)
 signal peer_count_changed(player_count: int, max_players: int)
+signal peer_display_name_changed(peer_id: int, display_name: String)
 signal join_authenticated
 signal server_lost
 signal remote_recovery_requested(peer_id: int, entry_position: Vector3)
@@ -275,10 +276,50 @@ func supports_server_capability(capability: StringName) -> bool:
 	if is_host():
 		return str(capability) in PackedStringArray([
 			"movement_v1", "fishing_v1", "sale_v1", "shop_v1",
+			"item_use_v1", "equipment_v1", "chat_v1",
 		])
 	return str(capability) in _server_capabilities
 
 
+func get_peer_record(peer_id: int) -> PeerRegistry.PeerRecord:
+	return _registry.get_peer(peer_id)
+
+
+func update_local_display_name(value: String) -> bool:
+	if _profile == null or not _profile.set_display_name(value):
+		return false
+	var peer_id := get_local_peer_id()
+	if is_host():
+		_apply_display_name(peer_id, _profile.display_name)
+		receive_display_name.rpc(peer_id, _profile.display_name)
+	elif is_joined_client() and supports_server_capability(&"chat_v1"):
+		submit_display_name.rpc_id(1, _profile.display_name)
+	return true
+
+
+@rpc("any_peer", "call_remote", "reliable", 0)
+func submit_display_name(value: String) -> void:
+	var sender_id := multiplayer.get_remote_sender_id()
+	if (
+		is_host()
+		and is_authenticated_peer(sender_id)
+		and NetworkProfilePreferences.is_valid_display_name(value)
+	):
+		_apply_display_name(sender_id, value.strip_edges())
+		receive_display_name.rpc(sender_id, value.strip_edges())
+
+
+@rpc("authority", "call_remote", "reliable", 0)
+func receive_display_name(peer_id: int, value: String) -> void:
+	if NetworkProfilePreferences.is_valid_display_name(value):
+		_apply_display_name(peer_id, value.strip_edges())
+
+
+func _apply_display_name(peer_id: int, value: String) -> void:
+	if _registry.update_display_name(peer_id, value):
+		peer_display_name_changed.emit(peer_id, value)
+
+
 func get_local_peer_id() -> int:
 	return multiplayer.get_unique_id() if is_gameplay_session_active() else 0
 
@@ -414,6 +455,12 @@ func submit_client_hello(data: Dictionary) -> void:
 		)
 		return
 	var display_name: String = data["display_name"]
+	if not NetworkProfilePreferences.is_valid_display_name(display_name):
+		_reject_peer(
+			sender_id,
+			NetworkProtocol.RejectionCode.MALFORMED_HANDSHAKE
+		)
+		return
 	if not _registry.add_peer(
 		sender_id,
 		profile_id,

network/peer_registry.gd

@@ -42,6 +42,14 @@ func get_peer(peer_id: int) -> PeerRecord:
 	return _records.get(peer_id)
 
 
+func update_display_name(peer_id: int, display_name: String) -> bool:
+	var record: PeerRecord = _records.get(peer_id)
+	if record == null or display_name.is_empty():
+		return false
+	record.display_name = display_name
+	return true
+
+
 func has_peer(peer_id: int) -> bool:
 	return _records.has(peer_id)