SRP 8 - Streams

08-board-partial.html.erb

<!-- app/views/boards/_board.html.erb -->
<ul class="..."
    data-controller="board"
    data-board-signed-resource-value="<%= board.to_sgid.to_s %>"
    id="<%= dom_id(board) %>">
  <!-- ... -->
</ul>

08-board-stimulus-controller.js

// app/javascript/controllers/board_controller.js

import { Controller } from "stimulus";
import CableReady from "cable_ready";
import consumer from "../channels/consumer";

export default class extends Controller {
  static values = { id: Number, signedResource: String };

  connect() {
    // ...
    this.presenceChannel = consumer.subscriptions.create(
      {
        channel: "PresenceChannel",
        signed_resource: this.signedResourceValue
      },
      {
        received(data) {
          if (data.cableReady) {
            CableReady.perform(data.operations);
          }
        }
      }
    );
  }

  disconnect() {
    this.presenceChannel.unsubscribe();
  }
}

08-development.rb

# ...

config.hosts << "mosoundic.eu.ngrok.io"

08-dropbox-controller.rb

# app/controllers/dropbox_controller.rb

class DropboxController < ApplicationController
  def webhook
    # verify
    render plain: params[:challenge] if params[:challenge].present?

    params["list_folder"]["accounts"].map do |account|
      ProcessDropboxWebhookJob.perform_later(account: account)
    end

    head :ok
  end
end

08-dropbox-routes.rb

# config/routes.rb

Rails.application.routes.draw do
  get "dropbox/webhook"
  post "dropbox/webhook"

  # ...
end

08-dropbox-sync-component-1.html.erb

<div class="flex-shrink flex justify-end items-center space-x-2 md:space-x-3 h-full" id="<%= dom_id(@embed) %>-dropbox-sync">
  <%= button_tag type: "button", class: klass do %>
    <i class="fab fa-dropbox fa-lg"></i>
  <% end %>
</div>

<div class="flex-shrink flex justify-end items-center space-x-2 md:space-x-3 h-full" id="<%= dom_id(@embed) %>-dropbox-sync">
  <%= button_tag type: "button", class: klass, data: {reflex: "click->DropboxSync#link", reflex_dataset: "combined"}, disabled: @embed.embeddable.syncing? do %>
    <i class="fab fa-dropbox fa-lg"></i>
  <% end %>
</div>

08-dropbox-sync-component.rb

class DropboxSyncComponent < ViewComponent::Base
  def initialize(embed:)
    @embed = embed
  end

  def render?
    @embed.embeddable.is_a? MediaAsset
  end

  def klass
    "inline-flex items-center px-1 py-0.5 rounded-md text-sm space-x-1 font-medium border h-full #{sync_state_klass}"
  end

  private

  def sync_state_klass
    case @embed.embeddable
    when ->(e) { e.unsynced? } then "text-gray-400"
    when ->(e) { e.syncing? } then "text-orange-400 animate-pulse"
    when ->(e) { e.synced? } then "text-green-400"
    end
  end
end

08-dropbox-sync-reflex.rb

# app/reflexes/dropbox_sync_reflex.rb

class DropboxSyncReflex < ApplicationReflex
  def link
    embed = element.signed[:sgid]

    DropboxUploadJob.perform_later(embed: embed)
    embed.embeddable.update(dropbox_sync_status: :syncing)

    morph dom_id(embed, embed.uuid), render(partial: "embeds/embed", locals: {embed: embed, user: current_user, board: embed.board})
  end
end

08-dropbox-upload-job

# app/jobs/dropbox_upload_job.rb

class DropboxUploadJob < ApplicationJob
  queue_as :default

  def perform(embed:)
    @dropbox_client = DropboxApi::Client.new(Rails.application.credentials[:dropbox_oauth_bearer])

    folder_name = embed.board.slug

    media_file = embed.embeddable.media_file
    filename = media_file.blob.filename

    begin
      @dropbox_client.list_folder "/#{folder_name}"
    rescue DropboxApi::Errors::NotFoundError
      @dropbox_client.create_folder "/#{folder_name}"
    end

    media_file.open do |file|
      @dropbox_client.upload_by_chunks "/#{folder_name}/#{filename}", file
    end
  end
end

08-embed-partial.html.erb

