DEV Community: Chris Bell

Building a real-time commenting app with Socket.io and React

Chris Bell — Fri, 17 Feb 2023 15:38:16 +0000

In this guide, we'll walk through how to build a real-time commenting system using React, Node.js, and Socket.io. We'll show you how to build a list of comments fed from an API, create new comments, and broadcast comments in real-time to connected users using Socket.io.

We'll also briefly introduce Knock, a developer platform for building a cross-channel notification system. Engineers use Knock for building powerful cross-channel notifications that can send to in-app and out-of-app channels like email, SMS, push, and Slack. All through a single API call.

Prerequisites

We recommend the following prerequisites before beginning:

Familiarity with JavaScript and the React library
Node installed on your machine

Getting started

In this section, we'll set up the basics of our application including the structure we'll use to house our Node.js server and our React front-end application.

First, create the project folder and two subfolders for the client and server of the application.

mkdir real-time-commenting-app
cd real-time-commenting-app
mkdir client server

Building our Node.js API server

Next, we're going to be creating the basis of our Node.js API server. The API server's job is to send information to the client for the comments we have available, and process any new comments created in the client. The server will also set up our Socket.io connection and expose a websocket that can be connected to from our React client.

Inside our server directory, we're going to install the dependencies we need to build our API server.

Express.js is a minimalist web framework for Node.js that provides a simple interface for building HTTP listeners as well as other functionality that makes it easy to build web applications in Node.js.
CORS is an NPM package that lets us process cross-origin requests. This is important as we'll be running a separate client and server application (e.g. they will be running on different ports).
Nodemon is a package that will automatically restart our Node.js server after detecting any changes, making local development easy.
Socket.io is a Javascript library that allows us to create real-time, bidirectional communication between web browsers (clients) and a Node.js server.

Let's go ahead and install our dependencies inside our server directory:

npm install express cors nodemon socket.io

The next step is to create the entry point of our web server as index.js:

touch index.js

Now inside the server/index.js file, we're going to setup our simple Node.js server using Express.js:

const express = require("express");
const app = express();

const http = require("http").Server(app);
const cors = require("cors");

const PORT = 4000;

app.use(cors());
app.use(express.json());

app.get("/users", (req, res) => {
  // TODO: implement
  res.json([]);
});

app.get("/comments", (req, res) => {
  // TODO: implement
  res.json([]);
});

app.post("/comments", async (req, res) => {
  // TODO: implement
  res.json(null);
});

http.listen(PORT, () => {
  console.log(`Server listening on ${PORT}`);
});

Here we're defining a few endpoints that we'll be filling in during the course of this guide:

GET /users: returns a list of available users
GET /comments: returns a list of posted comments
POST /comments: creates a new comment

We're also adding CORS support to our express server with app.use(cors()); and the ability to parse incoming JSON with app.use(express.json());.

Now let's add a start script so we can start our API server. Inside your server/package.json file, add the following:

"scripts": {
  "test": "echo \"Error: no test specified\" && exit 1",
  "start": "nodemon index.js"
}

You can now start the server on port 4000 by running:

npm start

We can verify our server is working correctly by going to http://localhost:4000/users where you should receive an empty list of users.

Implementing our users and commenting endpoints

In this next section, we're going to add the implementation for our HTTP endpoints.

💡 Note: in this example application we're going to be using static and in-memory data, which won't work for any production use cases. You'll need to use a persistent database, like MySQL or Postgres, if you want to prepare this for production.

First of all, let's add some information to power our GET /users endpoint which will return a static list of users that we can post comments as:

touch lib/users.js

And the contents of the lib/users.js file will be:

const users = [
  {
    id: "1",
    name: "Dennis Nedry",
    email: "dnedry@ingen.net",
    avatar:
      "https://cdn.vox-cdn.com/thumbor/4oha2z_P13Nc_Z_JUt1qpt133yw=/0x0:1440x653/920x613/filters:focal(314x95:544x325):format(webp)/cdn.vox-cdn.com/uploads/chorus_image/image/60017631/dennis_nedry_1440.0.jpg",
  },
  {
    id: "2",
    name: "John Hammond",
    email: "jhammond@ingen.net",
    avatar:
      "https://openpsychometrics.org/tests/characters/test-resources/pics/JP/4.jpg",
  },
  {
    id: "3",
    name: "Ellie Sattler",
    email: "esattler@ingen.net",
    avatar:
      "https://encrypted-tbn0.gstatic.com/images?q=tbn:ANd9GcRQKxKHuGTXeNrhoozgtgrxNOmN7l5AcExYBwthS3yyfGGWbE5ewg0IGvp3tNGdCDZamMg&usqp=CAU",
  },
];

