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 or batch 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)
  #
  # 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 oidc hash.
  #
  # Contains information needed for generating an OpenID Connect token. 
  # This type of authorization can be used for many scenarios, including calling Cloud Run, 
  # or endpoints where you intend to validate the token yourself. 
  # For the oidcs hash, the service_account_email can be found 
  # under the security details of the cloud run service.
  # 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 audience is provided
  # it will be inferred as the processor_host.
  # Note: If the oidc token is used for a google cloud run service make sure to include
  # the iam.serviceAccounts.actAs permission for the service account. 
  #
  # See https://openid.net/connect for more information on OpenID Connect tokens.
  #
  # Default: nil 
  #
  # config.oidc = { service_account_email: '[email protected]' }
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.
      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

# 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

# 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)