itchio-opengamepadui-plugin/core/itch_client.gd

extends NodeThread

## Godot interface for butlerd
##
## Provides a Godot interface to butlerd, itch.io's official JSON-RPC 2.0
## launcher daemon (`butler daemon --json ...`). This class relies on
## [InteractiveProcess] (same mechanism the Steam plugin uses for steamcmd)
## purely to spawn butlerd and read its single-line startup handshake off
## stdout. All *actual* communication after that happens over a raw TCP
## socket, per butlerd's wire protocol:
## https://itch.io/docs/butler/launcher-integration.html
##
## NOTE ON FIELD NAMES: the JSON field names used below follow butlerd's
## documented camelCase convention (mirroring the Go struct field names in
## https://pkg.go.dev/github.com/itchio/butler/butlerd). If itch.io ships a
## butler update that changes a field, diff against that spec.

const broth_base := "https://broth.itch.zone/butler"
## Separator between an upload name and its size in the install-options
## dropdown ("bubble-tower-linux.zip  (31.3 MB)"). Shared with the library so
## the value the user picks can be matched back to the upload it came from.
const UPLOAD_SIZE_SEPARATOR := "  ("
## Persistent storage for the downloaded butler binary + butler.db. Kept OUT
## of user://plugins/<id>/ because OGPU wipes that directory (and everything
## inside it) whenever the plugin is updated or re-extracted.
const butler_dir := "user://butler"
## Where installed games live. This is intentionally NOT user://plugins/itch:
## OGPU moves the entire extracted plugin directory (plugins/<id>/) to the
## trash on every plugin update, so any install kept under it silently loses
## its files on update while butler.db keeps the (now orphaned) cave, which
## used to break both fresh installs ("That upload is already installed!")
## and launches. Keeping games next to butler.db (user://butler) means they
## survive plugin updates untouched.
const games_dir := "user://butler/games"
const CACHE_DIR := "itch"

enum STATE {
	BOOT,
	WAITING_HANDSHAKE,
	CONNECTED,
}

enum LOGIN_STATUS {
	OK,
	FAILED,
	INVALID_KEY,
}

# butlerd process/socket signals
signal rpc_response(id: int, result: Variant, error: Variant)
signal rpc_notification(method: String, params: Variant)

# Main thread signals
signal bootstrap_finished
signal client_ready
signal logged_in(status: LOGIN_STATUS, profile: Dictionary)
signal install_progressed(cave_or_game_id: String, current: int, total: int)
signal install_failed(cave_or_game_id: String, game_title: String, message: String)
signal app_installed(cave_id: String, success: bool)
signal app_updated(cave_id: String, success: bool)
signal app_uninstalled(cave_id: String, success: bool)
signal launch_exited(cave_id: String)

var proc: InteractiveProcess
var socket: StreamPeerTCP
var state: STATE = STATE.BOOT
var client_started := false
var is_logged_in := false
var profile: Dictionary = {}

var _proc_buffer := ""
var _recv_buffer := ""
var _next_id := 1
var _install_location_id := ""

var logger := Log.get_logger("ItchClient", Log.LEVEL.INFO)


func _ready() -> void:
	add_to_group("itch_client")
	thread_group = SharedThread.new()
	thread_group.name = "ItchClient"
	bootstrap()


## Bootstraps the butler binary if it isn't present, then spawns butlerd.
func bootstrap() -> void:
	var butler_dir_global := ProjectSettings.globalize_path(butler_dir)
	_migrate_butler(butler_dir_global)
	var butler_bin := "/".join([butler_dir_global, "butler"])
	if not FileAccess.file_exists(butler_bin):
		logger.info("The butler binary wasn't found. Trying to install it.")
		var success := await _install_butler()
		if not success:
			logger.error("Unable to install butler")
			bootstrap_finished.emit()
			return
		logger.info("Successfully installed butler")

	var dbpath := "/".join([butler_dir_global, "butler.db"])
	var args := [
		"daemon",
		"--json",
		"--transport", "tcp",
		"--keep-alive",
		"--dbpath", dbpath,
		"--address", "https://itch.io",
		"--user-agent", "OpenGamepadUI-itch/0.1.0",
		"--destiny-pid", str(OS.get_process_id()),
	]

	proc = InteractiveProcess.new(butler_bin, args)
	if proc.start() != OK:
		logger.error("Unable to spawn butlerd")
		return
	client_started = true
	state = STATE.WAITING_HANDSHAKE
	bootstrap_finished.emit()


## Moves a butler install left over in the plugin directory (OGPU wipes
## plugins/<id>/ on every plugin update, which is why older builds kept
## losing butler) to the persistent user://butler location.
func _migrate_butler(dest_dir: String) -> void:
	var old_dir := ProjectSettings.globalize_path("user://plugins/itch/assets/butler")
	if old_dir == dest_dir:
		return
	if not DirAccess.dir_exists_absolute(old_dir):
		return
	if not FileAccess.file_exists("/".join([old_dir, "butler"])):
		return
	if FileAccess.file_exists("/".join([dest_dir, "butler"])):
		return
	DirAccess.make_dir_recursive_absolute(dest_dir)
	for file in ["butler", "7z.so", "libc7zip.so"]:
		var src := "/".join([old_dir, file])
		if FileAccess.file_exists(src):
			DirAccess.copy_absolute(src, "/".join([dest_dir, file]))
	logger.info("Migrated butler from " + old_dir + " to " + dest_dir)