function findUserById(userId) {
  return users.find((user) => user.id === userId);
}

module.exports = { users, findUserById };

We can now go back to our index.js server root and start returning this static list of users:

// Under the last require:
const { users } = require("./lib/users");

// Replace the get `/users` endpoint with
app.get("/users", (req, res) => {
  res.json(users);
});

This ensures that the static list of users is returned as JSON when the /users endpoint is requested.

Let's next implement our comments functions:

touch lib/comments.js

The contents of the lib/comments.js file will be:

const { findUserById, users } = require("./users");

let comments = [];
let lastId = 1;

function getCommentWithAuthor(comment) {
  return { ...comment, author: findUserById(comment.authorId) };
}

function createComment(params) {
  const comment = {
    id: lastId++,
    body: params.body,
    authorId: params.authorId,
    insertedAt: new Date().toISOString(),
  };

  comments.push(comment);

  const commentWithAuthor = getCommentWithAuthor(comment);
  return commentWithAuthor;
}

function listComments() {
  return comments.map((comment) => getCommentWithAuthor(comment));
}

module.exports = { createComment, listComments };

Here we're adding two new functions that we'll be calling from our endpoint: createComment and listComments. Our comments are implemented as a mutable array that we append to when we create a new comment, using an auto-incrementing id to generate a unique id for each comment created. The createComment function expects to be passed an object that contains a body and a authorId, which we'll implement within our React application shortly.

We're also "loading" the associated author with each comment when they are both created and listed, matching the authorId on the comment with a corresponding user that we have in our static user list.

Next, let's implement the endpoints in our index.js server entry point.

// Under the last require:
const { listComments, createComment } = require("./lib/comments");

app.get("/comments", (req, res) => {
  const comments = listComments();
  res.json(comments);
});
app.post("/comments", (req, res) => {
  const comment = createComment(req.body);
  res.json(comment);
});

Here we're creating a comment by passing the req.body, which will contain any JSON params that the client is sending up in the create comment request. These params expect a JSON encoded object to be sent that will look like { body: "My comment", authorId: "1" }.

It's worth checking at this stage that everything is working as expected. You can check out the endpoints we just added in an API testing tool like Postman. You should be able to create a new comment, and then retrieve the full list of comments with what we've built so far.

Note: since our comments are only stored in memory, when you refresh your Node server (or make changes) any comments will be reset.

Creating our comments app in React

In this section, we're going to build out the client portion of our commenting app. The goal is that we'll end up with a single-page React application that looks like this:

First of all, we're going to be working in our client directory for this section so let's move to that directory:

cd client

We're going to bootstrap our React application using the excellent create-react-app, which is a library from the React team that makes it easy to start building a single-page application in React.

npx create-react-app ./

Next we're going to install any additional dependencies we need. In our case, we're just going to need date-fns which is a library that makes it trivial to work with datetimes in JavaScript.

npm i date-fns

Now that we've bootstrapped our React application and installed our dependencies, we can start our client running in the browser. With create-react-app you'll get a live reloading application environment, so that any changes you're making should be automatically displayed in the browser.

npm run start

Your application should open on port 3000 by default in your browser.

Laying out our application through components

Next up we're going to start creating our components. For this application, we're breaking things down into Header and Comments components, where the Comments component renders individual Comment components.

Let's first make our empty component files. We'll be filling them in with code shortly.

touch src/components/Header.js
touch src/components/Comment.js
touch src/components/Comments.js

Creating the Header component

Inside our src/components/Header.js file:

const Header = ({ users, selectedUserId, setSelectedUserId }) => {
  return (
    <header className="Header" key={selectedUserId}>
      <div className="Header-user">
        <select
          value={selectedUserId}
          onChange={(e) => {
            setSelectedUserId(e.target.value);
          }}
        >
          <option>Select a user</option>
          {users.map((user) => (
            <option key={user.id} value={user.id}>
              {user.name}
            </option>
          ))}
        </select>
      </div>
    </header>
  );
};

export default Header;

Creating the Comment component

In our src/components/Comment.js component:

import parseISO from "date-fns/parseISO";
import formatDistance from "date-fns/formatDistance";

function formatDate(dateStr) {
  const date = parseISO(dateStr);
  return formatDistance(date, new Date(), { addSuffix: true });
}

const Comment = ({ comment, isYou }) => (
  <div className="Comment">
    <div className="Comment-header">
      <div className="Comment-avatar">
        <img src={comment.author.avatar} alt={comment.author.name} />
      </div>
      <span className="Comment-author">
        {isYou ? "You" : comment.author.name}
      </span>
      <span className="Comment-time">{formatDate(comment.insertedAt)}</span>
    </div>

    <div className="Comment-body">{comment.body}</div>
  </div>
);

