Skip to content

Commit 388e59c

Browse files
authored
Merge pull request #8 from cardmagic/agent/state-payload-broadcasts
feat: add personalized state payload broadcasts
2 parents 4385728 + ccb23bd commit 388e59c

25 files changed

Lines changed: 692 additions & 43 deletions

CHANGELOG.md

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,18 @@
11
# Changelog
22

3+
## 0.6.0 - 2026-08-09
4+
5+
- Add `broadcast_payload`, an actor DSL for sending one personalized JSON state
6+
payload over the actor stream a page already has open. The block runs once per
7+
subscriber with that subscriber's authorization context, so private state
8+
never crosses sessions. Payloads carry actor identity and the monotonic state
9+
revision, and both the channel and the browser drop stale revisions. Subscribe
10+
with `solid_object room, payloads: :playmat_state` and handle the
11+
`solid-objects:payload` DOM event. ERB component refreshes remain the default
12+
and are unchanged. A mutation that changes payload state without changing a
13+
declared observable still invalidates subscribers, through a revision-only
14+
broadcast that carries no observable value to the browser.
15+
316
## 0.5.2 - 2026-08-09
417

518
- Read the database clock once per transaction instead of once per step, and

Gemfile.lock

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
PATH
22
remote: .
33
specs:
4-
solid_objects (0.5.2)
4+
solid_objects (0.6.0)
55
actioncable (>= 8.0)
66
actionpack (>= 8.0)
77
actionview (>= 8.0)
@@ -373,7 +373,7 @@ CHECKSUMS
373373
rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d
374374
ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33
375375
securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1
376-
solid_objects (0.5.2)
376+
solid_objects (0.6.0)
377377
sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc
378378
sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d
379379
sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b
Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
const deliveredRevisions = new Map()
2+
3+
class SolidObjectsPayloadElement extends HTMLElement {
4+
connectedCallback() {
5+
if (this.dataset.started === "true") return
6+
7+
this.dataset.started = "true"
8+
this.deliver()
9+
}
10+
11+
deliver() {
12+
try {
13+
const name = this.dataset.name
14+
const revision = revisionFor(this)
15+
const scope = this.closest("[id]")
16+
if (!name || !revision || !scope) return
17+
18+
const key = `${scope.id}:${name}`
19+
if (!newerRevision(revision, deliveredRevisions.get(key))) return
20+
21+
const payload = JSON.parse(this.textContent)
22+
deliveredRevisions.set(key, revision)
23+
scope.dispatchEvent(
24+
new CustomEvent("solid-objects:payload", {
25+
bubbles: true,
26+
detail: {
27+
name,
28+
instanceId: revision[0],
29+
revision: revision[1],
30+
payload
31+
}
32+
})
33+
)
34+
} catch {
35+
this.dispatchEvent(
36+
new CustomEvent("solid-objects:payload-error", {
37+
bubbles: true,
38+
detail: { reason: "invalid_payload" }
39+
})
40+
)
41+
} finally {
42+
this.remove()
43+
}
44+
}
45+
}
46+
47+
function newerRevision(candidate, current) {
48+
if (!current) return true
49+
50+
return candidate[0] > current[0] ||
51+
(candidate[0] === current[0] && candidate[1] > current[1])
52+
}
53+
54+
function revisionFor(element) {
55+
const revision = element.dataset.revision
56+
if (!revision) return
57+
58+
const values = revision.split(":").map(Number)
59+
if (
60+
values.length !== 2 ||
61+
values.some((value) => !Number.isSafeInteger(value) || value < 0)
62+
) return
63+
64+
return values
65+
}
66+
67+
if (!customElements.get("solid-objects-payload")) {
68+
customElements.define("solid-objects-payload", SolidObjectsPayloadElement)
69+
}

app/helpers/solid_objects/actor_helper.rb

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,9 @@
22