## Downloads a butler binary for this platform from itch's broth distribution
## channel (the same one the official itch.io app uses to self-update).
func _install_butler() -> bool:
	var goos := "linux"
	var goarch := "amd64"
	if OS.get_name() == "Windows":
		goos = "windows"
	if OS.get_name() == "macOS":
		goos = "darwin"
	# Engine.get_architecture_name() is only available on newer Godot 4.x
	# builds; guard it so this still works if OpenGamepadUI is running on an
	# older engine version.
	if Engine.has_method("get_architecture_name"):
		var arch: String = Engine.get_architecture_name()
		if "arm64" in arch or "aarch64" in arch:
			goarch = "arm64"

	var platform_slug := goos + "-" + goarch
	var latest_url := "/".join([broth_base, platform_slug, "LATEST"])

	var http := HTTPRequest.new()
	add_child.call_deferred(http)
	await http.ready
	if http.request(latest_url) != OK:
		logger.error("Error fetching latest butler version: " + latest_url)
		remove_child(http)
		http.queue_free()
		return false
	var args: Array = await http.request_completed
	var result: int = args[0]
	var response_code: int = args[1]
	var body: PackedByteArray = args[3]
	if result != HTTPRequest.RESULT_SUCCESS or response_code != 200:
		logger.error("Unable to determine latest butler version for " + platform_slug)
		remove_child(http)
		http.queue_free()
		return false
	var version := body.get_string_from_utf8().strip_edges()

	var archive_url := "/".join([broth_base, platform_slug, version, "archive", "default"])
	if http.request(archive_url) != OK:
		logger.error("Error downloading butler: " + archive_url)
		remove_child(http)
		http.queue_free()
		return false
	args = await http.request_completed
	result = args[0]
	response_code = args[1]
	body = args[3]
	remove_child(http)
	http.queue_free()
	if result != HTTPRequest.RESULT_SUCCESS or response_code != 200:
		logger.error("butler couldn't be downloaded: " + archive_url)
		return false

	var globalized_dir := ProjectSettings.globalize_path(butler_dir)
	DirAccess.make_dir_recursive_absolute(globalized_dir)
	var zip_path := "/tmp/butler-" + version + ".zip"
	var file := FileAccess.open(zip_path, FileAccess.WRITE_READ)
	file.store_buffer(body)
	file.close()

	var out := []
	OS.execute("unzip", ["-o", zip_path, "-d", globalized_dir], out)
	OS.execute("chmod", ["+x", "/".join([globalized_dir, "butler"])], out)

	return true


# ---------------------------------------------------------------------------
# Wire protocol: spawn -> read handshake off stdout -> connect TCP -> auth
# ---------------------------------------------------------------------------

func _thread_process(_delta: float) -> void:
	if state == STATE.WAITING_HANDSHAKE and proc:
		_proc_buffer += proc.read()
		if not _proc_buffer.contains("\n"):
			return
		var lines := _proc_buffer.split("\n")
		_proc_buffer = lines[-1]
		for i in range(lines.size() - 1):
			var line: String = lines[i].strip_edges()
			if line == "":
				continue
			var parsed: Variant = _normalize_json(JSON.parse_string(line))
			if typeof(parsed) != TYPE_DICTIONARY:
				continue
			var msg: Dictionary = parsed
			if msg.get("type", "") != "butlerd/listen-notification":
				continue
			var secret: String = msg.get("secret", "")
			var address: String = msg.get("tcp", {}).get("address", "")
			_connect_and_authenticate.call_deferred(address, secret)
			return

	if state == STATE.CONNECTED and socket:
		socket.poll()
		var available := socket.get_available_bytes()
		if available <= 0:
			return
		var chunk := socket.get_partial_data(available)
		if chunk[0] != OK:
			return
		_recv_buffer += (chunk[1] as PackedByteArray).get_string_from_utf8()
		while _recv_buffer.contains("\n"):
			var idx := _recv_buffer.find("\n")
			var line := _recv_buffer.substr(0, idx)
			_recv_buffer = _recv_buffer.substr(idx + 1)
			if line.strip_edges() == "":
				continue
			var parsed: Variant = _normalize_json(JSON.parse_string(line))
			if typeof(parsed) != TYPE_DICTIONARY:
				continue
			var msg: Dictionary = parsed
			if "id" in msg:
				rpc_response.emit.call_deferred(int(msg["id"]), msg.get("result"), msg.get("error"))
			elif "method" in msg:
				rpc_notification.emit.call_deferred(msg["method"], msg.get("params", {}))


