Statum/src/StateService.vala

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

namespace Statum {

    /**
     * Error domain for state service operations.
     */
    public errordomain StateError {
        /** No slot is cached for the given key. */
        SLOT_NOT_FOUND,
    }

    public class StateService : Object {

        private const int RETRANSMIT_SECONDS = 60; // 1 minute
        private const int EXPIRY_SECONDS = 604800; // 7 days

        private Cryptography.EncryptionProvider encryption_provider = inject<Cryptography.EncryptionProvider>();
        private Cryptography.SigningProvider signing_provider = inject<Cryptography.SigningProvider>();

        /**
         * The realtime channel service. Injected (singleton) so that every state
         * mutation auto-pushes the resulting frame to subscribed channels. The
         * dependency is one-way: {@link ChannelService} never injects this
         * service, so there is no DI cycle.
         */
        private ChannelService channel_service = inject<ChannelService>();

        private Dictionary<string, Slot> slots = new Dictionary<string, Slot>();

        public Slot new_slot(Scope slot_scope, State initial_state) {
            var id = new uint8[32];
            Sodium.Random.random_bytes(id);

            initial_state.last_touched = new DateTime.now_utc();
            var slot = new Slot() {
                id = Base64.encode(id),
                scope = slot_scope,
                type_name = initial_state.type_name,
                current_state = initial_state,
            };

            slots.set(slot.id, slot);
            return slot;
        }

        /**
         * Reads the cached slot for `id` without mutating it.
         *
         * Liveness is now driven by the `transmit_after`/`invalid_after` windows
         * and explicit {@link update} calls, so reading a slot no longer touches
         * its `last_touched`.
         */
        public Slot? get_slot(string id) {
            Slot? slot = null;
            slots.try_get(id, out slot);
            return slot;
        }

        public Model.FrameDto? sign_slot(string id, bool unidirectional = false) throws GLib.Error {
            var slot = get_slot(id);
            if (slot == null) {
                return null;
            }
            return sign_slot_for((!)slot, unidirectional);
        }

        /**
         * Mutates the slot cached under `key` to `state`, signs the resulting
         * snapshot into a frame, auto-pushes that frame to any channel
         * subscribed to `key` (a no-op when nothing is subscribed), and returns
         * the signed frame so the caller can reuse it for its response `set`
         * directive (one sign → push + response).
         */
        public async Model.FrameDto update(string key, State state) throws GLib.Error {
            Slot? slot = null;
            if (!slots.try_get(key, out slot) || slot == null) {
                throw new StateError.SLOT_NOT_FOUND(@"No slot is cached for key \"$key\"");
            }

            state.last_touched = new DateTime.now_utc();
            slot.current_state = state;

            var frame = sign_slot_for(slot, false);
            var frame_json = serialize_frame(frame);
            yield channel_service.push_set(key, frame_json);
            return frame;
        }

        /**
         * Removes the slot cached under `key` and auto-pushes a `clear` event to
         * any channel subscribed to it (a no-op when nothing is subscribed).
         */
        public async void clear_slot(string key) {
            slots.remove(key);
            yield channel_service.push_clear(key);
        }

        // ------------------------------------------------------------------
        // Global slots (deterministic-key broadcast state)
        // ------------------------------------------------------------------

        public const string GLOBAL_KEY_PREFIX = "global:";

        /**
         * Creates (or updates) a global slot with the deterministic key
         * `"global:" + type_name`. Global slots share one key across all clients,
         * so every subscriber sees the same updates.
         */
        public Slot new_global_slot(string type_name, Scope slot_scope, State state) {
            var key = GLOBAL_KEY_PREFIX + type_name;
            state.last_touched = new DateTime.now_utc();
            state.type_name = type_name;

            Slot? existing = null;
            if (slots.try_get(key, out existing) && existing != null) {
                ((!)existing).current_state = state;
                return (!)existing;
            }

            var slot = new Slot() {
                id = key,
                scope = slot_scope,
                type_name = type_name,
                current_state = state
            };
            slots.set(key, slot);
            return slot;
        }

