/src/Plac/Connection.vala at 03aef5b83ae6eb0a66f45fc03fea09b47cd2ac74

// Copyright 2025 Shota FUJI
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

namespace Plac {
	public errordomain ConnectError {
		SERVER_NOT_FOUND,
		INVALID_RESPONSE,
		SERVER_MISMATCH,
		NETWORK_ERROR,
	}

	public errordomain RequestError {
		NETWORK_ERROR,
		UNEXPECTED_RESPONSE,
	}

	/**
	 * Roon forces application-layer ping/pong instead of WebSocket's ping/pong frame.
	 */
	private static void handle_ping(Soup.WebsocketConnection conn, int _type, GLib.Bytes bytes) {
		var message = (string) bytes.get_data();

		Moo.Metadata req_meta;
		Moo.Headers req_headers;
		try {
			req_meta = new Moo.Metadata.from_string(message);
			req_headers = new Moo.Headers.from_string(message, req_meta);
		} catch (GLib.Error error) {
			GLib.log("Plac", LEVEL_WARNING, "Got invalid MOO message: %s", error.message);
			return;
		}

		if (req_meta.service != "com.roonlabs.ping:1/ping") {
			return;
		}

		var res_meta = new Moo.Metadata("COMPLETE", "Success");
		var res_headers = new Moo.Headers();
		res_headers.write("Request-Id", @"$(req_headers.request_id)");

		conn.send_binary(@"$res_meta$res_headers".data);
	}

	private static async Response send_request(
		Soup.WebsocketConnection conn,
		string service,
		uint64 request_id,
		string? json_body = null
	) throws RequestError {
		SourceFunc callback = send_request.callback;
		Response? resp = null;

		var message_handler_id = conn.message.connect((c, type, bytes) => {
			var message = (string) bytes.get_data();

			Moo.Metadata res_meta;
			Moo.Headers res_headers;
			try {
				res_meta = new Moo.Metadata.from_string(message);
				res_headers = new Moo.Headers.from_string(message, res_meta);
			} catch (GLib.Error error) {
				GLib.log("Plac", LEVEL_WARNING, "Got invalid MOO message: %s", error.message);
				return;
			}

			if (res_meta.verb == "REQUEST") {
				// Application ping message, skip.
				return;
			}

			if (res_headers.request_id != request_id) {
				return;
			}

			resp = new Response(res_meta, res_headers, message);
			callback();
		});

		var error_handler_id = conn.error.connect((c, error) => {
			GLib.log("Plac", LEVEL_ERROR, "Failed to receive MOO message: %s", error.message);
			callback();
		});

		var req_meta = new Moo.Metadata("REQUEST", service);
		var req_headers = new Moo.Headers();
		req_headers.write("Request-Id", @"$request_id");

		if (json_body != null) {
			req_headers.write("Content-Type", "application/json");
			req_headers.write("Content-Length", @"$(json_body.length)");

			conn.send_binary(@"$req_meta$req_headers$json_body".data);
		} else {
			conn.send_binary(@"$req_meta$req_headers".data);
		}

		yield;

		conn.disconnect(message_handler_id);
		conn.disconnect(error_handler_id);

		if (resp == null) {
			throw new RequestError.NETWORK_ERROR("Server not found.");
		}

		return resp;
	}

	private static async Roon.Registry.Info.Response get_registry_info(
		Soup.WebsocketConnection conn, uint64 request_id = 1
	) throws ConnectError {
		Response resp;
		try {
			resp = yield send_request(conn, "com.roonlabs.registry:1/info", request_id);
		} catch (RequestError error) {
			if (error is RequestError.NETWORK_ERROR) {
				throw new ConnectError.NETWORK_ERROR(error.message);
			}

			throw new ConnectError.INVALID_RESPONSE(error.message);
		}

		if (resp.meta.service != "Success") {
			throw new ConnectError.INVALID_RESPONSE("Server error.");
		}

		if (resp.headers.content_type != "application/json") {
			throw new ConnectError.INVALID_RESPONSE("Not a JSON body.");
		}

		Roon.Registry.Info.Response info;
		try {
			var body = new Moo.JsonBody.from_string(resp.message, resp.headers);

			info = new Roon.Registry.Info.Response.from_json(body.data);
		} catch (GLib.Error error) {
			throw new ConnectError.INVALID_RESPONSE(error.message);
		}

		return info;
	}

