Statum/src/ChannelEndpoint.vala

using Invercargill;
using Invercargill.DataStructures;
using Astralis;

namespace Statum {

    /**
     * The realtime channel service.
     *
     * Registered as a singleton (as itself and as {@link ChannelService}).
     * Each `GET /_statum/channel` opens a long-lived SSE stream whose first event
     * is `event: channel` carrying a fresh channel id; subsequent `set`/`clear`
     * events are delivered only for the slot keys the channel has subscribed to
     * (via `PATCH /_statum/channel/{id}`).
     *
     * {@link StateService.update}/{@link StateService.clear_slot} call
     * {@link push_set}/{@link push_clear}, which fan out to the subscribed
     * streams — a no-op when nothing is subscribed, giving selective fan-out for
     * free.
     *
     * Resource bounds (SA-15): every connection is a long-lived stream with an
     * unbounded subscription set, so {@link MAX_TOTAL_CONNECTIONS},
     * {@link MAX_CONNECTIONS_PER_IP} and {@link MAX_SUBSCRIPTIONS_PER_CHANNEL}
     * cap how much of the server an anonymous client can pin down. Admission
     * control itself lives in {@link ChannelGatewayEndpoint}, which owns the
     * `/_statum/channel` route and calls {@link try_open_connection} before any
     * stream is created.
     */
    public class ChannelEndpoint : SseEndpoint, ChannelService {

        /**
         * Concurrent channel connections across all clients (SA-15). Each one
         * holds a stream, a connection object and its subscriptions open for
         * as long as the client stays connected.
         */
        public const int MAX_TOTAL_CONNECTIONS = 256;

        /**
         * Concurrent channel connections from a single remote address
         * (SA-15): one browser legitimately holds a handful of tabs, not
         * hundreds of EventSource streams.
         */
        public const int MAX_CONNECTIONS_PER_IP = 16;

        /**
         * Subscriptions one channel may hold (SA-15): each subscription adds
         * a fan-out write to every {@link push_set} for that key.
         */
        public const int MAX_SUBSCRIPTIONS_PER_CHANNEL = 64;

        public override uint retry_interval { get { return 3000; } }

        private Series<ChannelConnection> connections = new Series<ChannelConnection>();
        /**
         * Global/per-IP connection accounting (SA-15). Lives on the singleton
         * so every gateway admission decision sees the same counts.
         */
        public ConnectionTracker connection_tracker { get; private set; }

        construct {
            connection_tracker = new ConnectionTracker();
        }

        /**
         * Whether one more channel connection may be opened from `ip`, given
         * the SA-15 global and per-IP caps. Checked by the gateway before a
         * stream is created; the count is released on disconnect.
         */
        public bool can_accept_connection(string? ip) {
            drop_closed();
            return connection_tracker.can_accept(ip);
        }

        public override async void new_connection(HttpContext http_context, RouteContext route_context, SseStream stream) {
            var connection = new ChannelConnection() {
                stream = stream,
                id = generate_channel_id(),
                ip = http_context.request.remote_address
            };
            connections.add(connection);
            connection_tracker.acquired(connection.ip);

            stream.disconnected.connect(() => {
                connections.remove(connection);
                connection_tracker.released(connection.ip);
            });

            try {
                yield stream.send_event(new SseEvent.with_type("channel", connection.id));
            } catch (GLib.Error e) {
                stream.close();
            }
        }

        /** Drops closed connections and releases their counted slots. */
        private void drop_closed() {
            // Closed streams are normally removed by the disconnect handler;
            // sweep any that slipped through (e.g. a stream closed before its
            // handler ran) so the caps measure live connections only.
            var closed = connections.where(c => c.stream.is_closed).to_series();
            foreach (var connection in closed) {
                connections.remove(connection);
                connection_tracker.released(connection.ip);
            }
        }

        /** Pushes a `set` event carrying `frame_json` to subscribed channels. */
        public async void push_set(string key, string frame_json) {
            foreach (var connection in open_subscribers(key)) {
                try {
                    yield connection.stream.send_event(new SseEvent.with_type("set", frame_json));
                } catch (GLib.Error e) {
                    connection.stream.close();
                }
            }
        }

        /** Pushes a `clear` event for `key` to subscribed channels. */
        public async void push_clear(string key) {
            foreach (var connection in open_subscribers(key)) {
                try {
                    yield connection.stream.send_event(new SseEvent.with_type("clear", key));
                } catch (GLib.Error e) {
                    connection.stream.close();
                }
            }
        }

        /** Whether a channel with `id` is currently open. */
        public bool has_channel(string id) {
            return connections.any(c => !c.stream.is_closed && c.id == id);
        }

        /**
         * Adds `key` to channel `id`'s subscriptions, bounded by
         * {@link MAX_SUBSCRIPTIONS_PER_CHANNEL} (SA-15).
         *
         * @return `true` when the subscription was recorded.
         */
        public bool subscribe(string id, string key) {
            var connection = find_connection(id);
            if (connection != null) {
                if (((!)connection).subs.length >= MAX_SUBSCRIPTIONS_PER_CHANNEL) {
                    warning("[Statum] channel %s hit the %d-subscription cap; ignoring key \"%s\"",
                        id, MAX_SUBSCRIPTIONS_PER_CHANNEL, key);
                    return false;
                }
                ((!)connection).subs.add(key);
                return true;
            }
            return false;
        }

        /** Removes `key` from channel `id`'s subscriptions. */
        public void unsubscribe(string id, string key) {
            var connection = find_connection(id);
            if (connection != null) {
                ((!)connection).subs.remove(key);
            }
        }

        private ChannelConnection? find_connection(string id) {
            foreach (var connection in connections) {
                if (!connection.stream.is_closed && connection.id == id) {
                    return connection;
                }
            }
            return null;
        }

        private Lot<ChannelConnection> open_subscribers(string key) {
            return connections
                .where(c => !c.stream.is_closed && c.subs.contains(key))
                .to_immutable_buffer();
        }

        private static string generate_channel_id() {
            var bytes = new uint8[16];
            Sodium.Random.random_bytes(bytes);
            var builder = new StringBuilder();
            foreach (var b in bytes) {
                builder.append_printf("%02x", b);
            }
            return builder.str;
        }

        private class ChannelConnection : Object {
            public SseStream stream { get; set; }
            public string id { get; set; }
            /** Remote address, released from the per-IP count on disconnect. */
            public string? ip { get; set; }
            public Set<string> subs { get; private set; }

            public ChannelConnection() {
                subs = new HashSet<string>();
            }
        }

    }

}