diff --git a/enterprise/app/jobs/captain/documents/perform_sync_job.rb b/enterprise/app/jobs/captain/documents/perform_sync_job.rb new file mode 100644 index 000000000..baa9c03d2 --- /dev/null +++ b/enterprise/app/jobs/captain/documents/perform_sync_job.rb @@ -0,0 +1,97 @@ +class Captain::Documents::PerformSyncJob < MutexApplicationJob + queue_as :low + + LOCK_TIMEOUT = 10.minutes + + # Safety net for anything we didn't rescue by name — parser bugs, ActiveRecord blips, + # random infra issues. Three attempts lets a real hiccup recover. The exhaustion block + # absorbs the final exception so Sidekiq doesn't layer its own retry policy on top, and + # is the single place we report to Sentry — handle_unexpected_failure logs but does not + # capture, so a deterministic bug emits one Sentry event instead of one per attempt. + # Goes first because retry_on handlers dispatch bottom-to-top. + retry_on StandardError, wait: 5.seconds, attempts: 3 do |job, error| + document = job.arguments.first + ChatwootExceptionTracker.new(error, account: document.account).capture_exception + job.send(:log_sync_outcome, document, result: :unexpected_retry_exhausted, + error_code: 'sync_error', + exception_class: error.class.name) + end + + # Permanent errors (404, 403, empty content) — no point retrying, discard immediately. + # Document is already marked failed by SyncService before the exception reaches here. + discard_on(Captain::Documents::SyncService::PermanentSyncError) + + # TransientSyncError is raised by SyncService when the customer's site is unreachable — + # timeouts, TLS errors, 5xx, connection drops. Four attempts with backoff gives the site + # a chance to recover before we give up. + # + # The exhaustion block absorbs the exception so it doesn't propagate to Sentry — + # site flakiness isn't an application bug. + retry_on( + Captain::Documents::SyncService::TransientSyncError, + wait: ->(executions) { [30.seconds, 2.minutes, 5.minutes][executions - 1] || 5.minutes }, + attempts: 4 + ) do |job, error| + document = job.arguments.first + job.send(:log_sync_outcome, document, result: :transient_retry_exhausted, error_code: error.message) + end + + discard_on ActiveJob::DeserializationError + discard_on ActiveRecord::RecordNotFound + + def perform(document) + start_time = Time.current + return if document.pdf_document? + + with_lock(lock_key(document), LOCK_TIMEOUT) do + document.update!(sync_status: :syncing, last_sync_attempted_at: Time.current) + result = Captain::Documents::SyncService.new(document.reload).perform + log_sync_outcome(document, result: result, duration_ms: duration_ms_since(start_time)) + end + rescue LockAcquisitionError + log_sync_outcome(document, result: :already_syncing) + rescue Captain::Documents::SyncService::PermanentSyncError => e + log_failure_and_raise(document, :permanent_failure, e, start_time) + rescue Captain::Documents::SyncService::TransientSyncError => e + log_failure_and_raise(document, :transient_failure, e, start_time) + rescue StandardError => e + handle_unexpected_failure(document, e, start_time) + end + + private + + def log_sync_outcome(document, **fields) + payload = { + document_id: document.id, + account_id: document.account_id, + assistant_id: document.assistant_id + }.merge(fields) + Rails.logger.info("[Captain::Documents::PerformSyncJob] #{payload.to_json}") + end + + def log_failure_and_raise(document, result, error, start_time) + log_sync_outcome(document, result: result, error_code: error.message, + duration_ms: duration_ms_since(start_time)) + raise error + end + + def handle_unexpected_failure(document, error, start_time) + document.update!( + sync_status: :failed, + last_sync_error_code: 'sync_error', + last_sync_attempted_at: Time.current + ) + log_sync_outcome(document, result: :unexpected_failure, error_code: 'sync_error', + exception_class: error.class.name, + duration_ms: duration_ms_since(start_time)) + raise error + end + + def lock_key(document) + format(::Redis::Alfred::CAPTAIN_DOCUMENT_SYNC_MUTEX, document_id: document.id) + end + + def duration_ms_since(start_time) + ((Time.current - start_time) * 1000).round + end +end diff --git a/enterprise/app/jobs/captain/documents/response_builder_job.rb b/enterprise/app/jobs/captain/documents/response_builder_job.rb index 553cec110..60f1fa2cf 100644 --- a/enterprise/app/jobs/captain/documents/response_builder_job.rb +++ b/enterprise/app/jobs/captain/documents/response_builder_job.rb @@ -62,7 +62,7 @@ class Captain::Documents::ResponseBuilderJob < ApplicationJob end def reset_previous_responses(response_document) - response_document.responses.destroy_all + response_document.responses.where(edited: false).destroy_all end def create_response(faq, document) diff --git a/enterprise/app/models/captain/document.rb b/enterprise/app/models/captain/document.rb index d8e07a2d9..2abf18437 100644 --- a/enterprise/app/models/captain/document.rb +++ b/enterprise/app/models/captain/document.rb @@ -62,6 +62,7 @@ class Captain::Document < ApplicationRecord def pdf_document? return true if pdf_file.attached? && pdf_file.blob.content_type == 'application/pdf' + return true if external_link&.start_with?('PDF:') external_link&.ends_with?('.pdf') end @@ -90,6 +91,14 @@ class Captain::Document < ApplicationRecord self.metadata = (metadata || {}).merge('last_sync_error_code' => value) end + def sync_step + metadata&.dig('sync_step') + end + + def store_sync_step(step) + update!(metadata: (metadata || {}).merge('sync_step' => step)) + end + def openai_file_id metadata&.dig('openai_file_id') end diff --git a/enterprise/app/services/captain/documents/single_page_fetcher.rb b/enterprise/app/services/captain/documents/single_page_fetcher.rb new file mode 100644 index 000000000..115cd55b0 --- /dev/null +++ b/enterprise/app/services/captain/documents/single_page_fetcher.rb @@ -0,0 +1,80 @@ +class Captain::Documents::SinglePageFetcher + Result = Struct.new(:success, :title, :content, :error_code, keyword_init: true) + + CONTENT_MAX_LENGTH = 200_000 + TITLE_MAX_LENGTH = 255 # captain_documents.name is a varchar(255) + + def initialize(url) + @url = url + end + + def fetch + result = firecrawl_configured? ? fetch_with_firecrawl : fetch_with_fallback + validate_content(result) + rescue Net::ReadTimeout, Net::OpenTimeout, Errno::ETIMEDOUT + Result.new(success: false, error_code: 'timeout') + rescue SocketError, Errno::ECONNREFUSED, Errno::ECONNRESET, OpenSSL::SSL::SSLError + Result.new(success: false, error_code: 'fetch_failed') + end + + private + + def firecrawl_configured? + InstallationConfig.find_by(name: 'CAPTAIN_FIRECRAWL_API_KEY')&.value.present? + end + + def fetch_with_firecrawl + response = Captain::Tools::FirecrawlService.new.scrape(@url) + handle_firecrawl_response(response) + end + + def handle_firecrawl_response(response) + return Result.new(success: false, error_code: http_error_code(response.code)) unless response.success? + + data = response.parsed_response&.dig('data') + target_error = firecrawl_target_error_code(data) + return Result.new(success: false, error_code: target_error) if target_error + + Result.new( + success: true, + title: data&.dig('metadata', 'title')&.truncate(TITLE_MAX_LENGTH, omission: ''), + content: data&.dig('markdown')&.truncate(CONTENT_MAX_LENGTH, omission: '') + ) + end + + # Firecrawl returns API 200 even when the scraped page itself failed — + # the target page's real status lives in data.metadata.statusCode. + def firecrawl_target_error_code(data) + status = data&.dig('metadata', 'statusCode') + return nil if status.blank? || (200..299).cover?(status) + + http_error_code(status) + end + + def fetch_with_fallback + response = HTTParty.get(@url) + return Result.new(success: false, error_code: http_error_code(response.code)) unless response.success? + + parser = Captain::Tools::HtmlPageParser.new(response.body) + Result.new( + success: true, + title: parser.title&.truncate(TITLE_MAX_LENGTH, omission: ''), + content: parser.body_markdown&.truncate(CONTENT_MAX_LENGTH, omission: '') + ) + end + + def validate_content(result) + return result unless result.success && result.content.blank? + + Result.new(success: false, error_code: 'content_empty') + end + + def http_error_code(status_code) + case status_code + when 404 then 'not_found' + when 401, 403 then 'access_denied' + when 408, 504 then 'timeout' + else 'fetch_failed' + end + end +end diff --git a/enterprise/app/services/captain/documents/sync_service.rb b/enterprise/app/services/captain/documents/sync_service.rb new file mode 100644 index 000000000..5dcb52447 --- /dev/null +++ b/enterprise/app/services/captain/documents/sync_service.rb @@ -0,0 +1,76 @@ +class Captain::Documents::SyncService + class PermanentSyncError < StandardError + end + + class TransientSyncError < StandardError + end + + PERMANENT_ERROR_CODES = %w[not_found access_denied content_empty].freeze + + def initialize(document) + @document = document + end + + def perform + @document.store_sync_step('fetching') + result = Captain::Documents::SinglePageFetcher.new(@document.external_link).fetch + + unless result.success + mark_failed(result.error_code) + raise_for_error_code(result.error_code) + end + + @document.store_sync_step('comparing') + fingerprint = compute_fingerprint(result.content) + + if fingerprint == @document.content_fingerprint + mark_synced + return :unchanged + end + + @document.store_sync_step('updating') + update_content(result, fingerprint) + :updated + end + + private + + def compute_fingerprint(content) + Digest::SHA256.hexdigest(content.gsub(/\s+/, ' ').strip) + end + + def mark_failed(error_code) + @document.update!( + sync_status: :failed, + last_sync_error_code: error_code, + last_sync_attempted_at: Time.current + ) + end + + def mark_synced + @document.update!( + sync_status: :synced, + last_synced_at: Time.current, + last_sync_attempted_at: Time.current, + last_sync_error_code: nil + ) + end + + def update_content(result, fingerprint) + @document.update!( + content: result.content, + name: result.title.presence || @document.name, + content_fingerprint: fingerprint, + sync_status: :synced, + last_synced_at: Time.current, + last_sync_attempted_at: Time.current, + last_sync_error_code: nil + ) + end + + def raise_for_error_code(error_code) + raise PermanentSyncError, error_code if PERMANENT_ERROR_CODES.include?(error_code) + + raise TransientSyncError, error_code + end +end diff --git a/enterprise/app/services/captain/tools/firecrawl_service.rb b/enterprise/app/services/captain/tools/firecrawl_service.rb index 397c3ad63..fc7448593 100644 --- a/enterprise/app/services/captain/tools/firecrawl_service.rb +++ b/enterprise/app/services/captain/tools/firecrawl_service.rb @@ -1,4 +1,6 @@ class Captain::Tools::FirecrawlService + BASE_URL = 'https://api.firecrawl.dev/v1'.freeze + def initialize @api_key = InstallationConfig.find_by!(name: 'CAPTAIN_FIRECRAWL_API_KEY').value raise 'Missing API key' if @api_key.empty? @@ -6,7 +8,7 @@ class Captain::Tools::FirecrawlService def perform(url, webhook_url, crawl_limit = 10) HTTParty.post( - 'https://api.firecrawl.dev/v1/crawl', + "#{BASE_URL}/crawl", body: crawl_payload(url, webhook_url, crawl_limit), headers: headers ) @@ -14,6 +16,14 @@ class Captain::Tools::FirecrawlService raise "Failed to crawl URL: #{e.message}" end + def scrape(url) + HTTParty.post( + "#{BASE_URL}/scrape", + body: scrape_payload(url), + headers: headers + ) + end + private def crawl_payload(url, webhook_url, crawl_limit) @@ -31,6 +41,10 @@ class Captain::Tools::FirecrawlService }.to_json end + def scrape_payload(url) + { url: url, formats: ['markdown'], excludeTags: ['iframe'] }.to_json + end + def headers { 'Authorization' => "Bearer #{@api_key}", diff --git a/enterprise/app/services/captain/tools/html_page_parser.rb b/enterprise/app/services/captain/tools/html_page_parser.rb new file mode 100644 index 000000000..a7afb52cf --- /dev/null +++ b/enterprise/app/services/captain/tools/html_page_parser.rb @@ -0,0 +1,15 @@ +class Captain::Tools::HtmlPageParser + attr_reader :doc + + def initialize(html) + @doc = Nokogiri::HTML(html) + end + + def title + @doc.at_xpath('//title')&.text&.strip + end + + def body_markdown + ReverseMarkdown.convert(@doc.at_xpath('//body'), unknown_tags: :bypass, github_flavored: true) + end +end diff --git a/enterprise/app/services/captain/tools/simple_page_crawl_service.rb b/enterprise/app/services/captain/tools/simple_page_crawl_service.rb index 65731ad90..5eaee6e5e 100644 --- a/enterprise/app/services/captain/tools/simple_page_crawl_service.rb +++ b/enterprise/app/services/captain/tools/simple_page_crawl_service.rb @@ -3,7 +3,8 @@ class Captain::Tools::SimplePageCrawlService def initialize(external_link) @external_link = external_link - @doc = Nokogiri::HTML(HTTParty.get(external_link).body) + @parser = Captain::Tools::HtmlPageParser.new(HTTParty.get(external_link).body) + @doc = @parser.doc end def page_links @@ -11,12 +12,11 @@ class Captain::Tools::SimplePageCrawlService end def page_title - title_element = @doc.at_xpath('//title') - title_element&.text&.strip + @parser.title end def body_text_content - ReverseMarkdown.convert @doc.at_xpath('//body'), unknown_tags: :bypass, github_flavored: true + @parser.body_markdown end def meta_description diff --git a/lib/redis/redis_keys.rb b/lib/redis/redis_keys.rb index 59e33036d..893fb50ef 100644 --- a/lib/redis/redis_keys.rb +++ b/lib/redis/redis_keys.rb @@ -44,6 +44,7 @@ module Redis::RedisKeys SLACK_MESSAGE_MUTEX = 'SLACK_MESSAGE_LOCK::%s::%s'.freeze EMAIL_MESSAGE_MUTEX = 'EMAIL_CHANNEL_LOCK::%s'.freeze CRM_PROCESS_MUTEX = 'CRM_PROCESS_MUTEX::%s'.freeze + CAPTAIN_DOCUMENT_SYNC_MUTEX = 'CAPTAIN_DOCUMENT_SYNC_LOCK::%s'.freeze ## Auto Assignment Keys # Track conversation assignments to agents for rate limiting diff --git a/spec/enterprise/models/captain/document_spec.rb b/spec/enterprise/models/captain/document_spec.rb index e6543e0d7..af6c1cb58 100644 --- a/spec/enterprise/models/captain/document_spec.rb +++ b/spec/enterprise/models/captain/document_spec.rb @@ -55,6 +55,11 @@ RSpec.describe Captain::Document, type: :model do expect(doc.pdf_document?).to be true end + it 'returns true for PDF:-prefixed external links even when the blob is missing' do + doc = build(:captain_document, external_link: 'PDF: report_20250101120000') + expect(doc.pdf_document?).to be true + end + it 'returns false for non-PDF documents' do doc = build(:captain_document, external_link: 'https://example.com') expect(doc.pdf_document?).to be false