elements
Manual Channel

Channel

elements man channel Read as markdown

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 { Request, Response } from "@elements/app";
import { alerts } from "#app/shared/services/alerts";

export default function route(req: Request, res: Response) {
  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 { Request, Response, session } from "@elements/app";

export default function route(req: Request, res: Response) {
  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 { Request, Response, 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: Request, res: Response) {
  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. The message has to serialize to under 8000 bytes, the Postgres limit. A larger one throws an error naming the channel and the size, and nothing is sent. To move something bigger, notify with its id and read it on receipt.

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");

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 created while logged in belongs to that session. When the session ends (session.logout(), expiry, or session.revoke()), the server stops the listener and the browser drops it, so notifications after that point do not arrive. Listeners created by an anonymous request are public and keep running. The WebSocket stays open either way. To receive again after logging out, open a new listener. See elements man session.

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>

Connect and Disconnect

On the server, the listener listen() returns has two more events. They follow the listener's lifetime, not the socket's.

let listener = presence.listen({ filter: (e) => e.roomId === roomId })
  .on("connect", (l) => join(l.id, roomId, userId))
  .on("disconnect", (l) => leave(l.id, roomId));

on('connect', cb) fires once, the first time a browser attaches. A reattach after a dropped socket or a page restored from the back/forward cache does not fire it again.

on('disconnect', cb) fires once, when the listener stops, and only if connect fired. A page that closes or navigates away stops its listeners right away; a browser that vanishes without closing is stopped about a minute after its socket dies. A listener that never attached (a crawler, a prefetch) fires neither.

Each callback receives the listener, whose id is a key unique to that page view. Take everything else, the user included, from the route's own variables. Do not read session inside them: they run after the route has returned, from a socket or a timer, not from the request the route checked. They are not awaited, and a failure goes to on('error'). Both are server-only; calling them in a template is a type error. elements man recipes/presence builds a "who's here" list on them.

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, and an aggregate is not a row until you store one.

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 standing, their balance. Store it as a row per subscriber, keep the row current with a Postgres trigger on the table it counts, and open a LiveTable over those rows partitioned by user. A change then reaches only the users whose number moved, as one row each, and no browser re-runs the aggregate. elements man recipes/unread-counts builds this end to end.

Do not broadcast a "something changed" signal and have every browser call an @rpc to recompute its own value. Each notification then costs one query per connected browser, including every browser whose value did not change.

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.

A payload of 8000 bytes or more fails the pg_notify, and with it the write that fired the trigger. For a LiveTable channel, send { op, id } instead when octet_length(payload) >= 8000; see recipes/live-from-sql.

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.

Related

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