        /** Updates a global slot by type name (sign + push to all subscribers). */
        public async Model.FrameDto update_global(string type_name, State state) throws GLib.Error {
            return yield update(GLOBAL_KEY_PREFIX + type_name, state);
        }

        /**
         * Typed read-mutate-write of a global slot: maps its current public data
         * to `TPublic`, applies `mutator`, maps back, signs and pushes. Used from
         * background contexts (timers, webhooks) that have no request scope.
         *
         * Like {@link DirectiveBuilder.update_held}, JSON-array fields the typed
         * model cannot read back are carried forward unchanged rather than
         * dropped.
         */
        public async void update_global_typed<TPublic>(string type_name, owned UpdateMutator<TPublic> mutator) throws GLib.Error {
            var key = GLOBAL_KEY_PREFIX + type_name;
            var slot = get_slot(key);
            if (slot == null || ((!)slot).current_state == null) {
                throw new StateError.SLOT_NOT_FOUND(@"No global slot for type \"$type_name\"");
            }
            var current = ((!)slot).current_state;
            var model = GObjectMapping.from_properties_typed<TPublic>(typeof(TPublic),
                current.public_data ?? new PropertyDictionary());
            mutator(model);
            var new_state = new State() {
                type_name = current.type_name,
                public_data = GObjectMapping.preserving_arrays(current.public_data ?? new PropertyDictionary(),
                    GObjectMapping.to_properties((Object) model)),
                private_data = current.private_data ?? new PropertyDictionary()
            };
            yield update(key, new_state);
        }

        /** Clears a global slot by type name (remove + push clear to subscribers). */
        public async void clear_global(string type_name) {
            yield clear_slot(GLOBAL_KEY_PREFIX + type_name);
        }

        public bool restore_slot(Model.FrameDto frame) throws GLib.Error {
            signing_provider.verify_frame(frame);
            var content_element = new JsonElement.from_string(frame.content);
            var json_object = content_element.as<JsonObject>();
            var snapshot = Model.SnapshotDto.get_mapper().materialise(json_object);

            var slot = get_slot(snapshot.slot.key);
            if (slot != null) {
                // No-op: slot not empty.
                return false;
            }

            slot = new Slot() {
                id = snapshot.slot.key,
                type_name = snapshot.slot.slot_type,
                scope = snapshot.slot.scope,
                current_state = new State() {
                    type_name = snapshot.slot.slot_type,
                    public_data = snapshot.public,
                    private_data = encryption_provider.read_properties(snapshot.slot.key, snapshot.private),
                    last_touched = new DateTime.now_utc()
                }
            };
            slots.set(slot.id, slot);
            return true;
        }

        /**
         * Evicts slots whose `invalid_after` window has elapsed — i.e. slots that
         * have not been touched for {@link EXPIRY_SECONDS}. The client resends a
         * fresh frame via `/_statum/slots` when it next needs one of these.
         */
        public void cleanup() {
            var cutoff = new DateTime.now_utc().add_seconds(-EXPIRY_SECONDS);
            var to_remove = slots.values
                .where(s => s.current_state.last_touched.compare(cutoff) < 0)
                .to_series();
            foreach (var item in to_remove) {
                slots.remove(item.id);
            }
        }

        private Model.FrameDto sign_slot_for(Slot slot, bool unidirectional) throws GLib.Error {
            var slot_dto = new Model.SlotDto() {
                key = slot.id,
                scope = slot.scope,
                slot_type = slot.type_name
            };

            var timestamp = new DateTime.now_utc();
            var snapshot_dto = new Model.SnapshotDto() {
                as_at = timestamp,
                slot = slot_dto,
                transmit_after = unidirectional ? null : timestamp.add_seconds(RETRANSMIT_SECONDS),
                invalid_after = timestamp.add_seconds(EXPIRY_SECONDS),
                @private = encryption_provider.author_properties(slot.id, slot.current_state.private_data),
                @public = slot.current_state.public_data
            };

            var json = new JsonElement.from_properties(Model.SnapshotDto.get_mapper().map_from(snapshot_dto)).stringify();
            return signing_provider.author_frame(json);
        }

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

    }

}