export default Comment;

Here we're using the date-fns library to format the date and time in which the comment was posted as a relative date time (e.g. 3-days ago). Our formatDate function takes care of this behavior for us using the formatDistance function in date-fns to generate relative dates.

You'll also notice here that we're passing in an isYou prop which, when true, shows You as the author name rather than the full name.

Creating the Comments component

Next up, let's add the implementation for the src/components/Comments.js file. This component represents our comments sidebar and has the responsibility of showing a list of comments and displaying a form to post new comments.

The brains of our commenting application live in this component and it also houses the state about exactly which comments we have.

import { useEffect, useState } from "react";
import Comment from "./Comment";

const Comments = ({ selectedUserId }) => {
  const [comments, setComments] = useState([]);
  const [input, setInput] = useState("");

  async function sendComment() {
    const result = await fetch("http://localhost:4000/comments", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ body: input, authorId: selectedUserId }),
    });
    const comment = await result.json();
    setComments((comments) => [...comments, comment]);
  }

  async function fetchComments() {
    const result = await fetch("http://localhost:4000/comments");
    const comments = await result.json();
    setComments(comments);
  }

  useEffect(() => {
    fetchComments();
  }, []);

  return (
    <div className="Comments">
      <h3 className="Comments-title">
        {comments.length === 1 ? `1 comment` : `${comments.length} comments`}
      </h3>

      <div className="Comments-list">
        {comments.map((comment) => (
          <Comment
            key={comment.id}
            comment={comment}
            isYou={selectedUserId === comment.authorId}
          />
        ))}
      </div>
      <div className="Comments-box">
        <form
          className=""
          onSubmit={(e) => {
            e.preventDefault();
            sendComment(input);
            setInput("");
          }}
        >
          <textarea
            value={input}
            onChange={(e) => setInput(e.target.value)}
            name="body"
            className="Comments-box__input"
          />
          <button
            type="submit"
            disabled={selectedUserId === ""}
            className="Comments-box__btn"
          >
            Send
          </button>
        </form>
      </div>
    </div>
  );
};

export default Comments;

There's quite a bit happening in here, so let's walk through it:

We setup a piece of state (comments) to keep track of the currently fetched comments.
We fetch all of the existing comments from the API by calling the fetchComments function on mount.
We have a form where we can post new comments, which calls the sendComment function.
Once a comment has been successfully sent we store it in the comments state.

Updating our App component

Inside src/App.js we're going to set up the main component structure:

import { useEffect, useState } from "react";
import "./App.css";
import Comments from "./components/Comments";
import Header from "./components/Header";

function App() {
  const [users, setUsers] = useState([]);
  const [selectedUserId, setSelectedUserId] = useState("");

  async function fetchUsers() {
    const result = await fetch("http://localhost:4000/users");
    const users = await result.json();

    setUsers(users);
  }

  useEffect(() => {
    fetchUsers();
  }, []);

  return (
    <div className="App">
      <Header
        users={users}
        selectedUserId={selectedUserId}
        setSelectedUserId={setSelectedUserId}
      />
      <Comments selectedUserId={selectedUserId} />
    </div>
  );
}

export default App;

Again, there are a few things going on here so let's break it down:

We keep track of fetched users in state (as users).
When the component mounts we fetch all users from the API by calling the fetchUsers function.
We also keep track of the selectedUserId which represents which user we've selected.
We pass the selectedUserId and setSelectedUserId components down to the child components.

Styling our application

For our styling, you can replace the contents of the src/App.css file with:

.App {
  display: flex;
  flex-direction: column;
  height: 100vh;
  width: 100vw;
  background-color: #15151c;
}

.Header {
  display: flex;
  padding: 8px 16px;
  height: 32px;
  background-color: #20222b;
  border-bottom: 1px solid #0f1014;
}

.Header-notifications {
  margin-left: auto;
  position: relative;
}

.Header-user {
  display: flex;
  align-items: center;
}

.Comments {
  width: 390px;
  margin-left: auto;
  height: 100%;
  display: flex;
  flex-direction: column;
  background-color: #20222b;
}

.Comments-title {
  color: #fff;
  margin: 16px;
  font-size: 16px;
  font-weight: normal;
}

.Comments-list {
  overflow-y: auto;
  border-top: 1px solid #0f1014;
}

.Comments-box {
  margin-top: auto;
  border-top: 1px solid #0f1014;
  padding: 16px;
}

