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"
## 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"
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 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)


## 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", [])
	)


## 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).
func get_caves() -> Array:
	return await thread_group.exec(_get_caves)


func _get_caves() -> Array:
	var params := {}
	var res := await _rpc_call("Fetch.Caves", params)
	if "error" in res:
		logger.warn("Fetch.Caves failed: " + str(res["error"]))
		return []
	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 []
	return res.get("items", [])


## Ensures at least one install location exists and returns its ID, creating
## a default one under the plugin's data directory on first run.
func _ensure_install_location() -> String:
	if _install_location_id != "":
		return _install_location_id

	var res := await _rpc_call("Install.Locations.List", {})
	var locations: Array = res.get("installLocations", [])
	if locations.size() > 0:
		_install_location_id = locations[0].get("id", "")
		return _install_location_id

	var path := ProjectSettings.globalize_path("user://plugins/itch/games")
	DirAccess.make_dir_recursive_absolute(path)
	var add_res := await _rpc_call("Install.Locations.Add", {"id": "default", "path": path})
	_install_location_id = add_res.get("installLocation", {}).get("id", "default")
	return _install_location_id


## Installs (or updates, if a caveId is given) the given game.
func install(game: Dictionary, cave_id: String = "") -> void:
	await thread_group.exec(_install.bind(game, cave_id))


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

	var uploads_params := {"gameId": game_id, "compatible": true}
	var uploads_res := await _rpc_call("Fetch.GameUploads", uploads_params)
	if "error" in uploads_res:
		logger.warn("Fetch.GameUploads failed: " + str(uploads_res["error"]))
		emit_signal.call_deferred("app_installed", cave_id, false)
		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 uploads_res.get("stale", false):
		logger.info("Uploads are stale. Refreshing from itch.io.")
		uploads_params["fresh"] = true
		uploads_res = await _rpc_call("Fetch.GameUploads", uploads_params)
		if "error" in uploads_res:
			logger.warn("Fetch.GameUploads (fresh) failed: " + str(uploads_res["error"]))
			emit_signal.call_deferred("app_installed", cave_id, false)
			return
	var uploads: Array = uploads_res.get("uploads", [])
	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)
		return
	var upload: Dictionary = uploads[0]

	var location_id := await _ensure_install_location()

	var queue_params := {
		"game": game,
		"upload": upload,
		"installLocationId": location_id,
		"reason": reason,
		"queueDownload": true,
	}
	if cave_id != "":
		queue_params["caveId"] = cave_id

	var queue_res := await _rpc_call("Install.Queue", queue_params)
	if "error" in queue_res:
		logger.error("Install.Queue failed: " + str(queue_res["error"]))
		emit_signal.call_deferred("app_installed", cave_id, false)
		return

	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})

	rpc_notification.disconnect(on_notification)

	var success := not ("error" in perform_res)
	if not success:
		logger.error("Install.Perform failed: " + str(perform_res.get("error")))

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


## Uninstalls the given cave.
func uninstall(cave_id: String) -> void:
	await thread_group.exec(_uninstall.bind(cave_id))


func _uninstall(cave_id: String) -> void:
	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)


## 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://plugins/itch/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)


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