State service

js/statum.js

@@ -612,7 +612,8 @@
     });
     if (stale.length) {
       var directives = await postSlots(stale);
-      applyDirectives(directives, { originElement: null });
+      var pfResult = applyDirectives(directives, { originElement: null });
+      executeTerminal(pfResult);
     }
   }
 
@@ -636,6 +637,28 @@
     return headers;
   }
 
+  /**
+   * Collect the current frames held for the given slot keys, in the order the
+   * keys are first supplied. Used by the `transmit` directive (which forces a
+   * slot's frame to be posted as if its `transmit_after` had elapsed). Keys the
+   * client does not currently hold are skipped.
+   * @param {string[]} keys
+   * @returns {object[]}
+   */
+  function collectFramesForKeys(keys) {
+    var keySet = Object.create(null);
+    for (var i = 0; i < keys.length; i++) keySet[keys[i]] = true;
+    var frames = [];
+    store.forEach(function (entry) {
+      var key = entry.snapshot.slot && entry.snapshot.slot.key;
+      if (key && keySet[key]) {
+        frames.push(entry.frame);
+        delete keySet[key];
+      }
+    });
+    return frames;
+  }
+
   // ---------------------------------------------------------------------------
   // HTTP helpers
   // ---------------------------------------------------------------------------
@@ -667,19 +690,22 @@
   }
 
   /**
-   * Apply the directives from a response. set/clear/subscribe/unsubscribe/notify
-   * run first, then `render`, then terminal navigation (`navigate`, then `post`).
-   * Error directives are collected (not thrown) so the caller can route them
-   * through `reportFailure` (and thus `onErrorHandler`).
+   * Apply the non-terminal directives from a response (set/clear/subscribe/
+   * unsubscribe/notify/transmit-collection), then `render`.
+   *
+   * Terminal directives (`navigate`, `post`), error directives, and any
+   * transmit/retry intent are collected on the returned result rather than
+   * executed here: {@link handleResponse} performs the asynchronous
+   * transmission (and optional retry) and then runs terminal navigation, so
+   * that a `transmit` with `retry` can re-issue the request before the page is
+   * navigated away.
    * @param {object[]} directives
    * @param {{originElement?: HTMLElement}} [ctx]
-   * @returns {{errors: StatumError[]}}
+   * @returns {{errors: StatumError[], transmitKeys: string[], retry: boolean, navigate: object|null, post: object|null}}
    */
   function applyDirectives(directives, ctx) {
-    var result = { errors: [] };
+    var result = { errors: [], transmitKeys: [], retry: false, navigate: null, post: null };
     var list = Array.isArray(directives) ? directives : [];
-    var navigateDirective = null;
-    var postDirective = null;
 
     for (var i = 0; i < list.length; i++) {
       var d = list[i];
@@ -690,19 +716,28 @@
         case 'subscribe': channelSubscribe(d.key); break;
         case 'unsubscribe': channelUnsubscribe(d.key); break;
         case 'notify': triggerNotify(d.kind, d.message); break;
+        case 'transmit':
+          if (d.key != null) {
+            if (result.transmitKeys.indexOf(d.key) === -1) result.transmitKeys.push(d.key);
+            if (d.retry) result.retry = true;
+          }
+          break;
         case 'error': result.errors.push(new StatumError(d.message, d.code, d.data)); break;
-        case 'navigate': navigateDirective = d; break;
-        case 'post': postDirective = d; break;
+        case 'navigate': result.navigate = d; break;
+        case 'post': result.post = d; break;
       }
     }
 
     renderAll();
-
-    if (navigateDirective) doNavigate(navigateDirective);
-    if (postDirective) doPost(postDirective);
     return result;
   }
 