func _connect_and_authenticate(address: String, secret: String) -> void:
	var parts := address.split(":")
	var host: String = parts[0]
	var port: int = int(parts[1])

	socket = StreamPeerTCP.new()
	if socket.connect_to_host(host, port) != OK:
		logger.error("Unable to connect to butlerd at " + address)
		return

	# Wait for the connection to establish
	var timeout := 50
	while socket.get_status() == StreamPeerTCP.STATUS_CONNECTING and timeout > 0:
		socket.poll()
		OS.delay_msec(20)
		timeout -= 1
	if socket.get_status() != StreamPeerTCP.STATUS_CONNECTED:
		logger.error("Timed out connecting to butlerd")
		return

	state = STATE.CONNECTED

	var res := await _rpc_call("Meta.Authenticate", {"secret": secret})
	if "error" in res:
		logger.error("butlerd authentication failed: " + str(res["error"]))
		return

	logger.info("Connected and authenticated with butlerd")
	client_ready.emit.call_deferred()


# ---------------------------------------------------------------------------
# Low-level JSON-RPC helpers
# ---------------------------------------------------------------------------

func _send_line(text: String) -> void:
	if not socket:
		return
	socket.put_data((text + "\n").to_utf8_buffer())


## Godot's JSON.parse decodes every JSON number as a float (so 7670 becomes
## 7670.0), and JSON.stringify then re-serializes it as "7670.0". butlerd's Go
## structs use int64 for ids, and Go's encoding/json rejects "7670.0" for an
## int64 field (RPC error -32700). Normalizing integral floats back to ints at
## the parse boundary keeps every id (profileId, game.id, upload.id, ...)
## typed correctly so downstream RPC params serialize without the trailing
## ".0".
func _normalize_json(value: Variant) -> Variant:
	match typeof(value):
		TYPE_DICTIONARY:
			var out := {}
			for key in value:
				out[key] = _normalize_json(value[key])
			return out
		TYPE_ARRAY:
			var out := []
			for item in value:
				out.append(_normalize_json(item))
			return out
		TYPE_FLOAT:
			if is_finite(value) and value == floor(value):
				return int(value)
			return value
		_:
			return value


## Sends a JSON-RPC request and waits (via signal await, not a busy loop, so
## [method _thread_process] keeps ticking and can deliver the response) for
## the matching response. Mirrors the `_wait_for_command` pattern used by the
## Steam plugin's steamcmd wrapper, adapted for id-based JSON-RPC instead of
## line-scraping a REPL.
func _rpc_call(method: String, params: Dictionary = {}) -> Dictionary:
	var id := _next_id
	_next_id += 1
	var req := {"jsonrpc": "2.0", "id": id, "method": method, "params": params}
	# Params can carry game dicts that went through a JSON cache roundtrip
	# (Godot's JSON.parse_string decodes every number as a float), so an id
	# like 936521 comes back as 936521.0 and butlerd's int64 fields reject it
	# with RPC error -32700. Normalizing before serializing guarantees every
	# integral number is sent as an int, mirroring the inbound normalization.
	_send_line(JSON.stringify(_normalize_json(req)))

	var out: Array = [-1, null, null]
	while out[0] != id:
		out = await rpc_response

	if out[2] != null:
		return {"error": out[2]}
	var result: Variant = out[1]
	if typeof(result) == TYPE_DICTIONARY:
		return result
	return {}


# ---------------------------------------------------------------------------
# Public, high-level API. Each of these hops onto the SharedThread via
# thread_group.exec(), same convention the Steam/Epic plugins use.
# ---------------------------------------------------------------------------

func login_with_api_key(api_key: String) -> void:
	await thread_group.exec(_login_with_api_key.bind(api_key))


func _login_with_api_key(api_key: String) -> void:
	var res := await _rpc_call("Profile.LoginWithAPIKey", {"apiKey": api_key})
	if "error" in res:
		is_logged_in = false
		emit_signal.call_deferred("logged_in", LOGIN_STATUS.INVALID_KEY, {})
		return
	profile = res.get("profile", {})
	is_logged_in = true
	emit_signal.call_deferred("logged_in", LOGIN_STATUS.OK, profile)


func login_with_password(username: String, password: String) -> void:
	await thread_group.exec(_login_with_password.bind(username, password))


func _login_with_password(username: String, password: String) -> void:
	var params := {"username": username, "password": password}
	var res := await _rpc_call("Profile.LoginWithPassword", params)
	if "error" in res:
		is_logged_in = false
		emit_signal.call_deferred("logged_in", LOGIN_STATUS.INVALID_KEY, {})
		return
	profile = res.get("profile", {})
	is_logged_in = true
	emit_signal.call_deferred("logged_in", LOGIN_STATUS.OK, profile)


## Attempts to resume a previous login using butlerd's stored credentials.
## Returns true on success (profile is set, logged_in signal emitted).
func try_saved_login() -> bool:
	return await thread_group.exec(_try_saved_login)


