class SidekiqUniqueJobs::Locksmith

def ==(other)
  key == other.key && job_id == other.job_id
end

def argv
  [job_id, config.pttl, config.type, config.limit]
end

def brpoplpush(conn)
  # passing timeout 0 to brpoplpush causes it to block indefinitely
  conn.brpoplpush(key.queued, key.primed, timeout: config.timeout || 0)
end

def delete
  return if config.pttl.positive?
  delete!
end

def delete!
  call_script(:delete, key.to_a, [job_id, config.pttl, config.type, config.limit]).positive?
end

def drift(val)
  # Add 2 milliseconds to the drift to account for Redis expires
  # precision, which is 1 millisecond, plus 1 millisecond min drift
  # for small TTLs.
  (val.to_i * CLOCK_DRIFT_FACTOR).to_i + 2
end

def enqueue(conn)
  queued_token, elapsed = timed do
    call_script(:queue, key.to_a, argv, conn)
  end
  validity = config.pttl - elapsed - drift(config.pttl)
  return unless queued_token && (validity >= 0 || config.pttl.zero?)
  write_lock_info(conn)
  yield
end

def initialize(item, redis_pool = nil)
  @item        = item
  @key         = Key.new(item[LOCK_DIGEST] || item[UNIQUE_DIGEST]) # fallback until can be removed
  @job_id      = item[JID]
  @config      = LockConfig.new(item)
  @redis_pool  = redis_pool
end

def inspect
  to_s
end

def lock(&block)
  redis(redis_pool) do |conn|
    return lock_async(conn, &block) if block_given?
    lock_sync(conn) do
      return job_id
    end
  end
end

def lock_async(conn)
  return yield job_id if locked?(conn)
  enqueue(conn) do
    primed_async(conn) do
      locked_token = call_script(:lock, key.to_a, argv, conn)
      return yield if locked_token == job_id
    end
  end
ensure
  unlock!(conn)
end

def lock_info
  @lock_info ||= dump_json(
    WORKER => item[CLASS],
    QUEUE => item[QUEUE],
    LIMIT => item[LOCK_LIMIT],
    TIMEOUT => item[LOCK_TIMEOUT],
    TTL => item[LOCK_TTL],
    LOCK => config.type,
    LOCK_ARGS => item[LOCK_ARGS],
    TIME => now_f,
  )
end

def lock_sync(conn)
  return yield if locked?(conn)
  enqueue(conn) do
    primed_sync(conn) do
      locked_token = call_script(:lock, key.to_a, argv, conn)
      return yield if locked_token
    end
  end
end

def locked?(conn = nil)
  return taken?(conn) if conn
  redis { |rcon| taken?(rcon) }
end

def pop_queued(conn)
  if config.wait_for_lock?
    brpoplpush(conn)
  else
    rpoplpush(conn)
  end
end

def primed_async(conn)
  return yield if Concurrent::Promises
                  .future(conn) { |red_con| pop_queued(red_con) }
                  .value(drift(config.ttl))
  warn_about_timeout
end

def primed_sync(conn)
  return yield if pop_queued(conn)
  warn_about_timeout
end

def rpoplpush(conn)
  conn.rpoplpush(key.queued, key.primed)
end

def taken?(conn)
  conn.hexists(key.locked, job_id)
end

def to_s
  "Locksmith##{object_id}(digest=#{key} job_id=#{job_id}, locked=#{locked?})"
end

def unlock(conn = nil)
  return false unless locked?(conn)
  unlock!(conn)
end

def unlock!(conn = nil)
  call_script(:unlock, key.to_a, argv, conn)
end

def warn_about_timeout
  log_warn("Timed out after #{config.timeout}s while waiting for primed token (digest: #{key}, job_id: #{job_id})")
end

def write_lock_info(conn)
  return unless config.lock_info
  conn.set(key.info, lock_info)
end