	private static async Soup.WebsocketConnection ws_connect_async(
		GLib.InetSocketAddress address,
		uint16 port,
		GLib.Cancellable? cancellable = null
	) throws GLib.Error {
		var session_singleton = new Session();
		var session = session_singleton.session;

		var url = GLib.Uri.build(
			NONE, "ws:", null, address.address.to_string(), port,
			"/api", null, null
		);
		var msg = new Soup.Message.from_uri("GET", url);

		// Roon API does not specify WebSocket subprotocols.
		Soup.WebsocketConnection conn = yield session.websocket_connect_async(
			msg, null, null, GLib.Priority.DEFAULT, cancellable
		);

		conn.message.connect(handle_ping);

		return conn;
	}

	public class Connection : Object {
		/**
		 * Server closed a connection, or network error terminated a connection.
		 */
		public signal void closed();

		private size_t request_id = 1;

		public Soup.WebsocketConnection conn { get; construct; }
		public Server server { get; construct; }
		public string token { get; construct; }

		public Connection(Soup.WebsocketConnection conn, Server server, string token) {
			Object(conn: conn, server: server, token: token);
		}

		construct {
			conn.closed.connect(() => {
				this.closed();
			});
		}

		public static async Connection connect_async(
			GLib.InetSocketAddress address, uint16 http_port, string server_id,
			Roon.Registry.Register.Request req,
			GLib.Cancellable? cancellable = null
		) throws ConnectError {
			Soup.WebsocketConnection conn;
			try {
				conn = yield ws_connect_async(address, http_port, cancellable);
			} catch (GLib.Error error) {
				GLib.log("Plac", LEVEL_DEBUG, "Handshake error");
				throw new ConnectError.NETWORK_ERROR(error.message);
			}

			Roon.Registry.Info.Response info;
			try {
				info = yield get_registry_info(conn);
			} catch (ConnectError error) {
				throw new ConnectError.SERVER_NOT_FOUND(error.message);
			}

			if (info.core_id != server_id) {
				throw new ConnectError.SERVER_MISMATCH("Server at the location has different server ID.");
			}

			GLib.log("Plac", LEVEL_DEBUG, "Confirmed the Roon server is running. Registering extension.");

			Response register_resp;
			try {
				register_resp = yield send_request(conn, "com.roonlabs.registry:1/register", 2, req.to_json());
			} catch (RequestError error) {
				if (error is RequestError.NETWORK_ERROR) {
					throw new ConnectError.NETWORK_ERROR(error.message);
				}

				throw new ConnectError.INVALID_RESPONSE(error.message);
			}

			if (register_resp.meta.service != "Registered") {
				throw new ConnectError.INVALID_RESPONSE(@"$(register_resp.meta.service)");
			}

			Roon.Registry.Register.Response result;
			try {
				var body = new Moo.JsonBody.from_string(register_resp.message, register_resp.headers);

				result = new Roon.Registry.Register.Response.from_json(body.data);
			} catch (GLib.Error error) {
				throw new ConnectError.INVALID_RESPONSE(error.message);
			}

			GLib.log("Plac", LEVEL_DEBUG, @"Registered Roon extension at $(address.address):$(http_port)");

			return new Connection(
				conn,
				new Server(address, server_id, info.display_version, info.display_name, http_port),
				result.token
			);
		}

		public async Response json_request(string service, string body) throws RequestError {
			return yield send_request(conn, service, request_id ++, body);
		}
	}
}