feat: migrate dyte integration to cloudflare realtimekit (#14752)

Dyte is sunsetting its existing infrastructure after the Cloudflare
acquisition, so this migrates Chatwoot’s video call integration to
Cloudflare RealtimeKit.

The integration now uses Cloudflare Account ID, RealtimeKit App ID, and
a Cloudflare API token with Realtime Admin permissions. Meeting creation
and participant token generation now call Cloudflare’s RealtimeKit APIs,
while the existing Chatwoot call experience remains unchanged for agents
and customers.

This also adds setup-time credential validation, so admins get clearer
errors when the API token is invalid, the Cloudflare account or
permissions are incorrect, or the RealtimeKit App ID does not belong to
the selected account.

Fixes
https://linear.app/chatwoot/issue/PLA-176/migrate-dyte-integration-to-cloudflare-realtimekit

**How to test**

1. Go to Settings → Integrations → Cloudflare RealtimeKit.
2. Add a Cloudflare Account ID, RealtimeKit App ID, and API token with
Realtime Admin permissions.
3. Confirm the integration saves successfully with valid credentials.
4. Try invalid credentials and confirm the error identifies whether the
token, account/permissions, or app ID is wrong.
5. Start a video call from a conversation and confirm the RealtimeKit
meeting opens.

---------

Co-authored-by: Muhsin <12408980+muhsin-k@users.noreply.github.com>
Co-authored-by: Sony Mathew <sony@chatwoot.com>

lib/dyte.rb

@@ -1,13 +1,15 @@
 class Dyte
-  BASE_URL = 'https://api.dyte.io/v2'.freeze
+  BASE_URL = 'https://api.cloudflare.com/client/v4'.freeze
   API_KEY_HEADER = 'Authorization'.freeze
-  PRESET_NAME = 'group_call_host'.freeze
+  PRESET_NAME = 'group-call-host'.freeze
+  LEGACY_PRESET_NAME = 'group_call_host'.freeze
 
-  def initialize(organization_id, api_key)
-    @api_key = Base64.strict_encode64("#{organization_id}:#{api_key}")
-    @organization_id = organization_id
+  def initialize(account_id = nil, app_id = nil, api_token = nil)
+    @account_id = account_id
+    @app_id = app_id
+    @api_token = api_token
 
-    raise ArgumentError, 'Missing Credentials' if @api_key.blank? || @organization_id.blank?
+    raise ArgumentError, 'Missing Credentials' if @account_id.blank? || @app_id.blank? || @api_token.blank?
   end
 
   def create_a_meeting(title)
@@ -29,24 +31,65 @@ class Dyte
       'preset_name': PRESET_NAME
     }
     path = "meetings/#{meeting_id}/participants"
-    response = post(path, payload)
+    response = process_response(post(path, payload))
+    return response unless preset_not_found?(response)
+
+    payload[:preset_name] = LEGACY_PRESET_NAME
+    process_response(post(path, payload))
+  end
+
+  def refresh_participant_token(meeting_id, participant_id)
+    raise ArgumentError, 'Missing information' if meeting_id.blank? || participant_id.blank?
+
+    path = "meetings/#{meeting_id}/participants/#{participant_id}/token"
+    response = post(path)
+    process_response(response)
+  end
+
+  def fetch_participants(meeting_id)
+    raise ArgumentError, 'Missing information' if meeting_id.blank?
+
+    response = get("meetings/#{meeting_id}/participants")
     process_response(response)
   end
 
   private
 
   def process_response(response)
-    return response.parsed_response['data'].with_indifferent_access if response.success?
+    return { error: response.parsed_response, error_code: response.code } unless response.success?
 
-    { error: response.parsed_response, error_code: response.code }
+    data = parsed_data(response)
+    return data.with_indifferent_access if data.is_a?(Hash)
+    return data.map(&:with_indifferent_access) if data.is_a?(Array)
+
+    { error: :unexpected_response, error_code: response.code }
   end
 
