Redis #

gem install redis
gem install connection_pool   # for multi-threading

# Gemfile
gem 'redis',           '~> 5.0'
gem 'connection_pool', '~> 2.4'

require 'redis'

# Basic connection
redis = Redis.new(
  host:     "localhost",
  port:     6379,
  db:       0,             # databases 0-15
  password: ENV["REDIS_PASSWORD"],
  timeout:  1,             # connection timeout
  read_timeout:  1,        # read timeout
  write_timeout: 1         # write timeout
)

# Via URL (more concise)
redis = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))

# With TLS (Redis Cloud, production)
redis = Redis.new(
  url: ENV["REDIS_TLS_URL"],
  ssl_params: { verify_mode: OpenSSL::SSL::VERIFY_PEER }
)

# Test the connection
puts redis.ping   # => "PONG"
puts redis.info["redis_version"]

# Always close the connection
redis.close

require 'redis'
require 'connection_pool'

# Create a pool — one connection per thread
REDIS = ConnectionPool.new(size: 10, timeout: 5) do
  Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
end

# Use with a block
REDIS.with do |redis|
  redis.set("key", "value")
  redis.get("key")
end

# Or with Redis::Pool (Redis gem 5+)
redis = Redis.new(
  url:          ENV["REDIS_URL"],
  timeout:      1,
  pool_size:    10,    # integrated pool in Redis gem 5+
  pool_timeout: 5
)

redis = Redis.new

# SET and GET
redis.set("name", "Rina")
puts redis.get("name")    # => "Rina"

# SET with TTL (Time To Live) in seconds
redis.set("session_token", SecureRandom.hex(32), ex: 3600)   # expires in 1 hour
redis.set("otp",          "123456",              ex: 300)    # expires in 5 minutes

# SET with TTL in milliseconds
redis.set("lock", "1", px: 5000)   # expires in 5000ms = 5 seconds

# SETNX — set only if the key doesn't exist (returns true/false)
success = redis.setnx("unique_key", "value")
puts success   # => true if successful, false if it already exists

# SET with NX and EX together (atomic)
redis.set("distributed_lock", "process_id", nx: true, ex: 10)

# GET and SET at once (GETSET)
old_value = redis.getset("counter", "0")

# MSET and MGET — bulk operations
redis.mset("a", 1, "b", 2, "c", 3)
puts redis.mget("a", "b", "c").inspect   # => ["1", "2", "3"]

# Increment and Decrement (atomic!)
redis.set("view_count", 0)
redis.incr("view_count")         # => 1
redis.incrby("view_count", 10)   # => 11
redis.decr("view_count")         # => 10
redis.incrbyfloat("rating", 0.5) # => 0.5

# String operations
redis.append("log", "first line\n")
redis.append("log", "second line\n")
redis.strlen("log")   # string length

# TTL — remaining key lifetime
puts redis.ttl("session_token")    # => remaining seconds, -1 if no TTL, -2 if it doesn't exist
puts redis.pttl("lock")            # in milliseconds

# Extend or remove a TTL
redis.expire("session_token", 7200)   # extend to 2 hours
redis.persist("session_token")        # remove the TTL (becomes permanent)
redis.expireat("event", Time.now.to_i + 86400)  # expire at a specific timestamp

# HSET — set one or more fields
redis.hset("user:1", "name", "Rina", "email", "[email protected]", "age", 28)

# HGET and HMGET
puts redis.hget("user:1", "name")                        # => "Rina"
puts redis.hmget("user:1", "name", "email").inspect      # => ["Rina", "[email protected]"]

# HGETALL — all fields as a Ruby Hash
profile = redis.hgetall("user:1")
puts profile.inspect   # => {"name"=>"Rina", "email"=>"[email protected]", "age"=>"28"}

# HSETNX — set a field only if it doesn't exist
redis.hsetnx("user:1", "created_at", Time.now.iso8601)

# HINCRBY — increment a numeric field (atomic)
redis.hincrby("user:1", "login_count", 1)

