Add host-validated multiplayer shop purchases

fishing/fishing_spot.gd

@@ -226,7 +226,7 @@ func can_open_system_menu() -> bool:
 
 func can_open_fishing_shop() -> bool:
 	return (
-		_can_use_shared_gameplay()
+		_can_use_shop_gameplay()
 		and
 		_gameplay_input_enabled
 		and not _external_input_blocked
@@ -237,7 +237,7 @@ func can_open_fishing_shop() -> bool:
 
 func is_ready_for_shop_transaction() -> bool:
 	return (
-		_can_use_shared_gameplay()
+		_can_use_shop_gameplay()
 		and
 		_gameplay_input_enabled
 		and not _external_input_blocked
@@ -502,6 +502,17 @@ func _can_use_shared_gameplay() -> bool:
 	return _network_session == null or _network_session.can_use_host_gameplay()
 
 
+func _can_use_shop_gameplay() -> bool:
+	return (
+		_network_session == null
+		or _network_session.is_host()
+		or (
+			_network_session.is_joined_client()
+			and _network_session.supports_server_capability(&"shop_v1")
+		)
+	)
+
+
 func has_active_fishing_rod() -> bool:
 	if (
 		_local_bag == null

main/main.gd

@@ -49,6 +49,9 @@ const NetworkFishingServiceType = preload(
 const NetworkSaleServiceType = preload(
 	"res://network/network_sale_service.gd"
 )
+const NetworkShopServiceType = preload(
+	"res://network/network_shop_service.gd"
+)
 
 const TITLE_MUSIC_SILENCE_DB: float = -80.0
 
@@ -87,6 +90,7 @@ const TITLE_MUSIC_SILENCE_DB: float = -80.0
 	%NetworkFishingService
 )
 @onready var _network_sale: NetworkSaleServiceType = %NetworkSaleService
+@onready var _network_shop: NetworkShopServiceType = %NetworkShopService
 @onready var _players_root: Node3D = $Players
 
 var _gameplay_started: bool = false
@@ -169,6 +173,18 @@ func _ready() -> void:
 		pelican_buyer_profile,
 		_test_world.get_pelican_convenience_landmark()
 	)
+	_network_shop.setup(
+		_network_session,
+		_player_spawn_service,
+		_network_fishing,
+		_shop_interaction,
+		_player.wallet,
+		_player.bag,
+		item_catalog,
+		_player.fishing_upgrades,
+		_player.cooler_capacity,
+		_save_manager
+	)
 	_fishing_spot.setup(
 		_player,
 		_player.inventory,
@@ -200,7 +216,8 @@ func _ready() -> void:
 		_player.item_effects,
 		_player.cooler_capacity,
 		_network_session,
-		_network_sale
+		_network_sale,
+		_network_shop
 	)
 	_water_recovery.setup(
 		_player,

main/main.tscn

@@ -1,4 +1,4 @@
-[gd_scene load_steps=23 format=3]
+[gd_scene load_steps=24 format=3]
 
 [ext_resource type="PackedScene" path="res://world/test_world.tscn" id="1_world"]
 [ext_resource type="PackedScene" path="res://player/player.tscn" id="2_player"]
@@ -22,6 +22,7 @@
 [ext_resource type="Script" path="res://network/player_spawn_service.gd" id="20_spawn_service"]
 [ext_resource type="Script" path="res://network/network_fishing_service.gd" id="21_network_fishing"]
 [ext_resource type="Script" path="res://network/network_sale_service.gd" id="22_network_sale"]
+[ext_resource type="Script" path="res://network/network_shop_service.gd" id="23_network_shop"]
 
 [node name="Main" type="Node3D"]
 script = ExtResource("3_main")
@@ -54,6 +55,10 @@ script = ExtResource("21_network_fishing")
 unique_name_in_owner = true
 script = ExtResource("22_network_sale")
 
+[node name="NetworkShopService" type="Node" parent="."]
+unique_name_in_owner = true
+script = ExtResource("23_network_shop")
+
 [node name="TestWorld" parent="." instance=ExtResource("1_world")]
 
 [node name="Players" type="Node3D" parent="."]

network/network_protocol.gd

@@ -7,8 +7,10 @@ const MAX_DISPLAY_NAME_LENGTH: int = 48
 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.
+# snapshots, 3 fishing input, 4 fishing snapshots, 5 reliable sales,
+# 6 reliable shop transactions.
 const SALE_RELIABLE_CHANNEL: int = 5
+const SHOP_RELIABLE_CHANNEL: int = 6
 
 enum RejectionCode {
 	NONE,
@@ -106,6 +108,7 @@ static func make_server_hello(
 			"movement_v1",
 			"fishing_v1",
 			"sale_v1",
+			"shop_v1",
 		]),
 	}
 

network/network_session.gd

@@ -274,7 +274,7 @@ func get_operation_generation() -> int:
 func supports_server_capability(capability: StringName) -> bool:
 	if is_host():
 		return str(capability) in PackedStringArray([
-			"movement_v1", "fishing_v1", "sale_v1",
+			"movement_v1", "fishing_v1", "sale_v1", "shop_v1",
 		])
 	return str(capability) in _server_capabilities
 

network/network_shop_protocol.gd

@@ -0,0 +1,115 @@
+class_name NetworkShopProtocol
+extends RefCounted
+
+const CAPABILITY: StringName = &"shop_v1"
+const RELIABLE_CHANNEL: int = NetworkProtocol.SHOP_RELIABLE_CHANNEL
+const MAX_ID_LENGTH: int = 96
+const MAX_MESSAGE_LENGTH: int = 160
+const MAX_QUANTITY: int = 99
+
+enum ProductCategory {
+	SUPPLY,
+	ROD,
+	REEL_SPEED_UPGRADE,
+	BARRIER_POWER_UPGRADE,
+	COOLER_CAPACITY_UPGRADE,
+}
+
+
+static func validate_request(data: Variant) -> String:
+	if typeof(data) != TYPE_DICTIONARY:
+		return "Purchase could not be completed."
+	var payload: Dictionary = data
+	for key: String in [
+		"request_id",
+		"session_id",
+		"shop_id",
+		"product_id",
+		"category",
+		"quantity",
+		"wallet_balance",
+		"current_state",
+	]:
+		if not payload.has(key):
+			return "Purchase could not be completed."
+	if (
+		typeof(payload["request_id"]) != TYPE_STRING
+		or typeof(payload["session_id"]) != TYPE_STRING
+		or typeof(payload["shop_id"]) not in [TYPE_STRING, TYPE_STRING_NAME]
+		or typeof(payload["product_id"]) not in [
+			TYPE_STRING, TYPE_STRING_NAME
+		]
+		or typeof(payload["category"]) != TYPE_INT
+		or typeof(payload["quantity"]) != TYPE_INT
+		or typeof(payload["wallet_balance"]) != TYPE_INT
+		or typeof(payload["current_state"]) != TYPE_INT
+	):
+		return "Purchase could not be completed."
+	var request_id: String = payload["request_id"]
+	var session_id: String = payload["session_id"]
+	var shop_id: String = str(payload["shop_id"])
+	var product_id: String = str(payload["product_id"])
+	var category: int = payload["category"]
+	var quantity: int = payload["quantity"]
+	var wallet_balance: int = payload["wallet_balance"]
+	var current_state: int = payload["current_state"]
+	if (
+		request_id.is_empty()
+		or request_id.length() > MAX_ID_LENGTH
+		or session_id.is_empty()
+		or session_id.length() > MAX_ID_LENGTH
+		or shop_id.is_empty()
+		or shop_id.length() > MAX_ID_LENGTH
+		or product_id.is_empty()
+		or product_id.length() > MAX_ID_LENGTH
+		or category < 0
+		or category >= ProductCategory.size()
+		or quantity < 1
+		or quantity > MAX_QUANTITY
+		or wallet_balance < 0
+		or current_state < 0
+	):
+		return "Purchase could not be completed."
+	return ""
+
+
+static func validate_result(data: Variant) -> bool:
+	if typeof(data) != TYPE_DICTIONARY:
+		return false
+	var payload: Dictionary = data
+	return (
+		_valid_id(payload.get("result_id"))
+		and _valid_id(payload.get("request_id"))
+		and typeof(payload.get("session_id")) == TYPE_STRING
+		and typeof(payload.get("target_peer_id")) == TYPE_INT
+		and typeof(payload.get("accepted")) == TYPE_BOOL
+		and typeof(payload.get("product_id")) in [
+			TYPE_STRING, TYPE_STRING_NAME
+		]
+		and not str(payload["product_id"]).is_empty()
+		and str(payload["product_id"]).length() <= MAX_ID_LENGTH
+		and typeof(payload.get("category")) == TYPE_INT
+		and int(payload["category"]) >= 0
+		and int(payload["category"]) < ProductCategory.size()
+		and typeof(payload.get("quantity")) == TYPE_INT
+		and int(payload["quantity"]) >= 0
+		and int(payload["quantity"]) <= MAX_QUANTITY
+		and typeof(payload.get("total_cost")) == TYPE_INT
+		and int(payload["total_cost"]) >= 0
+		and typeof(payload.get("expected_wallet")) == TYPE_INT
+		and int(payload["expected_wallet"]) >= 0
+		and typeof(payload.get("expected_state")) == TYPE_INT
+		and int(payload["expected_state"]) >= 0
+		and typeof(payload.get("resulting_state")) == TYPE_INT
+		and int(payload["resulting_state"]) >= 0
+		and typeof(payload.get("message")) == TYPE_STRING
+		and str(payload["message"]).length() <= MAX_MESSAGE_LENGTH
+	)
+
+
+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_shop_protocol.gd.uid

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

network/network_shop_service.gd

@@ -0,0 +1,677 @@
+class_name NetworkShopService
+extends Node
+
+const FishingShopStockType = preload(
+	"res://economy/fishing_shop_stock.gd"
+)
+const ItemDataType = preload("res://items/item_data.gd")
+const OwnedItemType = preload("res://items/owned_item.gd")
+
+const SHOP_ID: StringName = &"main_fishing_shop"
+const REEL_PRODUCT_ID: StringName = &"reel_speed_upgrade"
+const BARRIER_PRODUCT_ID: StringName = &"barrier_power_upgrade"
+const COOLER_PRODUCT_ID: StringName = &"cooler_capacity_upgrade"
+const MAX_LEDGER_ENTRIES_PER_PEER: int = 64
+
+signal local_purchase_pending(request_id: String)
+signal local_purchase_finished(
+	request_id: String,
+	accepted: bool,
+	message: String,
+	product_id: StringName,
+	category: int,
+	quantity: int,
+	total_cost: int,
+)
+
+var _session: NetworkSession
+var _spawn_service: PlayerSpawnService
+var _network_fishing: NetworkFishingService
+var _interaction: FishingShopInteraction
+var _wallet: PlayerWallet
+var _bag: PlayerBag
+var _item_catalog: ItemCatalog
+var _upgrades: PlayerFishingUpgrades
+var _cooler_capacity: PlayerCoolerCapacity
+var _save_manager: PlayerSaveManager
+var _request_ledgers: Dictionary[int, Dictionary] = {}
+var _pending_by_peer: Dictionary[int, String] = {}
+var _result_owners: Dictionary[String, int] = {}
+var _acknowledged_results: Dictionary[String, bool] = {}
+var _received_results: Dictionary[String, bool] = {}
+var _applied_results: Dictionary[String, bool] = {}
+var _pending_local_request: Dictionary = {}
+
+
+func setup(
+	session: NetworkSession,
+	spawn_service: PlayerSpawnService,
+	network_fishing: NetworkFishingService,
+	interaction: FishingShopInteraction,
+	wallet: PlayerWallet,
+	bag: PlayerBag,
+	item_catalog: ItemCatalog,
+	upgrades: PlayerFishingUpgrades,
+	cooler_capacity: PlayerCoolerCapacity,
+	save_manager: PlayerSaveManager,
+) -> void:
+	_session = session
+	_spawn_service = spawn_service
+	_network_fishing = network_fishing
+	_interaction = interaction
+	_wallet = wallet
+	_bag = bag
+	_item_catalog = item_catalog
+	_upgrades = upgrades
+	_cooler_capacity = cooler_capacity
+	_save_manager = save_manager
+	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):
+		_session.state_changed.connect(_on_session_state_changed)
+
+
+func can_request_purchase() -> bool:
+	return (
+		_session != null
+		and _session.is_gameplay_session_active()
+		and (
+			_session.is_host()
+			or _session.supports_server_capability(
+				NetworkShopProtocol.CAPABILITY
+			)
+		)
+	)
+
+
+func is_local_purchase_pending() -> bool:
+	return not _pending_local_request.is_empty()
+
+
+func request_supply(item_id: StringName) -> String:
+	return _request_purchase(
+		item_id,
+		NetworkShopProtocol.ProductCategory.SUPPLY,
+		1,
+		_bag.get_quantity(item_id) if _bag != null else 0
+	)
+
+
+func request_reel_speed_upgrade() -> String:
+	return _request_purchase(
+		REEL_PRODUCT_ID,
+		NetworkShopProtocol.ProductCategory.REEL_SPEED_UPGRADE,
+		1,
+		_upgrades.get_reel_speed_level() if _upgrades != null else 0
+	)
+
+
+func request_barrier_power_upgrade() -> String:
+	return _request_purchase(
+		BARRIER_PRODUCT_ID,
+		NetworkShopProtocol.ProductCategory.BARRIER_POWER_UPGRADE,
+		1,
+		_upgrades.get_barrier_power_level() if _upgrades != null else 0
+	)
+
+
+func request_cooler_capacity_upgrade() -> String:
+	return _request_purchase(
+		COOLER_PRODUCT_ID,
+		NetworkShopProtocol.ProductCategory.COOLER_CAPACITY_UPGRADE,
+		1,
+		_cooler_capacity.get_level() if _cooler_capacity != null else 0
+	)
+
+
+func _request_purchase(
+	product_id: StringName,
+	category: int,
+	quantity: int,
+	current_state: int,
+) -> String:
+	if is_local_purchase_pending():
+		local_purchase_finished.emit(
+			"", false, "Purchasing…", product_id, category, 0, 0
+		)
+		return ""
+	if not can_request_purchase():
+		local_purchase_finished.emit(
+			"",
+			false,
+			"Purchases are not supported by this server.",
+			product_id,
+			category,
+			0,
+			0
+		)
+		return ""
+	var request_id: String = _new_id("shop")
+	var request: Dictionary = {
+		"request_id": request_id,
+		"session_id": _session.get_session_id(),
+		"shop_id": str(SHOP_ID),
+		"product_id": str(product_id),
+		"category": category,
+		"quantity": quantity,
+		"wallet_balance": _wallet.get_balance() if _wallet != null else 0,
+		"current_state": current_state,
+	}
+	_pending_local_request = request.duplicate(true)
+	local_purchase_pending.emit(request_id)
+	if _session.is_host():
+		_handle_purchase_request(_session.get_local_peer_id(), request)
+	else:
+		submit_purchase_request.rpc_id(1, request)
+	return request_id
+
+
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkShopProtocol.RELIABLE_CHANNEL
+)
+func submit_purchase_request(data: Dictionary) -> void:
+	var sender_id: int = multiplayer.get_remote_sender_id()
+	if (
+		_session == null
+		or not _session.is_host()
+		or not _session.is_authenticated_peer(sender_id)
+	):
+		return
+	_handle_purchase_request(sender_id, data)
+
+
+func _handle_purchase_request(peer_id: int, data: Dictionary) -> void:
+	var validation_error: String = NetworkShopProtocol.validate_request(data)
+	var request_id: String = str(data.get("request_id", ""))
+	var ledger: Dictionary = _request_ledgers.get(peer_id, {})
+	if (
+		not request_id.is_empty()
+		and request_id.length() <= NetworkShopProtocol.MAX_ID_LENGTH
+		and ledger.has(request_id)
+	):
+		_send_result(peer_id, ledger[request_id])
+		return
+	if not validation_error.is_empty():
+		_record_and_send(
+			peer_id,
+			_rejected_result(data, validation_error)
+		)
+		return
+	if str(data["session_id"]) != _session.get_session_id():
+		_record_and_send(
+			peer_id,
+			_rejected_result(data, "Purchase could not be completed.")
+		)
+		return
+	if (
+		_pending_by_peer.has(peer_id)
+		and _pending_by_peer[peer_id] != request_id
+	):
+		_record_and_send(
+			peer_id,
+			_rejected_result(data, "A purchase is already pending.")
+		)
+		return
+	if not _is_shop_available_for_peer(peer_id):
+		var avatar: Player = _spawn_service.get_avatar(peer_id)
+		var message: String = (
+			"The shop is unavailable."
+			if avatar == null or _interaction == null
+			else "Move closer to the shop."
+		)
+		_record_and_send(peer_id, _rejected_result(data, message))
+		return
+	var result: Dictionary = _build_authoritative_result(peer_id, data)
+	_pending_by_peer[peer_id] = request_id
+	_record_and_send(peer_id, result)
+
+
+func _build_authoritative_result(
+	peer_id: int,
+	request: Dictionary,
+) -> Dictionary:
+	var product_id := StringName(str(request["product_id"]))
+	var category: int = request["category"]
+	var quantity: int = request["quantity"]
+	var wallet_balance: int = request["wallet_balance"]
+	var current_state: int = request["current_state"]
+	var cost: int = -1
+	var resulting_state: int = current_state
+	var rejection: String = ""
+	if StringName(str(request["shop_id"])) != SHOP_ID:
+		rejection = "The shop is unavailable."
+	elif quantity != 1:
+		rejection = "Purchase could not be completed."
+	else:
+		match category:
+			NetworkShopProtocol.ProductCategory.SUPPLY:
+				var item: ItemDataType = _item_catalog.get_item_by_id(
+					product_id
+				) if _item_catalog != null else null
+				cost = FishingShopStockType.get_price(product_id)
+				if (
+					item == null
+					or not item.is_valid()
+					or product_id not in (
+						FishingShopStockType.get_stock_item_ids()
+					)
+					or item.category != ItemDataType.Category.CONSUMABLE
+					or not item.stackable
+					or not item.usable
+					or current_state >= item.max_stack
+				):
+					rejection = (
+						"Your Bag is full."
+						if item != null and current_state >= item.max_stack
+						else "Purchase could not be completed."
+					)
+				else:
+					resulting_state = current_state + 1
+			NetworkShopProtocol.ProductCategory.ROD:
+				rejection = "This item is not sold here."
+			NetworkShopProtocol.ProductCategory.REEL_SPEED_UPGRADE:
+				if (
+					product_id != REEL_PRODUCT_ID
+					or current_state
+					>= PlayerFishingUpgrades.MAX_REEL_SPEED_LEVEL
+				):
+					rejection = "Upgrade is already at maximum."
+				else:
+					cost = (
+						PlayerFishingUpgrades.REEL_SPEED_COSTS[
+							current_state
+						]
+					)
+					resulting_state = current_state + 1
+			NetworkShopProtocol.ProductCategory.BARRIER_POWER_UPGRADE:
+				if (
+					product_id != BARRIER_PRODUCT_ID
+					or current_state
+					>= PlayerFishingUpgrades.MAX_BARRIER_POWER_LEVEL
+				):
+					rejection = "Upgrade is already at maximum."
+				else:
+					cost = (
+						PlayerFishingUpgrades.BARRIER_POWER_COSTS[
+							current_state
+						]
+					)
+					resulting_state = current_state + 1
+			NetworkShopProtocol.ProductCategory.COOLER_CAPACITY_UPGRADE:
+				if (
+					product_id != COOLER_PRODUCT_ID
+					or current_state >= PlayerCoolerCapacity.MAX_LEVEL
+				):
+					rejection = "Upgrade is already at maximum."
+				else:
+					cost = (
+						PlayerCoolerCapacity.EXPANSION_COSTS[
+							current_state
+						]
+					)
+					resulting_state = current_state + 1
+	if rejection.is_empty() and (cost < 0 or wallet_balance < cost):
+		rejection = "Not enough fish coin."
+	if not rejection.is_empty():
+		return _rejected_result(request, rejection)
+	return {
+		"result_id": _new_id("shop_result"),
+		"request_id": str(request["request_id"]),
+		"session_id": _session.get_session_id(),
+		"target_peer_id": peer_id,
+		"accepted": true,
+		"product_id": str(product_id),
+		"category": category,
+		"quantity": quantity,
+		"total_cost": cost,
+		"expected_wallet": wallet_balance,
+		"expected_state": current_state,
+		"resulting_state": resulting_state,
+		"message": "Purchase complete.",
+	}
+
+
+func _rejected_result(request: Dictionary, message: String) -> Dictionary:
+	var request_id: String = str(request.get("request_id", ""))
+	if (
+		request_id.is_empty()
+		or request_id.length() > NetworkShopProtocol.MAX_ID_LENGTH
+	):
+		request_id = "invalid"
+	var category: int = clampi(
+		int(request.get("category", 0)),
+		0,
+		NetworkShopProtocol.ProductCategory.size() - 1
+	)
+	return {
+		"result_id": _new_id("shop_result"),
+		"request_id": request_id,
+		"session_id": _session.get_session_id() if _session != null else "",
+		"target_peer_id": 0,
+		"accepted": false,
+		"product_id": str(request.get("product_id", "invalid")).left(
+			NetworkShopProtocol.MAX_ID_LENGTH
+		),
+		"category": category,
+		"quantity": 0,
+		"total_cost": 0,
+		"expected_wallet": maxi(int(request.get("wallet_balance", 0)), 0),
+		"expected_state": maxi(int(request.get("current_state", 0)), 0),
+		"resulting_state": maxi(int(request.get("current_state", 0)), 0),
+		"message": message.left(NetworkShopProtocol.MAX_MESSAGE_LENGTH),
+	}
+
+
+func _record_and_send(peer_id: int, result: Dictionary) -> void:
+	result["target_peer_id"] = peer_id
+	var request_id: String = result["request_id"]
+	var ledger: Dictionary = _request_ledgers.get(peer_id, {})
+	ledger[request_id] = result.duplicate(true)
+	_bound_dictionary(ledger)
+	_request_ledgers[peer_id] = ledger
+	_result_owners[str(result["result_id"])] = peer_id
+	_bound_dictionary(_result_owners)
+	_send_result(peer_id, result)
+
+
+func _send_result(peer_id: int, result: Dictionary) -> void:
+	if peer_id == _session.get_local_peer_id():
+		_apply_purchase_result(result)
+	else:
+		receive_purchase_result.rpc_id(peer_id, result)
+
+
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkShopProtocol.RELIABLE_CHANNEL
+)
+func receive_purchase_result(data: Dictionary) -> void:
+	_apply_purchase_result(data)
+
+
+func _apply_purchase_result(data: Dictionary) -> void:
+	if (
+		not NetworkShopProtocol.validate_result(data)
+		or _session == null
+		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_result(data, _applied_results.has(result_id), "")
+		return
+	if (
+		_pending_local_request.is_empty()
+		or str(data["request_id"])
+		!= str(_pending_local_request.get("request_id", ""))
+	):
+		return
+	if not bool(data["accepted"]):
+		_received_results[result_id] = true
+		_bound_dictionary(_received_results)
+		_finish_local_purchase(data, false, str(data["message"]))
+		_acknowledge_result(data, false, str(data["message"]))
+		return
+	var validation_message: String = _validate_local_result(data)
+	if not validation_message.is_empty():
+		_fail_local_apply(data, validation_message)
+		return
+	var wallet_snapshot: int = _wallet.get_balance()
+	var bag_snapshot: Array[OwnedItemType] = _bag.get_all_items()
+	var reel_snapshot: int = _upgrades.get_reel_speed_level()
+	var barrier_snapshot: int = _upgrades.get_barrier_power_level()
+	var cooler_snapshot: int = _cooler_capacity.get_level()
+	var applied: bool = _apply_local_product(data)
+	if not applied or not _save_manager.save_if_dirty():
+		_bag.replace_all_items(bag_snapshot)
+		_upgrades.restore_levels(reel_snapshot, barrier_snapshot)
+		_cooler_capacity.restore_level(cooler_snapshot)
+		_wallet.restore_balance(wallet_snapshot)
+		_save_manager.save_if_dirty()
+		_fail_local_apply(data, "Purchase could not be completed.")
+		return
+	_applied_results[result_id] = true
+	_bound_dictionary(_applied_results)
+	_received_results[result_id] = true
+	_bound_dictionary(_received_results)
+	_finish_local_purchase(data, true, "Purchase complete.")
+	_acknowledge_result(data, true, "")
+
+
+func _validate_local_result(data: Dictionary) -> String:
+	if (
+		_wallet == null
+		or _bag == null
+		or _upgrades == null
+		or _cooler_capacity == null
+		or _save_manager == null
+		or _wallet.get_balance() != int(data["expected_wallet"])
+		or str(data["product_id"])
+		!= str(_pending_local_request.get("product_id", ""))
+		or int(data["category"])
+		!= int(_pending_local_request.get("category", -1))
+		or int(data["quantity"])
+		!= int(_pending_local_request.get("quantity", -1))
+	):
+		return "Purchase could not be completed."
+	var category: int = data["category"]
+	var product_id := StringName(str(data["product_id"]))
+	var expected_state: int = data["expected_state"]
+	var cost: int = data["total_cost"]
+	if int(data["resulting_state"]) != expected_state + 1:
+		return "Purchase could not be completed."
+	if not _wallet.can_afford(cost):
+		return "Not enough fish coin."
+	match category:
+		NetworkShopProtocol.ProductCategory.SUPPLY:
+			if _bag.get_quantity(product_id) != expected_state:
+				return "Purchase could not be completed."
+			if not _bag.can_add_item(product_id, data["quantity"]):
+				return "Your Bag is full."
+		NetworkShopProtocol.ProductCategory.REEL_SPEED_UPGRADE:
+			if _upgrades.get_reel_speed_level() != expected_state:
+				return "Purchase could not be completed."
+			if _upgrades.get_next_reel_speed_cost() != cost:
+				return "Purchase could not be completed."
+		NetworkShopProtocol.ProductCategory.BARRIER_POWER_UPGRADE:
+			if _upgrades.get_barrier_power_level() != expected_state:
+				return "Purchase could not be completed."
+			if _upgrades.get_next_barrier_power_cost() != cost:
+				return "Purchase could not be completed."
+		NetworkShopProtocol.ProductCategory.COOLER_CAPACITY_UPGRADE:
+			if _cooler_capacity.get_level() != expected_state:
+				return "Purchase could not be completed."
+			if _cooler_capacity.get_next_cost() != cost:
+				return "Purchase could not be completed."
+		_:
+			return "Purchase could not be completed."
+	return ""
+
+
+func _apply_local_product(data: Dictionary) -> bool:
+	var category: int = data["category"]
+	var product_id := StringName(str(data["product_id"]))
+	match category:
+		NetworkShopProtocol.ProductCategory.SUPPLY:
+			return (
+				_wallet.debit(data["total_cost"])
+				and _bag.add_item(product_id, data["quantity"])
+			)
+		NetworkShopProtocol.ProductCategory.REEL_SPEED_UPGRADE:
+			return _upgrades.purchase_reel_speed(_wallet)
+		NetworkShopProtocol.ProductCategory.BARRIER_POWER_UPGRADE:
+			return _upgrades.purchase_barrier_power(_wallet)
+		NetworkShopProtocol.ProductCategory.COOLER_CAPACITY_UPGRADE:
+			return _cooler_capacity.purchase(_wallet)
+	return false
+
+
+func _fail_local_apply(data: Dictionary, message: String) -> void:
+	_finish_local_purchase(data, false, message)
+	_acknowledge_result(data, false, message)
+
+
+func _finish_local_purchase(
+	data: Dictionary,
+	accepted: bool,
+	message: String,
+) -> void:
+	_pending_local_request.clear()
+	local_purchase_finished.emit(
+		str(data["request_id"]),
+		accepted,
+		message,
+		StringName(str(data["product_id"])),
+		int(data["category"]),
+		int(data["quantity"]),
+		int(data["total_cost"])
+	)
+
+
+func _acknowledge_result(
+	data: Dictionary,
+	applied: bool,
+	message: String,
+) -> void:
+	if _session.is_host():
+		_handle_acknowledgement(
+			_session.get_local_peer_id(),
+			str(data["result_id"]),
+			str(data["request_id"]),
+			applied,
+			message
+		)
+	else:
+		acknowledge_purchase_result.rpc_id(
+			1,
+			str(data["result_id"]),
+			str(data["request_id"]),
+			applied,
+			message.left(NetworkShopProtocol.MAX_MESSAGE_LENGTH)
+		)
+
+
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkShopProtocol.RELIABLE_CHANNEL
+)
+func acknowledge_purchase_result(
+	result_id: String,
+	request_id: String,
+	applied: bool,
+	message: String,
+) -> void:
+	var sender_id: int = multiplayer.get_remote_sender_id()
+	if (
+		_session == null
+		or not _session.is_host()
+		or not _session.is_authenticated_peer(sender_id)
+	):
+		return
+	_handle_acknowledgement(
+		sender_id, result_id, request_id, applied, message
+	)
+
+
+func _handle_acknowledgement(
+	peer_id: int,
+	result_id: String,
+	request_id: String,
+	_applied: bool,
+	message: String,
+) -> void:
+	if (
+		result_id.is_empty()
+		or result_id.length() > NetworkShopProtocol.MAX_ID_LENGTH
+		or request_id.is_empty()
+		or request_id.length() > NetworkShopProtocol.MAX_ID_LENGTH
+		or message.length() > NetworkShopProtocol.MAX_MESSAGE_LENGTH
+		or _result_owners.get(result_id, 0) != peer_id
+	):
+		return
+	_acknowledged_results[result_id] = true
+	_bound_dictionary(_acknowledged_results)
+	if _pending_by_peer.get(peer_id, "") == request_id:
+		_pending_by_peer.erase(peer_id)
+
+
+func _is_shop_available_for_peer(peer_id: int) -> bool:
+	if (
+		_session == null
+		or not _session.is_host()
+		or not _session.is_gameplay_session_active()
+		or _spawn_service == null
+		or _interaction == null
+		or not is_instance_valid(_interaction)
+	):
+		return false
+	var avatar: Player = _spawn_service.get_avatar(peer_id)
+	return (
+		avatar != null
+		and not avatar.is_water_recovery_active()
+		and (
+			_network_fishing == null
+			or not _network_fishing.has_peer_attempt(peer_id)
+		)
+		and _interaction.is_avatar_in_range(avatar)
+	)
+
+
+func _on_peer_removed(peer_id: int) -> void:
+	_request_ledgers.erase(peer_id)
+	_pending_by_peer.erase(peer_id)
+	for result_id: String in _result_owners.keys():
+		if _result_owners[result_id] == peer_id:
+			_result_owners.erase(result_id)
+			_acknowledged_results.erase(result_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 had_pending: bool = is_local_purchase_pending()
+		_clear_session_state()
+		if had_pending:
+			local_purchase_finished.emit(
+				"",
+				false,
+				"Connection lost.",
+				StringName(),
+				0,
+				0,
+				0
+			)
+
+
+func _clear_session_state() -> void:
+	_request_ledgers.clear()
+	_pending_by_peer.clear()
+	_result_owners.clear()
+	_acknowledged_results.clear()
+	_received_results.clear()
+	_applied_results.clear()
+	_pending_local_request.clear()
+
+
+func _bound_dictionary(values: Dictionary) -> void:
+	while values.size() > MAX_LEDGER_ENTRIES_PER_PEER:
+		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_shop_service.gd.uid

@@ -0,0 +1 @@
+uid://4ci4dbhyfc11

ui/fishing_shop.gd

@@ -77,6 +77,8 @@ var _interaction: ShopInteractionType
 var _bag: PlayerBagType
 var _item_catalog: ItemCatalogType
 var _cooler_capacity: PlayerCoolerCapacityType
+var _network_session: NetworkSession
+var _network_shop: NetworkShopService
 var _fish_selection := FishBatchSelectionType.new()
 var _confirmation_catch_ids: Array[StringName] = []
 var _confirmation_generation: int = -1
@@ -115,6 +117,8 @@ func setup(
 	bag: PlayerBagType,
 	item_catalog: ItemCatalogType,
 	cooler_capacity: PlayerCoolerCapacityType,
+	network_session: NetworkSession,
+	network_shop: NetworkShopService,
 ) -> void:
 	_player = player
 	_inventory = inventory
@@ -127,6 +131,20 @@ func setup(
 	_bag = bag
 	_item_catalog = item_catalog
 	_cooler_capacity = cooler_capacity
+	_network_session = network_session
+	_network_shop = network_shop
+	if (
+		_network_shop != null
+		and not _network_shop.local_purchase_pending.is_connected(
+			_on_network_purchase_pending
+		)
+	):
+		_network_shop.local_purchase_pending.connect(
+			_on_network_purchase_pending
+		)
+		_network_shop.local_purchase_finished.connect(
+			_on_network_purchase_finished
+		)
 	_fish_selection.clear()
 	if not _inventory.catches_changed.is_connected(_on_inventory_changed):
 		_inventory.catches_changed.connect(_on_inventory_changed)
@@ -158,7 +176,10 @@ func open_shop() -> bool:
 		return false
 	_generation += 1
 	_closing = false
-	_transaction_in_progress = false
+	_transaction_in_progress = (
+		_network_shop != null
+		and _network_shop.is_local_purchase_pending()
+	)
 	_prior_movement_enabled = _player.is_movement_enabled()
 	_prior_camera_enabled = _player.is_camera_input_enabled()
 	_prior_mouse_mode = Input.mouse_mode
@@ -353,6 +374,17 @@ func _update_sale_summary() -> void:
 		_selection_summary.text = "no fish selected"
 		_sell_button.disabled = true
 		return
+	if _network_session != null and _network_session.is_joined_client():
+		_selection_summary.text = (
+			"1 fish selected"
+			if selected_count == 1
+			else "%d fish selected" % selected_count
+		)
+		_selection_summary.text += (
+			"\nSell catches to the nearby pelicans."
+		)
+		_sell_button.disabled = true
+		return
 	var preview: FishSaleResultType = (
 		_sale_service.preview_batch(selected_ids, _buyer)
 		if _sale_service != null
@@ -543,6 +575,9 @@ func _on_fish_clicked(
 
 
 func _open_sale_confirmation() -> void:
+	if _network_session != null and _network_session.is_joined_client():
+		_feedback.text = "Sell catches to the nearby pelicans."
+		return
 	var selected_ids: Array[StringName] = _fish_selection.get_selected_ids()
 	if not _is_transaction_context_valid() or selected_ids.is_empty():
 		_feedback.text = "the fish selection is no longer available."
@@ -577,7 +612,13 @@ func _close_sale_confirmation() -> void:
 
 
 func _on_confirm_sale() -> void:
-	if _transaction_in_progress:
+	if (
+		_transaction_in_progress
+		or (
+			_network_session != null
+			and _network_session.is_joined_client()
+		)
+	):
 		return
 	var transaction_generation: int = _generation
 	var catch_ids: Array[StringName] = _confirmation_catch_ids.duplicate()
@@ -630,28 +671,16 @@ func _purchase_supply(item_id: StringName) -> void:
 	if _transaction_in_progress or not _is_transaction_context_valid():
 		_feedback.text = "unable to complete purchase."
 		return
-	var price: int = FishingShopStockType.get_price(item_id)
 	if not _bag.can_add_item(item_id, 1):
-		_feedback.text = "bag cannot accept this item."
+		_feedback.text = "Your Bag is full."
 		return
-	if not _wallet.can_afford(price):
+	if not _wallet.can_afford(FishingShopStockType.get_price(item_id)):
-		_feedback.text = "not enough money."
+		_feedback.text = "Not enough fish coin."
 		return
-	var transaction_generation: int = _generation
+	if _network_shop == null:
-	_transaction_in_progress = true
+		_feedback.text = "Purchase could not be completed."
-	var purchased: bool = FishingShopStockType.purchase_one(
-		item_id,
-		_wallet,
-		_bag,
-		_item_catalog
-	)
-	_transaction_in_progress = false
-	if transaction_generation != _generation or not visible:
 		return
-	_feedback.text = (
+	_network_shop.request_supply(item_id)
-		"item purchased." if purchased else "unable to complete purchase."
-	)
-	_refresh_all()
 
 
 func _purchase_cooler_capacity() -> void:
@@ -660,23 +689,15 @@ func _purchase_cooler_capacity() -> void:
 		return
 	var cost: int = _cooler_capacity.get_next_cost()
 	if cost < 0:
-		_feedback.text = "maximum level reached."
+		_feedback.text = "Upgrade is already at maximum."
 		return
 	if not _wallet.can_afford(cost):
-		_feedback.text = "not enough money."
+		_feedback.text = "Not enough fish coin."
 		return
-	var transaction_generation: int = _generation
+	if _network_shop == null:
-	_transaction_in_progress = true
+		_feedback.text = "Purchase could not be completed."
-	var purchased: bool = _cooler_capacity.purchase(_wallet)
-	_transaction_in_progress = false
-	if transaction_generation != _generation or not visible:
 		return
-	_feedback.text = (
+	_network_shop.request_cooler_capacity_upgrade()
-		"upgrade purchased."
-		if purchased
-		else "unable to complete purchase."
-	)
-	_refresh_all()
 
 
 func _purchase_upgrade(is_reel_speed: bool) -> void:
@@ -689,25 +710,43 @@ func _purchase_upgrade(is_reel_speed: bool) -> void:
 		else _upgrades.get_next_barrier_power_cost()
 	)
 	if cost < 0:
-		_feedback.text = "maximum level reached."
+		_feedback.text = "Upgrade is already at maximum."
 		return
 	if not _wallet.can_afford(cost):
-		_feedback.text = "not enough money."
+		_feedback.text = "Not enough fish coin."
 		return
-	var transaction_generation: int = _generation
+	if _network_shop == null:
+		_feedback.text = "Purchase could not be completed."
+		return
+	if is_reel_speed:
+		_network_shop.request_reel_speed_upgrade()
+	else:
+		_network_shop.request_barrier_power_upgrade()
+
+
+func _on_network_purchase_pending(_request_id: String) -> void:
 	_transaction_in_progress = true
-	var purchased: bool = (
+	if visible:
-		_upgrades.purchase_reel_speed(_wallet)
+		_feedback.text = "Purchasing…"
-		if is_reel_speed
+		_refresh_all()
-		else _upgrades.purchase_barrier_power(_wallet)
+
-	)
+
+func _on_network_purchase_finished(
+	_request_id: String,
+	accepted: bool,
+	message: String,
+	_product_id: StringName,
+	_category: int,
+	_quantity: int,
+	total_cost: int,
+) -> void:
 	_transaction_in_progress = false
-	if transaction_generation != _generation or not visible:
+	if not visible:
 		return
 	_feedback.text = (
-		"upgrade purchased."
+		"Purchase complete. $%d spent." % total_cost
-		if purchased
+		if accepted
-		else "unable to complete purchase."
+		else message
 	)
 	_refresh_all()
 

ui/game_ui.gd

@@ -103,6 +103,7 @@ func setup(
 	cooler_capacity: PlayerCoolerCapacityType,
 	network_session: NetworkSessionType,
 	network_sale_service: NetworkSaleService,
+	network_shop_service: NetworkShopService,
 ) -> void:
 	_fishing_spot = fishing_spot
 	_item_effects = item_effects
@@ -140,7 +141,9 @@ func setup(
 		shop_interaction,
 		bag,
 		item_catalog,
-		cooler_capacity
+		cooler_capacity,
+		network_session,
+		network_shop_service
 	)
 	_fishing_shop.menu_visibility_changed.connect(_on_shop_visibility_changed)
 

world/fishing_shop_interaction.gd

@@ -21,6 +21,14 @@ func is_local_player_in_range() -> bool:
 	return _local_player_in_range
 
 
+func is_avatar_in_range(avatar: Player) -> bool:
+	return (
+		avatar != null
+		and is_instance_valid(avatar)
+		and avatar in get_overlapping_bodies()
+	)
+
+
 func _refresh_overlaps() -> void:
 	var is_overlapping: bool = (
 		_local_player != null