<li class="..." id="<%= dom_id(embed, embed.uuid) %>" data-sgid="<%= embed.to_sgid.to_s %>">
  <!-- ... -->
  <%= render(DropboxSyncComponent.new(embed: embed)) %>
  <!-- ... -->
</li>

08-mediaasset-sync-status-enum

# migration
class AddDropboxSyncStatusToMediaAssets < ActiveRecord::Migration[6.1]
  def change
    add_column :media_assets, :dropbox_sync_status, :integer, default: 0

    MediaAsset.update_all(dropbox_sync_status: 0)
  end
end

# app/models/media_asset.rb
class MediaAsset < ApplicationRecord
  # ...

  enum dropbox_sync_status: %i[unsynced syncing synced], _default: :unsynced

  # ...
end

08-presence-channel.rb

class PresenceChannel < ApplicationCable::Channel
  def subscribed
    resource = GlobalID::Locator.locate_signed params[:signed_resource]
    if resource.present?
      stream_from "presence:#{resource.id}"
      resource.present_users.add(current_user.id)
    else
      reject
    end
  end

  def unsubscribed
    resource = GlobalID::Locator.locate_signed params[:signed_resource]
    return unless resource.present?

    resource.present_users.remove(current_user.id)
  end
end

08-presence-indicator.html.erb

<div id="presence-indicator">
  <%= render(UserAvatarGroupComponent.new(users: users)) %>
</div>

08-process-dropbox-upload-job.rb

# app/jobs/process_dropbox_webhook_job.rb

class ProcessDropboxWebhookJob < ApplicationJob
  queue_as :default

  def perform(account:)
    @dropbox_client = DropboxApi::Client.new(Rails.application.credentials[:dropbox_oauth_bearer])
    cursor = Rails.cache.read("dropbox_cursor:#{account}")
    has_more = true

    while has_more
      result = if cursor.present?
        @dropbox_client.list_folder_continue cursor
      else
        @dropbox_client.list_folder "", {recursive: true}
      end

      result.entries.map do |entry|
        if entry.is_a? DropboxApi::Metadata::File
          match = entry.path_lower.match /\/(?<board>.+)\/(?<filename>.+)/
          board = Board.friendly.find(match[:board])
          embed = board.embeds
            .joins("INNER JOIN media_assets ma ON ma.id = embeds.embeddable_id")
            .joins("INNER JOIN active_storage_attachments asa ON ma.id = asa.record_id")
            .joins("INNER JOIN active_storage_blobs asb on asa.blob_id = asb.id")
            .where("asb.filename = ?", match[:filename]).first

          embed.embeddable.update(dropbox_sync_status: :synced)
        end
      end

      cursor = result.cursor
      Rails.cache.write("dropbox_cursor:#{account}", cursor)

      has_more = result.has_more?
    end
  end
end

08-stream-presence-job.rb

# app/jobs/stream_presence_job.rb

class StreamPresenceJob < ApplicationJob
  include CableReady::Broadcaster
  queue_as :default

  def perform(resource:)
    cable_ready["presence:#{resource.id}"].outer_html(
      selector: "#presence-indicator",
      html: ApplicationController.render(partial: "shared/presence_indicator", locals: {users: User.where(id: resource.present_users.members)})
    ).broadcast
  end
end

08-user-notifiable.rb

# app/models/concerns/user_notifiable.rb

module UserNotifiable
  extend ActiveSupport::Concern
  include StimulusReflex::ConcernEnhancer
  include CableReady::Broadcaster

  included do
    after_commit :notify_users, unless: :skip_callbacks
  end

  def notify_users
    NotifyUsersJob.perform_later(changes: cable_ready_changes)
  end

  def cable_ready_changes
    users = if try(:user).present?
      self.users&.without(user)
    else
      self.users
    end

    users.map do |user|
      operations = yield user
      [user, operations]
    end
  end
end

# app/models/board.rb
class Board < ApplicationRecord
  # ...

  def cable_ready_changes
    super do |user|
      cable_car.morph(
        selector: dom_id(self),
        html: ApplicationController.render(self, locals: {embeds: embeds.order(created_at: :asc), embed_templates: user.embed_templates, user: user})
      ).dispatch
    end
  end
end

08-user-presence.rb

# app/models/concerns/user_presence.rb

module UserPresence
  extend ActiveSupport::Concern

  included do
    kredis_set :present_users, after_change: :stream_presence_later
  end

  def stream_presence_later
    StreamPresenceJob.perform_later(resource: self)
  end
end