33
module SolidObjects
44
module ActorHelper
5-
# @rbs (Reference, ?authorization_context: untyped) { (ActorView) -> untyped } -> untyped
6-
def solid_object(reference, authorization_context: self, &block)
5+
# @rbs (Reference, ?authorization_context: untyped, ?payloads: untyped) { (ActorView) -> untyped } -> untyped
6+
def solid_object(reference, authorization_context: self, payloads: nil, &block)
7+
payload_names = Array(payloads).map(&:to_s).uniq.presence
78
actor = ActorView.new(
89
reference:,
910
view_context: self,
@@ -13,7 +14,8 @@ def solid_object(reference, authorization_context: self, &block)
1314
subscription_data = {
1415
token: StreamToken.generate(
1516
reference,
16-
observables: actor.scalar_observable_names
17+
observables: actor.scalar_observable_names,
18+
payloads: payload_names
1719
)
1820
}
1921
if actor.component_tokens.any?
@@ -30,10 +32,17 @@ def solid_object(reference, authorization_context: self, &block)
3032
data: { turbo_track: "reload" }
3133
)
3234
end
35+
payload_client = if payload_names
36+
javascript_include_tag(
37+
"solid_objects/state_payload",
38+
type: "module",
39+
data: { turbo_track: "reload" }
40+
)
41+
end
3342

3443
content_tag(
3544
:div,
36-
safe_join([ refresh_client, subscription, content ].compact),
45+
safe_join([ refresh_client, payload_client, subscription, content ].compact),
3746
id: DomIdentity.scope(reference)
3847
)
3948
end

docs/realtime.md

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,93 @@ applications discover the namespaced engine asset. Applications created with
110110
explicitly serve the module. Turbo's normal morph rules still apply; use
111111
`data-turbo-permanent` for elements that must never be changed.
112112

113+
## Personalized state payloads
114+
115+
Reactive ERB components cost one browser request per changed component. When a
116+
single actor mutation changes several components, an application pays several
117+
round trips for one logical update. A payload broadcast collapses that into one
118+
message on the stream the page already has open.
119+
120+
Declare the payload on the actor. The block receives the actor and the
121+
subscriber's authorization context, and it runs **once per subscriber**, so two
122+
sessions watching the same actor never see each other's private state:
123+
124+
```ruby
125+
class PlaymatRoom < SolidObjects::Actor
126+
actor_type "playmat_room"
127+
128+
attribute :hands, default: -> { {} }
129+
attribute :turn, default: 1
130+
131+
observable :turn
132+
133+
broadcast_payload :playmat_state do |room, authorization_context|
134+
{
135+
"turn" => room.turn,
136+
"hand" => room.hands.fetch(authorization_context.session_id, [])
137+
}
138+
end
139+
end
140+
```
141+
142+
Subscribe the scope to it:
143+
144+
```erb
145+
<%= solid_object room, payloads: :playmat_state do |actor| %>
146+
<div data-playmat></div>
147+
<% end %>
148+
```
149+
150+
Handle it with any JavaScript. The gem dispatches a DOM event and requires no
151+
framework:
152+
153+
```javascript
154+
document.addEventListener("solid-objects:payload", (event) => {
155+
const { name, revision, payload } = event.detail
156+
if (name !== "playmat_state") return
157+
158+
renderPlaymat(payload)
159+
})
160+
```
161+
162+
### What the protocol guarantees
163+
164+
The payload travels as a Turbo Stream element on the existing actor stream, so
165+
applications do not run a second WebSocket system. Each message carries the
166+
actor identity plus the `instance_id` and monotonic `state_revision` that fence
167+
component refreshes, and both the channel and the browser drop a payload that
168+
is not newer than the last one delivered for that scope and name. A reconnecting
169+
client receives the current payload on subscribe.
170+
171+
Authorization is the same `authorize_query` boundary that components use, called
172+
with the payload name and the subscriber's Cable connection. A subscriber that
173+
fails the check is skipped rather than served a partial payload, and the payload
174+
name is signed into the stream token, so a browser cannot ask for a payload the
175+
server did not offer.
176+
177+
Payload blocks read committed actor state through the same snapshot components
178+
use. They cannot write application records, and the return value must be a JSON
179+
object or array so the wire format stays inspectable.
180+
181+
### mtg-playmat before and after
182+
183+
Before, one mutation that touched three observables produced three refresh
184+
elements and three HTTP requests:
185+
186+
```
187+
commit -> 3 Action Cable messages -> 3 GET /solid_objects/components -> 3 renders
188+
```
189+
190+
After, the same mutation delivers one personalized payload and the page renders
191+
once:
192+
193+
```
194+
commit -> 1 Action Cable message -> 0 HTTP requests -> 1 render
195+
```
196+
197+
Components remain the default. An actor with no `broadcast_payload` and a scope
198+
with no `payloads:` option behave exactly as before.
199+
113200
## Authorization
114201