# HDEL — delete fields
redis.hdel("user:1", "unneeded_field")

# HEXISTS, HKEYS, HVALS, HLEN
puts redis.hexists("user:1", "name")   # => true
puts redis.hkeys("user:1").inspect     # => ["name", "email", "age", ...]
puts redis.hlen("user:1")              # => number of fields

# Common pattern: store a model as a Redis Hash
class UserCache
  KEY_PREFIX = "user:"

  def self.save(user)
    key = "#{KEY_PREFIX}#{user.id}"
    Redis.new.hset(key,
      "id",        user.id,
      "name",      user.name,
      "email",     user.email,
      "role",      user.role,
      "cached_at", Time.now.iso8601
    )
    Redis.new.expire(key, 3600)   # 1 hour TTL
  end

  def self.fetch(id)
    data = Redis.new.hgetall("#{KEY_PREFIX}#{id}")
    return nil if data.empty?
    data.transform_keys(&:to_sym)
  end

  def self.delete(id)
    Redis.new.del("#{KEY_PREFIX}#{id}")
  end
end

# LPUSH / RPUSH — add to the left / right
redis.rpush("email_queue", "[email protected]")
redis.rpush("email_queue", "[email protected]", "[email protected]")
redis.lpush("activity_log", "login:user1")   # add to the front (newest on the left)

# LRANGE — fetch a range of elements
puts redis.lrange("email_queue", 0, -1).inspect   # all elements
puts redis.lrange("activity_log", 0, 9).inspect    # 10 newest

# LPOP / RPOP — fetch and remove from the left / right
email = redis.lpop("email_queue")    # fetch from the front (FIFO)
redis.rpop("email_queue")            # fetch from the back (LIFO)

# BLPOP — blocking pop (waits until an element is available)
# Very useful for queue consumers
result = redis.blpop("email_queue", timeout: 30)
# => ["email_queue", "[email protected]"] or nil on timeout

# LLEN — list length
puts redis.llen("email_queue")

# LINSERT — insert before/after a specific element
redis.linsert("list", :before, "target_element", "new_element")

# Cap the list length (keep only the N newest elements)
redis.lpush("user_activity:1", "new event")
redis.ltrim("user_activity:1", 0, 99)   # keep only the 100 newest

# LPOS — find an element's position (Redis 6.0.6+)
redis.lpos("list", "searched_value")

# SADD — add members
redis.sadd("tag:article:1", "ruby", "programming", "tips")
redis.sadd("tag:article:2", "ruby", "rails", "web")

# SMEMBERS — all members
puts redis.smembers("tag:article:1").inspect   # => Set {"ruby", "programming", "tips"}

# SISMEMBER — membership check (O(1))
puts redis.sismember("tag:article:1", "ruby")   # => true
puts redis.sismember("tag:article:1", "java")   # => false

# SCARD — number of members
puts redis.scard("tag:article:1")   # => 3

# Set operations
# SUNION — union
puts redis.sunion("tag:article:1", "tag:article:2").inspect
# => {"ruby", "programming", "tips", "rails", "web"}

# SINTER — intersection
puts redis.sinter("tag:article:1", "tag:article:2").inspect
# => {"ruby"}

# SDIFF — difference
puts redis.sdiff("tag:article:1", "tag:article:2").inspect
# => {"programming", "tips"}

# SREM — remove members
redis.srem("tag:article:1", "tips")

# SPOP — fetch and remove random members (useful for random sampling)
redis.spop("prize_pool", 3)   # pick 3 random winners

# Pattern: tracking online users
redis.sadd("online_users", user_id)
redis.expire("online_users", 300)   # reset every 5 minutes

# Pattern: daily unique visitor tracking
today = Date.today.strftime("%Y%m%d")
redis.sadd("visitor:#{today}", ip_address)
redis.expire("visitor:#{today}", 86400)
puts redis.scard("visitor:#{today}")   # unique visitors today

