elements
Manual Recipes Unread Counts

Unread Counts

elements man recipes/unread-counts Read as markdown

A per-user, per-channel unread badge that stays live without any browser asking for it. Each user's counts are rows in a channelMembers table, opened as a LiveTable partitioned on userId. A Postgres trigger bumps the count for the other members when a message is inserted, and a second trigger sends each changed row to its own user's partition. A new message costs one update per member and reaches each member's browser as one row, and nobody re-runs a count(*).

The recipe assumes a users table and a session that carries userId, as in elements man recipes/authentication.

Migration

elements create migration 'add unread counts' -tables=channels,messages,channelMembers

app/migrations/<timestamp>-add-unread-counts.migration.sql:

-- add unread counts

-- Auto-update updatedAt on row changes.
create or replace function touchUpdatedAt()
returns trigger
language plpgsql
as $$
begin
  new.updatedAt = now();
  return new;
end;
$$;

create table channels (
  id uuid primary key default uuidGenerateV7(),
  createdAt timestamptz not null default now(),
  updatedAt timestamptz not null default now(),
  name text not null
);

create trigger channelsTouchUpdatedAt
  before update on channels
  for each row execute function touchUpdatedAt();

create table messages (
  id uuid primary key default uuidGenerateV7(),
  createdAt timestamptz not null default now(),
  updatedAt timestamptz not null default now(),
  channelId uuid not null references channels(id) on delete cascade,
  userId uuid not null references users(id) on delete cascade,
  body text not null
);

create index messagesChannelIdx on messages (channelId, createdAt desc, id desc);

create trigger messagesTouchUpdatedAt
  before update on messages
  for each row execute function touchUpdatedAt();

create table channelMembers (
  id uuid primary key default uuidGenerateV7(),
  createdAt timestamptz not null default now(),
  updatedAt timestamptz not null default now(),
  userId uuid not null references users(id) on delete cascade,
  channelId uuid not null references channels(id) on delete cascade,
  lastReadAt timestamptz not null default now(),
  unread integer not null default 0,
  unique (userId, channelId)
);

create trigger channelMembersTouchUpdatedAt
  before update on channelMembers
  for each row execute function touchUpdatedAt();

-- A new message is unread for every other member of its channel.
create or replace function messagesCountUnread() returns trigger
language plpgsql as $$
begin
  update channelMembers
     set unread = unread + 1
   where channelId = new.channelId
     and userId <> new.userId;

  return new;
end;
$$;

create trigger messagesCountUnreadTrigger
  after insert on messages
  for each row execute function messagesCountUnread();

-- Send each changed membership to its own user's partition.
create or replace function channelMembersNotify() returns trigger
language plpgsql as $$
declare
  r record;
  payload text;
begin
  r := coalesce(new, old);

  payload := json_build_object(
    'op', lower(tg_op),
    'data', json_build_object(
      'id', r.id,
      'userId', r.userId,
      'channelId', r.channelId,
      'lastReadAt', json_build_object('$type', 'Date', '$value', (extract(epoch from r.lastReadAt) * 1000)::bigint),
      'unread', r.unread
    )
  )::text;

  -- NOTIFY takes under 8000 bytes. A larger row goes as its id, and the
  -- server reads the row back.
  if octet_length(payload) >= 8000 then
    payload := json_build_object('op', lower(tg_op), 'id', r.id)::text;
  end if;

  perform pg_notify(channel_name('channelMembers:userId=' || r.userId), payload);

  return r;
end;
$$;

create trigger channelMembersNotifyTrigger
  after insert or update or delete on channelMembers
  for each row execute function channelMembersNotify();

channelMembers is one row per user per channel, and unread is the badge. messagesCountUnread runs in the same transaction as the insert, so the message and its counts commit together. The sender's own row is skipped by userId <> new.userId.

The counting trigger writes with plain SQL, so nothing broadcasts it. channelMembersNotify does, in the shape elements man recipes/live-from-sql describes: the channel is channelMembers:userId=<id>, the partition key a view opened with view({ userId }) listens on, and data carries every field of the ChannelMember interface below, with the Date in its tagged form. Each member hears only their own row.

LiveTables

app/shared/services/channels.ts:

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

export interface Channel {
  id: string;
  name: string;
}

