Statum/src/ChannelSubscriptionEndpoint.vala

using Invercargill;
using Invercargill.DataStructures;
using InvercargillJson;
using Inversion;
using Astralis;

namespace Statum {

    /**
     * `PATCH /_statum/channel/{id}` — adds or removes a channel's subscriptions.
     *
     * The request body is `{ "subscribe": [keys…], "unsubscribe": [keys…] }`.
     * For each newly-subscribed key, if the {@link StateService} has the slot
     * cached, its current frame is pushed immediately so a fresh client sees the
     * latest state right away (mirroring `server.php`). Responds `204 No Content`.
     */
    public class ChannelSubscriptionEndpoint : Object, Endpoint {

        private ChannelEndpoint channel_endpoint = inject<ChannelEndpoint>();
        private StateService state_service = inject<StateService>();

        public async HttpResult handle_request(HttpContext http_context, RouteContext route_context) throws GLib.Error {
            var id = route_context.mapped_parameters.get_or_default("id");
            if (id == null || !channel_endpoint.has_channel((!)id)) {
                return new HttpStringResult("No such channel", StatusCode.NOT_FOUND);
            }

            var body = yield http_context.request.request_body.read_all();
            var body_str = body.to_raw_string();

            Enumerable<string> subscribe_keys = Iterate.nothing<string>();
            Enumerable<string> unsubscribe_keys = Iterate.nothing<string>();

            if (body_str.strip().length > 0) {
                var body_element = new JsonElement.from_string(body_str);
                var root = body_element.as<JsonObject>();
                subscribe_keys = read_string_array(root, "subscribe");
                unsubscribe_keys = read_string_array(root, "unsubscribe");
            }

            foreach (var key in unsubscribe_keys) {
                channel_endpoint.unsubscribe((!)id, key);
            }

            foreach (var key in subscribe_keys) {
                channel_endpoint.subscribe((!)id, key);

                var slot = state_service.get_slot(key);
                if (slot != null) {
                    var frame = state_service.sign_slot(key);
                    if (frame != null) {
                        var frame_json = serialize_frame((!)frame);
                        yield channel_endpoint.push_set(key, frame_json);
                    }
                }
            }

            return new HttpEmptyResult(StatusCode.NO_CONTENT);
        }

        private static string serialize_frame(Model.FrameDto frame) throws GLib.Error {
            return new JsonElement.from_properties(Model.FrameDto.get_mapper().map_from(frame)).stringify();
        }

        private static Enumerable<string> read_string_array(JsonObject root, string key) throws GLib.Error {
            var array = root.get_array(key);
            if (array == null) {
                return Iterate.nothing<string>();
            }
            var result = new Series<string>();
            foreach (var element in (!)array) {
                result.add(element.as<string>());
            }
            return result;
        }

    }

}