Channels (Pub/Sub)

PulsaChannel provides ephemeral pub/sub messaging over WebSocket, independent of any database table. All subscribers to a channel name receive messages in real time.

Create a Channel

const channel = pb.channel('room:123:typing');

Channel names are arbitrary strings. Use namespaced patterns (e.g., entity:id:event) to avoid collisions.

---

.on(event, callback)

Register a handler for a specific event type. Supports wildcard '*' to catch all events.

channel.on('typing', (payload) => {
  console.log(`${payload.user} is typing`);
});

channel.on('presence', (payload) => {
  console.log(`${payload.user} is ${payload.status}`);
});

// Catch all events on this channel
channel.on('*', (payload) => {
  console.log('Any event:', payload);
});

Returns this — chainable.

---

.subscribe()

Connect to the channel and start receiving events. Must be called after registering handlers with .on().

const channel = pb.channel('room:123');

channel.on('message', (data) => console.log(data));

await channel.subscribe(); // Opens WebSocket if not already open

---

.send(event, payload)

Broadcast an event to all subscribers of this channel.

await channel.send('typing', {
  user: pb.auth.getCachedUser()!.email,
  roomId: '123',
});

Parameter	Type	Description
event	string	Event name (arbitrary string)
payload	any	Data to broadcast

---

.unsubscribeChannel()

Stop listening and disconnect from the channel. Clears all registered handlers.

channel.unsubscribeChannel();

---

Full Example: Typing Indicator

// In your component
const channel = pb.channel(`room:${roomId}:typing`);

channel.on('typing', ({ userId, isTyping }) => {
  setTypingUsers((prev) =>
    isTyping ? [...prev, userId] : prev.filter((id) => id !== userId)
  );
});

await channel.subscribe();

// On input change
const handleInput = async () => {
  await channel.send('typing', {
    userId: pb.auth.getCachedUser()!.sub,
    isTyping: true,
  });
};

// On component unmount
channel.unsubscribeChannel();

Full Example: Presence

const presence = pb.channel(`room:${roomId}:presence`);

presence.on('join', ({ user }) => addOnlineUser(user));
presence.on('leave', ({ user }) => removeOnlineUser(user));

await presence.subscribe();

// Announce join
await presence.send('join', { user: { id: pb.auth.getCachedUser()!.sub, name: 'Jane' } });

// On unmount
window.addEventListener('beforeunload', async () => {
  await presence.send('leave', { user: { id: pb.auth.getCachedUser()!.sub } });
  presence.unsubscribeChannel();
});

Channels vs. Database Subscriptions

Feature	pb.channel()	pb.from(Model).on().listen()
Source	Custom pub/sub (NATS)	PostgreSQL WAL changes
Persistence	Ephemeral	Durable
Use case	Typing, presence, UI events	Data sync — row CRUD
RLS	Not enforced	Enforced via PostgreSQL
Payload structure	Arbitrary	ChangeMessage format