# ZADD — add members with scores
redis.zadd("leaderboard", 1500, "rina")
redis.zadd("leaderboard", 2300, "budi")
redis.zadd("leaderboard", 1800, "citra")
redis.zadd("leaderboard", 2300, "deni")   # same score as budi

# ZRANGE — fetch by rank (ascending)
puts redis.zrange("leaderboard", 0, -1, with_scores: true).inspect
# => [["rina", 1500.0], ["citra", 1800.0], ["budi", 2300.0], ["deni", 2300.0]]

# ZREVRANGE — descending (highest score first)
top3 = redis.zrevrange("leaderboard", 0, 2, with_scores: true)
top3.each_with_index do |(name, score), i|
  puts "##{i+1}: #{name} — #{score.to_i} points"
end

# ZSCORE — a member's score
puts redis.zscore("leaderboard", "rina")   # => 1500.0

# ZRANK / ZREVRANK — ranking position
puts redis.zrevrank("leaderboard", "budi")   # => 0 (rank 1 from the top)

# ZINCRBY — add to a score (atomic)
redis.zincrby("leaderboard", 200, "rina")   # rina is now 1700

# ZRANGEBYSCORE — fetch members within a score range
redis.zrangebyscore("leaderboard", 1500, 2000)

# Rate Limiting with a Sorted Set
def rate_limit_ok?(user_id, max_requests: 100, window: 60)
  key     = "rate:#{user_id}"
  now     = Time.now.to_f
  cutoff  = now - window

  redis.multi do |r|
    r.zadd(key, now, now.to_s)   # add this request's timestamp
    r.zremrangebyscore(key, "-inf", cutoff)  # remove expired entries
    r.zcard(key)                            # count requests in the window
    r.expire(key, window)
  end.last <= max_requests
end

# Delayed Queue — execute tasks at a specific time
def schedule(task, execution_time)
  redis.zadd("delayed_queue", execution_time.to_f, task.to_json)
end

def fetch_due_tasks
  now = Time.now.to_f
  redis.zrangebyscore("delayed_queue", "-inf", now, limit: [0, 10]).tap do |tasks|
    tasks.each { |t| redis.zrem("delayed_queue", t) }
  end
end

# Without pipelining: N round-trips to Redis
100.times { |i| redis.set("key:#{i}", i) }   # slow

# With pipelining: 1 round-trip for all commands
redis.pipelined do |pipe|
  100.times { |i| pipe.set("key:#{i}", i) }
end

# Or collect the responses
results = redis.pipelined do |pipe|
  pipe.get("a")
  pipe.get("b")
  pipe.incr("counter")
end
puts results.inspect   # => ["value_a", "value_b", 1]

# MULTI/EXEC — atomic transactions
# All commands in the block execute atomically
redis.multi do |r|
  r.set("balance:1", 500_000)
  r.set("balance:2", 1_500_000)
  r.incr("total_transactions")
end

# WATCH + MULTI/EXEC — optimistic locking
loop do
  redis.watch("balance:1") do
    balance = redis.get("balance:1").to_i

    break if balance < 100_000   # insufficient balance

    # If the balance changes between WATCH and EXEC → multi returns nil (failed)
    result = redis.multi do |r|
      r.set("balance:1", balance - 100_000)
      r.incrby("balance:2", 100_000)
    end

    break if result   # success
    # If nil (conflict) → retry the loop
  end
end

# Distributed lock implementation with SETNX + EXPIRE (atomic since Redis 2.6.12+)
class DistributedLock
  def initialize(redis, name, ttl: 10)
    @redis  = redis
    @key    = "lock:#{name}"
    @value  = "#{Process.pid}-#{Thread.current.object_id}-#{SecureRandom.hex(8)}"
    @ttl    = ttl
  end

  def acquire
    @redis.set(@key, @value, nx: true, ex: @ttl)
  end

  def release
    # Only delete if the value matches (we set the lock)
    # Lua script for atomic check-and-delete
    script = <<~LUA
      if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
      else
        return 0
      end
    LUA
    @redis.eval(script, keys: [@key], argv: [@value])
  end

  def with_lock
    raise "Failed to acquire lock: #{@key}" unless acquire
    begin
      yield
    ensure
      release
    end
  end