-  def post(path, payload)
+  def parsed_data(response)
+    response.parsed_response['data']
+  end
+
+  def preset_not_found?(response)
+    error = response[:error]
+    message = error.dig('error', 'message') if error.is_a?(Hash) && error['error'].is_a?(Hash)
+    message ||= error['message'] if error.is_a?(Hash)
+    message ||= error.to_s
+    message.include?('No preset found')
+  end
+
+  def post(path, payload = nil)
     HTTParty.post(
-      "#{BASE_URL}/#{path}", {
-        headers: { API_KEY_HEADER => "Basic #{@api_key}", 'Content-Type' => 'application/json' },
-        body: payload.to_json
-      }
+      "#{BASE_URL}/accounts/#{@account_id}/realtime/kit/#{@app_id}/#{path}", {
+        headers: { API_KEY_HEADER => "Bearer #{@api_token}", 'Content-Type' => 'application/json' },
+        body: payload&.to_json
+      }.compact
+    )
+  end
+
+  def get(path)
+    HTTParty.get(
+      "#{BASE_URL}/accounts/#{@account_id}/realtime/kit/#{@app_id}/#{path}",
+      headers: { API_KEY_HEADER => "Bearer #{@api_token}", 'Content-Type' => 'application/json' }
     )
   end
 end

lib/integrations/cloudflare/realtime_kit_credentials_validator.rb

@@ -0,0 +1,104 @@
+module Integrations::Cloudflare::RealtimeKitCredentialsValidator
+  Result = Data.define(:success?, :error)
+
+  BASE_URL = 'https://api.cloudflare.com/client/v4'.freeze
+  TIMEOUT_SECONDS = 5
+  APPS_PAGE_SIZE = 50
+
+  def self.valid?(account_id, app_id, api_token)
+    validate(account_id, app_id, api_token).success?
+  end
+
+  def self.validate(account_id, app_id, api_token)
+    return failure(:missing_credentials) if account_id.blank? || app_id.blank? || api_token.blank?
+
+    token_result = validate_token(api_token)
+    return token_result unless token_result.success?
+
+    validate_realtimekit_app(account_id, app_id, api_token)
+  rescue Faraday::Error => e
+    Rails.logger.warn("[cloudflare-realtimekit-credentials-validator] #{e.class}: #{e.message}")
+    failure(:verification_failed)
+  end
+
+  def self.validate_token(api_token)
+    response = connection.get("#{BASE_URL}/user/tokens/verify") do |req|
+      req.headers['Authorization'] = "Bearer #{api_token}"
+    end
+
+    return failure(:verification_failed) if transient_error?(response)
+
+    body = parse_response(response)
+    return success if response.status == 200 && body['success'] == true && body.dig('result', 'status') == 'active'
+
+    failure(:invalid_api_token)
+  end
+  private_class_method :validate_token
+
+  def self.validate_realtimekit_app(account_id, app_id, api_token)
+    page_no = 1
+
+    loop do
+      response = fetch_realtimekit_apps(account_id, api_token, page_no)
+      return failure(:verification_failed) if transient_error?(response)
+      return failure(:invalid_account_or_permissions) unless response.status == 200
+
+      body = parse_response(response)
+      apps = body['data'] || []
+      return success if apps.any? { |app| app['id'] == app_id }
+      break unless next_apps_page?(body, page_no, apps)
+
+      page_no += 1
+    end
+
+    failure(:app_not_found)
+  end
+  private_class_method :validate_realtimekit_app
+
+  def self.fetch_realtimekit_apps(account_id, api_token, page_no)
+    connection.get("#{BASE_URL}/accounts/#{account_id}/realtime/kit/apps") do |req|
+      req.headers['Authorization'] = "Bearer #{api_token}"
+      req.params['page_no'] = page_no
+      req.params['per_page'] = APPS_PAGE_SIZE
+    end
+  end
+  private_class_method :fetch_realtimekit_apps
+
+  def self.next_apps_page?(body, page_no, apps)
+    total_count = body.dig('paging', 'total_count') || body.dig('result_info', 'total_count')
+    return page_no * APPS_PAGE_SIZE < total_count.to_i if total_count.present?
+
+    apps.size == APPS_PAGE_SIZE
+  end
+  private_class_method :next_apps_page?
+
+  def self.connection
+    Faraday.new do |f|
+      f.options.timeout = TIMEOUT_SECONDS
+      f.options.open_timeout = TIMEOUT_SECONDS
+    end
+  end
+  private_class_method :connection
+
+  def self.parse_response(response)
+    JSON.parse(response.body)
+  rescue JSON::ParserError
+    {}
+  end
+  private_class_method :parse_response
+
+  def self.transient_error?(response)
+    response.status >= 500
+  end
+  private_class_method :transient_error?
+
+  def self.success
+    Result.new(true, nil)
+  end
+  private_class_method :success
+
+  def self.failure(error)
+    Result.new(false, error)
+  end
+  private_class_method :failure
+end

lib/integrations/dyte/processor_service.rb

