gem 'cloudtasker'

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  #
  # Adapt the server port to be the one used by your Rails web process
  #
  config.processor_host = 'http://localhost:3000'

  #
  # If you do not have any Rails secret_key_base defined, uncomment the following
  # This secret is used to authenticate jobs sent to the processing endpoint
  # of your application.
  #
  # config.secret = 'some-long-token'
end

# app/workers/dummy_worker.rb

class DummyWorker
  include Cloudtasker::Worker

  def perform(some_arg)
    logger.info("Job run with #{some_arg}. This is working!")
  end
end

  # Process job as soon as possible
  DummyWorker.perform_async('foo')

  # Process job in 60 seconds
  DummyWorker.perform_in(60, 'foo')

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  #
  # Adapt the server port to be the one used by your Rails web process
  #
  config.processor_host = 'http://localhost:3000'

  #
  # If you do not have any Rails secret_key_base defined, uncomment the following
  # This secret is used to authenticate jobs sent to the processing endpoint
  # of your application.
  #
  # config.secret = 'some-long-token'
end

# config/application.rb

require_relative 'boot'
require 'rails/all'

Bundler.require(*Rails.groups)

module Dummy
  class Application < Rails::Application
    # Initialize configuration defaults for originally generated Rails version.
    config.load_defaults 6.0

    # Settings in config/environments/* take precedence over those specified here.
    # Application configuration can go into files in config/initializers
    # -- all .rb files in that directory are automatically loaded after loading
    # the framework and any gems in your application.

    # Use cloudtasker as the ActiveJob backend:
    config.active_job.queue_adapter = :cloudtasker
  end
end

# app/jobs/example_job.rb

class ExampleJob < ApplicationJob
  queue_as :default

  def perform(some_arg)
    logger.info("Job run with #{some_arg}. This is working!")
  end
