Subscriptions

Pulsabase delivers real-time database change events to your frontend via WebSocket. Every INSERT, UPDATE, and DELETE on a published table is automatically pushed to connected subscribers.

Enable Publication

For a table to emit real-time events, it must have publication: true in its model’s static schema. This is registered with the backend on db:sync.

Basic Subscription

TypeScript

import pb from './pulsabase';
import Message from './models/Message';

// Listen for all INSERT eventsn on the messages table
const unsubscribe = pb.on(Message)
  .insert()
  .listen((change) => {
    console.log('Action:', change.action);   // "INSERT"
    console.log('New row:', change.data);    // The new record
  });

// Remove listener and close WebSocket if no other listeners remain
unsubscribe();

TypeScript (lower-level)

// Use pb.listen() for full control over filtering
const unsubscribe = pb.listen((change) => {
  console.log(change.action, change.data);
}, {
  filterTable: 'messages',             // Filter by table name
  filterAction: ['INSERT', 'UPDATE'],  // Filter action types
});

unsubscribe();

Dart

final unsub = pb.listen(
  (msg) {
    print('Action: ${msg.action}');   // "INSERT" | "UPDATE" | "DELETE"
    print('Data: ${msg.data}');
  },
  filterTable: 'messages',
  filterAction: 'INSERT',
);

// Unsubscribe
unsub();

Change Message Structure

Each event your callback receives has the following shape:

interface ChangeMessage {
  table: string;                       // e.g. "messages"
  action: 'INSERT' | 'UPDATE' | 'DELETE';
  data: Record<string, any>;           // The new/current row data
  old_data?: Record<string, any>;      // Previous data (UPDATE/DELETE only)
}

Filtering Events

Filter by Table and Action

pb.listen((change) => console.log(change), {
  filterTable: ['orders', 'payments'],     // Array of table names
  filterAction: 'INSERT',                  // Single action or array
});

Custom Filter Function

For more complex filtering logic:

pb.listen((change) => console.log(change), {
  filter: (change) => change.data.room_id === currentRoomId,
});

Note

When filter is provided, filterTable and filterAction are ignored. The filter function receives every message — do any heavy filtering on the client.

How It Works

Connection Management

The SDK manages the WebSocket lifecycle automatically:

• Auto-connect — the connection is established on the first listen() call
• Auto-reconnect — reconnects with exponential backoff on disconnection
• Auto-disconnect — the connection is closed when all listeners are removed

// All listeners are removed automatically when you call destroy()
pb.destroy();

// Or sign out clears listeners and closes the connection
await pb.auth.signOut();

Event Deduplication

If you have multiple mutations firing rapidly (e.g. many updates to the same record), the SDK can deduplicate change events using a built-in buffer:

const pb = new PulsaClient({
  // ... other options
  dedupe: {
    enabled: true,
    windowMs: 300,      // Collapse events within 300ms window
    keyFn: (msg) => msg.data.id,   // Group by record ID
  }
});

Deduplication can also be set per-listener:

pb.listen(callback, {
  filterTable: 'messages',
  dedupe: { enabled: true, windowMs: 500 },
});