115202
The HTML contains a signed actor identity token. Signing prevents modification;

lib/solid_objects.rb

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
require "solid_objects/component_subscriptions"
4141
require "solid_objects/component_view"
4242
require "solid_objects/component_renderer"
43+
require "solid_objects/payload_broadcast"
4344
require "solid_objects/state_snapshot"
4445
require "solid_objects/actor_view"
4546
require "solid_objects/actor_channel"

lib/solid_objects/actor.rb

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,13 @@ def observable(name, &block)
5757
definition.add_observable(name, block)
5858
end
5959

60+
# @rbs (Symbol | String) { (untyped, untyped) -> untyped } -> ActorDefinition::Handler
61+
def broadcast_payload(name, &block)
62+
raise InvalidActor, "payload broadcasts require a block" unless block
63+
64+
definition.add_payload_broadcast(name, block)
65+
end
66+
6067
# @rbs (?Integer) -> Integer
6168
def state_version(version = nil)
6269
definition.set_state_version(version) if version

lib/solid_objects/actor_channel.rb

Lines changed: 55 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,9 @@ def subscribed
1919

2020
@reference = Reference.new(actor_type:, actor_id:)
2121
@scalar_observables = identity["observables"]
22+
@payload_names = identity["payloads"]
2223
validate_scalar_observables!
24+
validate_payload_names!
2325
@component_subscriptions = ComponentSubscriptions.parse(
2426
params["components"],
2527
reference:
@@ -36,6 +38,7 @@ def subscribed
3638
)
3739
end
3840
refresh_outdated_components(snapshot)
41+
transmit_state_payloads(snapshot)
3942
rescue KeyError,
4043
JSON::ParserError,
4144
InvalidStreamToken,
@@ -46,21 +49,69 @@ def subscribed
4649

4750
private
4851

49-
attr_reader :reference, :component_subscriptions, :scalar_observables
52+
attr_reader :reference,
53+
:component_subscriptions,
54+
:scalar_observables,
55+
:payload_names
5056

5157
# @rbs (String) -> void
5258
def receive_broadcast(stream)
5359
invalidation = TurboStreamRenderer.invalidation(stream)
54-
if !invalidation ||
55-
scalar_observables.nil? ||
56-
scalar_observables.include?(invalidation.fetch("observable_name"))
60+
revision_only = invalidation &&
61+
invalidation.fetch("observable_name") == PayloadBroadcast::REVISION_OBSERVABLE
62+
if !revision_only &&
63+
(!invalidation ||
64+
scalar_observables.nil? ||
65+
scalar_observables.include?(invalidation.fetch("observable_name")))
5766
transmit stream
5867
end
5968
return unless invalidation
6069

6170
component_subscriptions
6271
.refreshes_for(invalidation)
6372
.each { |refresh| transmit refresh }
73+
transmit_state_payloads(ActorSnapshot.new(reference))
74+
end
75+
76+
# @rbs (ActorSnapshot) -> void
77+
def transmit_state_payloads(snapshot)
78+
return if payload_names.nil? || payload_names.empty?
79+
return unless newer_payload_revision?(snapshot)
80+
81+
payload_names.each do |name|
82+
payload = PayloadBroadcast.new(
83+
snapshot:,
84+
name:,
85+
authorization_context: connection
86+
).call
87+
transmit TurboStreamRenderer.state_payload(payload)
88+
rescue Unauthorized
89+
next
90+
end
91+
@payload_revision = [ snapshot.instance_id, snapshot.revision ]
92+
end
93+
94+
# @rbs (ActorSnapshot) -> bool
95+
def newer_payload_revision?(snapshot)
96+
current = @payload_revision
97+
return true unless current
98+
99+
(current <=> [ snapshot.instance_id, snapshot.revision ]) == -1
100+
end
101+
102+
# @rbs () -> void
103+
def validate_payload_names!
104+
return unless payload_names
105+
106+
broadcasts = SolidObjects
107+
.registry
108+
.fetch(reference.actor_type)
109+
.definition
110+
.payload_broadcasts
111+
unknown = payload_names.find { |name| !broadcasts.key?(name.to_sym) }
112+
return unless unknown
113+
114+
raise InvalidStreamToken, "unknown payload broadcast #{unknown.inspect}"
64115
end
65116

66117
# @rbs (ActorSnapshot) -> void

0 commit comments

Comments
 (0)