Add host-validated multiplayer fish sales

fish/fish_catch.gd

@@ -65,6 +65,17 @@ func to_save_dict() -> Dictionary:
 	}
 
 
+func to_network_dict() -> Dictionary:
+	return {
+		"catch_id": String(catch_id),
+		"fish_id": String(fish_id),
+		"weight_lb": weight_lb,
+		"display_scale": display_scale,
+		"sale_value": sale_value,
+		"is_favorited": is_favorited,
+	}
+
+
 static func from_save_dict(
 	data: Dictionary,
 	resolved_fish: FishDataType,

main/main.gd

@@ -46,6 +46,9 @@ const PlayerSpawnServiceType = preload(
 const NetworkFishingServiceType = preload(
 	"res://network/network_fishing_service.gd"
 )
+const NetworkSaleServiceType = preload(
+	"res://network/network_sale_service.gd"
+)
 
 const TITLE_MUSIC_SILENCE_DB: float = -80.0
 
@@ -83,6 +86,7 @@ const TITLE_MUSIC_SILENCE_DB: float = -80.0
 @onready var _network_fishing: NetworkFishingServiceType = (
 	%NetworkFishingService
 )
+@onready var _network_sale: NetworkSaleServiceType = %NetworkSaleService
 @onready var _players_root: Node3D = $Players
 
 var _gameplay_started: bool = false
@@ -153,6 +157,18 @@ func _ready() -> void:
 		item_catalog,
 		fish_catalog
 	)
+	_network_sale.setup(
+		_network_session,
+		_player_spawn_service,
+		_network_fishing,
+		_player.inventory,
+		_player.wallet,
+		_player.fish_sale_service,
+		_save_manager,
+		fish_catalog,
+		pelican_buyer_profile,
+		_test_world.get_pelican_convenience_landmark()
+	)
 	_fishing_spot.setup(
 		_player,
 		_player.inventory,
@@ -183,7 +199,8 @@ func _ready() -> void:
 		_shop_interaction,
 		_player.item_effects,
 		_player.cooler_capacity,
-		_network_session
+		_network_session,
+		_network_sale
 	)
 	_water_recovery.setup(
 		_player,

main/main.tscn

@@ -1,4 +1,4 @@
-[gd_scene load_steps=22 format=3]
+[gd_scene load_steps=23 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"]
@@ -21,6 +21,7 @@
 [ext_resource type="Script" path="res://network/saved_server_store.gd" id="19_saved_servers"]
 [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"]
 
 [node name="Main" type="Node3D"]
 script = ExtResource("3_main")
@@ -49,6 +50,10 @@ script = ExtResource("20_spawn_service")
 unique_name_in_owner = true
 script = ExtResource("21_network_fishing")
 
+[node name="NetworkSaleService" type="Node" parent="."]
+unique_name_in_owner = true
+script = ExtResource("22_network_sale")
+
 [node name="TestWorld" parent="." instance=ExtResource("1_world")]
 
 [node name="Players" type="Node3D" parent="."]

network/network_fishing_service.gd

@@ -150,6 +150,10 @@ func has_local_attempt() -> bool:
 	)
 
 
+func has_peer_attempt(peer_id: int) -> bool:
+	return _attempts.has(peer_id)
+
+
 func _process(delta: float) -> void:
 	if _session == null or not _session.is_host():
 		return

network/network_protocol.gd

@@ -6,6 +6,9 @@ const GAME_BUILD: String = "prealpha"
 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.
+const SALE_RELIABLE_CHANNEL: int = 5
 
 enum RejectionCode {
 	NONE,
@@ -102,6 +105,7 @@ static func make_server_hello(
 		"capability_flags": PackedStringArray([
 			"movement_v1",
 			"fishing_v1",
+			"sale_v1",
 		]),
 	}
 

network/network_sale_protocol.gd

@@ -0,0 +1,116 @@
+class_name NetworkSaleProtocol
+extends RefCounted
+
+const CAPABILITY: StringName = &"sale_v1"
+const RELIABLE_CHANNEL: int = NetworkProtocol.SALE_RELIABLE_CHANNEL
+const MAX_ID_LENGTH: int = 96
+const MAX_CATCH_ID_LENGTH: int = 160
+const MAX_CATCHES_PER_REQUEST: int = 64
+const MAX_MESSAGE_LENGTH: int = 160
+
+enum Rejection {
+	NONE,
+	MALFORMED,
+	STALE_SESSION,
+	UNAUTHENTICATED,
+	ALREADY_PENDING,
+	BUYER_UNAVAILABLE,
+	TOO_FAR,
+	INVALID_CATCH,
+	FAVORITED,
+	UNSUPPORTED,
+}
+
+
+static func validate_request(data: Variant) -> String:
+	if typeof(data) != TYPE_DICTIONARY:
+		return "Sale could not be completed."
+	var payload: Dictionary = data
+	for key: String in ["request_id", "session_id", "buyer_id", "catches"]:
+		if not payload.has(key):
+			return "Sale could not be completed."
+	if (
+		typeof(payload["request_id"]) != TYPE_STRING
+		or typeof(payload["session_id"]) != TYPE_STRING
+		or typeof(payload["buyer_id"]) not in [TYPE_STRING, TYPE_STRING_NAME]
+		or typeof(payload["catches"]) != TYPE_ARRAY
+	):
+		return "Sale could not be completed."
+	var request_id: String = payload["request_id"]
+	var session_id: String = payload["session_id"]
+	var buyer_id: String = str(payload["buyer_id"])
+	var catches: Array = payload["catches"]
+	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 buyer_id.is_empty()
+		or buyer_id.length() > MAX_ID_LENGTH
+		or catches.is_empty()
+		or catches.size() > MAX_CATCHES_PER_REQUEST
+	):
+		return "Sale could not be completed."
+	var seen: Dictionary[String, bool] = {}
+	for value: Variant in catches:
+		if typeof(value) != TYPE_DICTIONARY:
+			return "Sale could not be completed."
+		var evidence: Dictionary = value
+		for key: String in [
+			"catch_id", "fish_id", "weight_lb", "display_scale",
+			"sale_value", "is_favorited",
+		]:
+			if not evidence.has(key):
+				return "Sale could not be completed."
+		var catch_id: String = str(evidence["catch_id"])
+		if (
+			catch_id.is_empty()
+			or catch_id.length() > MAX_CATCH_ID_LENGTH
+			or seen.has(catch_id)
+			or typeof(evidence["fish_id"]) not in [
+				TYPE_STRING, TYPE_STRING_NAME
+			]
+			or typeof(evidence["is_favorited"]) != TYPE_BOOL
+		):
+			return "Sale could not be completed."
+		seen[catch_id] = true
+	return ""
+
+
+static func validate_result(data: Variant) -> bool:
+	if typeof(data) != TYPE_DICTIONARY:
+		return false
+	var payload: Dictionary = data
+	if not (
+		typeof(payload.get("result_id")) == TYPE_STRING
+		and not str(payload["result_id"]).is_empty()
+		and str(payload["result_id"]).length() <= MAX_ID_LENGTH
+		and typeof(payload.get("request_id")) == TYPE_STRING
+		and not str(payload["request_id"]).is_empty()
+		and str(payload["request_id"]).length() <= MAX_ID_LENGTH
+		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("catch_ids")) == TYPE_ARRAY
+		and payload["catch_ids"].size() <= MAX_CATCHES_PER_REQUEST
+		and typeof(payload.get("payout")) == TYPE_INT
+		and int(payload["payout"]) >= 0
+		and typeof(payload.get("base_value")) == TYPE_INT
+		and int(payload["base_value"]) >= 0
+		and typeof(payload.get("message")) == TYPE_STRING
+		and str(payload["message"]).length() <= MAX_MESSAGE_LENGTH
+	):
+		return false
+	var seen: Dictionary[String, bool] = {}
+	for value: Variant in payload["catch_ids"]:
+		if typeof(value) not in [TYPE_STRING, TYPE_STRING_NAME]:
+			return false
+		var catch_id: String = str(value)
+		if (
+			catch_id.is_empty()
+			or catch_id.length() > MAX_CATCH_ID_LENGTH
+			or seen.has(catch_id)
+		):
+			return false
+		seen[catch_id] = true
+	return true

network/network_sale_protocol.gd.uid

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

network/network_sale_service.gd

@@ -0,0 +1,553 @@
+class_name NetworkSaleService
+extends Node
+
+const FishCatchType = preload("res://fish/fish_catch.gd")
+const FishDataType = preload("res://fish/fish_data.gd")
+const FishPoolType = preload("res://fish/fish_pool.gd")
+const FishBuyerProfileType = preload("res://economy/fish_buyer_profile.gd")
+const FishSaleServiceType = preload("res://economy/fish_sale_service.gd")
+const FishSaleResultType = preload("res://economy/fish_sale_result.gd")
+
+const MAX_LEDGER_ENTRIES_PER_PEER: int = 64
+const PELICAN_INTERACTION_RANGE: float = 7.5
+
+signal local_sale_pending(request_id: String)
+signal local_sale_finished(
+	request_id: String,
+	accepted: bool,
+	message: String,
+	catch_ids: Array[StringName],
+	payout: int,
+)
+
+var _session: NetworkSession
+var _spawn_service: PlayerSpawnService
+var _network_fishing: NetworkFishingService
+var _inventory: FishInventory
+var _wallet: PlayerWallet
+var _sale_service: FishSaleServiceType
+var _save_manager: PlayerSaveManager
+var _fish_catalog: FishPoolType
+var _buyer: FishBuyerProfileType
+var _pelican_landmark: Node3D
+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 _applied_results: Dictionary[String, bool] = {}
+var _received_results: Dictionary[String, bool] = {}
+var _pending_local_request_id: String = ""
+var _pending_local_catch_ids: Array[StringName] = []
+
+
+func setup(
+	session: NetworkSession,
+	spawn_service: PlayerSpawnService,
+	network_fishing: NetworkFishingService,
+	inventory: FishInventory,
+	wallet: PlayerWallet,
+	sale_service: FishSaleServiceType,
+	save_manager: PlayerSaveManager,
+	fish_catalog: FishPoolType,
+	buyer: FishBuyerProfileType,
+	pelican_landmark: Node3D,
+) -> void:
+	_session = session
+	_spawn_service = spawn_service
+	_network_fishing = network_fishing
+	_inventory = inventory
+	_wallet = wallet
+	_sale_service = sale_service
+	_save_manager = save_manager
+	_fish_catalog = fish_catalog
+	_buyer = buyer
+	_pelican_landmark = pelican_landmark
+	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_sale() -> bool:
+	return (
+		_session != null
+		and _session.is_gameplay_session_active()
+		and (
+			_session.is_host()
+			or _session.supports_server_capability(
+				NetworkSaleProtocol.CAPABILITY
+			)
+		)
+	)
+
+
+func is_local_sale_pending() -> bool:
+	return not _pending_local_request_id.is_empty()
+
+
+func request_local_sale(catch_ids: Array[StringName]) -> String:
+	if is_local_sale_pending():
+		local_sale_finished.emit(
+			"", false, "Selling…", [], 0
+		)
+		return ""
+	if not can_request_sale():
+		local_sale_finished.emit(
+			"",
+			false,
+			"Selling is not supported by this server.",
+			[],
+			0
+		)
+		return ""
+	if catch_ids.is_empty() or _inventory == null:
+		local_sale_finished.emit(
+			"", false, "Sale could not be completed.", [], 0
+		)
+		return ""
+	var evidence: Array[Dictionary] = []
+	for catch_id: StringName in catch_ids:
+		var fish_catch: FishCatch = _inventory.get_catch_by_id(catch_id)
+		if fish_catch == null:
+			local_sale_finished.emit(
+				"", false, "That catch is no longer available.", [], 0
+			)
+			return ""
+		evidence.append(fish_catch.to_network_dict())
+	var request_id: String = _new_id("sale")
+	var request: Dictionary = {
+		"request_id": request_id,
+		"session_id": _session.get_session_id(),
+		"buyer_id": str(_buyer.id) if _buyer != null else "",
+		"catches": evidence,
+	}
+	_pending_local_request_id = request_id
+	_pending_local_catch_ids = catch_ids.duplicate()
+	local_sale_pending.emit(request_id)
+	if _session.is_host():
+		_handle_sale_request(_session.get_local_peer_id(), request)
+	else:
+		submit_sale_request.rpc_id(1, request)
+	return request_id
+
+
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkSaleProtocol.RELIABLE_CHANNEL
+)
+func submit_sale_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_sale_request(sender_id, data)
+
+
+func _handle_sale_request(peer_id: int, data: Dictionary) -> void:
+	var validation_error: String = NetworkSaleProtocol.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 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(
+			request_id, validation_error
+		))
+		return
+	if str(data["session_id"]) != _session.get_session_id():
+		_record_and_send(peer_id, _rejected_result(
+			request_id, "Sale 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(
+			request_id, "A sale is already pending."
+		))
+		return
+	if not _is_buyer_available_for_peer(peer_id):
+		var avatar: Player = _spawn_service.get_avatar(peer_id)
+		var message: String = (
+			"The buyer is unavailable."
+			if avatar == null or _pelican_landmark == null
+			else "Move closer to the buyer."
+		)
+		_record_and_send(peer_id, _rejected_result(request_id, message))
+		return
+	if _buyer == null or not _buyer.is_valid():
+		_record_and_send(peer_id, _rejected_result(
+			request_id, "The buyer is unavailable."
+		))
+		return
+	if StringName(str(data["buyer_id"])) != _buyer.id:
+		_record_and_send(peer_id, _rejected_result(
+			request_id, "The buyer is unavailable."
+		))
+		return
+	var result: Dictionary = _build_authoritative_result(
+		peer_id, request_id, data["catches"]
+	)
+	_pending_by_peer[peer_id] = request_id
+	_record_and_send(peer_id, result)
+
+
+func _build_authoritative_result(
+	peer_id: int,
+	request_id: String,
+	evidence_values: Array,
+) -> Dictionary:
+	if _fish_catalog == null or _buyer == null:
+		return _rejected_result(
+			request_id, "The buyer is unavailable."
+		)
+	var catch_ids: Array[String] = []
+	var base_value: int = 0
+	var payout: int = 0
+	for value: Variant in evidence_values:
+		var evidence: Dictionary = value
+		var fish_id := StringName(str(evidence.get("fish_id", "")))
+		var fish: FishDataType = _fish_catalog.get_fish_by_id(fish_id)
+		var decoded: FishCatch = FishCatchType.from_network_dict(
+			evidence, fish
+		)
+		if decoded == null:
+			return _rejected_result(
+				request_id, "Sale could not be completed."
+			)
+		if (
+			not str(decoded.catch_id).begins_with("%s:" % fish_id)
+			or decoded.weight_lb < fish.get_minimum_weight()
+			or decoded.weight_lb > fish.get_maximum_weight()
+			or decoded.sale_value < fish.sell_value_min
+			or decoded.sale_value > fish.sell_value_max
+		):
+			return _rejected_result(
+				request_id, "Sale could not be completed."
+			)
+		if bool(evidence.get("is_favorited", false)):
+			return _rejected_result(
+				request_id, "Favorite catches cannot be sold."
+			)
+		var offer: int = _buyer.get_offer(decoded.sale_value)
+		if (
+			offer < 0
+			or base_value > 9223372036854775807 - decoded.sale_value
+			or payout > 9223372036854775807 - offer
+		):
+			return _rejected_result(
+				request_id, "Sale could not be completed."
+			)
+		base_value += decoded.sale_value
+		payout += offer
+		catch_ids.append(str(decoded.catch_id))
+	return {
+		"result_id": _new_id("sale_result"),
+		"request_id": request_id,
+		"session_id": _session.get_session_id(),
+		"target_peer_id": peer_id,
+		"accepted": true,
+		"catch_ids": catch_ids,
+		"payout": payout,
+		"base_value": base_value,
+		"message": "Sale complete.",
+	}
+
+
+func _rejected_result(request_id: String, message: String) -> Dictionary:
+	var safe_request_id: String = (
+		request_id
+		if (
+			not request_id.is_empty()
+			and request_id.length() <= NetworkSaleProtocol.MAX_ID_LENGTH
+		)
+		else "invalid"
+	)
+	return {
+		"result_id": _new_id("sale_result"),
+		"request_id": safe_request_id,
+		"session_id": _session.get_session_id() if _session != null else "",
+		"target_peer_id": 0,
+		"accepted": false,
+		"catch_ids": [],
+		"payout": 0,
+		"base_value": 0,
+		"message": message.left(NetworkSaleProtocol.MAX_MESSAGE_LENGTH),
+	}
+
+
+func _record_and_send(peer_id: int, result: Dictionary) -> void:
+	result["target_peer_id"] = peer_id
+	var request_id: String = str(result["request_id"])
+	var ledger: Dictionary = _request_ledgers.get(peer_id, {})
+	ledger[request_id] = result.duplicate(true)
+	while ledger.size() > MAX_LEDGER_ENTRIES_PER_PEER:
+		ledger.erase(ledger.keys().front())
+	_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_sale_result(result)
+	else:
+		receive_sale_result.rpc_id(peer_id, result)
+
+
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkSaleProtocol.RELIABLE_CHANNEL
+)
+func receive_sale_result(data: Dictionary) -> void:
+	_apply_sale_result(data)
+
+
+func _apply_sale_result(data: Dictionary) -> void:
+	if (
+		not NetworkSaleProtocol.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 str(data["request_id"]) != _pending_local_request_id:
+		return
+	if not bool(data["accepted"]):
+		_received_results[result_id] = true
+		_bound_dictionary(_received_results)
+		_finish_local_sale(
+			data["request_id"], false, str(data["message"]), [], 0
+		)
+		_acknowledge_result(data, false, str(data["message"]))
+		return
+	var catch_ids: Array[StringName] = []
+	for value: Variant in data["catch_ids"]:
+		if typeof(value) not in [TYPE_STRING, TYPE_STRING_NAME]:
+			_fail_local_apply(data, "Sale could not be completed.")
+			return
+		catch_ids.append(StringName(str(value)))
+	if catch_ids != _pending_local_catch_ids:
+		_fail_local_apply(data, "Sale could not be completed.")
+		return
+	var preview: FishSaleResultType = _sale_service.preview_batch(
+		catch_ids, _buyer
+	)
+	if not preview.is_success():
+		var message: String = (
+			"Favorite catches cannot be sold."
+			if preview.status == FishSaleResultType.Status.FAVORITED
+			else "That catch is no longer available."
+		)
+		_fail_local_apply(data, message)
+		return
+	if (
+		preview.payout != int(data["payout"])
+		or preview.base_value != int(data["base_value"])
+	):
+		_fail_local_apply(data, "Sale could not be completed.")
+		return
+	var inventory_snapshot: Array[FishCatch] = _inventory.get_all_catches()
+	var sequence_snapshot: int = _inventory.get_next_catch_sequence()
+	var wallet_snapshot: int = _wallet.get_balance()
+	var local_result: FishSaleResultType = _sale_service.sell_batch(
+		catch_ids, _buyer
+	)
+	if not local_result.is_success() or not _save_manager.save_if_dirty():
+		_inventory.replace_all_catches(
+			inventory_snapshot, sequence_snapshot
+		)
+		_wallet.restore_balance(wallet_snapshot)
+		_save_manager.save_if_dirty()
+		_fail_local_apply(data, "Sale 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_sale(
+		data["request_id"],
+		true,
+		"Sale complete.",
+		catch_ids,
+		int(data["payout"])
+	)
+	_acknowledge_result(data, true, "")
+
+
+func _fail_local_apply(data: Dictionary, message: String) -> void:
+	_finish_local_sale(data["request_id"], false, message, [], 0)
+	_acknowledge_result(data, false, message)
+
+
+func _finish_local_sale(
+	request_id: String,
+	accepted: bool,
+	message: String,
+	catch_ids: Array[StringName],
+	payout: int,
+) -> void:
+	_pending_local_request_id = ""
+	_pending_local_catch_ids.clear()
+	local_sale_finished.emit(
+		request_id, accepted, message, catch_ids, payout
+	)
+
+
+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_sale_result.rpc_id(
+			1,
+			str(data["result_id"]),
+			str(data["request_id"]),
+			applied,
+			message.left(NetworkSaleProtocol.MAX_MESSAGE_LENGTH)
+		)
+
+
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkSaleProtocol.RELIABLE_CHANNEL
+)
+func acknowledge_sale_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() > NetworkSaleProtocol.MAX_ID_LENGTH
+		or request_id.is_empty()
+		or request_id.length() > NetworkSaleProtocol.MAX_ID_LENGTH
+		or _message.length() > NetworkSaleProtocol.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_buyer_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 _pelican_landmark == null
+		or not is_instance_valid(_pelican_landmark)
+	):
+		return false
+	var avatar: Player = _spawn_service.get_avatar(peer_id)
+	if (
+		avatar == null
+		or avatar.is_water_recovery_active()
+		or (
+			_network_fishing != null
+			and _network_fishing.has_peer_attempt(peer_id)
+		)
+	):
+		return false
+	return (
+		avatar.global_position.distance_to(
+			_pelican_landmark.global_position
+		) <= PELICAN_INTERACTION_RANGE
+	)
+
+
+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_sale_pending()
+		_clear_session_state()
+		if had_pending:
+			local_sale_finished.emit(
+				"", false, "Connection lost.", [], 0
+			)
+
+
+func _clear_session_state() -> void:
+	_request_ledgers.clear()
+	_pending_by_peer.clear()
+	_result_owners.clear()
+	_acknowledged_results.clear()
+	_applied_results.clear()
+	_received_results.clear()
+	_pending_local_request_id = ""
+	_pending_local_catch_ids.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_sale_service.gd.uid

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