export interface ChannelMember {
  id: string;
  userId: string;
  channelId: string;
  lastReadAt: Date;
  unread: number;
}

export interface Message {
  id: string;
  createdAt: Date;
  channelId: string;
  userId: string;
  body: string;
}

export function isMemberOrThrow(channelId: string) {
  session.isLoggedInOrThrow();

  let found = !sql(
    `select 1 from channelMembers
     where userId = ${session.getOrThrow("userId")} and channelId = ${channelId}`,
  ).empty();

  if (!found) {
    throw new ForbiddenError("not a member of this channel");
  }
}

export let channelMembers: LiveTable<ChannelMember> = new LiveTable<ChannelMember>({
  channel: (partition) =>
    partition ? `channelMembers:${partition}` : "channelMembers",

  insert: () => {
    throw new ForbiddenError();
  },

  // The one change a member makes to their own row: mark it read.
  update: (item) => {
    return sql<ChannelMember>(`
      update channelMembers
         set unread = 0, lastReadAt = now()
       where id = ${item.id} and userId = ${session.getOrThrow("userId")}
      returning *
    `).firstOrThrow();
  },

  delete: () => {
    throw new ForbiddenError();
  },
});

export let messages: LiveTable<Message> = new LiveTable<Message>({
  insert: (item) => {
    isMemberOrThrow(item.channelId!);
    return messages.insert({ ...item, userId: session.getOrThrow("userId") });
  },
});

channel pins the name the notify trigger hashes. Without it the compiler derives one from the declaration's location, which a migration cannot write.

A member never inserts or deletes their own membership row from the browser, so those handlers refuse. The update handler is the mark-read: whatever the browser sends, it sets unread = 0 and lastReadAt = now() on the caller's own row, so a user cannot set another count or touch another user's row. isMemberOrThrow is the membership guard the route and the message insert share; a partition is not a permission (elements man livetable/partitions).

Channel page

elements create page channel

app/pages/channel/index.ts:

import { Request, Response, session, sql } from "@elements/app";
import channel from "./template";
import { Channel, channelMembers, messages, isMemberOrThrow } from "#app/shared/services/channels";

export default function route(req: Request, res: Response) {
  let channelId = req.params.id;
  isMemberOrThrow(channelId);

  let me = session.getOrThrow("userId");

  let channels = sql<Channel>(`
    select c.id, c.name
    from channels c
    join channelMembers m on m.channelId = c.id
    where m.userId = ${me}
    order by c.name
  `).all();

  return new channel({
    channelId,
    channels,
    memberships: channelMembers.view({ userId: me }),
    messages: messages.view({ channelId }, { orderBy: "createdAt desc", limit: 50 }),
  });
}

The route checks membership, then opens two views. memberships is the user's own partition: a handful of rows, one per channel, each carrying its count. messages is the channel's feed, windowed to the newest fifty as in elements man recipes/chat-rooms. The channel names come from a plain query; names do not change often enough to stream.

app/pages/channel/template.ehtml:

import "./style.css";
import { LiveView } from "@elements/app";
import { Channel, ChannelMember, Message } from "#app/shared/services/channels";

interface SidebarRow {
  id: string;
  name: string;
  unread: number;
}

function sidebar(channels: Channel[], memberships: LiveView<ChannelMember>): SidebarRow[] {
  let byChannel = new Map<string, ChannelMember>();

  for (let m of memberships) {
    byChannel.set(m.channelId, m);
  }

  return channels.map((c) => ({
    id: c.id,
    name: c.name,
    unread: byChannel.get(c.id)?.unread ?? 0,
  }));
}

function unreadIn(memberships: LiveView<ChannelMember>, channelId: string): number {
  return memberships.find((m) => m.channelId === channelId)?.unread ?? 0;
}

function markRead(memberships: LiveView<ChannelMember>, channelId: string) {
  let m = memberships.find((m) => m.channelId === channelId);

  if (m && m.unread > 0) {
    memberships.update({ ...m, unread: 0, lastReadAt: new Date() });
  }
}

function onSend(feed: LiveView<Message>, draft: { value: string }) {
  feed.insert({ body: draft.value, createdAt: new Date() }, () => draft.value = "");
}