end

# Usage
lock = DistributedLock.new(redis, "payment_process:#{order_id}", ttl: 30)
lock.with_lock do
  # Only one process can enter here per order
  process_payment(order_id)
end

# With the redlock gem (a more robust Redlock algorithm implementation)
# gem install redlock
require 'redlock'
redlock = Redlock::Client.new(["redis://localhost:6379"])
redlock.lock("critical_resource", 10_000) do |locked|
  if locked
    # exclusive processing
  else
    puts "Could not acquire the lock"
  end
end

# PUBLISHER — send a message to a channel
publisher = Redis.new
publisher.publish("notification:user:99", JSON.generate({
  type: "new_message",
  from: "Budi",
  body: "Hello!"
}))

# SUBSCRIBER — listen to a channel in a separate thread
Thread.new do
  subscriber = Redis.new
  subscriber.subscribe("notification:user:99") do |on|
    on.message do |channel, message|
      data = JSON.parse(message)
      puts "#{channel}: #{data['type']} from #{data['from']}"
      # Send to WebSocket, push notification, etc.
    end

    on.subscribe do |channel, subscription_count|
      puts "Subscribed to #{channel} (total: #{subscription_count})"
    end
  end
end

# PSUBSCRIBE — subscribe with a wildcard pattern
Thread.new do
  subscriber = Redis.new
  subscriber.psubscribe("notification:*") do |on|
    on.pmessage do |pattern, channel, message|
      puts "Pattern #{pattern} matched #{channel}: #{message}"
    end
  end
end

sleep 1
publisher.publish("notification:user:99", "new message")
publisher.publish("notification:user:100", "message for another user")

# config/initializers/redis.rb
REDIS_CACHE = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/1"))

# Fetch pattern — read from cache, if absent compute and store
def fetch_with_cache(key, ttl: 3600)
  cached = REDIS_CACHE.get(key)
  return JSON.parse(cached, symbolize_names: true) if cached

  result = yield   # compute the value

  REDIS_CACHE.setex(key, ttl, JSON.generate(result))
  result
end

# Usage
stats = fetch_with_cache("stats:sales:today", ttl: 300) do
  Order.today.group(:status).count
end

# Cache with invalidation
def save_product(product)
  product.save!
  REDIS_CACHE.del("product:#{product.id}")      # invalidate this product's cache
  REDIS_CACHE.del("product:list:page:1")        # invalidate the first page
end

# Rails.cache with a Redis backend
# config/environments/production.rb
config.cache_store = :redis_cache_store, {
  url:             ENV["REDIS_URL"],
  expires_in:      1.hour,
  namespace:       "cache:#{Rails.env}",
  pool_size:       10,
  pool_timeout:    5,
  error_handler: ->(method:, returning:, exception:) {
    Rails.logger.error "Redis cache error: #{exception.message}"
  }
}

# Use Rails.cache
Rails.cache.fetch("product:#{id}", expires_in: 1.hour) do
  Product.find(id)
end

Rails.cache.write("setting:maintenance", false, expires_in: 5.minutes)
Rails.cache.read("setting:maintenance")
Rails.cache.delete("setting:maintenance")
Rails.cache.delete_matched("product:*")  # delete all product caches

# Script: add to a sorted set and cap the number of members
add_and_trim_script = <<~LUA
  local key = KEYS[1]
  local score = ARGV[1]
  local member = ARGV[2]
  local max_size = tonumber(ARGV[3])

  redis.call("zadd", key, score, member)
  local size = redis.call("zcard", key)
  if size > max_size then
    redis.call("zremrangebyrank", key, 0, size - max_size - 1)
  end
  return redis.call("zcard", key)
LUA

