Building Real-Time Applications with Remix.js Server-Sent Events and Job Queues

What you will learn

I hope that after reading this article you have enough background to create more complex event driven applications in Remix.js, taking advantage of SSE and Job Queues.

What does this article cover

We will cover several aspects, including:

• Remix.js Routing
• Quirrel Job Queues
• Remix.js Server-Sent Events

Prerequisites

Before starting the article, it is recommended that you have knowledge of React, Remix and that you have some notions of queues and server sent events.

Creating the Project

To initialize a project in Remix we execute the following command:

npx create-remix@latest nixy
cd nixy

We start the dev server with the following command:

npm run dev

We are going to use the Just the basics type, with a deployment target of Remix App Server and we are going to use TypeScript in the application.

Install the necessary dependencies:

npm install quirrel remix-utils superjson dayjs cuid
npm install -D concurrently

In the package.json file we change the dev script to the following:

{
  "dev": "concurrently 'remix dev' 'quirrel'",
}

This way, during the development of the application, Quirrel's server will run concurrently with Next's server. To access the Quirrell UI just run the following command:

npm run quirrel ui

Set up Event Emitter

In today's article we are going to have a simple example in which we are going to have only one process running which means that we are going to have a single instance available and for this use case the use of EventEmitter is ideal, but others should be considered options if you have many instances.

// @/app/common/emitter.ts
import { EventEmitter } from "events";

let emitter: EventEmitter;

declare global {
  var __emitter: EventEmitter | undefined;
}

if (process.env.NODE_ENV === "production") {
  emitter = new EventEmitter();
} else {
  if (!global.__emitter) {
    global.__emitter = new EventEmitter();
  }
  emitter = global.__emitter;
}

export { emitter };

With the EventEmitter instance created, we can move on to the next step.

Set up Job Queue

The queue we are going to create is quite easy to understand, first we define the data types of the Queue payload, which in this case we will need to add the identifier. Then, inside the callback, what we do is emit a new event taking into account the name of the queue and the data of the Job that we want to submit, which in this case need to be serialized.

// @/app/queues/add.server.ts
import { Queue } from "quirrel/remix";
import superjson from "superjson";

import { emitter } from "~/common/emitter";

export const addQueueEvtName = "addJobQueue";

export default Queue<{ identifier: string }>("queue/add", async (job) => {
  emitter.emit(
    addQueueEvtName,
    superjson.stringify({ identifier: job.identifier })
  );
});

With the queue created, we can move on to the next step.

Set up Routes

Now that we have everything that needs to be used ready, we can start defining our application's routes. The routes that we will have in the application are the following:

• _index.tsx - main route of the application, where all the real-time part will be visible.
• queue.add.ts - this route will expose the Queue that was created as an action, so that it can be exposed and consumed.
• sse.add.ts - this route will create an event stream that will push each of the events to the ui.

With the above routes created inside the routes/ folder, we can work on the route responsible for the event stream:

// @/app/routes/sse.add.ts
import type { LoaderFunction } from "@remix-run/node";
import { eventStream } from "remix-utils";

import { emitter } from "~/common/emitter";
import { addQueueEvtName } from "~/queues/add.server";

export const loader: LoaderFunction = ({ request }) => {
  // event stream setup
  return eventStream(request.signal, (send) => {
    // listener handler
    const listener = (data: string) => {
      // data should be serialized
      send({ data });
    };

    // event listener itself
    emitter.on(addQueueEvtName, listener);

    // cleanup
    return () => {
      emitter.off(addQueueEvtName, listener);
    };
  });
};

In the code snippet above, we use the event emitter instance to listen to each of the events that are emitted and using the eventStream function of the remix-utils dependency we can simplify the setup of live updates from the backend to the client.

Moving now to the Queue route registration, as mentioned earlier, it will be like this:

// @/app/routes/queue.add.ts
import addQueue from "~/queues/add.server";

export const action = addQueue;

In the code snippet above we imported the queue that was created in the past and was exposed using the action primitive of Remix.

Last but not least, we can now work on the client side of the application, where we can take advantage of everything created so far. The page will have a button that will invoke a server-side action to add a new Job to the Queue.

The Job is quite simple, just pass a unique identifier, so that visually we can identify that each event emitted is truly unique and in the queue we will add a delay of five seconds, to guarantee that the Job has truly entered the queue and that each of them is processed.

Then, on the UI side, we'll use the useEventSource hook to connect the component/page to the event stream that was created earlier and using a useEffect we'll add the messages of each emitted event to the component's local state. This in order to update the unordered list that we have in the JSX of the page. This way:

// @/app/routes/_index.tsx
import type { V2_MetaFunction } from "@remix-run/node";
import { Form } from "@remix-run/react";
import { useEffect, useState } from "react";
import { useEventSource } from "remix-utils";
import cuid from "cuid";
import superjson from "superjson";
import dayjs from "dayjs";

import addQueue from "~/queues/add.server";

export const meta: V2_MetaFunction = () => {
  return [{ title: "SSE and Quirrel" }];
};

export const action = async () => {
  const currentTime = dayjs();
  const newTime = currentTime.add(5, "second");

  await addQueue.enqueue({ identifier: cuid() }, { runAt: newTime.toDate() });

  return null;
};

export default function Index() {
  const [messages, setMessages] = useState<{ identifier: string }[]>([]);
  const lastMessage = useEventSource("/sse/add");

  useEffect(() => {
    setMessages((datums) => {
      if (lastMessage !== null) {
        return datums.concat(superjson.parse(lastMessage));
      }
      return datums;
    });
  }, [lastMessage]);

  return (
    <div>
      <h2>Server-sent events and Quirrel</h2>

      <ul>
        {messages.map((message, messageIdx) => (
          <li key={messageIdx}>{message.identifier}</li>
        ))}
      </ul>

      <Form method="POST">
        <button type="submit">Add New Job</button>
      </Form>
    </div>
  );
}

And with that I conclude the last step of this article. It is worth emphasizing that although the example is simple it ends up serving as a basis for the creation of more complex systems in which, for example, we can use queues to limit the number of insertions and updates that are made in the database to reduce the back pressure. Or create schedulers to have a set of tasks run like reminders, automatic messages, notifications, etc. From here the uses of this are various.

Conclusion

I hope you found this article helpful, whether you're using the information in an existing project or just giving it a try for fun.

Please let me know if you notice any mistakes in the article by leaving a comment. And, if you'd like to see the source code for this article, you can find it on the github repository linked below.

Github Repo