end

  # Process job as soon as possible
  ExampleJob.perform_later('foo')

  # Process job in 60 seconds
  ExampleJob.set(wait: 60).perform_later('foo')

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  #
  # If you do not have any Rails secret_key_base defined, uncomment the following.
  # This secret is used to authenticate jobs sent to the processing endpoint
  # of your application.
  #
  # Default with Rails: Rails.application.credentials.secret_key_base
  #
  # config.secret = 'some-long-token'

  #
  # Specify the details of your Google Cloud Task location. 
  # 
  # This is required when the mode of operation is set to :production
  # 
  config.gcp_location_id = 'us-central1' # defaults to 'us-east1'
  config.gcp_project_id = 'my-gcp-project'

  #
  # Specify the namespace for your Cloud Task queues.
  #
  # Specifying a namespace is optional but strongly recommended to keep
  # queues organised, especially in a micro-service environment.
  #
  # The gem assumes that a least a default queue named 'my-app-default'
  # exists in Cloud Tasks. You can create this default queue using the
  # gcloud SDK or via the `rake cloudtasker:setup_queue` task if you use Rails.
  #
  # Workers can be scheduled on different queues. The name of the queue
  # in Cloud Tasks is always assumed to be prefixed with the prefix below.
  #
  # E.g.
  # Setting `cloudtasker_options queue: 'critical'` on a worker means that
  # the worker will be pushed to 'my-app-critical' in Cloud Tasks.
  #
  # Specific queues can be created in Cloud Tasks using the gcloud SDK or
  # via the `rake cloudtasker:setup_queue name=<queue_name>` task.
  #
  config.gcp_queue_prefix = 'my-app'

  #
  # Specify the publicly accessible host for your application
  #
  # > E.g. in development, using the cloudtasker local server
  # config.processor_host = 'http://localhost:3000'
  #
  # > E.g. in development, using `config.mode = :production` and ngrok
  # config.processor_host = 'https://111111.ngrok.io'
  #
  config.processor_host = 'https://app.mydomain.com'

  #
  # Specify the mode of operation:
  # - :development => jobs will be pushed to Redis and picked up by the Cloudtasker local server
  # - :production => jobs will be pushed to Google Cloud Tasks. Requires a publicly accessible domain.
  #
  # Defaults to :development unless CLOUDTASKER_ENV or RAILS_ENV or RACK_ENV is set to something else.
  #
  # config.mode = Rails.env.production? || Rails.env.my_other_env? ? :production : :development

  #
  # Specify the logger to use
  #
  # Default with Rails: Rails.logger
  # Default without Rails: Logger.new(STDOUT)
  #
  # config.logger = MyLogger.new(STDOUT)

  #
  # Specify how many retries are allowed on jobs. This number of retries excludes any
  # connectivity error due to the application being down or unreachable.
  #
  # Default: 25
  #
  # config.max_retries = 10

  #
  # Specify the redis connection hash.
  #
  # This is ONLY required in development for the Cloudtasker local server and in
  # all environments if you use any cloudtasker extension (unique jobs, cron jobs,
  # batch jobs or storable jobs)
  #
  # See https://github.com/redis/redis-rb for examples of configuration hashes.
  #
  # Default: redis-rb connects to redis://127.0.0.1:6379/0
  #
  # config.redis = { url: 'redis://localhost:6379/5' }

  #
  # Set to true to store job arguments in Redis instead of sending arguments as part
  # of the job payload to Google Cloud Tasks.
  #
  # This is useful if you expect to process jobs with payloads exceeding 100KB, which
  # is the limit enforced by Google Cloud Tasks.
  #
  # You can set this configuration parameter to a KB value if you want to store jobs
  # args in redis only if the JSONified arguments payload exceeds that threshold.
  #
  # Supported since: v0.10.0
  #
  # Default: false
  #
  # Store all job payloads in Redis:
  # config.store_payloads_in_redis = true
  #
  # Store all job payloads in Redis exceeding 50 KB:
  # config.store_payloads_in_redis = 50

  #
  # Specify the dispatch deadline for jobs in Cloud Tasks, in seconds.
  # Jobs taking longer will be retried by Cloud Tasks, even if they eventually
  # complete on the server side.
  #
  # Note that this option is applied when jobs are enqueued job. Changing this value
  # will not impact already enqueued jobs.
  #
  # This option can also be configured on a per worker basis via
  # the cloudtasker_options directive.
  #
  # Supported since: v0.12.0
  #
  # Default: 600 seconds (10 minutes)
  # Min: 15 seconds
  # Max: 1800 seconds (30 minutes)
  #
  # config.dispatch_deadline = 600

  #
  # Specify a proc to be invoked every time a job fails due to a runtime
  # error.
  #
  # This hook is not invoked for DeadWorkerError. See on_dead instead.
  #
  # This is useful when you need to apply general exception handling, such
  # as reporting errors to a third-party service like Rollbar or Bugsnag.
  #
  # Note: the worker argument might be nil, such as when InvalidWorkerError is raised.
  #
  # Supported since: v0.12.0
  # 
  # Default: no operation
  #
  # config.on_error = ->(error, worker) { Rollbar.error(error) }

  #
  # Specify a proc to be invoked every time a job dies due to too many
  # retries.
  #
  # This is useful when you need to apply general exception handling, such
  # logging specific messages/context when a job dies.
  #
  # Supported since: v0.12.0
  # 
  # Default: no operation
  #
  # config.on_dead = ->(error, worker) { Rollbar.error(error) }

  #
  # Specify the Open ID Connect (OIDC) details to connect to a protected GCP service, such
  # as a private Cloud Run application.
  #
  # The configuration supports the following details:
  # - service_account_email: This is the "act as" user. It can be found under the security details
  #   of the Cloud Run service.
  # - audience: The audience is usually the publicly accessible host for the Cloud Run service
  #   (which is the same value configured as the processor_host). If no audiences are provided
  #   it will be set to the processor_host.
  #
  # Note: If the OIDC token is used for a Cloud Run service make sure to include the
  # `iam.serviceAccounts.actAs` permission on the service account.
  #
  # See https://cloud.google.com/tasks/docs/creating-http-target-tasks#sa for more information on
  # setting up service accounts for use with Cloud Tasks.
  #
  # Supported since: v0.14.0 (upcoming)
  #
  # Default: nil 
  #
  # config.oidc = { service_account_email: '[email protected]' }
  # config.oidc = { service_account_email: '[email protected]', audience: 'https://api.example.net' }

  #
  # Enable/disable the verification of SSL certificates on the local processing server when
  # sending tasks to the processor.
  #
  # Set to false to disable SSL verification (OpenSSL::SSL::VERIFY_NONE).
  #
  # Default: true
  #
  # config.local_server_ssl_verify = true
