feat: add personalized state payload broadcasts

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.6.0 - 2026-08-09
+
+- Add `broadcast_payload`, an actor DSL for sending one personalized JSON state
+  payload over the actor stream a page already has open. The block runs once per
+  subscriber with that subscriber's authorization context, so private state
+  never crosses sessions. Payloads carry actor identity and the monotonic state
+  revision, and both the channel and the browser drop stale revisions. Subscribe
+  with `solid_object room, payloads: :playmat_state` and handle the
+  `solid-objects:payload` DOM event. ERB component refreshes remain the default
+  and are unchanged. A mutation that changes payload state without changing a
+  declared observable still invalidates subscribers, through a revision-only
+  broadcast that carries no observable value to the browser.
+
 ## 0.5.2 - 2026-08-09
 
 - Read the database clock once per transaction instead of once per step, and

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_objects (0.5.2)
+    solid_objects (0.6.0)
       actioncable (>= 8.0)
       actionpack (>= 8.0)
       actionview (>= 8.0)
@@ -373,7 +373,7 @@ CHECKSUMS
   rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d
   ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33
   securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1
-  solid_objects (0.5.2)
+  solid_objects (0.6.0)
   sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc
   sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d
   sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b

app/assets/javascripts/solid_objects/state_payload.js

@@ -0,0 +1,69 @@
+const deliveredRevisions = new Map()
+
+class SolidObjectsPayloadElement extends HTMLElement {
+  connectedCallback() {
+    if (this.dataset.started === "true") return
+
+    this.dataset.started = "true"
+    this.deliver()
+  }
+
+  deliver() {
+    try {
+      const name = this.dataset.name
+      const revision = revisionFor(this)
+      const scope = this.closest("[id]")
+      if (!name || !revision || !scope) return
+
+      const key = `${scope.id}:${name}`
+      if (!newerRevision(revision, deliveredRevisions.get(key))) return
+
+      const payload = JSON.parse(this.textContent)
+      deliveredRevisions.set(key, revision)
+      scope.dispatchEvent(
+        new CustomEvent("solid-objects:payload", {
+          bubbles: true,
+          detail: {
+            name,
+            instanceId: revision[0],
+            revision: revision[1],
+            payload
+          }
+        })
+      )
+    } catch {
+      this.dispatchEvent(
+        new CustomEvent("solid-objects:payload-error", {
+          bubbles: true,
+          detail: { reason: "invalid_payload" }
+        })
+      )
+    } finally {
+      this.remove()
+    }
+  }
+}
+
+function newerRevision(candidate, current) {
+  if (!current) return true
+
+  return candidate[0] > current[0] ||
+    (candidate[0] === current[0] && candidate[1] > current[1])
+}
+
+function revisionFor(element) {
+  const revision = element.dataset.revision
+  if (!revision) return
+
+  const values = revision.split(":").map(Number)
+  if (
+    values.length !== 2 ||
+    values.some((value) => !Number.isSafeInteger(value) || value < 0)
+  ) return
+
+  return values
+}
+
+if (!customElements.get("solid-objects-payload")) {
+  customElements.define("solid-objects-payload", SolidObjectsPayloadElement)
+}

app/helpers/solid_objects/actor_helper.rb

@@ -2,8 +2,9 @@
 
 module SolidObjects
   module ActorHelper
-    # @rbs (Reference, ?authorization_context: untyped) { (ActorView) -> untyped } -> untyped
-    def solid_object(reference, authorization_context: self, &block)
+    # @rbs (Reference, ?authorization_context: untyped, ?payloads: untyped) { (ActorView) -> untyped } -> untyped
+    def solid_object(reference, authorization_context: self, payloads: nil, &block)
+      payload_names = Array(payloads).map(&:to_s).uniq.presence
       actor = ActorView.new(
         reference:,
         view_context: self,
@@ -13,7 +14,8 @@ def solid_object(reference, authorization_context: self, &block)
       subscription_data = {
         token: StreamToken.generate(
           reference,
-          observables: actor.scalar_observable_names
+          observables: actor.scalar_observable_names,
+          payloads: payload_names
         )
       }
       if actor.component_tokens.any?
@@ -30,10 +32,17 @@ def solid_object(reference, authorization_context: self, &block)
           data: { turbo_track: "reload" }
         )
       end
