Live collaboration with Yjs over ActionCable

All editors now work on a server document at the same time instead of
taking turns behind an edit lock.

- The document's content is a sequence of Yjs updates (document_updates);
  DocumentChannel stores and relays them immediately, checks access on
  subscribe and again whenever links or memberships change.
- app/frontend/collab/: the store stays position-based; a binding maps it
  to a Yjs form with stable row/column ids, so a row inserted by one
  person doesn't misplace what another is typing. Remote changes are
  applied only after pending local ones went into Yjs.
- Per-person undo (Y.UndoManager), presence via awareness: the others'
  selections as coloured boxes with names, avatars in the toolbar.
- Existing JSON documents convert on first open, deterministically, so
  two people opening one at once end up with the same Yjs state.
- Editors compact the stored updates after 200 and keep a JSON snapshot
  current for export and duplicating. Edit lock and polling are gone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/channels/application_cable/channel.rb

@@ -0,0 +1,4 @@
+module ApplicationCable
+  class Channel < ActionCable::Channel::Base
+  end
+end

app/channels/application_cable/connection.rb

@@ -0,0 +1,12 @@
+module ApplicationCable
+  # The WebSocket knows the signed-in user from the same session cookie as
+  # the rest of the app. Anonymous connections are allowed: a share link
+  # token (a subscription parameter) can grant access on its own.
+  class Connection < ActionCable::Connection::Base
+    identified_by :current_user
+
+    def connect
+      self.current_user = Session.find_by(id: cookies.signed[:session_id])&.user if cookies.signed[:session_id]
+    end
+  end
+end

app/channels/document_channel.rb

@@ -0,0 +1,114 @@
+# Live collaboration on one document (Yjs). Every edit is a Yjs update: the
+# server stores it and passes it on to everyone connected; it never needs
+# to understand what's inside. Binary data travels Base64-encoded.
+#
+# From the browser (perform):
+#   update     { data }                  an edit (editors only)
+#   awareness  { data }                  cursor/selection/name, not stored
+#   compact    { state, ids, snapshot }  replace the listed updates by one
+#                                        state; snapshot = JSON of the document
+#   snapshot   { content }               JSON of the document, for export etc.
+#
+# To the browser:
+#   init       { role, title, updates: [{ id, data }], content? }
+#              content (JSON) only while no update exists yet: the browser
+#              then builds the Yjs form and sends it as the first update
+#   update     { id, data, from }        from = the sending subscription
+#   awareness  { data, from }
+#   compacted  { removed: [ids], id }
+#   meta       { title }
+#   role       { role }                  access changed
+#   revoked                              access gone (or document deleted)
+class DocumentChannel < ApplicationCable::Channel
+  def subscribed
+    @document = Document.find_by(id: params[:id])
+    @token = params[:token].presence
+    @key = params[:key].to_s[0, 64]
+    @role = @document&.role_for(user: current_user, token: @token)
+    return reject unless @role
+
+    stream_for @document, coder: ActiveSupport::JSON do |message|
+      relay(message)
+    end
+    transmit init_message
+  end
+
+  def update(data)
+    return unless editor?
+    bytes = decode(data["data"]) or return
+    row = @document.updates.create!(data: bytes, user: current_user)
+    @document.update_columns(content_updated_at: Time.current, updated_by_id: current_user&.id)
+    broadcast({ type: "update", id: row.id, data: data["data"], from: @key })
+  rescue ActiveRecord::RecordInvalid
+    transmit({ type: "error", error: "too_large" })
+  end
+
+  def awareness(data)
+    return unless data["data"].is_a?(String) && data["data"].bytesize < 64.kilobytes
+    broadcast({ type: "awareness", data: data["data"], from: @key })
+  end
+
+  def compact(data)
+    return unless editor?
+    state = decode(data["state"]) or return
+    ids = Array(data["ids"]).map(&:to_i)
+    row = nil
+    @document.with_lock do
+      @document.updates.where(id: ids).delete_all
+      row = @document.updates.create!(data: state, user: current_user)
+      store_snapshot(data["snapshot"])
+    end
+    broadcast({ type: "compacted", removed: ids, id: row.id })
+  end
+
+  def snapshot(data)
+    return unless editor?
+    store_snapshot(data["content"])
+  end
+
+  private
+
+  def editor? = Document.role_at_least?(@role, :editor)
+
+  def broadcast(message) = DocumentChannel.broadcast_to(@document, message)
+
+  def relay(message)
+    case message["type"]
+    when "revalidate" then revalidate
+    when "awareness" then transmit(message) unless message["from"] == @key
+    else transmit(message)
+    end
+  end
+
+  # Membership or link changed: is this connection still allowed in?
+  def revalidate
+    role = Document.find_by(id: @document.id)&.role_for(user: current_user, token: @token)
+    if role.nil?
+      @role = nil
+      transmit({ type: "revoked" })
+      stop_all_streams
+    elsif role != @role
+      @role = role
+      transmit({ type: "role", role: })
+    end
+  end
+
+  def init_message
+    updates = @document.updates.order(:id).map(&:as_message)
+    message = { type: "init", role: @role, title: @document.title, updates: }
+    message[:content] = @document.content if updates.empty?
+    message
+  end
+
+  def decode(b64)
+    return unless b64.is_a?(String)
+    Base64.strict_decode64(b64)
+  rescue ArgumentError
+    nil
+  end
+
+  def store_snapshot(json)
+    return unless json.is_a?(String) && json.bytesize <= Document::MAX_CONTENT_BYTES && Document.valid_snapshot?(json)
+    @document.update_columns(content: json)
+  end
+end