Decode Action Cable broadcasts

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.4.2 - 2026-08-07
+
+- Decode Action Cable broadcast payloads before parsing observable invalidations
+  so scalar updates and component refreshes transmit as raw Turbo Stream HTML.
+
 ## 0.4.1 - 2026-08-07
 
 - Load `SolidObjects::ActorChannel` with the gem and pass stream and component

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_objects (0.4.1)
+    solid_objects (0.4.2)
       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.4.1)
+  solid_objects (0.4.2)
   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

lib/solid_objects/actor_channel.rb

@@ -24,7 +24,7 @@ def subscribed
         params["components"],
         reference:
       )
-      stream_from StreamName.for(reference) do |stream|
+      stream_from StreamName.for(reference), coder: ActiveSupport::JSON do |stream|
         receive_broadcast(stream)
       end
       snapshot = ActorSnapshot.new(reference)

lib/solid_objects/version.rb

@@ -1,5 +1,5 @@
 # rbs_inline: enabled
 
 module SolidObjects
-  VERSION = "0.4.1"
+  VERSION = "0.4.2"
 end

test/integration/actor_channel_test.rb

@@ -35,6 +35,15 @@ def update_missing
     end
   end
 
+  class CapturingActorChannel < SolidObjects::ActorChannel
+    attr_reader :stream_callback, :stream_coder
+
+    def stream_from(_broadcasting, callback = nil, coder: nil, &block)
+      @stream_callback = callback || block
+      @stream_coder = coder
+    end
+  end
+
   setup do
     SolidObjects.reset!
     ChannelActor.ensure_registered!
@@ -115,6 +124,71 @@ def update_missing
     worker&.stop
   end
 
+  test "decodes Action Cable broadcasts before reactive processing" do
+    reference = ChannelActor.ref("actor-1")
+    SolidObjects.configuration.authorize_subscription = ->(**) { true }
+    connection = ActionCable::Channel::ConnectionStub.new
+    channel = CapturingActorChannel.new(
+      connection,
+      "actor-channel",
+      {
+        token: SolidObjects::StreamToken.generate(
+          reference,
+          observables: %w[missing]
+        ),
+        components: JSON.generate(
+          [
+            component_token(
+              reference,
+              component_name: "summary",
+              dependencies: %w[missing],
+              revision: 0
+            )
+          ]
+        )
+      }.with_indifferent_access
+    )
+    channel.subscribe_to_channel
+
+    assert_equal ActiveSupport::JSON, channel.stream_coder
+    connection.transmissions.clear
+
+    reference.async(:update_missing)
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+    broadcast = SolidObjects::Broadcast.find_by!(observable_name: "missing")
+    SolidObjects::ActionCableBroadcastAdapter.new.call(broadcast)
+    stream_name = SolidObjects::StreamName.for(reference)
+    encoded_stream = ActionCable.server.pubsub.broadcasts(stream_name).sole
+    handler = channel.__send__(
+      :stream_handler,
+      stream_name,
+      channel.stream_callback,
+      coder: channel.stream_coder
+    )
+
+    handler.call(encoded_stream)
+
+    channel_transmissions = connection.transmissions.filter_map do |transmission|
+      transmission["message"]
+    end
+    scalar_target = SolidObjects::DomIdentity.observable(reference, :missing)
+    scalar_update = channel_transmissions.find do |transmission|
+      transmission.include?(scalar_target) &&
+        transmission.include?(">1</span>")
+    end
+    assert scalar_update
+    assert_equal 1,
+      component_refreshes(
+        reference,
+        :summary,
+        messages: channel_transmissions
+      ).length
+    refute channel_transmissions.any? { |transmission| transmission.start_with?("\"") }
+  ensure
+    worker&.stop
+  end
+
   test "streams only after token verification and host authorization" do
     reference = ChannelActor.ref("actor-1")
     SolidObjects.configuration.authorize_subscription = lambda do |actor_type:, actor_id:, authorization_context:|
@@ -371,9 +445,9 @@ def component_token(reference, component_name:, dependencies:, revision:)
     )
   end
 
-  def component_refreshes(reference, component_name)
+  def component_refreshes(reference, component_name, messages: transmissions)
     target = SolidObjects::DomIdentity.component(reference, component_name)
-    transmissions.select do |transmission|
+    messages.select do |transmission|
       transmission.include?(%(target="#{target}")) &&
         transmission.include?("<turbo-frame")
     end