feat: sync documents job [AI-142] (#14057)

# Pull Request Template

## Description

Document auto-sync job pipeline without wiring to controllers or FE

Fixes # (issue)

## Type of change

- [x] New feature (non-breaking change which adds functionality)

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide
instructions so we can reproduce. Please also list any relevant details
for your test configuration.
specs and locally

## Checklist:

- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my code
- [x] I have commented on my code, particularly in hard-to-understand
areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] Any dependent changes have been merged and published in downstream
modules
This commit is contained in:
Aakash Bakhle
2026-04-22 23:36:32 +05:30
committed by GitHub
parent ce34f93917
commit 2182165201
10 changed files with 303 additions and 6 deletions
@@ -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
@@ -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)
@@ -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
@@ -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
@@ -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
@@ -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}",
@@ -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
@@ -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
+1
View File
@@ -44,6 +44,7 @@ module Redis::RedisKeys
SLACK_MESSAGE_MUTEX = 'SLACK_MESSAGE_LOCK::%<conversation_id>s::%<reference_id>s'.freeze
EMAIL_MESSAGE_MUTEX = 'EMAIL_CHANNEL_LOCK::%<inbox_id>s'.freeze
CRM_PROCESS_MUTEX = 'CRM_PROCESS_MUTEX::%<hook_id>s'.freeze
CAPTAIN_DOCUMENT_SYNC_MUTEX = 'CAPTAIN_DOCUMENT_SYNC_LOCK::%<document_id>s'.freeze
## Auto Assignment Keys
# Track conversation assignments to agents for rate limiting
@@ -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