feat: implement CSV user import with real-time progress

- Background job for processing CSV imports asynchronously
- Real-time progress tracking via Action Cable channels
- Dashboard channel for live metrics updates
- Import progress channel for upload status
- Queue and cable configuration for background processing

Author: lucasleandro1

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,20 @@
+module ApplicationCable
+  class Connection < ActionCable::Connection::Base
+    identified_by :current_user
+
+    def connect
+      self.current_user = find_verified_user
+    end
+
+    private
+
+    def find_verified_user
+      # Try to find user from session
+      if verified_user = env["warden"]&.user
+        verified_user
+      else
+        reject_unauthorized_connection
+      end
+    end
+  end
+end

app/channels/dashboard_channel.rb

@@ -0,0 +1,15 @@
+class DashboardChannel < ApplicationCable::Channel
+  def subscribed
+    # Only allow admin users to subscribe to dashboard updates
+    if current_user&.admin?
+      stream_from "dashboard_updates"
+    else
+      reject
+    end
+  end
+
+  def unsubscribed
+    # Any cleanup needed when channel is unsubscribed
+    stop_all_streams
+  end
+end

app/channels/import_progress_channel.rb

@@ -0,0 +1,16 @@
+class ImportProgressChannel < ApplicationCable::Channel
+  def subscribed
+    return reject unless current_user&.admin?
+    return reject unless params[:import_id].present?
+
+    import = Import.find_by(id: params[:import_id])
+    return reject unless import
+
+    stream_from "import_#{import.id}"
+  end
+
+  def unsubscribed
+    # Any cleanup needed when channel is unsubscribed
+    stop_all_streams
+  end
+end

app/jobs/application_job.rb

@@ -0,0 +1,7 @@
+class ApplicationJob < ActiveJob::Base
+  # Automatically retry jobs that encountered a deadlock
+  # retry_on ActiveRecord::Deadlocked
+
+  # Most jobs are safe to ignore if the underlying records are no longer available
+  # discard_on ActiveJob::DeserializationError
+end

app/jobs/user_import_job.rb

