miggor-StreamGraph/rpc_renderer/rpc_renderer.gd

# (c) 2023-present Eroax
# (c) 2023-present Yagich
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
extends Node
class_name RPCRenderer
## A WebSocket API server.
##
## A renderer that exposes a subset of the core API to remote clients using WebSocket.

const SCOPES_DIR := "res://rpc_renderer/scopes/"

@export var default_port := 6907

var clients: Dictionary # Dictionary[int -> id, Client]

var _ws: WebSocketServer = WebSocketServer.new()

var scopes: Array[RPCScope]
var system_scope: RPCScopeSystem
var request_schema: Zodot


func load_scopes() -> void:
	var d := DirAccess.open(SCOPES_DIR)
	d.list_dir_begin()
	var current := d.get_next()
	while current != "":
		if !d.current_is_dir():
			var scope = load(SCOPES_DIR.path_join(current)).new() as RPCScope
			scopes.append(scope)
		
		current = d.get_next()
	
	for scope in scopes:
		if scope.name == "system":
			system_scope = scope

		scope.event.connect(
			func(event: RPCEvent):
				if event.to_peer != 0:
					send_frame(event.to_peer, event)
					return
				for client_id in clients:
					var client: Client = get_client(client_id)
					if not client.subscriptions.has(event.scope):
						continue
					if event.type in (client.subscriptions[event.scope] as Array):
						send_frame(client_id, event)
		)
		
		scope.response.connect(
			func(response: RPCRequestResponse):
				send_frame(response.peer_id, response)
				if response.event_counterpart != null:
					var event := response.event_counterpart
					for client_id in clients:
						if client_id == response.peer_id:
							continue
						var client: Client = get_client(client_id)
						if event.name in (client.subscriptions[event.scope] as Array):
							send_frame(client_id, event)
		)


func build_schema() -> void:
	var scope_names = scopes.map(
		func(scope: RPCScope):
			return scope.name
			)

	var scopes_schema: Array[Zodot]
	
	scopes_schema.assign(scope_names.map(
			func(scope_name: String):
				return Z.literal(scope_name)
				))
	
	request_schema = Z.schema({
		"request": Z.schema({
			"id": Z.string(),
			"scope": Z.union(scopes_schema),
			"operation": RPCOperation.schema(),
			"keep": Z.dictionary().nullable(),
		})
	})


func _ready() -> void:
	load_scopes()
	build_schema()
	
	add_child(_ws)
	_ws.client_connected.connect(_on_ws_client_connected)
	_ws.client_disconnected.connect(_on_ws_client_disconnected)
	_ws.message_received.connect(_on_ws_message)


func listen(port := default_port) -> void:
	if _ws.listen(port) != OK:
		pass


func stop() -> void:
	_ws.stop()


func send_frame(peer_id: int, frame: RPCFrame) -> void:
	_ws.send(peer_id, JSON.stringify(frame.to_dict(), "", false))


func get_client(peer_id: int) -> Client:
	return clients.get(peer_id)


func drop_client(peer_id: int, reason: String) -> void:
	var disconnect_data := {
		"disconnect": {
			"message": "You have been disconnected: %s" % reason
		}
	}
	
	_ws.send(peer_id, JSON.stringify(disconnect_data, "", false))
	_ws.peers.erase(peer_id)
	clients.erase(peer_id)


func _on_ws_message(peer_id: int, message: Variant) -> void:
	if not message is String:
		return
	
	var json := JSON.new()
	var err := json.parse(message)
	if err != OK:
		send_frame(peer_id, RPCError.new(json.get_error_message()))
		return
	var result = request_schema.parse(json.get_data())
	if not result.ok():
		send_frame(peer_id, RPCError.new(result.error))
		return
	
	var req := RPCRequest.from_dict(result.data, get_client(peer_id))
	if not get_client(peer_id).identified:
		if not (req.scope == "system" and req.operation.type == "identify"):
			drop_client(peer_id, "You must identify your client first.")
			return
		
		system_scope.identify(req)
		return

	var scope_idx := -1
	for i in scopes.size():
		var scope := scopes[i]
		if scope.name == req.scope:
			scope_idx = i
			break

	if scope_idx == -1:
		return # TODO: error
	
	var scope := scopes[scope_idx]
	if scope.can_handle_request(req):
		scope.handle_request(req)


func _on_ws_client_connected(peer_id: int) -> void:
	var c := Client.new()
	c.id = peer_id
	clients[peer_id] = c
	RPCSignalLayer.signals.client_connected.emit(c)


func _on_ws_client_disconnected(peer_id: int) -> void:
	clients.erase(peer_id)


class Client:
	var id: int
	var identified: bool = false
	var subscriptions: Dictionary = {}