.Comments-box__input {
  background-color: rgb(15 16 20);
  border: 1px solid transparent;
  font-family: inherit;
  font-size: 16px;
  color: #fff;
  display: block;
  box-sizing: border-box;
  width: 100%;
  height: 72px;
  padding: 8px;
  border-radius: 4px;
  margin-bottom: 8px;
}

.Comments-box__input:focus,
.Comments-box__input:active {
  border-color: rgb(91 83 255 / 75%);
  appearance: none;
}

.Comments-box__btn {
  background-color: rgb(91 83 255 / 75%);
  color: #fff;
  border: 1px solid transparent;
  padding: 4px 8px;
  display: inline-block;
  appearance: none;
  font-family: inherit;
  font-size: 14px;
  border-radius: 4px;
  margin-left: auto;
}

.Comments-box__btn:disabled {
  background-color: #ccc;
}

.Comment {
  padding: 16px;
  border-bottom: 1px solid #0f1014;
  color: #fff;
}

.Comment-header {
  margin-bottom: 4px;
  display: flex;
  align-items: center;
}

.Comment-avatar {
  width: 24px;
  height: 24px;
  background-color: #15151c;
  border-radius: 100%;
  margin-right: 8px;
  overflow: hidden;
}

.Comment-avatar img {
  width: 100%;
  height: 100%;
}

.Comment-author {
  font-weight: 600;
  font-size: 14px;
  margin-right: 8px;
}

.Comment-time {
  font-size: 14px;
  color: rgb(125 130 154);
}

.Comment-body {
  margin-left: 32px;
}

You should now have a functioning, styled React application that allows you to select a user in the top-left corner and post a comment as that user.

What you might notice at this stage is that our application does not post any sent comments in real-time to any other users who are looking at the application. We're going to fix this next.

Adding Socket.io to our application for real-time communication between React and Node.js

Our goal now is to add real-time support to our client and our server so that whenever a comment is posted, everyone viewing the comments page will receive the comment. This is where Socket.io is useful. We can add websocket support to our Node.js server, and have users connect to receive comments created in real-time.

Let's start by adding Socket.io in our webserver. Inside our server/index.js file we'll add the necessary socket.io setup code:

// Add this to where your other requires are
const io = require("socket.io")(http, {
  cors: {
    origin: "http://localhost:3000",
  },
});

// Above our `app.get("/users")` handler
io.on("connection", (socket) => {
  console.log(`⚡: ${socket.id} user just connected!`);

  socket.on("disconnect", () => {
    console.log("🔥: A user disconnected");
  });
});

This sets up our socket.io websocket server and allows a cross-origin websocket request to be made from our client application (which is running on port 3000). We'll also automatically log when a user connects and disconnects from the websocket, which happens when the websocket connection is terminated from the client.

The next step in our server is to broadcast any new comments that are created over a channel in socket.io so that we can receive them on the client. To do so, we'll change our create comment endpoint handler to emit the created comment:

// Replace our create comment handler
app.post("/comments", async (req, res) => {
  const comment = createComment(req.body);
  io.emit("new-comment", { comment });
  res.json(comment);
});

Here we're saying to socket.io to emit (broadcast) to all connected clients on the new-comment channel and send a payload that includes the created comment. When we listen for this event in our client application, we'll be able to receive the created comment and add it to the list of comments that we have available in state.

And now let's implement the changes required on the client.

Inside the client directory let's install the socket.io-client library which allows us to connect from the client to the web socket connection on the server.

npm install socket.io-client

And in client/src/App.js we're going to set up the websocket connection:

// Add with other imports
import io from "socket.io-client";

// Before the component definition:
const socket = io.connect("http://localhost:4000");