func _try_saved_login() -> bool:
	var list_res := await _rpc_call("Profile.List", {})
	if "error" in list_res:
		return false
	var profiles: Array = list_res.get("profiles", [])
	if profiles.is_empty():
		return false
	# Use the most recently connected profile.
	var best: Dictionary = profiles[0]
	var best_time: int = int(best.get("lastConnected", 0))
	for p in profiles:
		var t: int = int(p.get("lastConnected", 0))
		if t > best_time:
			best = p
			best_time = t
	var profile_id: int = int(best.get("id", 0))
	if profile_id == 0:
		return false
	var use_res := await _rpc_call("Profile.UseSavedLogin", {"profileId": profile_id})
	if "error" in use_res:
		logger.warn("Profile.UseSavedLogin failed: " + str(use_res["error"]))
		return false
	profile = use_res.get("profile", {})
	is_logged_in = true
	emit_signal.call_deferred("logged_in", LOGIN_STATUS.OK, profile)
	return true


## Returns every game the logged-in profile owns a download key for.
## Each item looks like: {"downloadKey": {...}, "game": {...}}
func get_owned_games() -> Array:
	return await thread_group.exec(_get_owned_games)


func _get_owned_games() -> Array:
	if not is_logged_in:
		return []
	var params := {
		"profileId": profile.get("user", {}).get("id", profile.get("id", 0)),
	}
	var res := await _rpc_call("Fetch.ProfileOwnedKeys", params)
	if "error" in res:
		logger.warn("Fetch.ProfileOwnedKeys failed: " + str(res["error"]))
		return []
	# butlerd serves Fetch.* results from its local cache by default and only
	# sets `stale` when the data hasn't been pulled from itch.io recently. On
	# a fresh database that cache is empty, so without this refresh the
	# library would stay empty until the daemon happens to refresh on its own.
	if res.get("stale", false):
		logger.info("Owned keys are stale. Refreshing from itch.io.")
		params["fresh"] = true
		res = await _rpc_call("Fetch.ProfileOwnedKeys", params)
		if "error" in res:
			logger.warn("Fetch.ProfileOwnedKeys (fresh) failed: " + str(res["error"]))
			return []
	return res.get("items", [])


## Returns every game present in any of the profile's collections, as full
## Game dicts (CollectionGame embeds the whole game object). These aren't
## necessarily owned: the user may have bookmarked free games in a collection.
func get_collection_games() -> Array:
	return await thread_group.exec(_get_collection_games)


func _get_collection_games() -> Array:
	if not is_logged_in:
		return []
	var params := {"profileId": profile.get("user", {}).get("id", profile.get("id", 0))}
	var res := await _rpc_call("Fetch.ProfileCollections", params)
	if "error" in res:
		logger.warn("Fetch.ProfileCollections failed: " + str(res["error"]))
		return []
	if res.get("stale", false):
		logger.info("Collections are stale. Refreshing from itch.io.")
		params["fresh"] = true
		res = await _rpc_call("Fetch.ProfileCollections", params)
		if "error" in res:
			logger.warn("Fetch.ProfileCollections (fresh) failed: " + str(res["error"]))
			return []
	return await _fetch_collection_games(
		profile.get("user", {}).get("id", profile.get("id", 0)),
		res.get("items", [])
	)


## Like get_collection_games, but returns {id, title, games} per collection
## so the library can filter by collection visibility settings.
func get_collection_groups() -> Array:
	return await thread_group.exec(_get_collection_groups)


func _get_collection_groups() -> Array:
	if not is_logged_in:
		return []
	var params := {"profileId": profile.get("user", {}).get("id", profile.get("id", 0))}
	var res := await _rpc_call("Fetch.ProfileCollections", params)
	if "error" in res:
		return []
	if res.get("stale", false):
		params["fresh"] = true
		res = await _rpc_call("Fetch.ProfileCollections", params)
		if "error" in res:
			return []
	var collections: Array = res.get("items", [])
	var groups := []
	for c in collections:
		var collection: Dictionary = c
		var col_id: int = collection.get("id", 0)
		var col_title: String = collection.get("title", "")
		var games := await _fetch_collection_games(
			profile.get("user", {}).get("id", profile.get("id", 0)),
			[collection]
		)
		groups.append({"id": col_id, "title": col_title, "games": games})
	return groups


## Paginates through Fetch.Collection.Games for every collection, collecting the
## embedded game objects. Each collection can span multiple pages (cursor), and
## a page served from butlerd's local cache is re-issued fresh.
func _fetch_collection_games(profile_id: int, collections: Array) -> Array:
	var games := []
	for c in collections:
		var collection: Dictionary = c
		var collection_id: int = collection.get("id", 0)
		var cursor: Variant = null
		while true:
			var params := {"profileId": profile_id, "collectionId": collection_id}
			if cursor != null:
				params["cursor"] = cursor
			var res := await _rpc_call("Fetch.Collection.Games", params)
			if "error" in res:
				logger.warn("Fetch.Collection.Games failed for collection " + str(collection_id) + ": " + str(res["error"]))
				break
			if res.get("stale", false):
				params["fresh"] = true
				res = await _rpc_call("Fetch.Collection.Games", params)
				if "error" in res:
					logger.warn("Fetch.Collection.Games (fresh) failed for collection " + str(collection_id) + ": " + str(res["error"]))
					break
			for cg in res.get("items", []):
				var collection_game: Dictionary = cg
				var game: Dictionary = collection_game.get("game", {})
				if not game.is_empty():
					games.append(game)
			cursor = res.get("nextCursor", null)
			if cursor == null:
				break
	return games


