# 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