end

# Worker will be processed as soon as possible
MyWorker.perform_async(arg1, arg2)

# Worker will be processed in 5 minutes
MyWorker.perform_in(5 * 60, arg1, arg2)
# or with Rails
MyWorker.perform_in(5.minutes, arg1, arg2)

# Worker will be processed on a specific date
MyWorker.perform_at(Time.parse('2025-01-01 00:50:00Z'), arg1, arg2)
# also with Rails
MyWorker.perform_at(3.days.from_now, arg1, arg2)

# With all options, including which queue to run the worker on.
MyWorker.schedule(args: [arg1, arg2], time_at: Time.parse('2025-01-01 00:50:00Z'), queue: 'critical')
# or
MyWorker.schedule(args: [arg1, arg2], time_in: 5 * 60, queue: 'critical')

# app/workers/fetch_resource_worker.rb

class FetchResourceWorker
  include Cloudtasker::Worker

  def perform(id)
    # ...do some logic...
    if some_condition
      # Stop and re-enqueue the job to be run again in 10 seconds.
      # Also see the section on Cloudtasker::RetryWorkerError for a different
      # approach on reenqueuing.
      return reenqueue(10)
    else
      # ...keep going...
    end
  end
end

# app/workers/critical_worker.rb

class CriticalWorker
  include Cloudtasker::Worker

  cloudtasker_options queue: :critical

  def perform(some_arg)
    logger.info("This is a critical job run with arg=#{some_arg}.")
  end
end

CriticalWorker.schedule(args: [1], queue: :important)

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  # ... other options

  # Push jobs to redis and let the Cloudtasker local server collect them
  # This is the default mode unless CLOUDTASKER_ENV or RAILS_ENV or RACK_ENV is set
  # to a non-development environment
  config.mode = :development
end

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  # Specify your Google Cloud Task queue configuration
  config.gcp_location_id = 'us-central1'
  config.gcp_project_id = 'my-gcp-project'
  config.gcp_queue_prefix = 'my-app'

  # Use your ngrok domain as the processor host
  config.processor_host = 'https://your-tunnel-id.ngrok.io'

  # Force Cloudtasker to use Google Cloud Tasks in development
  config.mode = :production
end

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  config.logger = SemanticLogger[Cloudtasker]
end

# app/workers/dummy_worker.rb

class DummyWorker
  include Cloudtasker::Worker

  def perform(some_arg)
    logger.info("Job run with #{some_arg}. This is working!")
  end
end

Cloudtasker::WorkerLogger.log_context_processor = ->(worker) { worker.to_h.slice(:worker, :job_id, :job_meta, :job_queue, :task_id) }

# config/initializers/cloudtasker.rb

Cloudtasker::WorkerLogger.log_context_processor = lambda { |worker|
  worker.to_h.slice(:worker, :job_id, :job_meta, :job_queue, :task_id).merge(app: 'my-app')
}

# app/workers/full_context_worker.rb

class FullContextWorker
  include Cloudtasker::Worker

  cloudtasker_options log_context_processor: ->(worker) { worker.to_h }

  def perform(some_arg)
    logger.info("This log entry will have full context!")
  end
end

Cloudtasker::WorkerLogger.log_context_processor = ->(worker) { worker.to_h }

# string_limit: The maximum size for strings. Default is 64. Set to -1 to disable.
# array_limit: The maximum length for arrays. Default is 10. Set to -1 to disable.
# max_depth: The maximum recursive depth. Default is 3. Set to -1 to disable.
Cloudtasker::WorkerLogger.truncate(payload, string_limit: 64, array_limit: 10, max_depth: 3)

