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"
const butler_dir := "user://plugins/itch/assets/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_bin := "/".join([ProjectSettings.globalize_path(butler_dir), "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([ProjectSettings.globalize_path("user://plugins/itch/assets"), "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()


## 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 = 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 = 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())


## 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}
	_send_line(JSON.stringify(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(fresh: bool = false) -> Array:
	return await thread_group.exec(_get_owned_games.bind(fresh))


func _get_owned_games(fresh: bool) -> Array:
	if not is_logged_in:
		return []
	var res := await _rpc_call("Fetch.ProfileOwnedKeys", {
		"profileId": profile.get("user", {}).get("id", profile.get("id", 0)),
		"fresh": fresh,
	})
	if "error" in res:
		logger.warn("Fetch.ProfileOwnedKeys failed: " + str(res["error"]))
		return []
	return res.get("items", [])


## 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 res := await _rpc_call("Fetch.Caves", {})
	if "error" in res:
		logger.warn("Fetch.Caves 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_res := await _rpc_call("Fetch.GameUploads", {"gameId": game_id, "compatible": true})
	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()