class RailsAiAdmin::Outbox::Dispatcher

# File lib/rails_ai_admin/outbox/dispatcher.rb, line 20
def run(work_ref)
  raise ConfigurationError, "Outbox delivery cannot run inside a database transaction" if WorkItem.connection.transaction_open?
  item = claim(work_ref)
  return item if item.is_a?(Result)
  return Result.new(work_ref, "missing", "NOT_FOUND") unless item
  definition = RailsAiAdmin.work_registry.fetch(item.kind)
  authorization = -> { authorize(item) }
  context, resource, record = ActiveRecord::Base.while_preventing_writes { authorization.call }
  values = InputSchema.new(definition.handler_class.input).coerce(item.payload, time_zone: context.time_zone)
  raise Blocked, "PAYLOAD_INVALID" unless values.valid? && InputSchema.digest(values.values) == item.payload_digest
  delivery = DeliveryContext.new(item, context, resource, record, authorization: authorization)
  ActiveRecord::Base.while_preventing_writes { definition.handler_class.new.perform(delivery, values.values) }
  raise ConfigurationError, "Work must complete exactly one ctx.effect" unless delivery.effect_completed?
  finish(item, "succeeded", "OK")
rescue AccessDenied
  delivery&.effect_completed? ? finish(item, "succeeded", "POST_EFFECT_CALLBACK_FAILED") : fail_delivery(item, "AUTHORIZATION_REVOKED", denied: true)
rescue Blocked => error
  delivery&.effect_completed? ? finish(item, "succeeded", "POST_EFFECT_CALLBACK_FAILED") : fail_delivery(item, error.code, reconcile: true)
rescue LeaseLost
  fence_lost(item)
rescue StandardError
  raise unless item.is_a?(WorkItem)
  delivery&.effect_completed? ? finish(item, "succeeded", "POST_EFFECT_CALLBACK_FAILED") : fail_delivery(item, "DELIVERY_FAILED")
end

# File lib/rails_ai_admin/outbox/dispatcher.rb, line 45
def scan_pending!(limit: 100, inline: false)
  raise ArgumentError, "limit must be between 1 and 1000" unless limit.is_a?(Integer) && (1..1000).cover?(limit)
  now = Time.current
  due = WorkItem.where(status: "pending").where("next_attempt_at IS NULL OR next_attempt_at <= ? OR deadline_at <= ?", now, now)
  stale = WorkItem.where(status: "running").where("lease_expires_at IS NULL OR lease_expires_at <= ?", now)
  ids = due.or(stale).order(:created_at, :id).limit(limit).pluck(:id)
  ids.count do |id|
    if inline
      run(id)
      true
    else
      !!DeliveryJob.perform_later(id)
    end
  end
end