+      payload_client = if payload_names
+        javascript_include_tag(
+          "solid_objects/state_payload",
+          type: "module",
+          data: { turbo_track: "reload" }
+        )
+      end
 
       content_tag(
         :div,
-        safe_join([ refresh_client, subscription, content ].compact),
+        safe_join([ refresh_client, payload_client, subscription, content ].compact),
         id: DomIdentity.scope(reference)
       )
     end

docs/realtime.md

@@ -110,6 +110,93 @@ applications discover the namespaced engine asset. Applications created with
 explicitly serve the module. Turbo's normal morph rules still apply; use
 `data-turbo-permanent` for elements that must never be changed.
 
+## Personalized state payloads
+
+Reactive ERB components cost one browser request per changed component. When a
+single actor mutation changes several components, an application pays several
+round trips for one logical update. A payload broadcast collapses that into one
+message on the stream the page already has open.
+
+Declare the payload on the actor. The block receives the actor and the
+subscriber's authorization context, and it runs **once per subscriber**, so two
+sessions watching the same actor never see each other's private state:
+
+```ruby
+class PlaymatRoom < SolidObjects::Actor
+  actor_type "playmat_room"
+
+  attribute :hands, default: -> { {} }
+  attribute :turn, default: 1
+
+  observable :turn
+
+  broadcast_payload :playmat_state do |room, authorization_context|
+    {
+      "turn" => room.turn,
+      "hand" => room.hands.fetch(authorization_context.session_id, [])
+    }
+  end
+end
+```
+
+Subscribe the scope to it:
+
+```erb
+<%= solid_object room, payloads: :playmat_state do |actor| %>
+  <div data-playmat></div>
+<% end %>
+```
+
+Handle it with any JavaScript. The gem dispatches a DOM event and requires no
+framework:
+
+```javascript
+document.addEventListener("solid-objects:payload", (event) => {
+  const { name, revision, payload } = event.detail
+  if (name !== "playmat_state") return
+
+  renderPlaymat(payload)
+})
+```
+
+### What the protocol guarantees
+
+The payload travels as a Turbo Stream element on the existing actor stream, so
+applications do not run a second WebSocket system. Each message carries the
+actor identity plus the `instance_id` and monotonic `state_revision` that fence
+component refreshes, and both the channel and the browser drop a payload that
+is not newer than the last one delivered for that scope and name. A reconnecting
+client receives the current payload on subscribe.
+
+Authorization is the same `authorize_query` boundary that components use, called
+with the payload name and the subscriber's Cable connection. A subscriber that
+fails the check is skipped rather than served a partial payload, and the payload
+name is signed into the stream token, so a browser cannot ask for a payload the
+server did not offer.
+
+Payload blocks read committed actor state through the same snapshot components
+use. They cannot write application records, and the return value must be a JSON
+object or array so the wire format stays inspectable.
+
+### mtg-playmat before and after
+
+Before, one mutation that touched three observables produced three refresh
+elements and three HTTP requests:
+
+```
+commit -> 3 Action Cable messages -> 3 GET /solid_objects/components -> 3 renders
+```
+
+After, the same mutation delivers one personalized payload and the page renders
+once:
+
+```
+commit -> 1 Action Cable message -> 0 HTTP requests -> 1 render
+```
+
+Components remain the default. An actor with no `broadcast_payload` and a scope
+with no `payloads:` option behave exactly as before.
+
 ## Authorization
 
 The HTML contains a signed actor identity token. Signing prevents modification;

lib/solid_objects.rb

