mirror of
https://github.com/we-promise/sure.git
synced 2026-07-28 12:42:14 +00:00
* feat(jobs): reap imports and exports stuck by lost background jobs A hard-killed worker (OOM, SIGKILL during deploy) loses its in-flight Sidekiq job permanently. Sync is the only model with a stale sweep; everything else wedges in a non-terminal status forever: - Import stuck "importing"/"reverting" (e.g. #2274) - ImportSession stuck "importing" — unrecoverable, publish_later refuses to re-publish while importing - FamilyExport stuck "pending"/"processing" — exports index polls every 3s indefinitely - PdfImport's AI-processing claim held forever (ProcessPdfJob's reclaim only runs if the job is redelivered) - Provider activities_fetch_pending flags stranded when a self-rescheduling fetch chain loses a link SyncCleanerJob (existing hourly cron) now sweeps all of them, each isolated so one failure doesn't block the rest, and records every reaped record as a DebugLogEntry with the family attached. Idempotency guards so a stray or redelivered job cannot corrupt a record the reaper (or a user retry) has since moved on: - Import#publish skips complete/reverting/revert_failed imports — a redelivered ImportJob after completion double-applies data for import types without row dedup. Failed imports deliberately stay publishable: their transaction rolled back, so a re-run is a retry. - FamilyDataExportJob refuses terminal exports but still allows processing ones through, since graceful-shutdown redelivery is what completes them. Reaper thresholds (6h imports, 2h exports) key off updated_at and dwarf legitimate runtimes, so live jobs are not swept in practice. * fix(jobs): make the reapers race-safe and commit-aware Review feedback on #2681 (jjmata, Codex, CodeRabbit): - Every sweep now mutates under record.with_lock with a staleness re-check, mirroring the guard Sync#perform gained in #2680 — a job finishing between the sweep query and the write can no longer be clobbered mid-flight - Import.clean distinguishes which side of import!'s single transaction the worker died on: rows attached means the data committed and only the status write was lost, so the record is finalized as complete (marking it failed invited a re-publish that double-imports types without row dedup, e.g. TradeImport); no rows means a clean rollback and the failed/try-again path stays - PdfImport.clean applies the same split: no rows → the AI-extraction claim died, reclaim to pending; rows → the publish died post-commit, finalize complete instead of letting the same extracted rows be published twice. Stuck PdfImport reverts (previously unswept by either clean) now go to revert_failed like every other import - ImportSession.clean reconciles chunks whose import! committed but never got the complete-status write before failing the session, so re-publish skips them instead of duplicating their rows via SureImport's split path - Import#publish redelivery skip is captured via DebugLogEntry instead of Rails.logger so support can see it in /settings/debug - The interrupted-error copy is i18n-backed (imports.errors.interrupted) * fix(jobs): round-2 review feedback on the reapers - Import.clean excludes session-owned chunks (import_session_id: nil) — SyncCleanerJob runs it before ImportSession.clean, so it could finalize or fail a SureImport chunk outside the session flow that owns its lifecycle. Regression test added (CodeRabbit major) - family.sync_later moved outside the row-lock transaction in both reap paths — Rails doesn't defer enqueues to after-commit by default, so Sidekiq could pick the job up before the status write was visible - Per-record rescue in Import.clean / PdfImport.clean so one bad record doesn't abort the rest of the hourly sweep - ImportSession.clean uses the importing? enum predicate; reconcile_committed_chunks! iterates with each (default-ordered association made find_each warn, and chunk counts are tiny) * fix(jobs): address round-3 reaper review (isolation + commit signal) Per-record rescue isolation for the three sweeps that lacked it — FamilyExport.clean, ImportSession.clean, and SyncCleanerJob's activity-flag loop — mirroring Import.clean/PdfImport.clean. One bad record (validation error, DB blip) no longer aborts the rest of that sweep for the hour; the activity-flag guard is per-record so a failure also stops skipping the models that follow. data_committed? now covers Category/Rule/Merchant imports, whose records hang off the family rather than the import (no entries or accounts), via the new Import#committed_by_named_records? helper. A committed one is reaped to complete instead of a retryable failed; nameless RuleImport rows carry no stable key, so a nameless-only file has no commit signal and stays retryable. * fix(schema): drop duplicate enable_banking_accounts columns from merge The merge of main into this branch re-appended product, credit_limit, and identification_hashes after updated_at, so schema.rb declared each twice and db:schema:load raised "you can't define an already defined column 'product'". Removed the duplicate declarations (kept the new treat_balance_as_available_credit column); the test database loads again.
340 lines
10 KiB
Ruby
340 lines
10 KiB
Ruby
class PdfImport < Import
|
|
has_one_attached :pdf_file, dependent: :purge_later
|
|
|
|
validates :document_type, inclusion: { in: DOCUMENT_TYPES }, allow_nil: true
|
|
validate :account_statement_matches_import
|
|
|
|
class << self
|
|
# PdfImport's importing status doubles as a processing claim: the AI
|
|
# extraction claim from process_with_ai_later (no rows yet) or a regular
|
|
# publish (rows being written). A lost job leaves the claim held forever —
|
|
# ProcessPdfJob's own reclaim only runs when the job is redelivered.
|
|
# Which claim died is observable from the data: no rows attached → the AI
|
|
# claim, reclaim to pending so the user can re-trigger; rows attached →
|
|
# the publish died after import!'s commit, so finalize as complete
|
|
# (pending would let the user publish the same extracted rows again —
|
|
# PdfImport#import! always builds new transactions). Stuck reverts go to
|
|
# revert_failed like every other import, keeping the retry path exposed.
|
|
def clean
|
|
where(status: [ :importing, :reverting ])
|
|
.where("updated_at < ?", Import::STUCK_AFTER.ago)
|
|
.includes(:family)
|
|
.find_each do |pdf_import|
|
|
reap_stuck!(pdf_import)
|
|
rescue => e
|
|
# One bad record must not abort the sweep for the rest.
|
|
Rails.logger.error("PdfImport.clean failed for #{pdf_import.id}: #{e.class}: #{e.message}")
|
|
Sentry.capture_exception(e) { |scope| scope.set_tags(record_type: name, record_id: pdf_import.id) } if defined?(Sentry)
|
|
end
|
|
end
|
|
|
|
def reap_stuck!(pdf_import)
|
|
needs_sync = false
|
|
# Read before the lock — see Import.reap_stuck!.
|
|
family = pdf_import.family
|
|
|
|
pdf_import.with_lock do
|
|
next unless pdf_import.reapable_since?(Import::STUCK_AFTER.ago)
|
|
|
|
previous_status = pdf_import.status
|
|
if previous_status == "reverting"
|
|
pdf_import.update!(status: :revert_failed, error: Import.interrupted_error_message)
|
|
elsif pdf_import.data_committed?
|
|
pdf_import.update!(status: :complete, error: nil)
|
|
needs_sync = true
|
|
else
|
|
pdf_import.update!(status: :pending)
|
|
end
|
|
|
|
DebugLogEntry.capture(
|
|
category: "background_jobs",
|
|
level: "warn",
|
|
message: "Reclaimed PdfImport stuck in #{previous_status} for over #{Import::STUCK_AFTER.inspect} (→ #{pdf_import.status})",
|
|
source: name,
|
|
family: family,
|
|
metadata: { record_type: name, record_id: pdf_import.id, previous_status: previous_status, new_status: pdf_import.status }
|
|
)
|
|
end
|
|
|
|
# Outside the row lock — see Import.reap_stuck!.
|
|
family.sync_later if needs_sync
|
|
end
|
|
|
|
def create_from_upload!(family:, file:, user:)
|
|
statement = AccountStatement.create_from_prepared_upload!(
|
|
family: family,
|
|
account: nil,
|
|
prepared_upload: AccountStatement.prepare_upload!(file)
|
|
)
|
|
|
|
create_from_statement!(statement: statement)
|
|
rescue AccountStatement::DuplicateUploadError => e
|
|
raise unless e.statement.manageable_by?(user)
|
|
|
|
create_from_statement!(statement: e.statement)
|
|
end
|
|
|
|
def create_from_statement!(statement:)
|
|
reusable_import = statement.latest_reusable_pdf_import
|
|
return reusable_import if reusable_import &&
|
|
reusable_import.account_id == statement.account_id &&
|
|
reusable_import.date_format == statement.family.date_format
|
|
|
|
create!(family: statement.family, account: statement.account, account_statement: statement, date_format: statement.family.date_format, status: :pending)
|
|
end
|
|
end
|
|
|
|
# A PdfImport's importing status is a processing claim (AI extraction or
|
|
# publish). Release a lost claim back to pending so the user can re-trigger
|
|
# processing, mirroring ProcessPdfJob's own reclaim; lost reverts keep the
|
|
# base revert_failed behavior.
|
|
def force_fail!(error_message = Import.lost_error_message)
|
|
return super if reverting?
|
|
|
|
with_lock do
|
|
return false unless presumed_lost?
|
|
|
|
update!(status: :pending)
|
|
end
|
|
|
|
true
|
|
end
|
|
|
|
def import!
|
|
raise "Account required for PDF import" unless account.present?
|
|
|
|
transaction do
|
|
mappings.each(&:create_mappable!)
|
|
|
|
new_transactions = rows.map do |row|
|
|
category = mappings.categories.mappable_for(row.category)
|
|
|
|
Transaction.new(
|
|
category: category,
|
|
entry: Entry.new(
|
|
account: account,
|
|
date: row.date_iso,
|
|
amount: row.signed_amount,
|
|
name: row.name,
|
|
currency: row.currency,
|
|
notes: row.notes,
|
|
import: self,
|
|
import_locked: true
|
|
)
|
|
)
|
|
end
|
|
|
|
Transaction.import!(new_transactions, recursive: true) if new_transactions.any?
|
|
end
|
|
end
|
|
|
|
def assign_account!(account)
|
|
transaction do
|
|
update!(account: account)
|
|
if (statement = account_statement)
|
|
statement.lock!
|
|
statement.link_to_account!(account) if statement.account_id != account.id
|
|
end
|
|
end
|
|
end
|
|
|
|
def pdf_uploaded?
|
|
statement_backed? || pdf_file.attached?
|
|
end
|
|
|
|
def ai_processed?
|
|
ai_summary.present?
|
|
end
|
|
|
|
def process_with_ai_later
|
|
return false unless with_lock { pending? && !ai_processed? && rows_count.zero? && pdf_uploaded? && update!(status: :importing) }
|
|
|
|
begin
|
|
ProcessPdfJob.perform_later(self)
|
|
true
|
|
rescue StandardError => e
|
|
Rails.logger.error("Failed to enqueue PDF processing for import #{id}: #{e.class.name} - #{e.message}")
|
|
reload.with_lock { update!(status: :pending) }
|
|
false
|
|
end
|
|
end
|
|
|
|
def process_with_ai
|
|
# Honors Setting.llm_provider (issue #2113) — Provider::Anthropic implements
|
|
# process_pdf (PR #1985).
|
|
provider = Provider::Registry.preferred_llm_provider
|
|
raise "AI provider not configured" unless provider
|
|
raise "AI provider does not support PDF processing" unless provider.supports_pdf_processing?
|
|
|
|
response = provider.process_pdf(
|
|
pdf_content: pdf_file_content,
|
|
family: family
|
|
)
|
|
|
|
unless response.success?
|
|
error_message = response.error&.message || "Unknown PDF processing error"
|
|
raise error_message
|
|
end
|
|
|
|
result = response.data
|
|
update!(
|
|
ai_summary: result.summary,
|
|
document_type: result.document_type
|
|
)
|
|
|
|
result
|
|
end
|
|
|
|
def extract_transactions
|
|
return unless statement_with_transactions?
|
|
|
|
# Honors Setting.llm_provider (issue #2113) — Provider::Anthropic implements
|
|
# extract_bank_statement (PR #1985).
|
|
provider = Provider::Registry.preferred_llm_provider
|
|
raise "AI provider not configured" unless provider
|
|
|
|
response = provider.extract_bank_statement(
|
|
pdf_content: pdf_file_content,
|
|
family: family
|
|
)
|
|
|
|
unless response.success?
|
|
error_message = response.error&.message || "Unknown extraction error"
|
|
raise error_message
|
|
end
|
|
|
|
update!(extracted_data: response.data)
|
|
response.data
|
|
end
|
|
|
|
def bank_statement?
|
|
document_type == "bank_statement"
|
|
end
|
|
|
|
def statement_with_transactions?
|
|
document_type.in?(%w[bank_statement credit_card_statement])
|
|
end
|
|
|
|
def has_extracted_transactions?
|
|
extracted_data.present? && extracted_data["transactions"].present?
|
|
end
|
|
|
|
def extracted_transactions
|
|
extracted_data&.dig("transactions") || []
|
|
end
|
|
|
|
def generate_rows_from_extracted_data
|
|
transaction do
|
|
rows.destroy_all
|
|
|
|
unless has_extracted_transactions?
|
|
update_column(:rows_count, 0)
|
|
return
|
|
end
|
|
|
|
currency = account&.currency || family.currency
|
|
|
|
mapped_rows = extracted_transactions.map.with_index(1) do |txn, index|
|
|
{
|
|
import_id: id,
|
|
source_row_number: index,
|
|
date: format_date_for_import(txn["date"]),
|
|
amount: txn["amount"].to_s,
|
|
name: txn["name"].to_s,
|
|
category: txn["category"].to_s,
|
|
notes: txn["notes"].to_s,
|
|
currency: currency
|
|
}
|
|
end
|
|
|
|
Import::Row.insert_all!(mapped_rows) if mapped_rows.any?
|
|
update_column(:rows_count, mapped_rows.size)
|
|
end
|
|
end
|
|
|
|
def send_next_steps_email(user)
|
|
PdfImportMailer.with(
|
|
user: user,
|
|
pdf_import: self
|
|
).next_steps.deliver_later
|
|
end
|
|
|
|
def uploaded?
|
|
pdf_uploaded?
|
|
end
|
|
|
|
def configured?
|
|
ai_processed? && rows_count > 0
|
|
end
|
|
|
|
def cleaned?
|
|
configured? && rows.all?(&:valid?)
|
|
end
|
|
|
|
def publishable?
|
|
account.present? && statement_with_transactions? && cleaned? && mappings.all?(&:valid?)
|
|
end
|
|
|
|
def cleaned_from_validation_stats?(invalid_rows_count:)
|
|
account.present? && statement_with_transactions? && super
|
|
end
|
|
|
|
def publishable_from_validation_stats?(invalid_rows_count:)
|
|
account.present? && statement_with_transactions? && super
|
|
end
|
|
|
|
def column_keys
|
|
%i[date amount name category notes]
|
|
end
|
|
|
|
def requires_csv_workflow?
|
|
false
|
|
end
|
|
|
|
def pdf_file_content
|
|
return @pdf_file_content if defined?(@pdf_file_content)
|
|
return @pdf_file_content = account_statement.original_file.download if statement_backed?
|
|
|
|
@pdf_file_content = pdf_file.download if pdf_file.attached?
|
|
end
|
|
|
|
def pdf_filename
|
|
return account_statement.filename if statement_backed?
|
|
|
|
pdf_file.filename.to_s if pdf_file.attached?
|
|
end
|
|
|
|
def statement_backed?
|
|
account_statement&.original_file&.attached?
|
|
end
|
|
|
|
def required_column_keys
|
|
%i[date amount]
|
|
end
|
|
|
|
def mapping_steps
|
|
base = []
|
|
# Only include CategoryMapping if rows have non-empty categories
|
|
base << Import::CategoryMapping if rows.where.not(category: [ nil, "" ]).exists?
|
|
# Note: PDF imports use direct account selection in the UI, not AccountMapping
|
|
# AccountMapping is designed for CSV imports where rows have different account values
|
|
base
|
|
end
|
|
|
|
private
|
|
|
|
def format_date_for_import(date_str)
|
|
return "" if date_str.blank?
|
|
|
|
Date.parse(date_str).strftime(date_format)
|
|
rescue ArgumentError
|
|
date_str.to_s
|
|
end
|
|
|
|
def account_statement_matches_import
|
|
return if account_statement.blank? || (account_statement.family_id == family_id && account_statement.pdf?)
|
|
|
|
errors.add(:account_statement, :invalid)
|
|
end
|
|
end
|