gem 'chrono_forge'

# Define your workflow class class OrderProcessingWorkflow < ApplicationJob prepend ChronoForge::Executor def perform(order_id:, customer_id:) # Workflow steps... end end

# Execute the workflow OrderProcessingWorkflow.perform_later( "order-123", # Unique workflow key order_id: "order-134", # Custom kwargs customer_id: "customer-456" # More custom kwargs )

class OrderProcessingWorkflow < ApplicationJob prepend ChronoForge::Executor def perform(order_id:) @order_id = order_id # Context can be used to pass and store data between executions context.set_once "execution_id", SecureRandom.hex # Wait until payment is confirmed wait_until :payment_confirmed? # Wait for potential fraud check wait 1.minute, :fraud_check_delay # Durably execute order processing durably_execute :process_order # Final steps durably_execute :complete_order end private def payment_confirmed? PaymentService.confirmed?(@order_id, context["execution_id"]) end def process_order OrderProcessor.process(@order_id, context["execution_id"]) context["processed_at"] = Time.current.iso8601 end def complete_order OrderCompletionService.complete(@order_id, context["execution_id"]) context["completed_at"] = Time.current.iso8601 end end

# Perform the workflow immediately OrderProcessingWorkflow.perform_now( "order-123", # Unique workflow key order_id: "O-123", # Custom parameter customer_id: "C-456" # Another custom parameter ) # Or queue it for background processing OrderProcessingWorkflow.perform_later( "order-123-async", # Unique workflow key order_id: "O-124", customer_id: "C-457" )

# Basic execution durably_execute :send_welcome_email # With custom retry attempts durably_execute :critical_payment_processing, max_attempts: 5 # With custom name for tracking multiple calls to same method durably_execute :upload_file, name: "profile_image_upload" # Complex example with error-prone operation class FileProcessingWorkflow < ApplicationJob prepend ChronoForge::Executor def perform(file_id:) @file_id = file_id # This might fail due to network issues, rate limits, etc. durably_execute :upload_to_s3, max_attempts: 5 # Process file after successful upload durably_execute :generate_thumbnails, max_attempts: 3 end private def upload_to_s3 file = File.find(@file_id) S3Client.upload(file.path, bucket: 'my-bucket') Rails.logger.info "Successfully uploaded file #{@file_id} to S3" end def generate_thumbnails ThumbnailService.generate(@file_id) end end

# Simple delays wait 30.minutes, "cooling_period" wait 1.day, "daily_batch_interval" # Complex workflow with multiple waits def user_onboarding_flow durably_execute :send_welcome_email wait 1.hour, "welcome_delay" durably_execute :send_tutorial_email wait 2.days, "tutorial_followup" durably_execute :send_feedback_request end

# Wait for external API wait_until :external_api_ready?, timeout: 30.minutes, check_interval: 1.minute # Wait with retry on specific errors wait_until :database_migration_complete?, timeout: 2.hours, check_interval: 30.seconds, retry_on: [ActiveRecord::ConnectionNotEstablished, Net::TimeoutError] # Complex condition example def third_party_service_ready? response = HTTParty.get("https://api.example.com/health") response.code == 200 && response.body.include?("healthy") end wait_until :third_party_service_ready?, timeout: 1.hour, check_interval: 2.minutes, retry_on: [Net::TimeoutError, Net::HTTPClientException]

# Basic usage - wait for webhook-driven state change continue_if :payment_confirmed? # With custom name for better tracking continue_if :payment_confirmed?, name: "stripe_webhook" # Wait for manual approval continue_if :document_approved? # Wait for external file processing continue_if :processing_complete? # Multiple waits with same condition but different contexts continue_if :external_system_ready?, name: "payment_gateway" # ... other steps ... continue_if :external_system_ready?, name: "inventory_system" # Complete workflow example class PaymentWorkflow < ApplicationJob prepend ChronoForge::Executor def perform(order_id:) @order_id = order_id # Initialize payment durably_execute :create_payment_request # Wait for external payment confirmation (webhook-driven) continue_if :payment_confirmed?, name: "stripe_confirmation" # Complete order after payment durably_execute :fulfill_order end private def payment_confirmed? PaymentService.confirmed?(@order_id) end end # Later, when webhook arrives: PaymentService.mark_confirmed(order_id) PaymentWorkflow.perform_later("order-#{order_id}", order_id: order_id)

