Channel

A channel is a way to broadcast a message from the server to every browser listening on it. Broadcasting on a channel is called notifying the channel. For example, when someone posts a new comment, the server notifies the comments channel with the new comment, and every browser listening on comments receives it as it arrives.

How It Works

Elements channels are built on Postgres LISTEN/NOTIFY. By default, you send a notification by calling the notify method on a channel. That sends a notification through the underlying Postgres channel, which every app server is listening on. Each server then forwards the notification to its connected browsers.

Postgres itself can also notify a channel. A trigger that calls pg_notify lets the database notify a channel when a row changes, so events that originate outside the app reach listeners too.

At a Glance

A route handler creates a listener and passes it to the rendered page.

// app/shared/services/alerts.ts
import { Channel } from "@elements/app";

export interface Alert {
  userId: string;
  level: "info" | "warning";
  text: string;
}

export const alerts = new Channel<Alert>("alerts");

// app/pages/home/index.ts
import { alerts } from "#app/shared/services/alerts";

export default function route(req, res) {
  return new html({
    alerts: alerts.listen(),
  });
}

The route handler returns a page with a listener attached. The listener serializes over the wire as part of the response. When the page attaches in the browser, the listener connects to the server over a WebSocket and starts receiving notifications.

The template subscribes to the listener and accumulates each notification into a reactive array.

import { alerts } from "#app/shared/services/alerts";
import type { Listener } from "@elements/app";

<html (alerts: Listener<Alert>, private messages: Alert[] = [])
      oninit={() => alerts.on("notify", (a) => messages.push(a))}>
  <li e:for={msg of messages}>[{msg.level}] {msg.text}</li>
</html>

Every listener on the channel receives the notification.

listen() returns a listener handle. Listener is a type-only export: import it with import type { Listener } from "@elements/app" to name the handle in a template attribute, as above. You get instances from channel.listen(); you never construct one yourself.

Creating a Channel

const chat = new Channel<ChatMessage>("chat");

The Channel constructor takes one argument: the channel name. Channels are server-only.

Listen

const listener = channel.listen();                                        // broadcast: every listener receives
const listener = channel.listen({ filter: (msg) => msg.roomId === id });  // per-room: only matching messages

listen() returns a listener handle. Returning it from an @rpc or route handler sends it to the browser as data, where it connects over the WebSocket and begins receiving notifications.

filter runs on the server before the message goes over the wire, so a particular client only sees the messages that match the predicate.

Omit filter to deliver every notification to every connected listener: the broadcast pattern, used for shared cursors, drawing strokes, presence updates, and anywhere else every listener needs every message.

Filter

channel.filter(fn) returns a derived channel with a server-side predicate applied. The original channel is unaffected. This is the idiomatic way to scope a channel per request before passing it into a page. Pass the filtered channel straight into an html attribute and Elements materializes the per-request listener for you.

import { alerts } from "#app/shared/services/alerts";
import { session } from "@elements/app";

export default function route(req, res) {
  return new html({
    alerts: alerts.filter((a) => a.userId === session.getOrThrow("userId")),
  });
}

filter composes: chaining another filter ANDs the predicates. Filters passed to listen({ filter }) and filters on the channel both apply.

Listen Before Select

When you want to merge a one-shot query with live notifications, call listen() first and the query second. listen() returns only after the postgres LISTEN has committed, so anything NOTIFY'd after that point is queued on the listener. The query runs against the same database; any row that arrived between LISTEN commit and the query reaching the server is in the query result. The browser-side store (e.g. LiveTable's row index, keyed by id) dedupes when both arrive. An insert echo for a row already present is a no-op.

import { Channel, session, sql } from "@elements/app";

interface Alert {
  id: string;
  userId: string;
  text: string;
}

const alerts = new Channel<Alert>("alerts");

export default function route(req, res) {
  const listener = alerts.listen();
  const initial  = sql<Alert>(`select * from alerts where userId = ${session.getOrThrow("userId")} order by createdAt`).all();

  return new html({ listener, alerts: initial });
}

The order matters. If you query first and then call listen(), a NOTIFY fired in the gap is lost: postgres has no listeners on the channel yet, and it is not in your SELECT either, because the inserting transaction committed before SELECT started. The row never reaches the user. Always call .listen() first.

Notify

channel.notify({ userId, level: "warning", text: "rate limit hit" });

channel.notify participates in the async/await transform. The compiler rewrites the call to await channel.notifyAsync(...) and propagates async up the call stack at build time. Write it sync-style. The same idiom applies to sql, tx, @rpc, and LiveTable.

The call notifies all listeners on all app servers via Postgres NOTIFY.

notify works the same in any server context: @rpc bodies, route handlers, Job.run(), even module-top-level scripts. The example below pushes computed stats from a background job to every connected dashboard:

import { Channel, Job, sql } from "@elements/app";

interface DashboardStats {
  activeUsers: number;
  revenue: number;
}

const dashboardChannel = new Channel<DashboardStats>("dashboard");

/** @job */
export class RefreshStatsJob extends Job {
  run() {
    let stats = sql<DashboardStats>(
      `select count(*) filter (where active) as activeUsers, sum(amount) as revenue from orders`,
    ).first()!;

    dashboardChannel.notify(stats);
  }
}