@@ -2,6 +2,8 @@ class Integrations::Dyte::ProcessorService
   pattr_initialize [:account!, :conversation!]
 
   def create_a_meeting(agent)
+    return missing_realtimekit_credentials_response if realtimekit_credentials_missing?
+
     title = I18n.t('integration_apps.dyte.meeting_name', agent_name: agent.available_name)
     response = dyte_client.create_a_meeting(title)
 
@@ -12,12 +14,31 @@ class Integrations::Dyte::ProcessorService
     message.push_event_data
   end
 
-  def add_participant_to_meeting(meeting_id, user)
-    dyte_client.add_participant_to_meeting(meeting_id, user.id, user.name, avatar_url(user))
+  def add_participant_to_meeting(meeting_id, user, message = nil)
+    return missing_realtimekit_credentials_response if realtimekit_credentials_missing?
+
+    client_id = realtimekit_client_id(user)
+    participant_id = realtimekit_participant_id(message, client_id)
+    response = participant_token_response(meeting_id, participant_id)
+    return response if response[:error].blank?
+
+    response = dyte_client.add_participant_to_meeting(meeting_id, client_id, user.name, avatar_url(user))
+    return store_participant_id_and_return(message, client_id, response) if response[:error].blank?
+
+    existing_participant_token_response(meeting_id, client_id, message) || response
   end
 
   private
 
+  def realtimekit_client_id(user)
+    "#{user.class.name}:#{user.id}"
+  end
+
+  def store_participant_id_and_return(message, client_id, response)
+    update_realtimekit_participant_id(message, client_id, response['id']) if response['id'].present?
+    response
+  end
+
   def create_a_dyte_integration_message(meeting, title, agent)
     @conversation.messages.create!(
       {
@@ -48,7 +69,65 @@ class Integrations::Dyte::ProcessorService
   end
 
   def dyte_client
-    credentials = dyte_hook.settings
-    @dyte_client ||= Dyte.new(credentials['organization_id'], credentials['api_key'])
+    @dyte_client ||= Dyte.new(*realtimekit_credentials)
+  end
+
+  def participant_token_response(meeting_id, participant_id)
+    return { error: :participant_id_missing } if participant_id.blank?
+
+    dyte_client.refresh_participant_token(meeting_id, participant_id)
+  end
+
+  def existing_participant_token_response(meeting_id, client_id, message)
+    participant_id = existing_realtimekit_participant_id(meeting_id, client_id)
+    return if participant_id.blank?
+
+    response = dyte_client.refresh_participant_token(meeting_id, participant_id)
+    update_realtimekit_participant_id(message, client_id, participant_id) if response[:error].blank?
+    response
+  end
+
+  def existing_realtimekit_participant_id(meeting_id, client_id)
+    participants = dyte_client.fetch_participants(meeting_id)
+    return if participants.blank? || participants.is_a?(Hash)
+
+    participants.find { |participant| participant['custom_participant_id'].to_s == client_id.to_s }&.dig('id')
+  end
+
+  def realtimekit_participant_id(message, client_id)
+    integration_message_data(message).dig(:participants, client_id.to_s)
+  end
+
+  def update_realtimekit_participant_id(message, client_id, participant_id)
+    return if message.blank?
+
+    attributes = message.content_attributes.with_indifferent_access
+    data = (attributes[:data] || {}).with_indifferent_access
+    participants = (data[:participants] || {}).with_indifferent_access
+    participants[client_id.to_s] = participant_id
+    data[:participants] = participants
+    attributes[:data] = data
+    message.update_columns(content_attributes: attributes.deep_stringify_keys, updated_at: Time.current) # rubocop:disable Rails/SkipsModelValidations
+  rescue StandardError => e
+    Rails.logger.warn("[dyte] Failed to store RealtimeKit participant ID for message #{message.id}: #{e.class}: #{e.message}")
+  end
+
+  def integration_message_data(message)
+    return {} if message.blank?
+
+    (message.content_attributes.with_indifferent_access[:data] || {}).with_indifferent_access
+  end
+
+  def realtimekit_credentials
+    credentials = dyte_hook.settings.with_indifferent_access
+    [credentials[:account_id], credentials[:app_id], credentials[:api_token]]
+  end
+
+  def realtimekit_credentials_missing?
+    realtimekit_credentials.any?(&:blank?)
+  end
+
+  def missing_realtimekit_credentials_response
+    { error: I18n.t('errors.dyte.realtimekit_credentials_required') }
   end
 end