## Returns every installed game (a "cave" in butlerd terminology).
## Returns `null` when butlerd is unreachable or the call errored, so callers
## can tell a genuine "no games installed" (`[]`) apart from a transient
## failure.
func get_caves() -> Variant:
	return await thread_group.exec(_get_caves)


func _get_caves() -> Variant:
	var params := {}
	var res := await _rpc_call("Fetch.Caves", params)
	if "error" in res:
		logger.warn("Fetch.Caves failed: " + str(res["error"]))
		return null
	if res.get("stale", false):
		logger.info("Caves are stale. Refreshing from itch.io.")
		params["fresh"] = true
		res = await _rpc_call("Fetch.Caves", params)
		if "error" in res:
			logger.warn("Fetch.Caves (fresh) failed: " + str(res["error"]))
			return null
	return res.get("items", [])


## Ensures at least one install location exists and returns its ID, creating
## one at the persistent user://butler/games path on first run. Older builds
## created the location under user://plugins/itch/games, which OGPU trashes on
## every plugin update; existing locations pointing there are left alone (their
## orphaned caves are cleaned up by [method _resolve_cave]) but never reused.
func _ensure_install_location() -> String:
	if _install_location_id != "":
		return _install_location_id

	var target_path := ProjectSettings.globalize_path(games_dir)
	var res := await _rpc_call("Install.Locations.List", {})
	var locations: Array = res.get("installLocations", [])
	for loc in locations:
		var location: Dictionary = loc
		if location.get("path", "") == target_path:
			_install_location_id = location.get("id", "")
			return _install_location_id

	DirAccess.make_dir_recursive_absolute(target_path)
	# Let butlerd generate the id so we never collide with a stale location
	# that happens to carry the same id under a different path.
	var add_res := await _rpc_call("Install.Locations.Add", {"path": target_path})
	if "error" in add_res:
		logger.error("Install.Locations.Add failed: " + str(add_res["error"]))
		return ""
	_install_location_id = add_res.get("installLocation", {}).get("id", "")
	return _install_location_id


## Returns every configured install location, each as:
## {"id": "...", "path": "...", "sizeInfo": {"installedSize", "freeSize", "totalSize"}}
## The preferred location (user://butler/games) is always listed first, so it
## becomes the default choice in OGPU's install-location picker (the picker
## focuses the first card). Locations pointing inside the plugin's own
## directory are dropped: games installed there get wiped on every plugin
## update (the plugin dir is moved to trash by OGPU when a new zip is loaded).
func get_install_locations() -> Array:
	return await thread_group.exec(_get_install_locations)


func _get_install_locations() -> Array:
	var res := await _rpc_call("Install.Locations.List", {})
	if "error" in res:
		logger.warn("Install.Locations.List failed: " + str(res["error"]))
		return []
	var locations: Array = res.get("installLocations", [])
	var preferred_path := ProjectSettings.globalize_path(games_dir)
	var plugin_dir := ProjectSettings.globalize_path("user://plugins/itch")
	var preferred: Array = []
	var others: Array = []
	for raw in locations:
		var location: Dictionary = raw
		var path: String = location.get("path", "")
		if path.begins_with(plugin_dir):
			logger.info("Ignoring install location inside the plugin directory: " + path)
			continue
		if path == preferred_path:
			preferred.append(location)
		else:
			others.append(location)
	return preferred + others


## Returns the uploads butlerd considers compatible with this machine for the
## given game. Each entry is a full Upload dict (id, filename, displayName,
## size, type, platforms, ...).
func get_compatible_uploads(game_id: int) -> Array:
	return await thread_group.exec(_get_compatible_uploads.bind(game_id))


func _get_compatible_uploads(game_id: int) -> Array:
	var params := {"gameId": game_id, "compatible": true}
	var res := await _rpc_call("Fetch.GameUploads", params)
	if "error" in res:
		logger.warn("Fetch.GameUploads failed for game " + str(game_id) + ": " + str(res["error"]))
		return []
	# Uploads are cached by butlerd just like owned keys, so retry fresh when
	# the cached result is stale, otherwise a cold database would report every
	# game as having no compatible upload.
	if res.get("stale", false):
		logger.info("Uploads are stale. Refreshing from itch.io.")
		params["fresh"] = true
		res = await _rpc_call("Fetch.GameUploads", params)
		if "error" in res:
			logger.warn("Fetch.GameUploads (fresh) failed for game " + str(game_id) + ": " + str(res["error"]))
			return []
	return res.get("uploads", [])