class NotificationWorkflow < ApplicationJob prepend ChronoForge::Executor def perform(user_id:) @user_id = user_id # Send reminders every 3 days until user completes onboarding durably_repeat :send_reminder_email, every: 3.days, till: :user_onboarded? # Critical payment processing every hour - fail workflow if it fails durably_repeat :process_pending_payments, every: 1.hour, till: :all_payments_processed?, on_error: :fail_workflow end private def send_reminder_email(scheduled_time = nil) # Optional parameter receives the scheduled execution time if scheduled_time lateness = Time.current - scheduled_time Rails.logger.info "Reminder scheduled for #{scheduled_time}, running #{lateness.to_i}s late" end UserMailer.onboarding_reminder(@user_id).deliver_now end def user_onboarded? User.find(@user_id).onboarded? end def process_pending_payments PaymentProcessor.process_pending_for_user(@user_id) end def all_payments_processed? Payment.where(user_id: @user_id, status: :pending).empty? end end

durably_repeat :generate_daily_report, every: 1.day, # Execution interval till: :reports_complete?, # Stop condition start_at: Date.tomorrow.beginning_of_day, # Custom start time (optional) max_attempts: 5, # Retries per execution (default: 3) timeout: 2.hours, # Catch-up timeout (default: 1.hour) on_error: :fail_workflow, # Error handling (:continue or :fail_workflow) name: "daily_reports" # Custom task name (optional)

# Without scheduled time parameter def cleanup_files FileCleanupService.perform end # With scheduled time parameter def cleanup_files(scheduled_time) # Use scheduled time for business logic cleanup_date = scheduled_time.to_date FileCleanupService.perform(date: cleanup_date) # Log timing information delay = Time.current - scheduled_time Rails.logger.info "Cleanup was #{delay.to_i} seconds late" end

# Set context values context[:user_name] = "John Doe" context[:status] = "processing" # Read context values user_name = context[:user_name] # Using the fetch method (returns default if key doesn't exist) status = context.fetch(:status, "pending") # Set a value with the set method (alias for []=) context.set(:total_amount, 99.99) # Set a value only if the key doesn't already exist context.set_once(:created_at, Time.current.iso8601) # Check if a key exists if context.key?(:user_id) # Do something with the user ID end

class MyWorkflow < ApplicationJob prepend ChronoForge::Executor private def should_retry?(error, attempt_count) case error when NetworkError attempt_count < 5 # Retry network errors up to 5 times when ValidationError false # Don't retry validation errors else attempt_count < 3 # Default retry policy end end end

group :test do gem 'chaotic_job' end

# test_helper.rb require 'chrono_forge' require 'minitest/autorun' require 'chaotic_job'

class WorkflowTest < ActiveJob::TestCase include ChaoticJob::Helpers def test_workflow_completion # Enqueue the job with a unique key and custom parameters OrderProcessingWorkflow.perform_later( "order-test-123", order_id: "O-123", customer_id: "C-456" ) # Perform all enqueued jobs perform_all_jobs # Assert workflow completed successfully workflow = ChronoForge::Workflow.find_by(key: "order-test-123") assert workflow.completed? # Check workflow context assert workflow.context["processed_at"].present? assert workflow.context["completed_at"].present? end end

workflow = ChronoForge::Workflow.find_by(key: "order-123") if workflow.stalled? || workflow.failed? job_class = workflow.job_class.constantize # Retry immediately job_class.retry_now(workflow.key) # Or retry asynchronously job_class.retry_later(workflow.key) end

# Find workflows running for too long long_running = ChronoForge::Workflow.where(state: :running) .where('locked_at < ?', 30.minutes.ago) long_running.each do |workflow| # Log potential issues for investigation Rails.logger.warn "Workflow #{workflow.key} has been running for >30 minutes" # Optionally force unlock if you suspect deadlock # CAUTION: Only do this if you're certain the job is stuck # workflow.update!(locked_at: nil, locked_by: nil, state: :idle) end