function App() {

Next, we're going to pass this newly created socket to our Comments component in our src/App.js component so we have it accessible in the Comments component:

<Comments socket={socket} selectedUserId={selectedUserId} />

And now in src/components/Comments.js we'll add our real-time support for creating and receiving a comment.

Let's extract the passed-in socket prop in our component definition:

const Comments = ({ socket, selectedUserId }) => {
  // Same component code

And now we want to connect to our new-comment channel when the component mounts so we'll add a new useEffect hook that will handle the socket setup code for us.

useEffect(() => {
  socket.on("new-comment", ({ comment }) => {
    // Only insert comments when the current user is not the author
    if (comment.author.id !== selectedUserId) {
      setComments((comments) => [...comments, comment]);
    }
  });

  return () => {
    socket.off("new-comment");
  };
}, [setComments, socket, selectedUserId]);

Here we're saying that every time a new-comment event is received we want to append that comment to the list of comments that we already have in our comments state. We use the function setter to do this so that any existing comments in state are preserved. Note that we're also returning a cleanup function to stop listening to the channel once the component unmounts (the return from the useEffect does this).

One other important line to note here is that we're not appending the comment to our list of comments when the selected user is the author. This avoids the duplicate comment problem where we add the comment to the list of comments in state twice; once when it's been created in the sendComment handler and once when it's broadcast in the new-comment channel.

And with that we should now have a fully functioning real-time commenting app 🎉. You can try it out by opening up two browser windows, selecting different users, and posting a few comments. You should see the comments coming in real-time to each connected user.

Using Knock to send cross-channel notifications

Commenting is a common use case for where you'll likely want to add some notifications to your product. In a collaboration product with commenting, you can imagine needing to add both an in-app notification feed of comments that have been left as well as an email summarizing any missed comments while the user was away from the product and did not see their notifications. This is where Knock can help.

Knock is a developer tool for building cross-channel notification systems. With Knock, you can create complex cross-channel notifications for your product without the need to build out the system yourself. You call Knock's API and you can send notifications with a single API call to in-app and out-of-app channels like email, push, SMS, and Slack.

When it comes to integrating an in-app notification feed, Knock's pre-built React feed components comes ready out of the box with real-time behavior, live updating badge counts, and the ability to mark notifications as read or seen. You can use the in-app notification components to effortlessly get in-app notification inboxes, feeds, or toasts up and running in a matter of hours, all powered by Knock's scalable and reliable notification engine.

In addition to powering in-app notifications, Knock takes care of batching (collapsing multiple notifications about a single topic into one), intelligent orchestration (send to email only if the in-app notification has been seen), managing user preferences, and gives you unparalleled visibility into the notifications your product is sending.

If you'd like to try it out, you can sign up for a free account or book a demo. 👋

Controlling Elixir supervisors at runtime with feature flags

Chris Bell — Wed, 22 Jun 2022 21:12:40 +0000

👋 We’re Knock, the most powerful notification system you'll never build. Developers use Knock to schedule, batch, and deliver notifications, without the need to write custom application code.

At Knock, we use LaunchDarkly to power feature flags. Feature flags are powerful because they enable us to control, at runtime, the behavior of different parts of the system. Most of the time, this means controlling access to features on our client-side application, or controlling the rollout of new features across our application. However, we recently adopted a circuit breaker pattern built around feature flags. This pattern helps our services be more reliable when things fail.

In this post, we will look at using feature flags to start and stop parts of our Elixir application at runtime. This will give us more options to gracefully degrade and recover from runtime errors and too much load on our service.

The problem

Like many applications, our infrastructure relies on queues to decouple various components. In our system we use AWS Kinesis as a data stream, consumed by Broadway consumers for some critical parts of our infrastructure. We have found that sometimes our Broadway consumers for AWS Kinesis fail in ways that do not gracefully recover when they crash. For example, each Kinesis shard has its own supervision tree managed by the Kinesis Broadway consumer. We found that if a shard consumer experienced a crash-inducing error, the shard would not restart and the crash would not cascade up to the Broadway producer. While we have worked on contributing to this consumer library, we decided that it would be important to have runtime control over stopping and starting consumers to respond to such failures just in case.

Likewise, we value having many options for tuning our service at runtime so that we can always serve critical requests - even if that means temporarily degrading less critical parts of our application. Temporarily stopping queue consumption is one such strategy.

We knew that we needed a way, at runtime, to start and stop these queue consumers. Although we could have reached for other configuration management tools like Hashicorp Consul or AWS AppConfig, we already use LaunchDarkly at Knock to control the runtime behavior of our frontend and backend applications. LaunchDarkly feature flags seemed like a great way to control this starting and stopping process without adding new dependencies or complexity to our stack.

The solution

We built an Elixir supervisor called Circuit Breaker that takes inspiration from the typical circuit breaker pattern and applies it to supervision trees (but, to be clear, our circuit breaker is not the same thing as the typical circuit breaker pattern you may be used to).

The basic structure of our supervisor circuit breaker pattern.

Our circuit breaker responds to two signals:

If a child process is crashing, let it crash, don’t bring it back up, and notify our team using our observability tools.

Illustration showing a child of our circuit breaker in a crashed state.
If a feature flag is toggled on or off, start or stop a corresponding child process accordingly.

Illustration showing a switched off child of our circuit breaker.

Every second, the Circuit Breaker checks the current state of our feature flags and starts/stops the corresponding child processes.

If a process has crashed and its flag is on, we don’t restore the process.
If a process is on but its flag is off, we gracefully shut it down.
If a process is off (but not crashed) and its flag is on, then we start it back up again.

This approach gives us a lot of flexibility in dealing with failures and extra load in our system. By turning off consumers strategically, we can intentionally degrade our service in order to maintain uptime on more critical parts of our infrastructure. Once traffic spikes have passed, we can resume consumption and fully restore service. Having options like these is part of a “defense in depth” strategy for scaling. Instead of relying 100% on queues, autoscaling, or the Elixir supervision tree, we can also intervene in the system at runtime without fiddling with lower levels of infrastructure (e.g. Kubernetes or environment variables).

Typically in an Elixir application, if a supervisor is crashing repeatedly, such crashes can eventually cascade up and cause the entire application to restart. By trapping the crashes within the circuit breaker, we have a way to ensure that such crashes cannot cascade up to the root of the application - and we have a way to start up the crashing supervisor again on the terms we want to work with.

How it works

There is a top-level supervisor CircuitBreaker that we start in our application supervisor with a list of children to monitor. This list includes the subsequent child specs and the feature flag to watch for each child.

   # How we start the Circuit Breaker, including the list of child specs
   # and their corresponding feature flags to watch
   defmodule MyApp.Application do
     @moduledoc false

     use Application

     def start(_type, _args) do
       children = [
         # ...
         {
           CircuitBreaker,
           children: [
             %{
               id: SomeService.Supervisor,
               start: {SomeService.Supervisor, :start_link, [[]]},
               restart: :temporary,
               ld_key: "some-feature-flag-on-launch-darkly"
             },
             %{
               id: AnotherService.Supervisor,
               start: {AnotherService.Supervisor, :start_link, [[]]},
               restart: :temporary,
               ld_key: "a-different-feature-flag"
             }
           ]
         }
         # ...
       ]

       opts = [strategy: :one_for_one, name: MyApp.Supervisor]
       Supervisor.start_link(children, opts)
     end
   end

   defmodule CircuitBreaker do
     @moduledoc """
     Supervisor for starting/stopping/restarting supervisors using feature flags.
     If a child completly shuts down, it remains shut down until the entire
     application restarts, or the LaunchDarkly flag is cycled.
     """
     use Supervisor

     def start_link(init_arg) do
       Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
     end

     @impl true
     def init(init_arg) do
       monitored_children = Keyword.get(init_arg, :children, [])

       children = [
         {DynamicSupervisor, name: __MODULE__.DynamicSupervisor, strategy: :one_for_one},
         {__MODULE__.Monitor, children: monitored_children}
       ]

       Supervisor.init(children, strategy: :one_for_one)
     end

     def add_child(spec) do
       GenServer.call(__MODULE__.Monitor, {:add_child, spec})
     end

     defdelegate pid_for_spec(spec), to: __MODULE__.Monitor
     defdelegate spec_to_id(spec), to: __MODULE__.Monitor
   end

The CircuitBreaker supervisor has two children:
1. A Dynamic Supervisor
2. CircuitBreaker.Monitor, which is given the list of child specs.
The monitor loops through the list of child specs and checks the feature flags for each of them. Based on the state of the flags, it adds the child specs to the supervisor and monitors them, trapping exits.
If a process crashes, the Monitor tracks that signal and marks that child as crashed. We clear the crashed state when we see that the feature flag has been turned off.
If the Monitor sees a flag off but a child is running, it tells the dynamic supervisor to terminate the child process.
If the Monitor sees a flag as on, but the child is not running (and not crashed), then it starts the process under the dynamic supervisor.

A basic implementation (along with instructions for integration into your own projects) is available in this GitHub gist. Internally, our implementation includes support for Telemetry and additional logging & observability.

Monitoring Failures

We use the Elixir/Erlang Telemetry module extensively at Knock to monitor runtime performance. Each time the Monitor checks the flag status for each process, it reports how many child processes are running, how many are off, and how many are in a crashed state. As processes crash, it also emits log messages capturing the crash (without the crash cascading up to the parent supervisor). When we see increases in the crash metric or those log messages, we get a notification in Slack so that we can intervene as appropriate.

Future development

From here, there are a few improvements we can make to our implementation:

We are still making improvements to our Kinesis consumer library so that shard-level failures are handled correctly without manually toggling feature flags. Even so, being able to manually control consumers at runtime gives us more options when dealing with incidents.
Abstracting our dependency on LaunchDarkly so we can swap out different feature flag implementations.
Release what we’ve built as open source. Today’s blog post is a step towards releasing a fully built, plug-and-play library. With some time and attention, this tool could be generalized so anyone could drop it into their project for controlling the starting/stopping of supervision trees at runtime.

Building a batched notification engine

Chris Bell — Wed, 08 Jun 2022 17:19:56 +0000

👋 We’re Knock. We run flexible notifications infrastructure that developers use to schedule, batch, and deliver notifications, without the need to write custom application code.

In this blog post we'll take a deep dive into building a batched notification engine, including some of the technical details on how you might design and construct it yourself.

What are batched notifications?

A batched notification collapses together one or more relevant notifications into a single notification message delivered to a recipient across a window of time.

In a naive notification system each event maps to a single notification sent to a user. A common example: a document collaboration app where every comment results in an email notification (looking at you, overly noisy Notion default notification settings!). Contrast that to a batched notification system, where all comments left on a document within a window of time are batched together and sent as a single notification message.

When you batch notifications, you send your customers fewer, higher-information-density notifications, which leads to increased engagement on the notifications you do send and, ultimately, happier customers with better retention rates.

Note: you might also think of a batched notification as a kind of "notification digest". However, at Knock we think about digests as operating on a heterogenous set of notification types. In our document commenting example, a digest could contain notifications about comments, likes, and replies. A batch instead operates on a single notification type. We would build separate batches for comments, likes, and replies.

Designing a batched notification system

There are, broadly speaking, two different approaches we can take in the design of a batched notification system:

Batch on write: notifications are accumulated into batches per recipient when an event has occurred. Batches are "flushed" at the end of a batch window to become a notification.
Batch on read: notification batches are lazily generated by periodically running a task (usually via a cron job) that finds all notifications that have not been sent, collapsing them into batches, and sending notifications.

The biggest difference between these two approaches is how they scale: a batch on write system trades storage for building an optimized lookup table of what needs to be batched and when. A batch on read system must (fairly inefficiently) query ever increasing amounts of data to determine what to batch and when.

In our experience building a batch on write system is more effort but is generally worth it to future proof your notification system. Of course this approach also isn't perfect, and it has it's own scaling challenges to overcome. We touch on some of those later on in the post.

Table design

For this example we'll model our system using a good ol' fashion relational database. Our table design may therefore look like:

A notifications table to keep track of the individual notifications that a user should receive.
A notification_batches table to keep track of all of the batched notifications for a recipient.
A notification_batch_notifications table to keep track of the individual notifications per batch (our entries on the queue).

CREATE TABLE `notifications` (
  `id` serial PRIMARY KEY,
  `type` varchar(255) NOT NULL,
  `actor_id` INT NOT NULL,
  `recipient_id` INT NOT NULL,
  `object_id` INT NOT NULL,
  `object_type` varchar(255) NOT NULL,
  `inserted_at` TIMESTAMP NOT NULL
);

CREATE TABLE `notification_batches` (
  `id` serial PRIMARY KEY,
  `type` varchar(255) NOT NULL,
  `recipient_id` INT NOT NULL,
  `batch_key` varchar(255) NOT NULL,
  `object_id` INT NOT NULL,
  `object_type` varchar(255) NOT NULL,
  `closes_at` TIMESTAMP NOT NULL,
  `processed_at` TIMESTAMP,
  `inserted_at` TIMESTAMP NOT NULL
);

CREATE TABLE `notification_batch_notifications` (
  `notification_batch_id` INT NOT NULL,
  `notification_id` INT NOT NULL,
  `inserted_at` TIMESTAMP NOT NULL,
  PRIMARY KEY (notification_batch_id, notification_id),
  FOREIGN KEY (notification_batch_id) REFERENCES notification_batches (id),
  FOREIGN KEY (notification_id) REFERENCES notifications (id),
);

A few details on the design of our tables:

We use a polymorphic design with object_id and object_type to reference the object attached to a notification
We use a batch_key on our notification batches table, which we'll use as a lookup key to accumulate items into open batches. For example, if we want to batch all comments in the document for a single recipient our batch_key would be an identifier that includes the document_id
We keep a closes_at timestamp to store when the batch window should close
We store a processed_at timestamp to keep track of the batches that we've flushed

Batching notifications

In order to batch our notifications for our users, we'll want to (per recipient):

Generate a batch_key to use to accumulate notifications into a batch for a window of time
Create a notification entry to keep track of the action that occurred, as well tracking the object that the action occurred on
Find an "open" notification batch using the batch_key where the batch window has not closed. If there isn't an open batch, then create one using the batch key and set the closes_at window to now() + batch_window

Let's see what this might look like in practice in our codebase using our document commenting example (granular implementation details omitted):

// Create our comment for the document
const comment = await Comments.createComment(
  document,
  { text: commentText },
  user
);

// Find all of the recipients for the document (excluding the user who created the comment)
const recipients = await Documents.getCollaborators(document);
const recipientsToNotify = recipients.filter((recipient) => recipient.id !== user.id);

// The key we want to query an open batch for
const batchKey = `document:${document.id}:comments`;

// How long do we want this batch window to be open? (5 mins)
const batchWindow = 60 * 5;

recipientsToNotify.forEach((recipient) => {
  // For each recipient, generate a notification and add it to the batch
  const notification = await Notifications.createNotification(
    "new-comment",
    { object: comment, actor: user },
    recipient
  );

  // Find an open batch by the key given for this recipient
  // SELECT * FROM notification_batches AS nb
  // WHERE nb.recipient_id == ? AND nb.batch_key == ? AND nb.closes_at <= now();
  const batch = await Notifications.findOrCreateBatch(
    recipient,
    batchKey,
    { object: document, type: "new-comment", batchWindow }
  );

  // Add the notification to the batch
  const batchedNotification = await Notifications.addNotificationToBatch(batch, notification);
});

Flushing closed batches

We'll next need a way to "flush" our batches at the end of the batch window to produce a notification message per recipient. There are two separate approaches we can leverage to do this:

Enqueue a job to be executed at the end of the batch window once the batch has been created
Have a cron task that runs every minute to find any batch windows that are closed but not yet sent

If you're dealing with an ephemeral job queue (like something Redis backed) then the first option might be a non-starter for you given that you could end up dropping scheduled jobs in the event of a failure. Similarly, not all job queues support future scheduled jobs. Instead lets take a look at some code for executing a cron job to flush our batches:

// SELECT * FROM 'notification_batches' AS batch where batch.closes_at >= now() AND batch.processed_at is NULL;
const batches = await Notifications.getBatchesToBeProcessed();

batches.forEach((batch) => {
  // For each batch, generate a notification message
  const { template, subjectLine } = await Notifications.generateEmailFromBatch(batch);

  // Send our email
  await Notifications.sendEmail(batch.recipient, subjectLine, template);

  // Mark the batch as processed
  await Notifications.markBatchAsProcessed(batch);
});

Notice here that we're also keeping track of a processed_at field for our batches as well, so that we know if we need to reprocess any of batches in the event of an issue with the cron job.

Generating our batched notification message

Now that we have our batched notifications, we'll use them to generate actual notification messages. This is the code inside our Notifications.generateEmailFromBatch function in the example above.

Note: one important consideration you'll want to think through here is the total number of items fetched in the batch. In our example, the number of items in the batch can theoretically be unbounded, which may lead to poor performance when fetching and rendering a notification template.

In our document commenting example, we might have the following template (here, written in Liquid for simplicity) to show the available comments for the document:

<h1>Comments for {{ batch.object.name }}</h1>

{% for notification in batch.notifications %}
  <p>
    <strong>{{ notification.object.author.name }}</strong> said at {{ notification.inserted_at }}:
  </p>

  <blockquote>
    <p>{{ notification.object.text }}</p>
  </blockquote>
{% endfor %}

Preparing this design for production

The above design is a naive implementation of a batching system, and there are a few important details to consider when taking this design to production:

Protecting against race conditions whereby two comments can be created at the same time, leading to multiple batches being generated
Ensuring each batch executes only once, so that we don't send duplicate messages
Handling retries with the delivery of our email notifications

Extending our batched notification system

Building upon our design, we may want to handle more advanced scenarios:

Keeping a count of the total number of items that are stored in the batch. This is useful for when you want to display a subset of the items in the batch, but still have the ability to show the total number of items that was added within the window (e.g. "There were 15 new comments on X").
Adding the ability to flush a batch window early. When a batch hits a certain size, flush the batch window early to ensure users get notified sooner than later at specified thresholds of activity.
Removing one or more items from a batch. To return to our document collaboration example, if users are allowed to delete comments, we'll want to remove those comments from our batch before the batch window closes and a notification is sent to users.
Enabling user-specific batch windows. Your users may wish to customize the duration of their batch window, such that they can determine shorter or longer frequencies in which to receive notifications. This is especially helpful for digesting use cases, where some users will want a daily digest and others will want them once a week.
Partitioning cron jobs to flush batches to handle large numbers of users. Most applications won't need this level of optimization, but if your product does serve very large numbers of users this can become an interesting challenge as your notifications scale.

Don't want to build this yourself?

If you've read the above and thought that this sounds like a large lift for you or your engineering team, you're not alone. That's exactly why we built Knock.

Knock is a complete solution for powering product notifications that handles batching out of the box with no cron jobs or job queues to set up. We even support removing items from batches. You can use Knock to orchestrate notifications across multiple channels, manage user preferences, and keep your notification templates in a single place that's visible to your whole team.

If you want to try out Knock to power your batched notifications (and much more!), you can sign up for free here. We have a generous free tier you can use to get started.