## Looks for updates to installed caves. When cave_ids is empty, butlerd checks
## every cave and respects per-cave snooze (matching the official itch.io app).
## Returns the list of GameUpdate dicts (each with a `caveId`).
func check_updates(cave_ids: Array = []) -> Array:
	return await thread_group.exec(_check_updates.bind(cave_ids))


func _check_updates(cave_ids: Array) -> Array:
	var params := {}
	if not cave_ids.is_empty():
		params["caveIds"] = cave_ids
	var res := await _rpc_call("CheckUpdate", params)
	if "error" in res:
		logger.warn("CheckUpdate failed: " + str(res["error"]))
		return []
	return res.get("updates", [])


## Installs (or updates, if a caveId is given) the given game.
## `options` may carry:
##   - "install_location_id": the InstallLocation id to install into (fresh installs)
##   - "upload": the upload filename/displayName chosen by the user (defaults to
##     butlerd's first compatible upload otherwise)
## Returns true when the operation finished without errors.
func install(game: Dictionary, cave_id: String = "", options: Dictionary = {}) -> bool:
	return await thread_group.exec(_install.bind(game, cave_id, options))


## Reconciles an install/update request against what butlerd actually has
## installed. OGPU moves the entire extracted plugin directory (including any
## games previously installed under user://plugins/itch/games) to the trash
## whenever the plugin is updated, leaving "orphan" caves in butler.db that
## point at folders which no longer exist. Those caves make a fresh install
## fail with butlerd's "That upload is already installed!" error while the
## game can't actually be launched, so we detect them here and uninstall them
## first (the folder is already gone, so nothing is lost).
##
## Returns the cave id to pass to Install.Queue: a valid existing cave when
## one is found (so the request becomes an update), or an empty string when
## the game needs a fresh install.
func _resolve_cave(game: Dictionary, cave_id: String) -> String:
	var game_id: int = game.get("id", 0)
	var res := await _rpc_call("Fetch.Caves", {"filters": {"gameId": game_id}})
	if "error" in res:
		logger.warn("Fetch.Caves failed while reconciling game " + str(game_id) + ": " + str(res["error"]))
		return cave_id

	var matched: Array = []
	for c in res.get("items", []):
		var cave: Dictionary = c
		if cave_id == "" or cave.get("id", "") == cave_id:
			matched.append(cave)
	if matched.is_empty():
		return cave_id

	for cave in matched:
		var install_folder: String = cave.get("installInfo", {}).get("installFolder", "")
		if DirAccess.dir_exists_absolute(install_folder):
			return cave.get("id", cave_id)

	for cave in matched:
		var orphan_id: String = cave.get("id", "")
		logger.info("Uninstalling orphaned cave " + orphan_id + " for game " + str(game_id) + " (install folder is missing)")
		await _rpc_call("Uninstall.Perform", {"caveId": orphan_id})
	return ""


## Asks butlerd to queue an install of `upload` for `game`.
func _queue_install(
	game: Dictionary, upload: Dictionary, location_id: String, reason: String, cave_id: String
) -> Dictionary:
	var queue_params := {
		"game": game,
		"upload": upload,
		"installLocationId": location_id,
		"reason": reason,
		"queueDownload": true,
	}
	if cave_id != "":
		queue_params["caveId"] = cave_id
	return await _rpc_call("Install.Queue", queue_params)


## itch.io's answer when butlerd asks for a build without a download key the
## account holds: the game is in the library but butlerd never learned its key.
const ITCHIO_MISSING_DOWNLOAD_KEY_ERROR := "you must buy this game to download"


func _is_missing_download_key_error(queue_res: Dictionary) -> bool:
	if not queue_res.has("error"):
		return false
	return ITCHIO_MISSING_DOWNLOAD_KEY_ERROR in _rpc_error_message(queue_res)


## butlerd only learns the itch.io download keys from Fetch.ProfileOwnedKeys.
## When that call never ran (or ran long ago), butlerd queues installs without
## credentials and itch.io rejects them, so pull a fresh copy of the keys.
func _sync_owned_keys() -> void:
	if not is_logged_in:
		return
	var profile_id: int = int(profile.get("user", {}).get("id", profile.get("id", 0)))
	if profile_id == 0:
		return
	var res := await _rpc_call("Fetch.ProfileOwnedKeys", {"profileId": profile_id, "fresh": true})
	if "error" in res:
		logger.warn("Fetch.ProfileOwnedKeys (fresh) failed: " + str(res["error"]))


