Subscriptions & Live Queries

SUBSCRIBE TO <namespace>.<table_name>
[WHERE <condition>]
[OPTIONS (last_rows=<n>, batch_size=<n>, from=<n>)];

-- Subscribe to all messages in a conversation
SUBSCRIBE TO chat.messages
WHERE conversation_id = 1
OPTIONS (last_rows=50);
 
-- Subscribe to typing events with batching
SUBSCRIBE TO chat.typing_events
WHERE conversation_id = 1
OPTIONS (last_rows=10, batch_size=5);
 
-- Resume from a specific point
SUBSCRIBE TO chat.messages
WHERE conversation_id = 1
OPTIONS (from=1234);

KILL LIVE QUERY '<subscription_id>';

import { createClient, Auth } from '@kalamdb/client';
 
const client = createClient({
  url: 'http://localhost:8080',
  authProvider: async () => Auth.basic('admin', 'AdminPass123!')
});
 
// Subscribe with SQL WHERE clause
const unsub = await client.subscribeWithSql(
  'SELECT * FROM chat.messages WHERE conversation_id = 1',
  (event) => {
    if (event.type === 'change') {
      console.log('New message:', event.rows);
    }
  },
  { batch_size: 50 }
);
 
// Simple table subscription
const unsubTyping = await client.subscribe(
  'chat.typing_events',
  (event) => {
    console.log('Typing:', event.change_type, event.rows);
  }
);
 
// Check active subscription count
console.log(`Active: ${client.getSubscriptionCount()}`);
 
// Cleanup
await unsub();
await unsubTyping();
await client.disconnect();

kalam> \subscribe SELECT * FROM chat.messages WHERE conversation_id = 1 OPTIONS (last_rows=20);