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(); private Cryptography.SigningProvider signing_provider = inject(); /** * 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(); private Dictionary slots = new Dictionary(); 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(string type_name, owned UpdateMutator 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(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(); 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(); } } }