The Listener

The browser-side listener handle exposes a small event API and a stop method.

listener.on("notify", (data) => { ... });
listener.on("error", (err) => { ... });
listener.off("notify", callback);
listener.stop();

on('notify', cb) fires for each notification as it arrives.

on('error', cb) fires when the listener fails to connect or its subscription errors. Use it to surface a failure to the user or to retry.

stop() ends the subscription. Listeners also clean up automatically when their WebSocket closes (page navigation, tab close), so explicit stop() is only needed when ending a subscription mid-page.

A listener handle is also a reactive value, so reading it inside a template binding tracks the dependency and the rendered DOM updates on each notification.

<p>{listener}</p>

The right pattern depends on what each notification represents.

When a notification is one item in a stream: a chat message, a new comment: push it to a reactive array.

import { messages as msgChannel } from "#app/shared/services/messages";
import type { Listener } from "@elements/app";

<html (listener: Listener<Msg>, private messages: Msg[] = [])
      oninit={() => listener.on("notify", (m) => messages.push(m))}>
  <li e:for={m of messages}>{m.text}</li>
</html>

When a notification replaces the previous value: updated stats, the current user list: reassign it.

import { dashboardChannel } from "#app/shared/services/dashboard";
import type { Listener } from "@elements/app";

<html (listener: Listener<DashboardStats>,
       stats: DashboardStats,
       private current: DashboardStats = stats)
      oninit={() => listener.on("notify", (s) => current = s)}>
  <p>{current.activeUsers} active users</p>
  <p>{current.revenue} revenue</p>
</html>

Live Aggregates

Some data isn't a stream of rows or a single replaceable value: it's an aggregate computed from many rows (count, sum, group by, top-N). LiveTable streams row changes; aggregates don't fit.

Two patterns, depending on whether the aggregate is the same for every subscriber.

Same value for every subscriber. A background job computes the aggregate, broadcasts the result. Every connected listener gets the same payload: the RefreshStatsJob example above. The job's sql(...) runs once, every browser receives the new value, the template reassigns.

Different value per subscriber. Each user sees their own aggregate (their unread count, their team's leaderboard, their account balance). The server broadcasts a "something changed" signal; each browser re-fetches its own value via @rpc.

// app/pages/inbox/services.ts
import { Channel, session, sql } from "@elements/app";

export interface InboxStats {
  unread: number;
  flagged: number;
}

export const inboxChannel = new Channel<{ event: "refresh" }>("inbox");

/** @rpc */
export function loadInboxStats(): InboxStats {
  session.isLoggedInOrThrow();

  return sql<InboxStats>(
    `select count(*) filter (where read = false) as unread,
            count(*) filter (where flagged = true) as flagged
     from messages
     where userId = ${session.getOrThrow("userId")}`,
  ).first()!;
}

// app/pages/inbox/index.ts
import { session } from "@elements/app";
import inbox from "./template";
import { inboxChannel, loadInboxStats } from "./services";

export default function route(req, res) {
  session.isLoggedInOrThrow();

  return new inbox({
    listener: inboxChannel.listen(),
    stats: loadInboxStats(),
  });
}

<!-- app/pages/inbox/template.html -->
import { inboxChannel, loadInboxStats } from "./services";
import type { Listener } from "@elements/app";
import type { InboxStats } from "./services";

<html (listener: Listener<{ event: "refresh" }>,
       stats: InboxStats,
       private current: InboxStats = stats)
      oninit={() => listener.on("notify", () => current = loadInboxStats())}>
  <p>{current.unread} unread</p>
  <p>{current.flagged} flagged</p>
</html>

When a new message arrives, the writer (@rpc or Job.run()) calls inboxChannel.notify({ event: "refresh" }). Every connected inbox page re-runs loadInboxStats() for its own user.

Database Triggers

Postgres can also notify a channel directly. A trigger that calls pg_notify lets the database notify a channel when a row changes. This is useful when events originate outside the app, such as a psql command, a scheduled job, or another service writing to the database.

Elements hashes channel names through the channel_name(text) SQL function so Postgres' 63-char identifier limit doesn't truncate them. Triggers that fire on Elements channels must use the same hash.

A trigger written directly in psql uses real Postgres function and identifier names (snake_case). The camelCase-to-snake_case translation Elements applies to app SQL and migrations does not run here:

create or replace function notify_alerts() returns trigger as $$
begin
  perform pg_notify(
    channel_name('alerts'),
    json_build_object(
      'op', TG_OP,
      'data', row_to_json(case when TG_OP = 'DELETE' then old else new end)
    )::text
  );
  return new;
end;
$$ language plpgsql;

create trigger alerts_notify after insert or update or delete
  on alerts for each row execute function notify_alerts();

channel_name(text) ships with the Elements baseline schema and produces the same hash the runtime uses, so triggers and the app route to the same channel.

Channel Name Uniqueness

Two Channel instances with the same name in different files share the same Postgres channel. To make two declarations independent, give them different names.

livetable: higher-level CRUD with optimistic UI on top of Channel.
database: the Postgres LISTEN/NOTIFY wire Elements uses.