Using SSE on Fly.io with Elysia

app.ts

import { Elysia, t } from "elysia";
import { nanoid } from "nanoid";
import { Emitter } from "strict-event-emitter";

// Initialize event emitter
const emitter = new Emitter();

// Create Elysia app
const app = new Elysia().get(
  "/:id/stream",
  async ({ params, request }) => {
    // Manually implemented SSE stream instead of using Elysia's stream plugin (@elysiajs/stream).
    // Fly.io terminates connections if no data is sent within ~10 seconds,
    // so we include a ping event every 3 seconds to keep the connection alive.
    let listener: (event: any) => void;
    const removeListener = () => {
      if (listener) {
        emitter.off("dataUpdate", listener);
      }
    };

    // Clean up on client abort
    request.signal.addEventListener("abort", removeListener);

    // Create SSE stream
    const stream = new ReadableStream({
      async start(controller) {
        // Set up event listener for data updates
        listener = ({ id, data }: { id: string; data: string }) => {
          if (id === params.id) {
            controller.enqueue(
              new TextEncoder().encode(
                [`id: ${nanoid()}`, "event: data-update", `data: ${data}`]
                  .join("\n")
                  .concat("\n\n")
              )
            );
          }
        };

        emitter.on("dataUpdate", listener);

        // Ping every 3 seconds to keep connection alive
        const pingInterval = setInterval(() => {
          if (request.signal.aborted) {
            clearInterval(pingInterval);
            controller.close();
            return;
          }

          controller.enqueue(
            new TextEncoder().encode(
              [
                `id: ${nanoid()}`,
                "event: ping",
                `data: ${new Date().toISOString()}`,
              ]
                .join("\n")
                .concat("\n\n")
            )
          );
        }, 3_000);

        // Clean up on stream abort
        request.signal.addEventListener("abort", () => {
          clearInterval(pingInterval);
          removeListener();
          controller.close();
        });
      },
      cancel() {
        removeListener();
      },
    });

    // Return SSE response
    return new Response(stream, {
      headers: {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        Connection: "keep-alive",
      },
    });
  },
  {
    params: t.Object({
      id: t.String(),
    }),
    detail: {
      summary: "Stream data updates",
      description:
        "Streams data updates for a specific ID using Server-Sent Events.",
    },
  }
);

// Example: Trigger an event (for demonstration purposes)
setInterval(() => {
  emitter.emit("dataUpdate", {
    id: "example-id",
    data: `Update at ${new Date().toISOString()}`,
  });
}, 10_000);

// Start the server
app.listen(3000, () => {
  console.log("Elysia server running on http://localhost:3000");
});