Riddle/src/lib/Client.vala

using Invercargill;

namespace Riddle {

    public class Client : Object {

        private Invercargill.Sequence<PeerDiscoverer> discoverers = new Invercargill.Sequence<PeerDiscoverer> ();
        private Invercargill.Sequence<InetSocketAddress> servers = new Invercargill.Sequence<InetSocketAddress>();
        private Invercargill.Sequence<Message> join_messages = new Invercargill.Sequence<Message>();
        private InetSocketAddress? self_server = null;

        public Client() {
            discoverers.add (new KnownHostDiscoverer());
            discoverers.add (new LanDiscoverer());
            setup();
        }

        public Client.with_server(InetSocketAddress server_address) {
            var lan_discoverer = new LanDiscoverer();
            discoverers.add (new KnownHostDiscoverer());
            discoverers.add (lan_discoverer);
            self_server = server_address;

            setup();

            lan_discoverer.advertise(self_server);
        }

        private void setup() {
            foreach (var discoverer in discoverers) {
                discoverer.peer_discovered.connect(new_server_found);
                discoverer.begin();
            }
        }

        private void new_server_found(InetSocketAddress address) {
            if(self_server != null && server_equals(address, self_server)) {
                // Ignore self_server address
                return;
            }

            if(servers.any(s => server_equals(s, address))) {
                // Ignore already discovered server
                return;
            }

            Enumerable<Message> j_msgs;
            lock(join_messages) {
                servers.add(address);
                j_msgs = join_messages.to_sequence();
            }

            foreach (var message in j_msgs) {
                raw_request(message, single(address), 5000, MessageType.OK);
            }
        }

        public int join(string group, uint port) {
            lock(join_messages) {
                var msg = new Message(MessageType.JOIN, new string[] { group, port.to_string() }, new string[0]);
                var responses = raw_request(msg, servers, 10000, MessageType.OK);
                join_messages.add(msg);
                return responses.where(r => r.message_type == MessageType.OK).count();
            }
        }

        public int leave(string group, uint port) {
            lock(join_messages) {
                join_messages = join_messages.where(m => m.arguments[0] != group && m.arguments[1] != port.to_string()).to_sequence();
            }
            var msg = new Message(MessageType.LEAVE, new string[] { group, port.to_string() }, new string[0]);
            var responses = raw_request(msg, servers, 10000, MessageType.OK);
            return responses.where(r => r.message_type == MessageType.OK).count();
        }

        public int propogate(NameInfo info, string? idempotency_token = null) {
            var token = idempotency_token ?? GLib.Uuid.string_random();
            var msg = new Message(MessageType.PROPOGATE, new string[] { info.name, "10", token }, new string[] { info.get_encoded() });
            var responses = raw_request(msg, servers, 10000, MessageType.OK);
            return responses.where(r => r.message_type == MessageType.OK).count();
        }

        public Enumerable<InetSocketAddress> who_is_in(string group) {
            var who_in = new Message(MessageType.WHO_IN, new string[] { group }, new string[0]);
            var responses = raw_request(who_in, servers, 2000, MessageType.ANSWER);
            return responses.select_many<InetSocketAddress>(r => ate(r.items).select<InetSocketAddress>(i => parse_address(i)));
        }

        public Enumerable<NameInfo> who_is(string domain) {
            var who_is = new Message(MessageType.WHO_IS, new string[] { domain }, new string[0]);
            var responses = raw_request(who_is, servers, 2000, MessageType.ANSWER);

            if(domain.has_suffix(".rns")) {
                return responses.select_many<NameInfo>(r => ate(r.items).select<NameInfo>(i => new DecentralisedNameInfo.from_string(i)));
            }

            return responses.select_many<NameInfo>(r => ate(r.items).select<NameInfo>(i => new CertifiedNameInfo.from_string(i)));
        }

        public int riddle(RiddleEnvelope riddle, Enumerable<InetSocketAddress> servers) {
            var msg = riddle.to_message();
            var responses = raw_request(msg, servers, 10000, MessageType.OK);
            return responses.where(r => r.message_type == MessageType.OK).count();
        }

        public Enumerable<Message> raw_request(Message msg, Enumerable<InetSocketAddress> servers, int64 timeout = 10000, MessageType? filter = null) {
            var manager = new RequestManager(servers);
            manager.start_request(msg);
            return manager.get_responses(timeout, filter);
        }

        public InetSocketAddress? callback(SolutionEnvelope solution, InetSocketAddress server, uint8[] author_signing_key, uint8[] reply_public_key, uint8[] reply_secret_key) throws Error {
            var socket = new SocketClient().connect(server);
            var dis = new DataInputStream(socket.input_stream);
            var dos = new DataOutputStream(socket.output_stream);

            solution.to_message().send(dos);
            var reply = Message.from_stream(dis);

            if(reply.message_type == MessageType.NOT_ACCEPTED) {
                warning(@"Got NOT-ACCEPTED response from $(server.address): $(reply.arguments[0]) $(reply.arguments[1])");
                return null;
            }

            var verified_reply = Solution.verify_solved_response(reply, author_signing_key);
            if(verified_reply == null) {
                return null;
            }

            return Solution.decrypt_connection_details(verified_reply, reply_public_key, reply_secret_key);

        }

        public void sync(InetSocketAddress server, NameInfoStore store) throws Error {
            var socket = new SocketClient().connect(server);
            var dis = new DataInputStream(socket.input_stream);
            var dos = new DataOutputStream(socket.output_stream);

            var existing = store.get_all().select<string>(i => @"$(i.name) $(i.effective.format_iso8601())");
            var request = new Message(MessageType.SYNC, new string[0], existing.to_array());

            request.send(dos);
            var reply = Message.from_stream(dis);

            if(reply.message_type != MessageType.DOMAINS) {
                warning(@"Bad response from $(server.address) for SYNC request.");
            }

            foreach (var item in reply.items) {
                var parts = item.split(" ", 2);
                var name = parts[0];
                var data = parts[1];
                
                NameInfo info;
                if(name.has_suffix(".rns")) {
                    info = new DecentralisedNameInfo.from_string(data);
                }
                else {
                    info = new CertifiedNameInfo.from_string(data);
                }

                store.save_name(info);
            }
        }

        internal static bool server_equals(InetSocketAddress a1, InetSocketAddress a2) {
            return (a1.address.equal(a2.address) && a1.port == a2.port);
        }

        private InetSocketAddress parse_address(string address) {
            var parts = address.split(" ", 3);
            var isa = new InetSocketAddress.from_string(parts[0], uint.parse(parts[1]));
            return isa;
        }
    }

}