Sse

// src/index.ts
const encoder = new TextEncoder();

export default {
  fetch(request: Request): Response {
    const url = new URL(request.url);
    if (url.pathname !== "/events") return new Response("ok");

    let timer: ReturnType<typeof setInterval>;
    const stream = new ReadableStream<Uint8Array>({
      start(controller) {
        // An SSE frame is `data: <payload>\n\n`. A line starting with `:` is a
        // comment — used here as a heartbeat to keep the stream from going idle.
        controller.enqueue(encoder.encode("data: hello\n\n"));
        timer = setInterval(
          () => controller.enqueue(encoder.encode(": ping\n\n")),
          25_000,
        );
      },
      // cancel() fires when the client disconnects.
      cancel() {
        clearInterval(timer);
      },
    });

    return new Response(stream, {
      headers: {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache, no-transform",
        Connection: "keep-alive",
      },
    });
  },
};

// src/index.ts
import { Hono } from "hono";
import { cors } from "hono/cors";

const app = new Hono();
app.use("*", cors({ origin: process.env.WEB_ORIGIN ?? "*" })); // EventSource is cross-origin from a SPA

app.get("/events", (c) => {
  const stream = new ReadableStream<Uint8Array>({
    start(controller) {
      controller.enqueue(new TextEncoder().encode("data: connected\n\n"));
      // ...register `controller` in a broadcast set; see fan-out below.
    },
  });
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache, no-transform",
    },
  });
});

export default app;

import { attachDatabasePool } from "@neon/functions";
import { Pool, Client } from "pg";

const encoder = new TextEncoder();
const clients = new Set<ReadableStreamDefaultController<Uint8Array>>();
const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 5 });
attachDatabasePool(pool);
const CHANNEL = "events";

// One dedicated DIRECT connection per isolate to receive events (LISTEN needs a
// real session — use DATABASE_URL_UNPOOLED, not the pooled URL).
// Don't call attachDatabasePool here: it would silence the idle drop that killed the feed.
// The error listener keeps the process alive; reconnect the client on error in production (omitted here).
const listener = new Client({
  connectionString: process.env.DATABASE_URL_UNPOOLED,
});
listener.on("error", (err) => {
  console.error(err);
});
listener.connect().then(() => listener.query(`LISTEN ${CHANNEL}`));
listener.on("notification", (msg) => {
  if (!msg.payload) return;
  const frame = encoder.encode(`data: ${msg.payload}\n\n`);
  for (const controller of clients) {
    try {
      controller.enqueue(frame); // enqueue is synchronous — no concurrent-await hazard
    } catch {
      clients.delete(controller); // controller already closed
    }
  }
});

// Anywhere you mutate state, NOTIFY so every isolate pushes to its own streams.
function publish(payload: unknown) {
  return pool.query("SELECT pg_notify($1, $2)", [
    CHANNEL,
    JSON.stringify(payload),
  ]);
}

data: a one-line payload\n\n
event: count\ndata: 42\n\n          # named event → addEventListener("count", …)
id: 7\ndata: resumable\n\n            # sets EventSource.lastEventId for resume
: this is a comment / heartbeat\n\n   # ignored by the client; keeps the stream warm
retry: 5000\n\n                       # tells the client how long to wait before reconnecting

const source = new EventSource(`${FUNCTION_URL}/events`); // GET only
source.onmessage = (e) => console.log("update", e.data);
source.onerror = () => {
  /* EventSource auto-reconnects; nothing to do */
};
// source.close() to stop.