@@ -40,6 +40,7 @@
 require "solid_objects/component_subscriptions"
 require "solid_objects/component_view"
 require "solid_objects/component_renderer"
+require "solid_objects/payload_broadcast"
 require "solid_objects/state_snapshot"
 require "solid_objects/actor_view"
 require "solid_objects/actor_channel"

lib/solid_objects/actor.rb

@@ -57,6 +57,13 @@ def observable(name, &block)
         definition.add_observable(name, block)
       end
 
+      # @rbs (Symbol | String) { (untyped, untyped) -> untyped } -> ActorDefinition::Handler
+      def broadcast_payload(name, &block)
+        raise InvalidActor, "payload broadcasts require a block" unless block
+
+        definition.add_payload_broadcast(name, block)
+      end
+
       # @rbs (?Integer) -> Integer
       def state_version(version = nil)
         definition.set_state_version(version) if version

lib/solid_objects/actor_channel.rb

@@ -19,7 +19,9 @@ def subscribed
 
       @reference = Reference.new(actor_type:, actor_id:)
       @scalar_observables = identity["observables"]
+      @payload_names = identity["payloads"]
       validate_scalar_observables!
+      validate_payload_names!
       @component_subscriptions = ComponentSubscriptions.parse(
         params["components"],
         reference:
@@ -36,6 +38,7 @@ def subscribed
         )
       end
       refresh_outdated_components(snapshot)
+      transmit_state_payloads(snapshot)
     rescue KeyError,
       JSON::ParserError,
       InvalidStreamToken,
@@ -46,21 +49,69 @@ def subscribed
 
     private
 
-    attr_reader :reference, :component_subscriptions, :scalar_observables
+    attr_reader :reference,
+      :component_subscriptions,
+      :scalar_observables,
+      :payload_names
 
     # @rbs (String) -> void
     def receive_broadcast(stream)
       invalidation = TurboStreamRenderer.invalidation(stream)
-      if !invalidation ||
-          scalar_observables.nil? ||
-          scalar_observables.include?(invalidation.fetch("observable_name"))
+      revision_only = invalidation &&
+        invalidation.fetch("observable_name") == PayloadBroadcast::REVISION_OBSERVABLE
+      if !revision_only &&
+          (!invalidation ||
+            scalar_observables.nil? ||
+            scalar_observables.include?(invalidation.fetch("observable_name")))
         transmit stream
       end
       return unless invalidation
 
       component_subscriptions
         .refreshes_for(invalidation)
         .each { |refresh| transmit refresh }
+      transmit_state_payloads(ActorSnapshot.new(reference))
+    end
+
+    # @rbs (ActorSnapshot) -> void
+    def transmit_state_payloads(snapshot)
+      return if payload_names.nil? || payload_names.empty?
+      return unless newer_payload_revision?(snapshot)
+
+      payload_names.each do |name|
+        payload = PayloadBroadcast.new(
+          snapshot:,
+          name:,
+          authorization_context: connection
+        ).call
+        transmit TurboStreamRenderer.state_payload(payload)
+      rescue Unauthorized
+        next
+      end
+      @payload_revision = [ snapshot.instance_id, snapshot.revision ]
+    end
+
+    # @rbs (ActorSnapshot) -> bool
+    def newer_payload_revision?(snapshot)
+      current = @payload_revision
+      return true unless current
+
+      (current <=> [ snapshot.instance_id, snapshot.revision ]) == -1
+    end
+
+    # @rbs () -> void
+    def validate_payload_names!
+      return unless payload_names
+
+      broadcasts = SolidObjects
+        .registry
+        .fetch(reference.actor_type)
+        .definition
+        .payload_broadcasts
+      unknown = payload_names.find { |name| !broadcasts.key?(name.to_sym) }
+      return unless unknown
+
+      raise InvalidStreamToken, "unknown payload broadcast #{unknown.inspect}"
     end
 
     # @rbs (ActorSnapshot) -> void