Async Redis key mutation notifications in Rails
Tags: rails, redis
Published: 2021-06-27 14:55:42

---
title: Async Redis key mutation notifications in Rails
published: true
description: Run code when your Redis data changes
tags: rails, redis
cover_image: https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ni0asktoojdq7hfxrv4r.jpg
---

*Update 10/7/2022*: I prepared [a simple Rails 7 app](https://github.com/leastbad/tracker) that demonstrates the functionality described in this article.

---

I am a huge fan of [Kredis](https://github.com/rails/kredis). It allows Rails developers to see Redis as far more than just a fragment cache and "where jobs are".

Working with Kredis made me want to be able to run [arbitrary operations](https://cableready.stimulusreflex.com/reference/operations) in my Rails app when specific keys are modified via specific Redis commands. [Redis has an excellent pub/sub infrastructure](https://redis.io/topics/pubsub), and all [Redis commands publish messages](https://redis.io/topics/notifications).

# Why would someone want this?

While it's true that changes to data that occur within a typical Rails app are already well covered by model callbacks, state machines and other standard tooling, an entire world of real-time stream processing, ETL and multi-application use cases open up when you can run `redis-cli set leastbad rules` on your terminal and pick it up in your app.

Problem #1: Listening for messages blocks execution.
Solution #1: Spin up a thread!

Problem #2: Every dyno/server is going to receive the same messages, causing mayhem as developers respond to those messages with database updates. Side-effect chaos!
Solution #2: A standalone process that can be registered as a `worker` in `Procfile`... *sort of like Sidekiq.*

At first, I was just planning on borrowing 95% of Mike Perham's battle-hardened code. Then I realized that the Venn diagram of "people who want a Redis changeset firehose" and "Sidekiq users" is close to 100%... **so I just bolted what I needed onto Sidekiq**.

# Try it out!

> What follows is the MVP of my new gem. In fact, it's not a gem, yet: it's an initializer! It has no tests and is hours old. My janky code would make poor Mike [bleed out](https://youtu.be/W7JyjZI3LUM?t=333). The goal is to see if folks actually need/want this to exist. **I'm looking for feedback on what the ideal Rails-side API would actually look like**.

Your Rails app needs to be up and running with Sidekiq. Just stick this in `config/initializers/sidekiq.rb`:
```rb
module Sidekiq
  class Subscriber
    include ::Sidekiq::Util

    def initialize
      @done = false
      @thread = nil
    end

    def start
      @thread ||= safe_thread("subscriber") {
        until @done
          Sidekiq.redis do |conn|
            # https://redis.io/topics/notifications#configuration
            conn.config(:set, "notify-keyspace-events", "E$lshz")
            # https://redis.io/topics/notifications#events-generated-by-different-commands
            conn.psubscribe("__key*__:*") do |on|
              on.psubscribe do
                @firehose = Firehose.new
              end
              on.pmessage do |pattern, command, key|
                @firehose.process(command.split(":").last.to_sym, key)
              end
              on.punsubscribe do
                @firehose = nil
              end
            end
          end
        end
        Sidekiq.logger.info("Subscriber exiting...")
      }
    end

    def terminate
      @done = true
      if @thread
        t = @thread
        Thread.kill(@thread)
        @thread = nil
        t.value
      end
    end
  end
end

module CoreExtensions
  module Sidekiq
    module Launcher
      attr_accessor :subscriber

      def initialize(options)
        @subscriber = ::Sidekiq::Subscriber.new
        super(options)
      end
  
      def run
        super
        subscriber.start
      end
  
      def quiet
        subscriber.terminate
        super
      end
  
      def stop
        subscriber.terminate
        super
      end
    end
  end
end

Sidekiq.configure_server do
  require "sidekiq/launcher"
  ::Sidekiq::Launcher.prepend(CoreExtensions::Sidekiq::Launcher)
end
```

I'm using [CableReady](https://cableready.stimulusreflex.com) to send console log notifications to the Console Inspector whenever a key is updated with the Redis `SET` command. I have a simple `AllUsers` ActionCable Channel in play for testing. This lives in `app/lib/firehose.rb`:
```rb
class Firehose
  include CableReady::Broadcaster

  attr_reader :redis

  def initialize
    @redis = ::ActionCable.server.pubsub.redis_connection_for_subscriptions
  end

  def process(command, key)
    case command # https://github.com/rails/kredis#examples
    when :set    # string, integer, json
      cable_ready["all_users"].console_log(message: "#{key} was just updated to #{redis.get(key)}").broadcast
    when :rpush  # list
    when :lrem   # unique_list
    when :sadd   # set
    when :incr
    when :decr
    when :incrby
    when :decrby
    when :exists
    when :del
      cable_ready["all_users"].console_log(message: "#{key} was deleted").broadcast
    else
    end
  end
end
```

As Seinfeld would say, **is this anything?**