func _install(game: Dictionary, cave_id: String, options: Dictionary) -> bool:
	var game_id: int = game.get("id", 0)
	cave_id = await _resolve_cave(game, cave_id)
	var reason := "update" if cave_id != "" else "install"

	var uploads := await _get_compatible_uploads(game_id)
	if uploads.is_empty():
		logger.error("No compatible uploads found for game id " + str(game_id))
		emit_signal.call_deferred("app_installed", cave_id, false)
		emit_signal.call_deferred(
			"install_failed",
			cave_id if cave_id != "" else str(game_id),
			game.get("title", "Unknown itch.io game"),
			"No compatible upload found for this game."
		)
		return false
	var upload: Dictionary = uploads[0]
	# Respect the upload the user picked in the install-options dialog, when
	# one was offered (the game has more than one compatible upload).
	var chosen_upload: Variant = options.get("upload", null)
	if chosen_upload != null:
		var picked := _pick_upload(uploads, str(chosen_upload))
		if picked.is_empty():
			logger.warn("Selected upload '" + str(chosen_upload) + "' not found. Using default upload.")
		else:
			upload = picked

	var location_id: String = options.get("install_location_id", "") as String
	if location_id == "":
		location_id = await _ensure_install_location()

	var queue_res := await _queue_install(game, upload, location_id, reason, cave_id)
	# butlerd sends the build request without an itch.io download key when its
	# database doesn't hold one for this game, and itch.io answers "you must buy
	# this game to download". Sync the owned keys and give it one more try.
	if _is_missing_download_key_error(queue_res):
		logger.info("Install of game " + str(game_id) + " was rejected for lack of a download key. Syncing owned keys and retrying.")
		await _sync_owned_keys()
		queue_res = await _queue_install(game, upload, location_id, reason, cave_id)
	if "error" in queue_res:
		var queue_error: String = _rpc_error_message(queue_res)
		logger.error("Install.Queue failed: " + str(queue_res["error"]))
		logger.error("Install.Queue failed (message): " + queue_error)
		emit_signal.call_deferred("app_installed", cave_id, false)
		emit_signal.call_deferred(
			"install_failed",
			cave_id if cave_id != "" else str(game_id),
			game.get("title", "Unknown itch.io game"),
			queue_error
		)
		return false

	var task_id: String = queue_res.get("id", "")
	var staging_folder: String = queue_res.get("stagingFolder", "")

	var on_notification := func(method: String, params: Variant) -> void:
		if method != "Progress":
			return
		var p: Dictionary = params
		var progress: float = p.get("progress", 0.0)
		install_progressed.emit.call_deferred(
			cave_id if cave_id != "" else str(game_id),
			int(progress * 100),
			100
		)
	rpc_notification.connect(on_notification)

	var perform_res := await _rpc_call("Install.Perform", {"id": task_id, "stagingFolder": staging_folder})

	# butler can panic ("runtime error: slice bounds out of range") when
	# resuming a download whose checkpoint recorded an invalid offset (a
	# stale/corrupt partial download in the staging folder). Clear just the
	# download checkpoint and partial copy and retry once so the download
	# starts from scratch. The rest of the staging folder (the queue metadata
	# that holds the game/upload) must be left intact, or butler fails the
	# retry with "Corrupted download info (missing game)".
	if "error" in perform_res and staging_folder != "" and _is_download_resume_panic(perform_res.get("error")):
		logger.warn("butler panicked resuming download; clearing download state and retrying once")
		_clear_download_state(staging_folder)
		perform_res = await _rpc_call("Install.Perform", {"id": task_id, "stagingFolder": staging_folder})
		if "error" in perform_res:
			logger.error("Install.Perform retry after clearing download state also failed: " + str(perform_res.get("error")))
		else:
			logger.info("Install.Perform recovered after clearing corrupt download state")

	rpc_notification.disconnect(on_notification)

	var success := not ("error" in perform_res)
	if not success:
		var error_msg: String = _rpc_error_message(perform_res)
		logger.error("Install.Perform failed: " + str(perform_res.get("error")))
		logger.error("Install.Perform failed (message): " + error_msg)
		emit_signal.call_deferred(
			"install_failed",
			cave_id if cave_id != "" else str(game_id),
			game.get("title", "Unknown itch.io game"),
			error_msg
		)

	if reason == "update":
		emit_signal.call_deferred("app_updated", cave_id, success)
	else:
		emit_signal.call_deferred("app_installed", cave_id, success)
	return success


## Extracts a concise, human-readable message from a butlerd RPC error dict
## (the "message" field, falling back to the whole error when absent).
func _rpc_error_message(res: Dictionary) -> String:
	var error: Variant = res.get("error", null)
	if error == null:
		return "Unknown error"
	if error is Dictionary:
		var msg: String = error.get("message", "")
		if msg != "":
			return msg
	return str(error)


## True when butler panicked resuming a partially-downloaded game. The panic
## message ("runtime error: slice bounds out of range") comes from savior's
## seekSource when a corrupt checkpoint leaves the read offset beyond the
## section size, so the only reliable way forward is a fresh download.
func _is_download_resume_panic(error: Variant) -> bool:
	return str(error).contains("slice bounds out of range")