Cloudtasker::WorkerLogger.log_context_processor = lambda do |worker|
  payload = worker.to_h

  # Using default options
  payload[:job_args] = Cloudtasker::WorkerLogger.truncate(payload[:job_args])

  # Using custom options
  # payload[:job_args] = Cloudtasker::WorkerLogger.truncate(payload[:job_args], string_limit: 32, array_limit: 5, max_depth: 2)

  # Return the payload to log
  payload
end

Cloudtasker::WorkerLogger.log_context_processor = lambda do |worker|
  payload = worker.to_h

  # Adjust the log payload based on the lifecycle of the job
  payload[:job_args] = if worker.perform_started_at
                         # The job start has already been logged. Log the job primitive arguments without depth.
                         # Arrays and hashes will be masked.
                         Cloudtasker::WorkerLogger.truncate(payload[:job_args], max_depth: 0)
                       else
                         # This is the job start. Log a more complete version of the job args.
                         Cloudtasker::WorkerLogger.truncate(payload[:job_args])
                       end

  # Return the payload to log
  payload
end

# app/workers/handle_error_worker.rb

class HandleErrorWorker
  include Cloudtasker::Worker

  def perform
    raise(ArgumentError)
  end

  # The runtime error is passed as an argument.
  def on_error(error)
    logger.error("The following error happened: #{error}")
  end

  # The job has been retried too many times and will be removed
  # from the queue.
  def on_dead(error)
    logger.error("The job died with the following error: #{error}")
  end
end

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  #
  # Report runtime and dead worker errors to Rollbar
  #
  config.on_error = -> (error, _worker) { Rollbar.error(error) }
  config.on_dead = -> (error, _worker) { Rollbar.error(error) }
end

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  #
  # Specify how many retries are allowed on jobs. This number of retries excludes any
  # connectivity error that would be due to the application being down or unreachable.
  #
  # Default: 25
  #
  config.max_retries = 10
end

# app/workers/some_error_worker.rb

class SomeErrorWorker
  include Cloudtasker::Worker

  # This will override the global setting
  cloudtasker_options max_retries: 3

  def perform
    raise(ArgumentError)
  end
end

# app/workers/some_error_worker.rb

class SomeErrorWorker
  include Cloudtasker::Worker

  # Return the number of max retries based on
  # worker arguments.
  #
  # If this method returns nil then max_retries
  # will delegate to the class `max_retries` setting or Cloudtasker
  # `max_retries` configuration otion.
  def max_retries(arg1, arg2)
    arg1 == 'foo' ? 13 : nil
  end

  def perform(arg1, arg2)
    raise(ArgumentError)
  end
end

# app/workers/my_worker.rb

class MyWorker
  include Cloudtasker::Worker

  def perform(project_id)
    # Abort if project does not exist
    return unless (project = Project.find_by(id: project_id))

    # Trigger a retry if the project is still in "discovering" status
    # This error will NOT log an error. It only triggers a retry.
    raise Cloudtasker::RetryWorkerError if project.status == 'discovering'

    # The previous approach was to use `reenqueue`. This works but since it
    # does not increase the number of retries, you may end up with forever running
    # jobs
    # return reenqueue(10) if project.status == 'discovering'

    # Do stuff when project is not longer discovering
    do_some_stuff
  end

  # You can then specify what should be done if we've been waiting for too long
  def on_dead(error)
    logger.error("Looks like the project is forever discovering. Time to give up.")

    # This is of course an imaginary method
    send_slack_notification_to_internal_support_team(worker: self.class, args: job_args)
  end
end

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  #
  # Specify the dispatch deadline for jobs in Cloud Tasks, in seconds. 
  # Jobs taking longer will be retried by Cloud Tasks, even if they eventually
  # complete on the server side.
  #
  # Note that this option is applied when jobs are enqueued job. Changing this value
  # will not impact already enqueued jobs.
  #
  # Default: 600 (10 minutes)
  #
  config.dispatch_deadline = 20 * 60 # 20 minutes
