-
Database Settings
+
Database Settings
{databaseType === 'MongoDB'
- ? 'Configure MongoDB replica set read preferences, write concern, and read concern levels for your deployment.'
+ ? 'Configure live updates, replica set read preferences, write concern, and read concern.'
: 'Database replication settings for your deployment.'}
@@ -53,6 +57,7 @@ export const Settings = ({ data, databaseType }: Props) => {
setEdit={setEdit}
reset={reset}
databaseType={databaseType}
+ realtimeStatus={realtimeStatus}
/>
diff --git a/src/components/database/settings/settingsForm.tsx b/src/components/database/settings/settingsForm.tsx
index 570567f9c..ff07ca7b2 100644
--- a/src/components/database/settings/settingsForm.tsx
+++ b/src/components/database/settings/settingsForm.tsx
@@ -14,7 +14,9 @@ import {
SelectValue,
} from '@/components/ui/select';
import { Alert, AlertDescription } from '@/components/ui/alert';
+import { Switch } from '@/components/ui/switch';
import { SettingsFormActions } from '@/components/settings/SettingsFormActions';
+import type { DatabaseRealtimeStatus } from '@/lib/models/database';
interface Props {
control: any;
@@ -23,6 +25,36 @@ interface Props {
setEdit: (arg0: boolean) => void;
reset: any;
databaseType: string;
+ realtimeStatus?: DatabaseRealtimeStatus | null;
+}
+
+function realtimeStatusCopy(status: DatabaseRealtimeStatus | null | undefined) {
+ if (!status) {
+ return 'Runtime status is unavailable for this backend.';
+ }
+ switch (status.status) {
+ case 'live':
+ return `Live · ${status.activeSchemaCount} opted-in ${
+ status.activeSchemaCount === 1 ? 'schema' : 'schemas'
+ }`;
+ case 'starting':
+ return 'Starting the change-stream listener.';
+ case 'idle':
+ return (
+ status.message ??
+ 'Waiting for a replica set and at least one opted-in schema.'
+ );
+ case 'disabled':
+ return 'Live updates are turned off for this module.';
+ case 'degraded':
+ return status.message ?? 'The change-stream listener is degraded.';
+ case 'unsupported':
+ return status.message ?? 'Live updates require MongoDB.';
+ default: {
+ const _exhaustive: never = status.status;
+ return _exhaustive;
+ }
+ }
}
export const SettingsForm = ({
@@ -32,13 +64,14 @@ export const SettingsForm = ({
setEdit,
reset,
databaseType,
+ realtimeStatus,
}: Props) => {
if (databaseType !== 'MongoDB') {
return (
- Replica set read preferences are only available for MongoDB
- deployments. Your current database type is{' '}
+ Live updates and replica set read preferences are only available for
+ MongoDB deployments. Your current database type is{' '}
{databaseType}.
@@ -47,7 +80,37 @@ export const SettingsForm = ({
return (
<>
-
+
+
(
+
+
+
+ Live updates
+
+
+ Watch opted-in schemas through MongoDB change streams. Schemas
+ must also enable live updates. Data Explorer never
+ auto-refreshes.
+
+
+ {realtimeStatusCopy(realtimeStatus)}
+
+
{
- return await (await getApiClient())
- .get<{ indexes: unknown[] }>(`/database/schemas/${schemaId}/indexes`)
+ const data = await (await getApiClient())
+ .get(`/database/schemas/${schemaId}/indexes`)
.then(res => res.data);
+ return { indexes: normalizeSchemaIndexResponse(data) };
};
export const deleteSchemaIndexes = async (
@@ -366,6 +369,13 @@ export const getDatabaseSettings = async () => {
return res.data;
};
+export const getDatabaseRealtimeStatus = async () => {
+ const res = await (
+ await getApiClient()
+ ).get('/database/realtime/status');
+ return res.data;
+};
+
export const patchDatabaseSettings = async (data: Partial) => {
await (
await getApiClient()
diff --git a/src/lib/api/realtime/ticket.ts b/src/lib/api/realtime/ticket.ts
new file mode 100644
index 000000000..9a83359be
--- /dev/null
+++ b/src/lib/api/realtime/ticket.ts
@@ -0,0 +1,28 @@
+'use server';
+
+import { getApiClient } from '@/lib/api';
+import { getCurrentEnvironment } from '@/lib/logic/EnvManager';
+import { deriveAdminSocketUrl } from '@/lib/realtime/adminSocketUrl';
+
+export type AdminRealtimeTicket = {
+ token: string;
+ expiresIn: number;
+ socketUrl: string;
+ namespace: string;
+ path: string;
+};
+
+export async function issueAdminRealtimeTicket(): Promise {
+ const env = await getCurrentEnvironment();
+ const { data } = await (
+ await getApiClient()
+ ).post<{ token: string; expiresIn: number }>('/realtime/ticket');
+
+ return {
+ token: data.token,
+ expiresIn: data.expiresIn,
+ socketUrl: deriveAdminSocketUrl(env.baseUrl, env.socketUrl),
+ namespace: '/database/',
+ path: '/realtime',
+ };
+}
diff --git a/src/lib/database/schema-indexes.test.ts b/src/lib/database/schema-indexes.test.ts
new file mode 100644
index 000000000..a5b5441f3
--- /dev/null
+++ b/src/lib/database/schema-indexes.test.ts
@@ -0,0 +1,38 @@
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+import { normalizeSchemaIndexResponse } from './schema-indexes.ts';
+
+const idIndex = {
+ fields: ['_id'],
+ types: [1],
+ options: { name: '_id_' },
+};
+
+const customIndex = {
+ fields: ['owner'],
+ types: [1],
+ options: { name: 'owner_1' },
+};
+
+describe('normalizeSchemaIndexResponse', () => {
+ it('accepts the raw array returned by the Database module', () => {
+ assert.deepEqual(normalizeSchemaIndexResponse([idIndex, customIndex]), [
+ idIndex,
+ customIndex,
+ ]);
+ });
+
+ it('accepts the documented { indexes } envelope', () => {
+ assert.deepEqual(
+ normalizeSchemaIndexResponse({ indexes: [idIndex, customIndex] }),
+ [idIndex, customIndex]
+ );
+ });
+
+ it('returns an empty array when indexes are missing', () => {
+ assert.deepEqual(normalizeSchemaIndexResponse(undefined), []);
+ assert.deepEqual(normalizeSchemaIndexResponse(null), []);
+ assert.deepEqual(normalizeSchemaIndexResponse({}), []);
+ assert.deepEqual(normalizeSchemaIndexResponse({ indexes: undefined }), []);
+ });
+});
diff --git a/src/lib/database/schema-indexes.ts b/src/lib/database/schema-indexes.ts
new file mode 100644
index 000000000..798b9be0e
--- /dev/null
+++ b/src/lib/database/schema-indexes.ts
@@ -0,0 +1,56 @@
+export type SchemaIndexPayload = {
+ fields: string[];
+ options?: Record;
+ types?: string[] | string;
+};
+
+function isIndexRecord(value: unknown): value is Record {
+ return Boolean(value) && typeof value === 'object' && !Array.isArray(value);
+}
+
+function toIndexPayload(value: Record): SchemaIndexPayload {
+ return {
+ ...value,
+ fields: Array.isArray(value.fields)
+ ? value.fields.filter(
+ (field): field is string => typeof field === 'string'
+ )
+ : [],
+ options:
+ value.options &&
+ typeof value.options === 'object' &&
+ !Array.isArray(value.options)
+ ? (value.options as Record)
+ : undefined,
+ types:
+ typeof value.types === 'string' || Array.isArray(value.types)
+ ? (value.types as string[] | string)
+ : undefined,
+ };
+}
+
+function collectIndexRecords(data: unknown): Record[] {
+ if (Array.isArray(data)) {
+ return data.filter(isIndexRecord);
+ }
+
+ if (
+ data &&
+ typeof data === 'object' &&
+ Array.isArray((data as { indexes?: unknown }).indexes)
+ ) {
+ return (data as { indexes: unknown[] }).indexes.filter(isIndexRecord);
+ }
+
+ return [];
+}
+
+/**
+ * GET /database/schemas/:id/indexes is documented as `{ indexes: [...] }`,
+ * but the Database module handler returns the index array itself.
+ */
+export function normalizeSchemaIndexResponse(
+ data: unknown
+): SchemaIndexPayload[] {
+ return collectIndexRecords(data).map(toIndexPayload);
+}
diff --git a/src/lib/hooks/use-database-live-updates.ts b/src/lib/hooks/use-database-live-updates.ts
new file mode 100644
index 000000000..a1fd0489d
--- /dev/null
+++ b/src/lib/hooks/use-database-live-updates.ts
@@ -0,0 +1,145 @@
+'use client';
+
+import * as React from 'react';
+import { io, type Socket } from 'socket.io-client';
+import { issueAdminRealtimeTicket } from '@/lib/api/realtime/ticket';
+import {
+ parseDatabaseChangeEvent,
+ rememberResumeToken,
+ shouldCountChange,
+} from '@/lib/realtime/change-events';
+
+export type LiveConnectionState =
+ | 'idle'
+ | 'connecting'
+ | 'live'
+ | 'reconnecting'
+ | 'error';
+
+const RETRY_BASE_MS = 1_000;
+const RETRY_MAX_MS = 15_000;
+
+export function useDatabaseLiveUpdates(options: {
+ schemaName: string;
+ enabled: boolean;
+}) {
+ const { schemaName, enabled } = options;
+ const [connectionState, setConnectionState] =
+ React.useState('idle');
+ const [pendingUpdates, setPendingUpdates] = React.useState(0);
+ const [errorMessage, setErrorMessage] = React.useState(null);
+
+ React.useEffect(() => {
+ if (!enabled || schemaName.trim() === '') {
+ setConnectionState('idle');
+ setPendingUpdates(0);
+ setErrorMessage(null);
+ return;
+ }
+
+ let cancelled = false;
+ let socket: Socket | undefined;
+ let retryTimer: ReturnType | undefined;
+ let attempt = 0;
+ let everConnected = false;
+
+ const disconnectSocket = () => {
+ if (!socket) return;
+ socket.removeAllListeners();
+ socket.disconnect();
+ socket = undefined;
+ };
+
+ const connect = async () => {
+ if (cancelled) return;
+ setConnectionState(everConnected ? 'reconnecting' : 'connecting');
+ try {
+ const ticket = await issueAdminRealtimeTicket();
+ if (cancelled) return;
+ disconnectSocket();
+ socket = io(`${ticket.socketUrl}${ticket.namespace}`, {
+ path: ticket.path,
+ auth: { token: ticket.token },
+ transports: ['websocket', 'polling'],
+ reconnection: false,
+ });
+
+ socket.on('connect', () => {
+ everConnected = true;
+ attempt = 0;
+ setErrorMessage(null);
+ setConnectionState('live');
+ socket?.emit('subscribe', { schema: schemaName });
+ });
+
+ const seenResumeTokens = new Set();
+ socket.on('change', (data: unknown) => {
+ const event = parseDatabaseChangeEvent(data);
+ if (!event || !shouldCountChange(event, schemaName)) return;
+ if (!rememberResumeToken(seenResumeTokens, event.resumeToken)) return;
+ setPendingUpdates(count => count + 1);
+ });
+
+ socket.on('conduit_error', (data: unknown) => {
+ const message =
+ data &&
+ typeof data === 'object' &&
+ 'message' in data &&
+ typeof data.message === 'string'
+ ? data.message
+ : 'Live updates failed';
+ setErrorMessage(message);
+ setConnectionState('error');
+ });
+
+ socket.on('connect_error', err => {
+ setErrorMessage(
+ err instanceof Error ? err.message : 'Live updates failed'
+ );
+ scheduleReconnect();
+ });
+
+ socket.on('disconnect', reason => {
+ if (cancelled || reason === 'io client disconnect') return;
+ scheduleReconnect();
+ });
+ } catch (err) {
+ setErrorMessage(
+ err instanceof Error ? err.message : 'Live updates failed'
+ );
+ scheduleReconnect();
+ }
+ };
+
+ const scheduleReconnect = () => {
+ if (cancelled || retryTimer) return;
+ disconnectSocket();
+ setConnectionState(everConnected ? 'reconnecting' : 'error');
+ const delay = Math.min(RETRY_MAX_MS, RETRY_BASE_MS * 2 ** attempt);
+ attempt += 1;
+ retryTimer = setTimeout(() => {
+ retryTimer = undefined;
+ void connect();
+ }, delay);
+ };
+
+ void connect();
+
+ return () => {
+ cancelled = true;
+ if (retryTimer) clearTimeout(retryTimer);
+ disconnectSocket();
+ };
+ }, [enabled, schemaName]);
+
+ const consumePendingUpdates = React.useCallback(() => {
+ setPendingUpdates(0);
+ }, []);
+
+ return {
+ connectionState,
+ pendingUpdates,
+ errorMessage,
+ consumePendingUpdates,
+ };
+}
diff --git a/src/lib/logic/EnvManager.ts b/src/lib/logic/EnvManager.ts
index f74d69060..005d7c584 100644
--- a/src/lib/logic/EnvManager.ts
+++ b/src/lib/logic/EnvManager.ts
@@ -7,6 +7,7 @@ export interface Environment {
name: string;
baseUrl: string;
masterKey: string;
+ socketUrl?: string;
lokiUrl?: string;
promUrl?: string;
namespace?: string;
@@ -43,13 +44,14 @@ function getConfigFingerprint(): string {
mode,
defaultEnvironment: process.env.DEFAULT_ENVIRONMENT,
apiBaseUrl: process.env.API_BASE_URL,
+ adminSocketUrl: process.env.ADMIN_SOCKET_URL,
masterKeyHash: hashSecret(process.env.MASTER_KEY),
});
}
const environments: Record<
string,
- { baseUrl?: string; masterKeyHash: string }
+ { baseUrl?: string; socketUrl?: string; masterKeyHash: string }
> = {};
Object.keys(envVars)
.filter(key => key.endsWith('_API_BASE_URL'))
@@ -58,6 +60,7 @@ function getConfigFingerprint(): string {
const prefix = key.replace('_API_BASE_URL', '').toUpperCase();
environments[prefix] = {
baseUrl: envVars[`${prefix}_API_BASE_URL`],
+ socketUrl: envVars[`${prefix}_ADMIN_SOCKET_URL`],
masterKeyHash: hashSecret(envVars[`${prefix}_MASTER_KEY`]),
};
});
@@ -110,6 +113,7 @@ async function getSingleEnvironmentConfig(
name: envName,
baseUrl: process.env.API_BASE_URL!,
masterKey: process.env.MASTER_KEY!,
+ socketUrl: process.env.ADMIN_SOCKET_URL,
lokiUrl: process.env.LOKI_URL,
promUrl: process.env.PROMETHEUS_URL,
namespace: process.env.NAMESPACE,
@@ -158,6 +162,7 @@ async function getMultiEnvironmentConfig(
name: envName,
baseUrl: envVars[`${prefix}_API_BASE_URL`]!,
masterKey: envVars[`${prefix}_MASTER_KEY`]!,
+ socketUrl: envVars[`${prefix}_ADMIN_SOCKET_URL`],
lokiUrl: envVars[`${prefix}_LOKI_URL`],
promUrl: envVars[`${prefix}_PROMETHEUS_URL`],
namespace: envVars[`${prefix}_NAMESPACE`],
diff --git a/src/lib/models/database/schemas.ts b/src/lib/models/database/schemas.ts
index 34601f427..7c80f4b8d 100644
--- a/src/lib/models/database/schemas.ts
+++ b/src/lib/models/database/schemas.ts
@@ -33,6 +33,9 @@ export type SchemaOptions = {
authorization?: {
enabled: boolean;
};
+ realtime?: {
+ enabled: boolean;
+ };
/** MongoDB only: overrides module default read preference for this schema's reads */
readPreference?: string;
};
@@ -65,6 +68,9 @@ export type PatchSchemaRequest = Partial> & {
cms?: Record;
authorization?: Record;
permissions?: Record;
+ realtime?: {
+ enabled?: boolean;
+ };
/** MongoDB only; empty string clears schema-level override */
readPreference?: string;
};
diff --git a/src/lib/models/database/settings.ts b/src/lib/models/database/settings.ts
index a81e88187..7c33d5b51 100644
--- a/src/lib/models/database/settings.ts
+++ b/src/lib/models/database/settings.ts
@@ -12,8 +12,27 @@ export type ReadConcern =
| 'linearizable'
| 'snapshot';
+export type DatabaseRealtimeStatusCode =
+ | 'unsupported'
+ | 'disabled'
+ | 'idle'
+ | 'starting'
+ | 'live'
+ | 'degraded';
+
+export type DatabaseRealtimeStatus = {
+ status: DatabaseRealtimeStatusCode;
+ engine: string;
+ activeSchemaCount: number;
+ lastEventAt?: string;
+ message?: string;
+};
+
export type DatabaseConfig = {
readPreference: ReadPreference;
writeConcern: WriteConcern;
readConcern: ReadConcern;
+ realtime?: {
+ enabled: boolean;
+ };
};
diff --git a/src/lib/realtime/adminSocketUrl.test.ts b/src/lib/realtime/adminSocketUrl.test.ts
new file mode 100644
index 000000000..e74d44f47
--- /dev/null
+++ b/src/lib/realtime/adminSocketUrl.test.ts
@@ -0,0 +1,59 @@
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+import {
+ deriveAdminSocketUrl,
+ deriveClientSocketUrl,
+} from './adminSocketUrl.ts';
+
+describe('deriveAdminSocketUrl', () => {
+ it('uses an explicit socket URL when provided', () => {
+ assert.equal(
+ deriveAdminSocketUrl(
+ 'http://localhost:3030',
+ 'http://sockets.example:4000/'
+ ),
+ 'http://sockets.example:4000'
+ );
+ });
+
+ it('maps the default admin API port 3030 to 3031', () => {
+ assert.equal(
+ deriveAdminSocketUrl('http://localhost:3030'),
+ 'http://localhost:3031'
+ );
+ });
+
+ it('keeps proxied origins that are not on port 3030', () => {
+ assert.equal(
+ deriveAdminSocketUrl('https://admin.example.com'),
+ 'https://admin.example.com'
+ );
+ assert.equal(
+ deriveAdminSocketUrl('http://localhost:8080/v1'),
+ 'http://localhost:8080'
+ );
+ });
+});
+
+describe('deriveClientSocketUrl', () => {
+ it('maps the default client API port 3000 to 3001', () => {
+ assert.equal(
+ deriveClientSocketUrl('http://localhost:3000'),
+ 'http://localhost:3001'
+ );
+ });
+
+ it('maps standalone router hostUrl on 3030 to client socket 3001', () => {
+ assert.equal(
+ deriveClientSocketUrl('http://localhost:3030'),
+ 'http://localhost:3001'
+ );
+ });
+
+ it('keeps proxied client origins that are not on port 3000', () => {
+ assert.equal(
+ deriveClientSocketUrl('https://api.example.com'),
+ 'https://api.example.com'
+ );
+ });
+});
diff --git a/src/lib/realtime/adminSocketUrl.ts b/src/lib/realtime/adminSocketUrl.ts
new file mode 100644
index 000000000..a6a509eb1
--- /dev/null
+++ b/src/lib/realtime/adminSocketUrl.ts
@@ -0,0 +1,42 @@
+export function deriveAdminSocketUrl(
+ apiBaseUrl: string,
+ explicit?: string | null
+): string {
+ return deriveSocketUrl(apiBaseUrl, { '3030': '3031' }, explicit);
+}
+
+/** Maps Client REST `:3000` (and standalone Router hostUrl on `:3030`) to Socket.IO `:3001`. */
+export function deriveClientSocketUrl(
+ apiBaseUrl: string,
+ explicit?: string | null
+): string {
+ return deriveSocketUrl(
+ apiBaseUrl,
+ { '3000': '3001', '3030': '3001' },
+ explicit
+ );
+}
+
+export function deriveSocketUrl(
+ apiBaseUrl: string,
+ restToSocketPort: Record,
+ explicit?: string | null
+): string {
+ if (explicit && explicit.trim() !== '') {
+ return trimTrailingSlash(explicit.trim());
+ }
+
+ const url = new URL(apiBaseUrl);
+ const mapped = restToSocketPort[url.port];
+ if (mapped) {
+ url.port = mapped;
+ }
+ url.pathname = '';
+ url.search = '';
+ url.hash = '';
+ return trimTrailingSlash(url.toString());
+}
+
+function trimTrailingSlash(value: string): string {
+ return value.replace(/\/+$/, '');
+}
diff --git a/src/lib/realtime/change-events.test.ts b/src/lib/realtime/change-events.test.ts
new file mode 100644
index 000000000..004af6436
--- /dev/null
+++ b/src/lib/realtime/change-events.test.ts
@@ -0,0 +1,57 @@
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+import {
+ parseDatabaseChangeEvent,
+ rememberResumeToken,
+ shouldCountChange,
+} from './change-events.ts';
+
+const validEvent = {
+ version: 1,
+ operation: 'insert',
+ schema: 'Order',
+ documentId: '64b64c4c4c4c4c4c4c4c4c4c',
+ occurredAt: '2026-01-02T00:00:00.000Z',
+ resumeToken: '{"_data":"token"}',
+};
+
+describe('parseDatabaseChangeEvent', () => {
+ it('accepts a metadata-only payload', () => {
+ assert.deepEqual(parseDatabaseChangeEvent(validEvent), validEvent);
+ assert.deepEqual(
+ parseDatabaseChangeEvent(JSON.stringify(validEvent)),
+ validEvent
+ );
+ });
+
+ it('rejects document fields, wrong versions, and malformed payloads', () => {
+ assert.equal(parseDatabaseChangeEvent(null), null);
+ assert.equal(parseDatabaseChangeEvent('{'), null);
+ assert.equal(parseDatabaseChangeEvent({ ...validEvent, version: 2 }), null);
+ assert.equal(
+ parseDatabaseChangeEvent({ ...validEvent, operation: 'drop' }),
+ null
+ );
+ assert.equal(parseDatabaseChangeEvent({ ...validEvent, schema: '' }), null);
+ });
+});
+
+describe('shouldCountChange', () => {
+ it('counts only events for the subscribed schema', () => {
+ const event = parseDatabaseChangeEvent(validEvent)!;
+ assert.equal(shouldCountChange(event, 'Order'), true);
+ assert.equal(shouldCountChange(event, 'User'), false);
+ });
+});
+
+describe('rememberResumeToken', () => {
+ it('counts a token once and evicts the oldest past the limit', () => {
+ const seen = new Set();
+ assert.equal(rememberResumeToken(seen, 'a', 2), true);
+ assert.equal(rememberResumeToken(seen, 'a', 2), false);
+ assert.equal(rememberResumeToken(seen, 'b', 2), true);
+ assert.equal(rememberResumeToken(seen, 'c', 2), true);
+ assert.equal(seen.has('a'), false);
+ assert.equal(rememberResumeToken(seen, 'a', 2), true);
+ });
+});
diff --git a/src/lib/realtime/change-events.ts b/src/lib/realtime/change-events.ts
new file mode 100644
index 000000000..3f1f8b3a7
--- /dev/null
+++ b/src/lib/realtime/change-events.ts
@@ -0,0 +1,89 @@
+const DATABASE_CHANGE_EVENT_VERSION = 1;
+
+const DATABASE_CHANGE_OPERATIONS = [
+ 'insert',
+ 'update',
+ 'replace',
+ 'delete',
+] as const;
+
+export type DatabaseChangeOperation =
+ (typeof DATABASE_CHANGE_OPERATIONS)[number];
+
+export type DatabaseChangeEvent = {
+ version: number;
+ operation: DatabaseChangeOperation;
+ schema: string;
+ documentId: string;
+ occurredAt: string;
+ resumeToken: string;
+};
+
+export function parseDatabaseChangeEvent(
+ data: unknown
+): DatabaseChangeEvent | null {
+ const raw = typeof data === 'string' ? safeParseJson(data) : data;
+ if (!raw || typeof raw !== 'object' || Array.isArray(raw)) {
+ return null;
+ }
+
+ const event = raw as Record;
+ if (event.version !== DATABASE_CHANGE_EVENT_VERSION) return null;
+ if (
+ typeof event.operation !== 'string' ||
+ !DATABASE_CHANGE_OPERATIONS.includes(
+ event.operation as DatabaseChangeOperation
+ )
+ ) {
+ return null;
+ }
+ if (typeof event.schema !== 'string' || event.schema.trim() === '') {
+ return null;
+ }
+ if (typeof event.documentId !== 'string' || event.documentId.trim() === '') {
+ return null;
+ }
+ if (typeof event.occurredAt !== 'string') return null;
+ if (typeof event.resumeToken !== 'string') return null;
+
+ return {
+ version: DATABASE_CHANGE_EVENT_VERSION,
+ operation: event.operation as DatabaseChangeOperation,
+ schema: event.schema,
+ documentId: event.documentId,
+ occurredAt: event.occurredAt,
+ resumeToken: event.resumeToken,
+ };
+}
+
+export function shouldCountChange(
+ event: DatabaseChangeEvent,
+ schemaName: string
+): boolean {
+ return event.schema === schemaName;
+}
+
+const DEFAULT_SEEN_TOKEN_LIMIT = 500;
+
+/** Returns true the first time `token` is seen. Oldest entries are dropped past `limit`. */
+export function rememberResumeToken(
+ seen: Set,
+ token: string,
+ limit: number = DEFAULT_SEEN_TOKEN_LIMIT
+): boolean {
+ if (seen.has(token)) return false;
+ seen.add(token);
+ if (seen.size > limit) {
+ const oldest = seen.values().next().value;
+ if (oldest !== undefined) seen.delete(oldest);
+ }
+ return true;
+}
+
+function safeParseJson(value: string): unknown {
+ try {
+ return JSON.parse(value);
+ } catch {
+ return null;
+ }
+}
diff --git a/src/lib/realtime/connection-snippets.test.ts b/src/lib/realtime/connection-snippets.test.ts
new file mode 100644
index 000000000..e92f9e251
--- /dev/null
+++ b/src/lib/realtime/connection-snippets.test.ts
@@ -0,0 +1,70 @@
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+import {
+ buildAdminJavascriptSnippet,
+ buildClientJavascriptSnippet,
+ buildPostmanConnection,
+ socketEndpoint,
+ subscribePayload,
+} from './connection-snippets.ts';
+
+describe('subscribePayload', () => {
+ it('scopes client authorized schemas to a document id', () => {
+ assert.deepEqual(subscribePayload('Order', true), {
+ schema: 'Order',
+ documentId: '',
+ });
+ assert.deepEqual(subscribePayload('Order', false), { schema: 'Order' });
+ });
+});
+
+describe('buildClientJavascriptSnippet', () => {
+ it('targets the schema namespace and subscribe payload', () => {
+ const snippet = buildClientJavascriptSnippet({
+ socketUrl: 'http://localhost:3001',
+ schemaName: 'LiveUpdateProbe',
+ documentScoped: false,
+ });
+ assert.match(snippet, /http:\/\/localhost:3001\/database\//);
+ assert.match(snippet, /path: "\/realtime"/);
+ assert.match(snippet, /"schema":"LiveUpdateProbe"/);
+ });
+});
+
+describe('buildAdminJavascriptSnippet', () => {
+ it('uses the admin socket origin', () => {
+ const snippet = buildAdminJavascriptSnippet({
+ socketUrl: 'http://localhost:3031',
+ schemaName: 'LiveUpdateProbe',
+ });
+ assert.match(snippet, /http:\/\/localhost:3031\/database\//);
+ assert.match(snippet, /adminToken/);
+ });
+});
+
+describe('buildPostmanConnection', () => {
+ it('matches the Socket.IO handshake used in Data Explorer', () => {
+ const connection = buildPostmanConnection({
+ adminSocketUrl: 'http://localhost:3031',
+ schemaName: 'LiveUpdateProbe',
+ });
+ assert.equal(connection.serverUrl, 'http://localhost:3031/database/');
+ assert.equal(connection.path, '/realtime');
+ assert.equal(connection.clientVersion, 'v4');
+ assert.equal(connection.subscribeEvent, 'subscribe');
+ assert.match(connection.subscribeBody, /LiveUpdateProbe/);
+ assert.deepEqual(
+ connection.headers.map(header => header.key),
+ ['masterkey', 'Authorization']
+ );
+ });
+});
+
+describe('socketEndpoint', () => {
+ it('always ends with the database namespace', () => {
+ assert.equal(
+ socketEndpoint('http://localhost:3031/'),
+ 'http://localhost:3031/database/'
+ );
+ });
+});
diff --git a/src/lib/realtime/connection-snippets.ts b/src/lib/realtime/connection-snippets.ts
new file mode 100644
index 000000000..12aeda524
--- /dev/null
+++ b/src/lib/realtime/connection-snippets.ts
@@ -0,0 +1,116 @@
+const DATABASE_SOCKET_NAMESPACE = '/database/';
+export const DATABASE_SOCKET_PATH = '/realtime';
+export const DATABASE_SOCKET_LISTEN_EVENTS = [
+ 'change',
+ 'connected',
+ 'conduit_error',
+] as const;
+
+type SubscribePayload = {
+ schema: string;
+ documentId?: string;
+};
+
+export function socketEndpoint(socketUrl: string): string {
+ return `${socketUrl.replace(/\/+$/, '')}${DATABASE_SOCKET_NAMESPACE}`;
+}
+
+export function subscribePayload(
+ schemaName: string,
+ documentScoped: boolean
+): SubscribePayload {
+ if (documentScoped) {
+ return { schema: schemaName, documentId: '' };
+ }
+ return { schema: schemaName };
+}
+
+export function buildClientJavascriptSnippet(options: {
+ socketUrl: string;
+ schemaName: string;
+ documentScoped: boolean;
+}): string {
+ const url = socketEndpoint(options.socketUrl);
+ const payload = subscribePayload(options.schemaName, options.documentScoped);
+ return `import { io } from 'socket.io-client';
+
+const socket = io(${JSON.stringify(url)}, {
+ path: ${JSON.stringify(DATABASE_SOCKET_PATH)},
+ auth: { token: accessToken },
+ transports: ['websocket', 'polling'],
+});
+
+socket.on('connect', () => {
+ socket.emit('subscribe', ${JSON.stringify(payload)});
+});
+
+socket.on('change', (event) => {
+ // { version, operation, schema, documentId, occurredAt, resumeToken }
+ console.log(event);
+});
+
+socket.on('conduit_error', (error) => {
+ console.error(error);
+});
+`;
+}
+
+export function buildAdminJavascriptSnippet(options: {
+ socketUrl: string;
+ schemaName: string;
+}): string {
+ const url = socketEndpoint(options.socketUrl);
+ const payload = subscribePayload(options.schemaName, false);
+ return `import { io } from 'socket.io-client';
+
+const socket = io(${JSON.stringify(url)}, {
+ path: ${JSON.stringify(DATABASE_SOCKET_PATH)},
+ auth: { token: adminToken },
+ transports: ['websocket', 'polling'],
+});
+
+socket.on('connect', () => {
+ socket.emit('subscribe', ${JSON.stringify(payload)});
+});
+
+socket.on('change', (event) => {
+ console.log(event);
+});
+
+socket.on('conduit_error', (error) => {
+ console.error(error);
+});
+`;
+}
+
+type PostmanConnection = {
+ serverUrl: string;
+ path: string;
+ clientVersion: 'v4';
+ headers: { key: string; value: string }[];
+ listenFor: string[];
+ subscribeEvent: 'subscribe';
+ subscribeBody: string;
+};
+
+export function buildPostmanConnection(options: {
+ adminSocketUrl: string;
+ schemaName: string;
+}): PostmanConnection {
+ return {
+ serverUrl: socketEndpoint(options.adminSocketUrl),
+ path: DATABASE_SOCKET_PATH,
+ clientVersion: 'v4',
+ headers: [
+ { key: 'masterkey', value: '{{masterKey}}' },
+ { key: 'Authorization', value: 'Bearer {{adminToken}}' },
+ ],
+ listenFor: [...DATABASE_SOCKET_LISTEN_EVENTS],
+ subscribeEvent: 'subscribe',
+ subscribeBody: JSON.stringify(
+ subscribePayload(options.schemaName, false),
+ null,
+ 2
+ ),
+ };
+}