@@ -0,0 +1,155 @@
+class UserImportJob < ApplicationJob
+  queue_as :default
+
+  def perform(import)
+    import.update!(status: "processing", processed_rows: 0, successful_rows: 0, failed_rows: 0)
+
+    begin
+      process_import(import)
+      import.update!(status: "completed")
+    rescue StandardError => e
+      import.update!(status: "failed")
+      import.add_error("Import failed: #{e.message}")
+      raise e
+    end
+  end
+
+  private
+
+  def process_import(import)
+    file_path = download_file(import)
+
+    # Use Roo to parse the spreadsheet
+    spreadsheet = open_spreadsheet(file_path, import.file_name)
+    headers = spreadsheet.row(1)
+
+    validate_headers(headers, import)
+
+    total_rows = spreadsheet.last_row - 1 # Exclude header row
+    import.update!(total_rows: total_rows)
+
+    (2..spreadsheet.last_row).each_with_index do |row_num, index|
+      row = spreadsheet.row(row_num)
+      process_row(row, headers, import)
+
+      # Update progress every 10 rows or on last row
+      if (index + 1) % 10 == 0 || (index + 1) == total_rows
+        import.update_progress!
+        # Broadcast progress via ActionCable
+        broadcast_progress(import)
+      end
+    end
+
+    # Clean up temporary file
+    File.delete(file_path) if File.exist?(file_path)
+  end
+
+  def download_file(import)
+    temp_file = Tempfile.new([ import.file_name, File.extname(import.file_name) ])
+    temp_file.binmode
+    temp_file.write(import.file.download)
+    temp_file.close
+    temp_file.path
+  end
+
+  def open_spreadsheet(file_path, filename)
+    case File.extname(filename).downcase
+    when ".csv"
+      Roo::CSV.new(file_path)
+    when ".xls"
+      Roo::Excel.new(file_path)
+    when ".xlsx"
+      Roo::Excelx.new(file_path)
+    else
+      raise "Unknown file type: #{filename}"
+    end
+  end
+
+  def validate_headers(headers, import)
+    required_headers = [ "full_name", "email" ]
+    optional_headers = [ "role", "avatar_url" ]
+
+    missing_headers = required_headers - headers.map(&:to_s).map(&:downcase)
+
+    if missing_headers.any?
+      raise "Missing required headers: #{missing_headers.join(', ')}"
+    end
+  end
+
+  def process_row(row, headers, import)
+    begin
+      user_data = build_user_data(row, headers)
+
+      user = User.find_by(email: user_data[:email])
+
+      if user
+        # Update existing user
+        user.update!(user_data.except(:email))
+        import.increment!(:successful_rows)
+      else
+        # Create new user
+        user = User.create!(user_data.merge(password: generate_password))
+        import.increment!(:successful_rows)
+      end
+
+      import.increment!(:processed_rows)
+
+    rescue StandardError => e
+      import.increment!(:failed_rows)
+      import.increment!(:processed_rows)
+      import.add_error("Row #{import.processed_rows}: #{e.message}")
+    end
+  end
+
+  def build_user_data(row, headers)
+    data = {}
+
+    headers.each_with_index do |header, index|
+      value = row[index]
+      next if value.blank?
+
+      case header.to_s.downcase
+      when "full_name"
+        data[:full_name] = value.to_s.strip
+      when "email"
+        data[:email] = value.to_s.strip.downcase
+      when "role"
+        role = value.to_s.strip.downcase
+        data[:role] = %w[admin user].include?(role) ? role : "user"
+      when "avatar_url"
+        data[:avatar_url] = value.to_s.strip if valid_url?(value.to_s.strip)
+      end
+    end
+
+    data
+  end
+
+  def generate_password
+    SecureRandom.alphanumeric(12)
+  end
+
+  def valid_url?(url)
+    uri = URI.parse(url)
+    %w[http https].include?(uri.scheme)
+  rescue URI::InvalidURIError
+    false
+  end
+
+  def broadcast_progress(import)
+    ActionCable.server.broadcast(
+      "import_#{import.id}",
+      {
+        type: "progress_update",
+        import: {
+          id: import.id,
+          progress: import.progress,
+          processed_rows: import.processed_rows,
+          total_rows: import.total_rows,
+          successful_rows: import.successful_rows,
+          failed_rows: import.failed_rows,
+          status: import.status
+        }
+      }
+    )
+  end
+end

config/cable.yml

@@ -0,0 +1,17 @@
+# Async adapter only works within the same process, so for manually triggering cable updates from a console,
+# and seeing results in the browser, you must do so from the web console (running inside the dev process),
+# not a terminal started via bin/rails console! Add "console" to any action or any ERB template view
+# to make the web console appear.
+development:
+  adapter: async
+
+test:
+  adapter: test
+
+production:
+  adapter: solid_cable
+  connects_to:
+    database:
+      writing: cable
+  polling_interval: 0.1.seconds
+  message_retention: 1.day

config/queue.yml

@@ -0,0 +1,18 @@
+default: &default
+  dispatchers:
+    - polling_interval: 1
+      batch_size: 500
+  workers:
+    - queues: "*"
+      threads: 3
+      processes: <%= ENV.fetch("JOB_CONCURRENCY", 1) %>
+      polling_interval: 0.1
+
+development:
+  <<: *default
+
+test:
+  <<: *default
+
+production:
+  <<: *default

config/recurring.yml

@@ -0,0 +1,15 @@
+# examples:
+#   periodic_cleanup:
+#     class: CleanSoftDeletedRecordsJob
+#     queue: background
+#     args: [ 1000, { batch_size: 500 } ]
+#     schedule: every hour
+#   periodic_cleanup_with_command:
+#     command: "SoftDeletedRecord.due.delete_all"
+#     priority: 2
+#     schedule: at 5am every day
+
+production:
+  clear_solid_queue_finished_jobs:
+    command: "SolidQueue::Job.clear_finished_in_batches(sleep_between_batches: 0.3)"
+    schedule: every hour at minute 12