+  /** Execute the terminal navigation directives collected by {@link applyDirectives}. */
+  function executeTerminal(result) {
+    if (result.navigate) doNavigate(result.navigate);
+    if (result.post) doPost(result.post);
+  }
+
   /** Execute a `navigate` directive as a full page load. */
   function doNavigate(d) {
     if (d.uri) window.location.href = d.uri;
@@ -730,8 +765,15 @@
   /**
    * Handle a fetch response: if it is a Statum response, apply its directives;
    * otherwise throw on HTTP errors and no-op on other successful responses.
+   *
+   * Directive ordering: non-terminal directives are applied and the page
+   * rendered; then any `transmit` directives force their slot frames to be
+   * posted (applying the directives the slot endpoint returns); then, if a
+   * transmit requested a retry, the originating request (`ctx.retry`) is
+   * re-issued and this response's terminal directives and errors are skipped.
+   * Otherwise terminal navigation runs, then error directives are surfaced.
    * @param {Response} res
-   * @param {{originElement?: HTMLElement}} [ctx]
+   * @param {{originElement?: HTMLElement, retry?: Function}} [ctx]
    */
   async function handleResponse(res, ctx) {
     var el = ctx && ctx.originElement;
@@ -742,6 +784,22 @@
     var directives;
     try { directives = await res.json(); } catch (e) { directives = []; }
     var result = applyDirectives(directives, ctx);
+
+    if (result.transmitKeys && result.transmitKeys.length) {
+      var frames = collectFramesForKeys(result.transmitKeys);
+      if (frames.length) {
+        var txDirectives = await postSlots(frames);
+        applyDirectives(txDirectives, { originElement: el });
+      }
+    }
+
+    if (result.retry && ctx && typeof ctx.retry === 'function') {
+      await ctx.retry();
+      return;
+    }
+
+    executeTerminal(result);
+
     var errors = result.errors || [];
     for (var i = 0; i < errors.length; i++) {
       await reportFailure(errors[i], el);
@@ -754,22 +812,31 @@
 
   /**
    * Fetch the entrypoint for the current URL, applying its directives. Carries
-   * slot headers (and pre-flights) like any other request.
+   * slot headers (and pre-flights) like any other request. The request is
+   * wrapped in an `attempt` closure passed as the response `retry` callback, so
+   * a `transmit` directive with `retry` can re-issue it after transmission.
    * @returns {Promise<void>}
    */
-  async function loadEntrypoint() {
-    await prepareSlots();
-    var headers = new Headers();
-    headers.set('Accept', STATUM_CONTENT_TYPE);
-    buildSlotHeaders().forEach(function (v) { headers.append(SLOT_HEADER, v); });
-    var url = config.entrypointUrl + '?uri=' + encodeURIComponent(window.location.href);
-    var res = await fetch(url, { headers: headers, credentials: 'same-origin' });
-    await handleResponse(res, { originElement: null });
+  function loadEntrypoint() {
+    var attempt = async function () {
+      await prepareSlots();
+      var headers = new Headers();
+      headers.set('Accept', STATUM_CONTENT_TYPE);
+      buildSlotHeaders().forEach(function (v) { headers.append(SLOT_HEADER, v); });
+      var url = config.entrypointUrl + '?uri=' + encodeURIComponent(window.location.href);
+      var res = await fetch(url, { headers: headers, credentials: 'same-origin' });
+      await handleResponse(res, { originElement: null, retry: attempt });
+    };
+    return attempt();
   }
 
   /**
    * Perform an action request.
    *
+   * The request is wrapped in an `attempt` closure passed as the response
+   * `retry` callback, so a `transmit` directive with `retry` can re-issue the
+   * same action after the transmission (rebuilding headers from current state).
+   *
    * @param {object} action `{ uri, method, private }`
    * @param {object} [data] Key/value parameters. Encoded as query string for
    *   GET/HEAD, otherwise as `application/x-www-form-urlencoded`.
@@ -778,34 +845,37 @@
    * @returns {Promise<void>} Resolves once the request completes and its
    *   directives have run; rejects on HTTP failure or an `error` directive.
    */
-  async function performAction(action, data, originElement) {
-    await prepareSlots();
-    var headers = new Headers();
-    headers.set('Accept', STATUM_CONTENT_TYPE);
-    buildSlotHeaders().forEach(function (v) { headers.append(SLOT_HEADER, v); });
-    if (action.private) headers.set(PRIVATE_HEADER, action.private);
-
-    var method = (action.method || 'GET').toUpperCase();
-    var params = data || {};
-    var init = { method: method, headers: headers, credentials: 'same-origin' };
-    var url = action.uri;
-
-    if (method === 'GET' || method === 'HEAD') {
-      var qs = new URLSearchParams(params).toString();
-      if (qs) url += (url.indexOf('?') !== -1 ? '&' : '?') + qs;
-    } else {
-      init.body = new URLSearchParams(params).toString();
-      headers.set('Content-Type', 'application/x-www-form-urlencoded');
-    }
+  function performAction(action, data, originElement) {
+    var attempt = async function () {
+      await prepareSlots();
+      var headers = new Headers();
+      headers.set('Accept', STATUM_CONTENT_TYPE);
+      buildSlotHeaders().forEach(function (v) { headers.append(SLOT_HEADER, v); });
+      if (action.private) headers.set(PRIVATE_HEADER, action.private);
+
+      var method = (action.method || 'GET').toUpperCase();
+      var params = data || {};
+      var init = { method: method, headers: headers, credentials: 'same-origin' };
+      var url = action.uri;
+
+      if (method === 'GET' || method === 'HEAD') {
+        var qs = new URLSearchParams(params).toString();
+        if (qs) url += (url.indexOf('?') !== -1 ? '&' : '?') + qs;
+      } else {
+        init.body = new URLSearchParams(params).toString();
+        headers.set('Content-Type', 'application/x-www-form-urlencoded');
+      }
 
-    var res;
-    try {
-      res = await fetch(url, init);
-    } catch (networkErr) {
-      await reportFailure(new StatumError('Network error'), originElement);
-      return;
-    }
-    await handleResponse(res, { originElement: originElement });
+      var res;
+      try {
+        res = await fetch(url, init);
+      } catch (networkErr) {
+        await reportFailure(new StatumError('Network error'), originElement);
+        return;
+      }
+      await handleResponse(res, { originElement: originElement, retry: attempt });
+    };
+    return attempt();
   }
 
   // ---------------------------------------------------------------------------

model.md

@@ -46,6 +46,7 @@ A statum snapshot captures an aspect of the application state at a particular po
     - `"subscribe"`: Used to subscribe the client's realtime channel to a slot.
     - `"unsubscribe"`: Used to unsubscribe the client's realtime channel from a slot.
     - `"notify"`: Used to surface a transient notification to the user.
+    - `"transmit"`: Used to force the client to transmit a slot's frame to the server immediately, optionally retrying the originating request afterward.
 
 Depending on the type, it will have other properties.
 
@@ -97,6 +98,17 @@ Removes a slot from the client's realtime channel subscriptions. Subscriptions a
 
 Surfaces a transient message to the user. The default handler calls the browser's `alert()` with the `message`; this can be customised via `statum.onNotifyHandler` (see [Handlers](#handlers)). Notify runs in the same pass as `set` and `clear`.
 
+### Transmit Directive
+
+- `key`: The key of the slot whose frame should be transmitted.
+- `retry`: A boolean (optional, defaults to `false`).
+
+Forces the client to transmit the named slot's current frame to the slot post endpoint immediately — exactly as a pre-flight does for a snapshot that has passed its `transmit_after` window (see [Snapshot freshness decision](#snapshot-freshness-decision)). The client POSTs the held frame, applies any `set` directives the slot endpoint returns (refreshing the snapshot), and then continues. If the client does not currently hold a frame for `key`, the transmission is skipped.
+
+When `retry` is `true`, the originating request (the one that returned this directive) is re-issued once the transmission and its response directives have been applied, so the server can re-evaluate the request against the freshly transmitted state. The retry carries rebuilt slot headers reflecting the transmitted (and possibly refreshed) snapshots. A retried response is processed like any other, so combining `transmit` with `retry` is how the server forces a state sync followed by a fresh evaluation of the same request.
+
+Transmit is not terminal: it runs in the same pass as `set` and `clear`. The actual transmission (and any retry) occurs after the response's other non-terminal directives have been applied and the page re-rendered. When a retry is performed, the original response's terminal directives (`navigate`, `post`) and `error` directives are not applied — the retried response supersedes them.
+
 # Statum HTTP API
 
 All endpoints that return directives do so as a JSON array of Statum Directives, served with the Content-Type `application/vnd.statum+json`. The frontend uses this content type to distinguish a Statum directive response from an ordinary HTML or JSON response.

src/Cryptography/SigningProvider.vala

@@ -81,9 +81,12 @@ namespace Statum.Cryptography {
          *        {@link FrameDto.content} must already be set; the `signer` and
          *        `signature` fields are overwritten.
          */
-        public void author_frame(FrameDto frame) {
+        public FrameDto author_frame(string content) {
+            var frame = new FrameDto();
+            frame.content = content;
             frame.signer = signer_id;
             frame.signature = sign(frame.content);
+            return frame;
         }
 
         /**

src/Model/DirectiveDto.vala

@@ -68,6 +68,7 @@ namespace Statum.Model {
                 case "subscribe": return SubscribeDirectiveDto.get_mapper().materialise(json);
                 case "unsubscribe": return UnsubscribeDirectiveDto.get_mapper().materialise(json);
                 case "notify": return NotifyDirectiveDto.get_mapper().materialise(json);
+                case "transmit": return TransmitDirectiveDto.get_mapper().materialise(json);
                 default:
                     throw new ModelError.UNKNOWN_DIRECTIVE(@"Unrecognised directive type \"$((!)type)\"");
             }
@@ -360,4 +361,48 @@ namespace Statum.Model {
 
     }
 
+    /**
+     * Forces the client to transmit a slot's frame to the server immediately,
+     * exactly as a pre-flight does for a snapshot that has passed its
+     * {@link SnapshotDto.transmit_after} window. Optionally re-issues the
+     * originating request once the transmission is complete.
+     */
+    public class TransmitDirectiveDto : DirectiveDto {
+
+        /** The key of the slot whose frame should be transmitted. */
+        public string key { get; set; }
+
+        /**
+         * When `true`, the request that returned this directive is re-issued
+         * once the transmission (and the directives returned by the slot
+         * endpoint) have been applied, so the server can re-evaluate it against
+         * the freshly transmitted state. Defaults to `false`.
+         */
+        public bool retry { get; set; }
+
+        public override string discriminator { get { return "transmit"; } }
+
+        protected override Properties build_field_properties() throws GLib.Error {
+            return TransmitDirectiveDto.get_mapper().map_from(this);
+        }
+
+        public TransmitDirectiveDto() { }
+
+        public TransmitDirectiveDto.with(string key, bool retry) {
+            this.key = key;
+            this.retry = retry;
+        }
+
+        public static PropertyMapper<TransmitDirectiveDto> get_mapper() {
+            return PropertyMapper.build_for<TransmitDirectiveDto>(cfg => {
+                cfg.map<string>("key", o => o.key, (o, v) => o.key = v);
+                cfg.map<bool>("retry", o => o.retry, (o, v) => o.retry = v)
+                    .undefined_when(o => !o.retry)
+                    .when_undefined(o => o.retry = false);
+                cfg.set_constructor(() => new TransmitDirectiveDto());
+            });
+        }
+
+    }
+
 }

src/Model/Scope.vala

@@ -1,4 +1,4 @@
-namespace Statum.Model {
+namespace Statum {
 
     /**
      * The persistence scope of a {@link Slot}.

src/Slot.vala

@@ -0,0 +1,13 @@
+using Invercargill;
+namespace Statum {
+
+    public class Slot : Object {
+
+        public string id { get; set; }
+        public string type_name { get; set; }
+        public Scope scope { get; set; }
+        public State current_state { get; set; }
+        
+    }
+
+}

src/State.vala

@@ -0,0 +1,13 @@
+using Invercargill;
+namespace Statum {
+
+    public class State : Object {
+
+        public string type_name { get; set; }
+        public Properties public_data { get; set; }
+        public Properties private_data { get; set; }
+        public DateTime last_touched { get; set; }
+
+    }
+
+}

src/StateService.vala

@@ -0,0 +1,102 @@
+using Invercargill.DataStructures;
+using InvercargillJson;
+using Inversion;
+
+namespace Statum {
+
+    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>();
+
+        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;
+        }
+
+        public Slot? get_slot(string id) {
+            Slot? slot = null;
+            if(slots.try_get(id, out slot)) {
+                slot.current_state.last_touched = new DateTime.now_utc();
+            }
+            return slot;
+        }
+
+        public Model.FrameDto? sign_slot(string id, bool unidirectional = false) throws Error {
+            var slot = get_slot(id);
+            if(slot == null) {
+                return null;
+            }
+
+            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);
+        }
+
+        public bool restore_slot(Model.FrameDto frame) throws Error {
+            signing_provider.verify_frame(frame);
+            var json_object = new JsonElement.from_string(frame.content).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;
+        }
+
+        public void cleanup() {
+            var cutoff = new DateTime.now_utc().add_seconds(-RETRANSMIT_SECONDS).to_unix();
+            var to_remove = slots.values.where(s => s.current_state.last_touched.to_unix() > cutoff).to_series();
+            foreach (var item in to_remove) {
+                slots.remove(item.id);
+            }
+        }
+    }
+
+}

src/meson.build

@@ -8,6 +8,9 @@ sources = files(
     'Model/DirectiveDto.vala',
     'Cryptography/SigningProvider.vala',
     'Cryptography/EncryptionProvider.vala',
+    'StateService.vala',
+    'State.vala',
+    'Slot.vala'
 )
 
 library_version = meson.project_version()