end

# app/workers/some_error_worker.rb

class SomeFasterWorker
  include Cloudtasker::Worker

  # This will override the global setting
  cloudtasker_options dispatch_deadline: 5 * 60

  def perform
    # ... do things ...
  end
end

require 'cloudtasker/testing'

# Mode 1 (default): Push jobs to Google Cloud Tasks (env != development) or Redis (env == development)
Cloudtasker::Testing.enable!

# Mode 2: Push jobs to an in-memory queue. Jobs will not be processed until you call
# Cloudtasker::Worker.drain_all (process all jobs) or MyWorker.drain (process jobs for specific worker)
Cloudtasker::Testing.fake!

# Mode 3: Push jobs to an in-memory queue. Jobs will be processed immediately.
Cloudtasker::Testing.inline!

Cloudtasker::Testing.enabled?
Cloudtasker::Testing.fake?
Cloudtasker::Testing.inline?

# Enable fake mode for all tests
Cloudtasker::Testing.fake!

# Enable inline! mode temporarily for a given test
Cloudtasker.inline! do
   MyWorker.perform_async(1,2)
end

# Remove all middlewares
Cloudtasker.configure do |c|
  c.client_middleware.clear
  c.server_middleware.clear
end

# Remove all unique job middlewares
Cloudtasker.configure do |c|
  c.client_middleware.remove(Cloudtasker::UniqueJob::Middleware::Client)
  c.server_middleware.remove(Cloudtasker::UniqueJob::Middleware::Server)
end

# Perform all jobs in queue
Cloudtasker::Worker.drain_all

# Remove all jobs in queue
Cloudtasker::Worker.clear_all

# Perform all jobs in queue for a specific worker type
MyWorker.drain

# Return the list of jobs in queue for a specific worker type
MyWorker.jobs

describe 'worker scheduling'
  subject(:enqueue_job) { MyWorker.perform_async(1,2) }

  it { expect { enqueue_job }.to change(MyWorker.jobs, :size).by(1) }
end

describe 'worker calls api'
  subject { Cloudtasker::Testing.inline! { MyApiWorker.perform_async(1,2) } }

  before { expect(MyApi).to receive(:fetch).and_return([]) }
  it { is_expected.to be_truthy }
end

# app/workers/user_email_worker.rb

class UserEmailWorker
  include Cloudtasker::Worker

  def perform(user)
    user.reload.send_email
  end
end

# app/workers/user_email_worker.rb

class UserEmailWorker
  include Cloudtasker::Worker

  def perform(user_id)
    User.find_by(id: user_id)&.send_email
  end
end

# Enqueuing a job with:
MyWorker.perform_async({ foo: 'bar', 'baz' => { key: 'value' } })

# will be processed as
MyWorker.new.perform({ 'foo' => 'bar', 'baz' => { 'key' => 'value' } })

# app/workers/user_email_worker.rb

class UserEmailWorker
  include Cloudtasker::Worker

  cloudtasker_options lock: :until_executed

  def perform(user_id, time_at = Time.now.iso8601)
    User.find_by(id: user_id)&.send_email(Time.parse(time_at))
  end
end

# config/initializers/cloudtasker.rb

Cloudtasker.configure do |config|
  # Enable Redis support. Specify your redis connection
  config.redis = { url: 'redis://localhost:6379/5' }

  # Store all job payloads in Redis:
  config.store_payloads_in_redis = true

  # OR: store all job payloads in Redis exceeding 50 KB:
  # config.store_payloads_in_redis = 50
end

# app/workers/big_payload_worker.rb

class BigPayloadWorker
  include Cloudtasker::Worker

  def perform(payload_id)
    data = Rails.cache.fetch(payload_id)
    # ...do some processing...
  end
end

# Fetch and store the payload
data = ApiClient.fetch_thousands_of_records
payload_id = SecureRandom.uuid
Rails.cache.write(payload_id, data)

# Enqueue the processing job
BigPayloadWorker.perform_async(payload_id)