# Run the script
redis.eval(
  add_and_trim_script,
  keys:  ["top_scores"],
  argv:  [1500, "rina", 100]   # max 100 members
)

# Cache the script with SHA for efficiency (EVALSHA)
sha = redis.script(:load, add_and_trim_script)
redis.evalsha(sha, keys: ["top_scores"], argv: [1600, "budi", 100])

# Gemfile
# gem 'sidekiq', '~> 7.2'

# config/initializers/sidekiq.rb
Sidekiq.configure_server do |config|
  config.redis = {
    url:      ENV.fetch("REDIS_URL", "redis://localhost:6379/0"),
    pool_size: ENV.fetch("SIDEKIQ_CONCURRENCY", 10).to_i + 5
  }
end

Sidekiq.configure_client do |config|
  config.redis = { url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0") }
end

# app/workers/send_email_worker.rb
class SendEmailWorker
  include Sidekiq::Worker

  sidekiq_options(
    queue:    :email,
    retry:    5,         # retry up to 5 times
    backtrace: true
  )

  def perform(user_id, template, data = {})
    user = User.find(user_id)
    NotificationMailer.send(template, user, data).deliver_now
  end
end

# Enqueue jobs
SendEmailWorker.perform_async(user.id, :welcome)
SendEmailWorker.perform_in(1.hour, user.id, :reminder)
SendEmailWorker.perform_at(Time.now + 2.days, user.id, :followup)

# Add elements
redis.pfadd("unique_visitor:20240815", "ip1", "ip2", "ip3", "ip1")   # ip1 duplicate ignored

# Count the estimate (accuracy ~0.81%)
puts redis.pfcount("unique_visitor:20240815")   # => ~3

# Merge several HyperLogLogs
redis.pfmerge("unique_visitor:this_week",
  "unique_visitor:20240812",
  "unique_visitor:20240813",
  "unique_visitor:20240814",
  "unique_visitor:20240815"
)
puts redis.pfcount("unique_visitor:this_week")   # estimated unique visitors this week

# Add an event to the stream
id = redis.xadd(
  "order_events",
  "*",   # auto-generate ID based on the timestamp
  "event",     "order_created",
  "order_id",  "12345",
  "total",     "150000",
  "user_id",   "99"
)
puts "Event ID: #{id}"   # => "1723724400000-0"

# Read the newest events
events = redis.xrange("order_events", "-", "+", count: 10)
events.each do |id, fields|
  puts "#{id}: #{fields}"
end

# Consumer group — distribute events across several consumers
redis.xgroup("CREATE", "order_events", "worker_group", "$", mkstream: true)

# Consumers read from the group
messages = redis.xreadgroup(
  "GROUP", "worker_group", "worker-1",
  COUNT: 10, BLOCK: 5000,
  "order_events" => ">"   # ">" = messages not yet assigned
)

# Acknowledge after processing
redis.xack("order_events", "worker_group", id)

# Redis info
info = redis.info
puts "Redis version:  #{info['redis_version']}"
puts "Memory used:    #{info['used_memory_human']}"
puts "Connected clients: #{info['connected_clients']}"
puts "Commands/sec:   #{info['instantaneous_ops_per_sec']}"
puts "Hit rate:       #{info['keyspace_hits'].to_f / (info['keyspace_hits'].to_f + info['keyspace_misses'].to_f) * 100}%"

# Monitor the most frequently accessed keys
redis.debug(:sleep, 0)   # make sure debug is available

# Slowlog — slow queries
slowlog = redis.slowlog(:get, 10)
slowlog.each do |entry|
  puts "#{entry[0]}: #{entry[2]}μs — #{entry[3].join(' ')}"
end

# DBSIZE — number of keys
puts redis.dbsize

# Scan keys with a pattern (non-blocking unlike KEYS)
cursor = "0"
loop do
  cursor, keys = redis.scan(cursor, match: "product:*", count: 100)
  keys.each { |k| puts k }
  break if cursor == "0"
end