Batched Background operations

Settings.cron_jobs['inactive_projects_deletion_cron_worker'] ||= {}
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['cron'] ||= '*/10 * * * *'
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['job_class'] = 'Projects::InactiveProjectsDeletionCronWorker'

def perform
  project_id = last_processed_project_id

  Project.where('projects.id > ?', project_id).each_batch(of: 100) do |batch|
    inactive_projects = batch.inactive.not_aimed_for_deletion

    inactive_projects.each do |project|
      if over_time?
        save_last_processed_project_id(project.id)
        raise TimeoutError
      end

      with_context(project: project, user: admin_bot) do
        deletion_warning_email_sent_on = notified_inactive_projects["project:#{project.id}"]

        if deletion_warning_email_sent_on.blank?
          send_notification(project)
          log_audit_event(project, admin_bot)
        elsif grace_period_is_over?(deletion_warning_email_sent_on)
          Gitlab::DormantProjectsDeletionWarningTracker.new(project.id).reset
          delete_project(project, admin_bot)
        end
      end
    end
  end
  reset_last_processed_project_id
rescue TimeoutError
  # no-op
end

def save_last_processed_project_id(project_id)
  with_redis do |redis|
    redis.set(LAST_PROCESSED_INACTIVE_PROJECT_REDIS_KEY, project_id)
  end
end


def reset_last_processed_project_id
  with_redis do |redis|
    redis.del(LAST_PROCESSED_INACTIVE_PROJECT_REDIS_KEY)
  end
end

Settings.cron_jobs['inactive_projects_deletion_cron_worker'] ||= {}
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['cron'] ||= '*/10 * * * *'
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['job_class'] = 'EnqueueBatchedBackgroundOperationWorker'
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['job_arguments'] = { bbo_job_class_name: 'Projects::InactiveProjectsDeletionCronWorker', min_cursor: [<last_processed_project_id>] }

Settings.cron_jobs['inactive_projects_deletion_cron_worker'] ||= {}
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['cron'] ||= '*/10 * * * *'
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['job_class'] = 'Projects::InactiveProjectsDeletionCronWorker'
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['job_arguments'] = { min_cursor: [<last_processed_project_id>] }
Settings.cron_jobs['inactive_projects_deletion_cron_worker']['batched_operation'] = true

module Gitlab
  module BackgroundOperation
    # BatchedOperationJob will have the sub-batch iteration logic, similar to BBM
    class Project::DeleteInactiveProjects < Gitlab::Database::BatchedOperationJob
      feature_category <feature_category>

      # 'last_processed_project_id' from current design can be set as the min_cursor
      cursor(:id)

      def perform
        each_sub_batch do |sub_batch|
          inactive_projects = sub_batch.inactive.not_aimed_for_deletion

          inactive_projects.each do |project|
            with_context(project: project, user: admin_bot) do
              deletion_warning_email_sent_on = notified_inactive_projects["project:#{project.id}"]

              if deletion_warning_email_sent_on.blank?
                send_notification(project)
                log_audit_event(project, admin_bot)
              elsif grace_period_is_over?(deletion_warning_email_sent_on)
                Gitlab::DormantProjectsDeletionWarningTracker.new(project.id).reset
                delete_project(project, admin_bot)
              end
            end
          end
        end
      end
    end
  end
end

module Mutations
  module Todos
    class DeleteAllDone < ::Mutations::BaseMutation
      def resolve(updated_before: nil)
        delete_until = (updated_before || Time.now).utc.to_datetime.to_s

        ::Todos::DeleteAllDoneWorker.perform_async(current_user.id, delete_until)

        {
          message: format(_('Your request has succeeded. Results will be visible in a couple of minutes.')),
          errors: []
        }
      end
    end
  end
end

module Todos
  class DeleteAllDoneWorker
    include ApplicationWorker
    include EachBatch

    LOCK_TIMEOUT = 1.hour
    BATCH_DELETE_SIZE = 10_000
    SUB_BATCH_DELETE_SIZE = 100
    SLEEP_INTERVAL = 100
    MAX_RUNTIME = 2.minutes

    def perform(user_id, time)
      runtime_limiter = Gitlab::Metrics::RuntimeLimiter.new(MAX_RUNTIME)
      delete_until = time.to_datetime
      pause_ms = SLEEP_INTERVAL

      in_lock("#{self.class.name.underscore}_#{user_id}", ttl: LOCK_TIMEOUT, retries: 0) do
        Todo.where(user_id: user_id)
            .with_state(:done)
            .each_batch(of: BATCH_DELETE_SIZE) do |batch|
              batch.each_batch(of: SUB_BATCH_DELETE_SIZE) do |sub_batch|
                sql = <<~SQL
                      WITH sub_batch AS MATERIALIZED (
                        #{sub_batch.select(:id, :updated_at).limit(SUB_BATCH_DELETE_SIZE).to_sql}
                      ), filtered_relation AS MATERIALIZED (
                      SELECT id FROM sub_batch WHERE updated_at < '#{delete_until.to_fs(:db)}' LIMIT #{SUB_BATCH_DELETE_SIZE}
                    )
                    DELETE FROM todos WHERE id IN (SELECT id FROM filtered_relation)
                    SQL

                Todo.connection.exec_query(sql)

                sleep(pause_ms * 0.001) # Avoid hitting the database too hard
              end

              next unless runtime_limiter.over_time?

              self.class.perform_in(MAX_RUNTIME, user_id, time)

              break
            end
      end
    end
  end