network/network_session.gd

@@ -59,6 +59,7 @@ var _last_server_max_players: int = DEFAULT_SESSION_MAX_PLAYERS
 var _last_server_player_count: int = 0
 var _last_server_display_name: String = ""
 var _last_server_protocol_version: int = 0
+var _server_capabilities: PackedStringArray = PackedStringArray()
 var _profile_ready: bool = false
 
 
@@ -270,6 +271,14 @@ func get_operation_generation() -> int:
 	return _operation_generation
 
 
+func supports_server_capability(capability: StringName) -> bool:
+	if is_host():
+		return str(capability) in PackedStringArray([
+			"movement_v1", "fishing_v1", "sale_v1",
+		])
+	return str(capability) in _server_capabilities
+
+
 func get_local_peer_id() -> int:
 	return multiplayer.get_unique_id() if is_gameplay_session_active() else 0
 
@@ -503,6 +512,16 @@ func receive_server_hello(data: Dictionary) -> void:
 	_last_server_protocol_version = int(data.get(
 		"protocol_version", NetworkProtocol.PROTOCOL_VERSION
 	))
+	_server_capabilities = PackedStringArray()
+	var advertised_capabilities: Variant = data.get(
+		"capability_flags", PackedStringArray()
+	)
+	if typeof(advertised_capabilities) in [
+		TYPE_ARRAY, TYPE_PACKED_STRING_ARRAY
+	]:
+		for value: Variant in advertised_capabilities:
+			if typeof(value) in [TYPE_STRING, TYPE_STRING_NAME]:
+				_server_capabilities.append(str(value))
 	var local_peer_id: int = multiplayer.get_unique_id()
 	_registry.clear()
 	_registry.add_peer(
@@ -855,3 +874,4 @@ func _teardown_peer() -> void:
 	_connection_deadline = 0.0
 	_input_accumulator = 0.0
 	_snapshot_accumulator = 0.0
+	_server_capabilities = PackedStringArray()

ui/game_ui.gd

@@ -102,6 +102,7 @@ func setup(
 	item_effects: PlayerItemEffectsType,
 	cooler_capacity: PlayerCoolerCapacityType,
 	network_session: NetworkSessionType,
+	network_sale_service: NetworkSaleService,
 ) -> void:
 	_fishing_spot = fishing_spot
 	_item_effects = item_effects
@@ -124,7 +125,8 @@ func setup(
 		hotbar,
 		item_catalog,
 		cooler_capacity,
-		network_session
+		network_session,
+		network_sale_service
 	)
 	_hotbar_ui.setup(hotbar, bag, item_catalog, fishing_spot)
 	_fishing_shop.setup(

ui/player_menu.gd

@@ -227,6 +227,7 @@ var _collection_log: CollectionLogType
 var _wallet: PlayerWalletType
 var _sale_service: FishSaleServiceType
 var _network_session: NetworkSessionType
+var _network_sale_service: NetworkSaleService
 var _default_buyer: FishBuyerProfileType
 var _catalog: FishPoolType
 var _fishing_spot: FishingSpotType
@@ -356,6 +357,7 @@ func setup(
 	item_catalog: ItemCatalogType,
 	cooler_capacity: PlayerCoolerCapacityType,
 	network_session: NetworkSessionType,
+	network_sale_service: NetworkSaleService,
 ) -> void:
 	_player = player
 	_inventory = inventory
@@ -370,6 +372,19 @@ func setup(
 	_item_catalog = item_catalog
 	_cooler_capacity = cooler_capacity
 	_network_session = network_session
+	_network_sale_service = network_sale_service
+	if (
+		_network_sale_service != null
+		and not _network_sale_service.local_sale_pending.is_connected(
+			_on_network_sale_pending
+		)
+	):
+		_network_sale_service.local_sale_pending.connect(
+			_on_network_sale_pending
+		)
+		_network_sale_service.local_sale_finished.connect(
+			_on_network_sale_finished
+		)
 	_fish_selection.clear()
 	if not _inventory.catches_changed.is_connected(_on_inventory_changed):
 		_inventory.catches_changed.connect(_on_inventory_changed)
@@ -2341,20 +2356,42 @@ func _update_sale_summary() -> void:
 		_sale_unavailable.text = ""
 		_sale_unavailable.visible = false
 		return
-	if not _can_use_shared_world_actions():
+	if (
+		_sale_in_progress
+		or (
+			_network_sale_service != null
+			and _network_sale_service.is_local_sale_pending()
+		)
+	):
 		_selection_summary.text = (
 			"1 fish selected"
 			if selected_count == 1
 			else "%d fish selected" % selected_count
 		)
 		_selection_status.set_content("selected", str(selected_count))
-		_offer_status.set_content("pelican offer", "host only")
+		_offer_status.set_content("pelican offer", "pending")
+		_sell_button.disabled = true
+		_sell_bubble.disabled = true
+		_sale_unavailable.text = "Selling…"
+		_sale_unavailable.visible = true
+		return
+	if (
+		_network_sale_service == null
+		or not _network_sale_service.can_request_sale()
+	):
+		_selection_summary.text = (
+			"1 fish selected"
+			if selected_count == 1
+			else "%d fish selected" % selected_count
+		)
+		_selection_status.set_content("selected", str(selected_count))
+		_offer_status.set_content("pelican offer", "unavailable")
 		_sell_button.disabled = true
 		_sell_bubble.disabled = true
 		_sell_bubble.persistent_mark = false
 		_sell_bubble.refresh_ink_state()
 		_sale_unavailable.text = (
-			"Selling in joined games is coming in a later multiplayer phase."
+			"Selling is not supported by this server."
 		)
 		_sale_unavailable.visible = true
 		return
@@ -2428,9 +2465,21 @@ func _on_favorite_pressed() -> void:
 
 
 func _on_sell_pressed() -> void:
-	if not _can_use_shared_world_actions():
+	if (
+		_sale_in_progress
+		or (
+			_network_sale_service != null
+			and _network_sale_service.is_local_sale_pending()
+		)
+	):
+		_transaction_feedback.text = "Selling…"
+		return
+	if (
+		_network_sale_service == null
+		or not _network_sale_service.can_request_sale()
+	):
 		_transaction_feedback.text = (
-			"Selling in joined games is coming in a later multiplayer phase."
+			"Selling is not supported by this server."
 		)
 		return
 	var selected_ids: Array[StringName] = _fish_selection.get_selected_ids()
@@ -2481,7 +2530,8 @@ func _on_sell_pressed() -> void:
 func _on_confirm_sale_pressed() -> void:
 	if (
 		_sale_in_progress
-		or not _can_use_shared_world_actions()
+		or _network_sale_service == null
+		or not _network_sale_service.can_request_sale()
 		or _sale_service == null
 		or _confirmation_catch_ids.is_empty()
 		or _confirmation_buyer == null
@@ -2494,30 +2544,58 @@ func _on_confirm_sale_pressed() -> void:
 		_confirmation_catch_ids.duplicate()
 	)
 	var transaction_generation: int = _menu_generation
-	var transaction_buyer: FishBuyerProfileType = _confirmation_buyer
 	_confirm_sale_button.disabled = true
-	var result: FishSaleResultType = _sale_service.sell_batch(
-		requested_catch_ids,
-		transaction_buyer
-	)
 	_close_sale_confirmation()
-	_sale_in_progress = false
+	var request_id: String = _network_sale_service.request_local_sale(
-	if transaction_generation != _menu_generation or not visible:
+		requested_catch_ids
+	)
+	if request_id.is_empty():
+		_sale_in_progress = false
 		return
-	_transaction_feedback.text = result.get_message()
+	if (
-	if result.is_success():
+		_network_sale_service.is_local_sale_pending()
-		_fish_selection.remove_ids(requested_catch_ids)
+		and transaction_generation == _menu_generation
-	_refresh_all()
+		and visible
-	_inventory_tab.grab_focus()
+	):
+		_transaction_feedback.text = "Selling…"
 
 
 func _can_use_shared_world_actions() -> bool:
 	return (
-		_network_session == null
+		_network_sale_service != null
-		or _network_session.can_use_host_gameplay()
+		and _network_sale_service.can_request_sale()
 	)
 
 
+func _on_network_sale_pending(_request_id: String) -> void:
+	_sale_in_progress = true
+	if visible:
+		_transaction_feedback.text = "Selling…"
+		_update_sale_summary()
+
+
+func _on_network_sale_finished(
+	_request_id: String,
+	accepted: bool,
+	message: String,
+	catch_ids: Array[StringName],
+	payout: int,
+) -> void:
+	_sale_in_progress = false
+	if accepted:
+		_fish_selection.remove_ids(catch_ids)
+	if not visible:
+		return
+	_transaction_feedback.text = (
+		"Sale complete. $%d received." % payout
+		if accepted
+		else message
+	)
+	_refresh_all()
+	if _current_section == Section.COOLER:
+		_inventory_tab.grab_focus()
+
+
 func _close_sale_confirmation() -> void:
 	var was_visible: bool = _sale_confirmation.visible
 	_confirmation_catch_ids.clear()

world/interactables/pelican_buyer_world.tscn

@@ -50,7 +50,7 @@ no_depth_test = true
 
 [node name="ConvenienceLabel" type="Label3D" parent="Visual"]
 position = Vector3(0, 1.65, -1.15)
-text = "available anywhere • 0.25x"
+text = "sell nearby • 0.25x"
 font_size = 20
 outline_size = 6
 billboard = 1