<html class="page-channel" (
  channelId: string,
  channels: Channel[],
  memberships: LiveView<ChannelMember>,
  messages: LiveView<Message>,
  private draft: { value: string } = { value: "" },
)>
  <main class="page-shell">
    <nav>
      <ul>
        <li e:for={row of sidebar(channels, memberships)} e:key={(row: SidebarRow) => row.id}>
          <a href={`/channels/${row.id}`}>{row.name}</a>
          <span e:if={row.unread > 0 && row.id !== channelId} class="badge">{row.unread}</span>
        </li>
      </ul>
    </nav>

    <span e:if={unreadIn(memberships, channelId) > 0}
          oninsert={() => markRead(memberships, channelId)}
          hidden></span>

    <button e:if={messages.hasMore} onclick={() => messages.more()}>load older</button>

    <ul class="feed">
      <li e:for={m of messages.toReversed()}>{m.body}</li>
    </ul>

    <form onsubmit={() => onSend(messages, draft)}>
      <input type="text" value={draft.value} required>
      <button type="submit">send</button>
    </form>
  </main>
</html>

sidebar() groups the memberships into a Map once and looks each channel up, so the badge list is one pass over each list, never a find per channel. Each row is a wrapper holding primitives, keyed by channel id, so a count change repaints one badge.

The hidden <span> is the mark-read. It exists only while the open channel has unread messages, and oninsert runs each time it appears: when the page loads with a count, and again when a message arrives while the user is reading. markRead updates through the view, so the badge clears at once, the server zeroes the row, and the notify trigger carries the zero to the user's other tabs.

Test

app/pages/channel/test.ts:

import { test, equal, sql, session } from "@elements/app";
import { channelMembers, messages } from "#app/shared/services/channels";

test("unread counts", () => {
  let ada = sql<{ id: string }>(`insert into users (handle) values ('ada') returning id`).firstOrThrow();
  let bo = sql<{ id: string }>(`insert into users (handle) values ('bo') returning id`).firstOrThrow();
  let general = sql<{ id: string }>(`insert into channels (name) values ('general') returning id`).firstOrThrow();

  sql(`insert into channelMembers (userId, channelId)
       values (${ada.id}, ${general.id}), (${bo.id}, ${general.id})`);

  function unread(userId: string): number {
    return sql<{ unread: number }>(
      `select unread from channelMembers where userId = ${userId} and channelId = ${general.id}`,
    ).firstOrThrow().unread;
  }

  test("a message counts for every other member", () => {
    session.login({ userId: ada.id, userName: "ada" });

    messages.view({ channelId: general.id }).insert({ body: "hi" });

    equal(unread(bo.id), 1);
    equal(unread(ada.id), 0);
  });

  test("marking read resets the count", () => {
    sql(`insert into messages (channelId, userId, body) values (${general.id}, ${ada.id}, 'hi')`);
    session.login({ userId: bo.id, userName: "bo" });

    let mine = channelMembers.view({ userId: bo.id });
    let row = mine.find((m) => m.channelId === general.id)!;
    equal(row.unread, 1);

    mine.update({ ...row, unread: 0 });

    equal(unread(bo.id), 0);
  });
});

Both writes run inside the test's transaction, so the triggers fire and the rows roll back with the test. The second test writes the message with raw SQL, the path a job or an import would take, and still sees the count.

Routes

import channel from "#app/pages/channel";

// ...
app.route("/channels/:id", channel);

Notes

  • Why not a refresh signal. The alternative broadcasts "something changed" and has every open browser call an @rpc that counts its unread rows. With a thousand users online that is a thousand count(*) queries per message, most of them for users who are not in the channel. Here a message writes one row per member, and each browser receives only its own changed row.
  • Joining a channel. Insert the channelMembers row from the @rpc or job that adds the member. The notify trigger sends it to that user's partition, so their memberships view gains the row. The route's channel list is a query, so a new channel's name appears on the next navigation.
  • lastReadAt for a divider. The row keeps the time of the last mark-read. Read it before marking read to place a "new messages" line in the feed.

Related

  • recipes/live-from-sql: the notify trigger and its payload.
  • livetable/partitions: one slice of a table per user.
  • livetable/windows: the feed's newest-fifty window.
  • channel: aggregates that are the same for every subscriber.