end

module Mutations
  module Todos
    class DeleteAllDone < ::Mutations::BaseMutation
      def resolve(updated_before: nil)
        delete_until = (updated_before || Time.now).utc.to_datetime.to_s

        queue_batched_background_operation(
          'Gitlab::Database::BackgroundOperation::DeleteAllDoneTodos',
          :todos,
          :id,
          user: current_user,
          job_arguments: {
            delete_until: delete_until
          }
        )

        {
          message: format(_('Your request has succeeded. Results will be visible in a couple of minutes.')),
          errors: []
        }
      end
    end
  end
end

def queue_batched_background_operation(job_class_name, table_name, column_name, user: nil, job_arguments: {})
  user_specific_args = user.present? ? { user_id: user.id, organization_id: user.organization_id } : {}

  # Having only crucial args for simplicity, it will also assign other attrs during the development (eg: batch_size, sub_batch_size, cursors, etc.,)
  Gitlab::Database::BatchedBackgroundOperation.create!({
    job_class_name: job_class_name,
    table_name: table_name,
    column_name: column_name,
    job_arguments: job_arguments
  }.merge(user_specific_args))
end

module Gitlab
  module BackgroundOperation
    # BatchedOperationJob will have the sub-batch iteration logic, similar to BBM
    class DeleteAllDoneTodos < BatchedOperationJob
      feature_category <feature_category>

      cursor(:id)

      scope_to ->(relation) do
        relation
          .with_state(:done)
          .where("user_id = ? AND updated_at < ?", user_id, delete_until.to_fs(:db))
      end

      def perform
        each_sub_batch do |sub_batch|
          sub_batch.delete_all
        end
      end
    end
  end
end

module Ci
  class DestroyOldPipelinesWorker
    def perform_work(*)
      Project.find_by_id(fetch_next_project_id).try do |project|
        ...
        # This pushes the next ID to the redis store
        Ci::DestroyPipelineService.new(project, nil).unsafe_execute(pipelines)
      end
    end

    def fetch_next_project_id
      Gitlab::Redis::SharedState.with do |redis|
        redis.lpop(queue_key)
      end
    end
  end

module Ci
  class UnlockPipelinesInQueueWorker
    def perform_work(*_)
      # `next!` fetches the next ID from cache
      pipeline_id, enqueue_timestamp = Ci::UnlockPipelineRequest.next!
      return log_extra_metadata_on_done(:remaining_pending, 0) unless pipeline_id

      Ci::Pipeline.find_by_id(pipeline_id).try do |pipeline|
        log_extra_metadata_on_done(:pipeline_id, pipeline.id)
        log_extra_metadata_on_done(:project, pipeline.project.full_path)

        result = Ci::UnlockPipelineService.new(pipeline).execute
        ...
      end
    end
  end
end

module WorkItems
  class CopyTimelogsWorker
    include ApplicationWorker
    urgency :high

    BATCH_SIZE = 100

    def perform(from_issue_id, to_issue_id)
      ...
      ...
      from_issue = Issue.find_by_id(from_issue_id)

      reset_attributes = { project_id: to_issue.project_id, issue_id: to_issue.id }
      ApplicationRecord.transaction do
        from_issue.timelogs.each_batch(of: BATCH_SIZE) do |timelogs|
          new_timelogs_attributes = timelogs.map do |timelog|
          timelog.attributes.except('id').merge(reset_attributes)
        end

        Timelog.insert_all!(new_timelogs_attributes)
      end
    end
  end
end

PARTITION_DURATION = 7.days

partitioned_by :partition_id, strategy: :sliding_list,
  next_partition_if: ->(active_partition) do
    oldest_record_in_partition = BatchedBackgroundOperation
      .select(:id, :created_at)
      .for_partition(active_partition.value)
      .order(:id)
      .limit(1)
      .take

    oldest_record_in_partition.present? && oldest_record_in_partition.created_at < PARTITION_DURATION.ago
  end,
  detach_partition_if: ->(partition) do
    !BatchedBackgroundOperation
      .for_partition(partition.value)
      .executable # with status [active, paused]
      .exists?
  end
end