## Removes butler's download checkpoint file(s) and the partially-downloaded
## install source from the staging folder, so the next Install.Perform attempt
## downloads from scratch. Only the *download* state is cleared: the queue
## metadata butler persisted alongside it (which holds the game/upload the
## install was queued with) must survive, or the retry fails with "Corrupted
## download info (missing game)".
func _clear_download_state(staging_folder: String) -> void:
	if staging_folder == "" or not DirAccess.dir_exists_absolute(staging_folder):
		return
	var dir := DirAccess.open(staging_folder)
	if dir == null:
		logger.warn("Could not open staging folder to clear download state: " + staging_folder)
		return
	dir.list_dir_begin()
	var fname := dir.get_next()
	while fname != "":
		if fname.begins_with("downsource-") and fname.ends_with("-state.dat"):
			DirAccess.remove_absolute(staging_folder + "/" + fname)
			logger.info("Removed corrupt download checkpoint: " + fname)
		fname = dir.get_next()
	dir.list_dir_end()
	# The partially-downloaded copy butler was resuming from also has to go.
	if DirAccess.dir_exists_absolute(staging_folder + "/install-source"):
		_wipe_dir(staging_folder + "/install-source")
		logger.info("Cleared partial download copy in staging install-source")


## Recursively removes the contents of a directory, leaving the directory
## itself in place.
func _wipe_dir(path: String) -> void:
	if path == "" or not DirAccess.dir_exists_absolute(path):
		return
	var dir := DirAccess.open(path)
	if dir == null:
		logger.warn("Could not open staging folder to clear: " + path)
		return
	dir.list_dir_begin()
	var fname := dir.get_next()
	while fname != "":
		if fname != "." and fname != "..":
			var full: String = path + "/" + fname
			if dir.current_is_dir():
				_wipe_dir(full)
				DirAccess.remove_absolute(full)
			else:
				DirAccess.remove_absolute(full)
		fname = dir.get_next()
	dir.list_dir_end()


## Returns the upload whose displayName or filename matches the user's pick
## from the install-options dropdown, or {} when nothing matches. The dropdown
## renders each upload as "<name>  (<size>)", so the bare name prefix counts as
## a match too.
static func _pick_upload(uploads: Array, chosen: String) -> Dictionary:
	for u in uploads:
		var upload: Dictionary = u
		if upload_name_matches(upload, chosen):
			return upload
	return {}


## True when `chosen` (an install-options dropdown value) names `upload`.
static func upload_name_matches(upload: Dictionary, chosen: String) -> bool:
	for name_key in ["displayName", "filename"]:
		var name: String = upload.get(name_key, "")
		if name == "":
			continue
		if chosen == name or chosen.begins_with(name + UPLOAD_SIZE_SEPARATOR):
			return true
	return false


## Uninstalls the given cave. Returns true on success.
func uninstall(cave_id: String) -> bool:
	return await thread_group.exec(_uninstall.bind(cave_id))


func _uninstall(cave_id: String) -> bool:
	var res := await _rpc_call("Uninstall.Perform", {"caveId": cave_id})
	var success := not ("error" in res)
	if not success:
		logger.error("Uninstall.Perform failed: " + str(res.get("error")))
	emit_signal.call_deferred("app_uninstalled", cave_id, success)
	return success


## Launches the given cave. This call blocks (on the background thread) for
## the entire lifetime of the game, so callers should not await it directly
## from a UI-facing context; connect to `launch_exited` instead if needed.
func launch(cave_id: String) -> void:
	await thread_group.exec(_launch.bind(cave_id))


func _launch(cave_id: String) -> void:
	var prereqs_dir := ProjectSettings.globalize_path("user://butler/prereqs")
	DirAccess.make_dir_recursive_absolute(prereqs_dir)

	# `Launch` may send interactive requests mid-call (PickManifestAction,
	# AcceptLicense, ShellLaunch, HTMLLaunch). This client answers the common
	# ones with sane non-interactive defaults; anything fancier (HTML5
	# games, EULAs) is a good next step but out of scope for a first pass.
	var on_notification := func(method: String, params: Variant) -> void:
		match method:
			"PrereqsStarted":
				logger.info("Installing prerequisites for cave " + cave_id)
			"LaunchRunning":
				logger.info("Game running: " + cave_id)
	rpc_notification.connect(on_notification)

	var res := await _rpc_call("Launch", {"caveId": cave_id, "prereqsDir": prereqs_dir})

	rpc_notification.disconnect(on_notification)

	if "error" in res:
		logger.error("Launch failed: " + str(res["error"]))
	launch_exited.emit.call_deferred(cave_id)


## Logs out the current profile by clearing butlerd's stored credentials
## and resetting the client state. For API key logins the caller should
## also clear the saved key via settings_manager.
func logout() -> void:
	await thread_group.exec(_logout)


func _logout() -> void:
	if not is_logged_in:
		return
	var profile_id: int = int(profile.get("user", {}).get("id", profile.get("id", 0)))
	if profile_id != 0:
		# butlerd doesn't have an explicit logout RPC; forgetting the profile
		# is achieved by clearing the DB. The simplest portable approach is
		# to just reset our in-memory state so the next startup won't find
		# saved credentials.
		logger.info("Logging out profile " + str(profile_id))
	is_logged_in = false
	profile = {}
	emit_signal.call_deferred("logged_in", LOGIN_STATUS.FAILED, {})


func _exit_tree() -> void:
	if socket:
		socket.disconnect_from_host()
	if not proc:
		return
	proc.stop()