From 56fe33f8cd24e28f5cc818cce953df0bac8ceb59 Mon Sep 17 00:00:00 2001 From: John Chantzigoulas Date: Fri, 4 Sep 2026 14:38:02 +0300 Subject: [PATCH 1/6] feat(router): add event relays administration page Give operators CRUD for mapping exact bus events to ReBAC-protected socket rooms, with a local template preview before save. --- package.json | 1 + .../(modules)/router/event-relays/page.tsx | 32 +++ src/app/(dashboard)/(modules)/router/page.tsx | 15 +- src/components/navigation/navList.config.ts | 1 + .../router/event-relays/event-relay-form.tsx | 214 +++++++++++++++ .../router/event-relays/event-relay-list.tsx | 259 ++++++++++++++++++ src/components/router/event-relays/zod.ts | 72 +++++ src/lib/api/router/index.ts | 44 +++ src/lib/event-relays/path.ts | 56 ++++ src/lib/event-relays/preview.test.ts | 31 +++ src/lib/event-relays/preview.ts | 33 +++ src/lib/event-relays/template.ts | 65 +++++ src/lib/models/Router.ts | 32 +++ 13 files changed, 854 insertions(+), 1 deletion(-) create mode 100644 src/app/(dashboard)/(modules)/router/event-relays/page.tsx create mode 100644 src/components/router/event-relays/event-relay-form.tsx create mode 100644 src/components/router/event-relays/event-relay-list.tsx create mode 100644 src/components/router/event-relays/zod.ts create mode 100644 src/lib/event-relays/path.ts create mode 100644 src/lib/event-relays/preview.test.ts create mode 100644 src/lib/event-relays/preview.ts create mode 100644 src/lib/event-relays/template.ts diff --git a/package.json b/package.json index 59e6619fa..32e45a065 100644 --- a/package.json +++ b/package.json @@ -9,6 +9,7 @@ "start": "next start -p 8080", "lint": "eslint .", "test:semantic-colors": "node --experimental-strip-types --test src/lib/semantic-colors.test.ts src/lib/reactflow-edge-colors.test.ts", + "test:event-relays": "node --experimental-strip-types --test src/lib/event-relays/preview.test.ts", "build:docker": "docker build --platform linux/amd64 -t ghcr.io/conduitplatform/conduit-ui:latest .", "prepare": "husky", "release": "standard-version" diff --git a/src/app/(dashboard)/(modules)/router/event-relays/page.tsx b/src/app/(dashboard)/(modules)/router/event-relays/page.tsx new file mode 100644 index 000000000..509f7094a --- /dev/null +++ b/src/app/(dashboard)/(modules)/router/event-relays/page.tsx @@ -0,0 +1,32 @@ +import { getEventRelays, getRouterSettings } from '@/lib/api/router'; +import { EventRelayList } from '@/components/router/event-relays/event-relay-list'; + +export default async function EventRelaysPage(props: { + searchParams: Promise<{ + skip?: string; + limit?: string; + search?: string; + }>; +}) { + const searchParams = await props.searchParams; + const skip = Number(searchParams.skip ?? 0); + const limit = Number(searchParams.limit ?? 10); + const [{ relays, count }, { config }] = await Promise.all([ + getEventRelays({ + skip, + limit, + search: searchParams.search, + }), + getRouterSettings(), + ]); + + return ( +
+ +
+ ); +} diff --git a/src/app/(dashboard)/(modules)/router/page.tsx b/src/app/(dashboard)/(modules)/router/page.tsx index 64eaa8897..11e104ea1 100644 --- a/src/app/(dashboard)/(modules)/router/page.tsx +++ b/src/app/(dashboard)/(modules)/router/page.tsx @@ -1,5 +1,12 @@ import React from 'react'; -import { BarChart3, Network, Route, Settings, Shield } from 'lucide-react'; +import { + BarChart3, + Network, + Radio, + Route, + Settings, + Shield, +} from 'lucide-react'; import { ModuleDashboard } from '@/components/dashboard/ModuleDashboard'; import { getModuleStatus, @@ -66,6 +73,12 @@ export default async function RouterDashboard() { icon: , href: '/router/vizualize', }, + { + title: 'Event Relays', + description: 'Forward bus events to socket subscribers', + icon: , + href: '/router/event-relays', + }, { title: 'Settings', description: 'Router module configuration', diff --git a/src/components/navigation/navList.config.ts b/src/components/navigation/navList.config.ts index ff9d42677..017235613 100644 --- a/src/components/navigation/navList.config.ts +++ b/src/components/navigation/navList.config.ts @@ -133,6 +133,7 @@ export const navGroups: NavGroup[] = [ items: [ { title: 'Visualize', url: '/router/vizualize' }, { title: 'Security', url: '/router/security' }, + { title: 'Event Relays', url: '/router/event-relays' }, { title: 'Settings', url: '/router/settings' }, ], }, diff --git a/src/components/router/event-relays/event-relay-form.tsx b/src/components/router/event-relays/event-relay-form.tsx new file mode 100644 index 000000000..833aa3db3 --- /dev/null +++ b/src/components/router/event-relays/event-relay-form.tsx @@ -0,0 +1,214 @@ +'use client'; + +import { useMemo } from 'react'; +import { useForm, useWatch } from 'react-hook-form'; +import { rhfZodResolver } from '@/lib/zod-form'; +import { Form } from '@/components/ui/form'; +import { InputField } from '@/components/ui/form-inputs/InputField'; +import SwitchField from '@/components/ui/form-inputs/SwitchField'; +import { CodeField } from '@/components/ui/form-inputs/CodeField'; +import { Button } from '@/components/ui/button'; +import { Alert, AlertDescription, AlertTitle } from '@/components/ui/alert'; +import { + EventRelayFormSchema, + EventRelayFormValues, + parseJsonField, +} from '@/components/router/event-relays/zod'; +import { previewEventRelay } from '@/lib/event-relays/preview'; +import { EventRelay, EventRelayWriteRequest } from '@/lib/models/Router'; + +const DEFAULT_TEMPLATE = '{\n "id": "{{payload._id}}"\n}'; +const DEFAULT_SAMPLE = + '{\n "_id": "64f1c0a2b4d0e1f2a3b4c5d6",\n "status": "paid"\n}'; + +interface EventRelayFormProps { + relay?: EventRelay | null; + onSubmit: (data: EventRelayWriteRequest) => Promise; + onCancel: () => void; + isSaving?: boolean; +} + +export function EventRelayForm({ + relay, + onSubmit, + onCancel, + isSaving, +}: EventRelayFormProps) { + const form = useForm({ + resolver: rhfZodResolver(EventRelayFormSchema), + defaultValues: { + name: relay?.name ?? '', + notes: relay?.notes ?? '', + active: relay?.active ?? true, + busEvent: relay?.busEvent ?? '', + socketEvent: relay?.socketEvent ?? '', + resourceType: relay?.resourceType ?? '', + resourceIdPath: relay?.resourceIdPath ?? '_id', + permission: relay?.permission ?? 'read', + messageTemplate: relay + ? JSON.stringify(relay.messageTemplate, null, 2) + : DEFAULT_TEMPLATE, + samplePayload: DEFAULT_SAMPLE, + }, + }); + + const watched = useWatch({ control: form.control }); + const preview = useMemo(() => { + try { + const template = parseJsonField( + watched.messageTemplate ?? '', + 'Message template' + ); + const sample = parseJsonField( + watched.samplePayload && watched.samplePayload.trim() !== '' + ? watched.samplePayload + : '{}', + 'Sample payload' + ); + return previewEventRelay({ + resourceIdPath: watched.resourceIdPath || '_id', + messageTemplate: template, + samplePayload: sample, + }); + } catch (err) { + return { + error: err instanceof Error ? err.message : String(err), + }; + } + }, [watched.messageTemplate, watched.samplePayload, watched.resourceIdPath]); + + const handleSubmit = form.handleSubmit(async values => { + const messageTemplate = parseJsonField( + values.messageTemplate, + 'Message template' + ); + await onSubmit({ + name: values.name, + notes: values.notes?.trim() ? values.notes.trim() : undefined, + active: values.active, + busEvent: values.busEvent, + socketEvent: values.socketEvent, + resourceType: values.resourceType, + resourceIdPath: values.resourceIdPath, + permission: values.permission, + messageTemplate, + }); + }); + + return ( +
+ +
+ + +
+ +
+ + +
+
+ + + +
+ + +
+

Preview

+

+ Local only. Nothing is published to the bus. +

+ {preview.error ? ( +

{preview.error}

+ ) : ( +
+

+ Resource{' '} + + {watched.resourceType || 'Type'}:{preview.resourceId} + +

+ 
+                {JSON.stringify(preview.payload, null, 2)}
+
+
+ )} +
+ + Client contract + +

+ Connect to /events/ with{' '} + path: /realtime and a bearer + token. Then emit{' '} + subscribe(relayId, resourceId). +

+ 
+              {`const socket = io(\`\${SOCKET_URL}/events/\`, {
+  path: '/realtime',
+  extraHeaders: { authorization: \`Bearer \${accessToken}\` },
+});
+socket.emit('subscribe', relayId, resourceId);
+socket.on('${watched.socketEvent || 'your-event'}', payload => {});
+socket.emit('unsubscribe', relayId, resourceId);`}
+
+ + +
+ + +
+ + + ); +} diff --git a/src/components/router/event-relays/event-relay-list.tsx b/src/components/router/event-relays/event-relay-list.tsx new file mode 100644 index 000000000..162e8a0a5 --- /dev/null +++ b/src/components/router/event-relays/event-relay-list.tsx @@ -0,0 +1,259 @@ +'use client'; + +import { useCallback, useMemo, useState } from 'react'; +import { ColumnDef } from '@tanstack/react-table'; +import { Plus, Radio, Zap } from 'lucide-react'; +import { Button } from '@/components/ui/button'; +import { Badge } from '@/components/ui/badge'; +import { DataTable } from '@/components/ui/data-table'; +import { EmptyState } from '@/components/ui/empty-state'; +import { Alert, AlertDescription, AlertTitle } from '@/components/ui/alert'; +import { + Dialog, + DialogContent, + DialogDescription, + DialogHeader, + DialogTitle, +} from '@/components/ui/dialog'; +import { + PageActions, + PageDescription, + PageHeader, + PageTitle, +} from '@/components/ui/page-header'; +import { SearchInput } from '@/components/ui/form-inputs/SearchInput'; +import { DeleteAlert } from '@/components/helpers/delete'; +import { EventRelayForm } from '@/components/router/event-relays/event-relay-form'; +import { EventRelay, EventRelayWriteRequest } from '@/lib/models/Router'; +import { + createEventRelay, + deleteEventRelay, + patchEventRelay, +} from '@/lib/api/router'; +import { useSettingsSave } from '@/lib/hooks/use-settings-save'; +import { useRouter } from 'next/navigation'; + +interface EventRelayListProps { + relays: EventRelay[]; + count: number; + socketsEnabled: boolean; +} + +export function EventRelayList({ + relays, + count, + socketsEnabled, +}: EventRelayListProps) { + const router = useRouter(); + const [isCreateOpen, setIsCreateOpen] = useState(false); + const [editing, setEditing] = useState(null); + const { save, isSaving } = useSettingsSave('Event Relay'); + + const refresh = useCallback(() => { + router.refresh(); + }, [router]); + + const handleCreate = async (data: EventRelayWriteRequest) => { + const result = await save({ + action: async () => { + await createEventRelay(data); + await refresh(); + }, + successMessage: 'Event relay created', + }); + if (result.ok) { + setIsCreateOpen(false); + } + }; + + const handleUpdate = async (data: EventRelayWriteRequest) => { + if (!editing) return; + const result = await save({ + action: async () => { + await patchEventRelay(editing._id, data); + await refresh(); + }, + successMessage: 'Event relay updated', + }); + if (result.ok) { + setEditing(null); + } + }; + + const columns = useMemo[]>( + () => [ + { + accessorKey: 'name', + header: 'Name', + cell: ({ row }) => ( +
+

{row.original.name}

+ {row.original.notes ? ( +

+ {row.original.notes} +

+ ) : null} +
+ ), + }, + { + accessorKey: 'busEvent', + header: 'Bus event', + cell: ({ row }) => ( + + {row.original.busEvent} + + ), + }, + { + accessorKey: 'socketEvent', + header: 'Socket event', + cell: ({ row }) => ( + + {row.original.socketEvent} + + ), + }, + { + accessorKey: 'resourceType', + header: 'Resource', + cell: ({ row }) => ( + + {row.original.resourceType}:{row.original.permission} + + ), + }, + { + accessorKey: 'active', + header: 'Status', + cell: ({ row }) => ( + + {row.original.active ? 'Active' : 'Disabled'} + + ), + }, + { + id: 'actions', + cell: ({ row }) => ( +
+ + + save({ + action: async () => { + await deleteEventRelay(row.original._id); + await refresh(); + }, + successMessage: 'Event relay deleted', + }) + } + /> +
+ ), + }, + ], + [save, refresh] + ); + + return ( +
+ +
+ Event Relays + + Send a templated socket message when an exact bus event arrives. + Subscribers join resource rooms on /events/ after a ReBAC check. + +
+ + + + + + {!socketsEnabled ? ( + + + WebSockets are disabled + + Enable sockets in Router Settings before clients can subscribe to + relays. + + + ) : null} + +
+ +
+ + {relays.length === 0 ? ( + setIsCreateOpen(true)}> + + New relay + + } + /> + ) : ( + + )} + + + + + Create event relay + + Match one bus channel and emit a JSON template to the related + resource room. + + + setIsCreateOpen(false)} + isSaving={isSaving} + /> + + + + { + if (!open) setEditing(null); + }} + > + + + Edit event relay + + Changes apply immediately to new bus events. Delivery is ephemeral + and is not replayed. + + + {editing ? ( + setEditing(null)} + isSaving={isSaving} + /> + ) : null} + + +
+ ); +} diff --git a/src/components/router/event-relays/zod.ts b/src/components/router/event-relays/zod.ts new file mode 100644 index 000000000..bbffddbc4 --- /dev/null +++ b/src/components/router/event-relays/zod.ts @@ -0,0 +1,72 @@ +import { z } from 'zod'; +import { parseDotPath, RESERVED_SOCKET_EVENTS } from '@/lib/event-relays/path'; + +export const EventRelayFormSchema = z.object({ + name: z + .string() + .trim() + .min(1, 'Name is required') + .max(64, 'Name must be at most 64 characters') + .regex( + /^[A-Za-z0-9][A-Za-z0-9 _.-]{0,63}$/, + 'Name must start with a letter or number' + ), + notes: z.string().max(256).optional(), + active: z.boolean(), + busEvent: z + .string() + .trim() + .min(1, 'Bus event is required') + .max(128) + .regex( + /^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$/, + 'Use an exact channel name with no wildcards' + ) + .refine(value => !value.includes('*'), 'Wildcards are not supported'), + socketEvent: z + .string() + .trim() + .min(1, 'Socket event is required') + .max(64) + .regex(/^[A-Za-z][A-Za-z0-9_:-]{0,63}$/, 'Socket event name is invalid') + .refine( + value => !RESERVED_SOCKET_EVENTS.has(value), + 'This socket event name is reserved' + ), + resourceType: z + .string() + .trim() + .min(1, 'Resource type is required') + .regex(/^[A-Za-z][A-Za-z0-9_]{0,63}$/, 'Resource type is invalid'), + resourceIdPath: z + .string() + .trim() + .min(1, 'Resource ID path is required') + .superRefine((value, ctx) => { + try { + parseDotPath(value); + } catch (err) { + ctx.addIssue({ + code: 'custom', + message: err instanceof Error ? err.message : 'Invalid path', + }); + } + }), + permission: z + .string() + .trim() + .min(1, 'Permission is required') + .regex(/^[A-Za-z][A-Za-z0-9_]{0,63}$/, 'Permission is invalid'), + messageTemplate: z.string().trim().min(1, 'Message template is required'), + samplePayload: z.string().optional(), +}); + +export type EventRelayFormValues = z.infer; + +export function parseJsonField(value: string, label: string): unknown { + try { + return JSON.parse(value); + } catch { + throw new Error(`${label} must be valid JSON`); + } +} diff --git a/src/lib/api/router/index.ts b/src/lib/api/router/index.ts index 006227974..cc5e0b427 100644 --- a/src/lib/api/router/index.ts +++ b/src/lib/api/router/index.ts @@ -1,6 +1,11 @@ 'use server'; import { getApiClient } from '@/lib/api'; import { RouterSettings } from '@/lib/models/Router'; +import type { + EventRelay, + EventRelaysResponse, + EventRelayWriteRequest, +} from '@/lib/models/Router'; import { afterPatchServing } from '@/lib/api/modules/afterPatchServing'; import { PatchSettingsOptions } from '@/lib/api/modules/patch-settings-options'; @@ -94,3 +99,42 @@ export const patchAppRouteMiddlewares = async ( ); return res.data; }; + +export const getEventRelays = async (params?: { + skip?: number; + limit?: number; + search?: string; +}) => { + const res = await ( + await getApiClient() + ).get('/router/event-relays', { params }); + return res.data; +}; + +export const getEventRelay = async (id: string) => { + const res = await ( + await getApiClient() + ).get(`/router/event-relays/${id}`); + return res.data; +}; + +export const createEventRelay = async (data: EventRelayWriteRequest) => { + const res = await ( + await getApiClient() + ).post('/router/event-relays', data); + return res.data; +}; + +export const patchEventRelay = async ( + id: string, + data: Partial +) => { + const res = await ( + await getApiClient() + ).patch(`/router/event-relays/${id}`, data); + return res.data; +}; + +export const deleteEventRelay = async (id: string) => { + await (await getApiClient()).delete(`/router/event-relays/${id}`); +}; diff --git a/src/lib/event-relays/path.ts b/src/lib/event-relays/path.ts new file mode 100644 index 000000000..47929b5bb --- /dev/null +++ b/src/lib/event-relays/path.ts @@ -0,0 +1,56 @@ +export const RESERVED_SOCKET_EVENTS = new Set([ + 'connect', + 'disconnect', + 'connect_error', + 'error', + 'join-room', + 'leave-room', + 'conduit_error', + 'subscribe', + 'unsubscribe', + 'ping', + 'pong', +]); + +export const FORBIDDEN_PATH_SEGMENTS = new Set([ + '__proto__', + 'constructor', + 'prototype', +]); + +export const MAX_PATH_SEGMENTS = 8; +export const MAX_TEMPLATE_BYTES = 16 * 1024; + +const PATH_SEGMENT = /^[A-Za-z_][A-Za-z0-9_]*$/; + +export function parseDotPath(path: string): string[] { + const trimmed = path.trim(); + if (!trimmed) { + throw new Error('Path is required'); + } + const segments = trimmed.split('.'); + if (segments.length > MAX_PATH_SEGMENTS) { + throw new Error(`Path exceeds ${MAX_PATH_SEGMENTS} segments`); + } + for (const segment of segments) { + if (FORBIDDEN_PATH_SEGMENTS.has(segment) || !PATH_SEGMENT.test(segment)) { + throw new Error('Path contains an invalid segment'); + } + } + return segments; +} + +export function lookupOwnPath(source: unknown, path: string): unknown { + const segments = parseDotPath(path); + let current: unknown = source; + for (const segment of segments) { + if (current === null || typeof current !== 'object') { + return undefined; + } + if (!Object.prototype.hasOwnProperty.call(current, segment)) { + return undefined; + } + current = (current as Record)[segment]; + } + return current; +} diff --git a/src/lib/event-relays/preview.test.ts b/src/lib/event-relays/preview.test.ts new file mode 100644 index 000000000..50ecbd9df --- /dev/null +++ b/src/lib/event-relays/preview.test.ts @@ -0,0 +1,31 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import { lookupOwnPath } from './path.ts'; +import { renderMessageTemplate } from './template.ts'; +import { previewEventRelay } from './preview.ts'; + +describe('event relay preview helpers', () => { + it('resolves the resource id and rendered payload', () => { + const result = previewEventRelay({ + resourceIdPath: '_id', + messageTemplate: { id: '{{payload._id}}', status: '{{payload.status}}' }, + samplePayload: { _id: 'order-1', status: 'paid' }, + }); + assert.equal(result.error, undefined); + assert.equal(result.resourceId, 'order-1'); + assert.deepEqual(result.payload, { id: 'order-1', status: 'paid' }); + }); + + it('fails closed on missing fields and prototype paths', () => { + assert.equal(lookupOwnPath({ a: 1 }, 'b'), undefined); + assert.throws(() => + renderMessageTemplate({ x: '{{payload.missing}}' }, {}) + ); + const preview = previewEventRelay({ + resourceIdPath: '__proto__', + messageTemplate: { id: '{{payload._id}}' }, + samplePayload: { _id: '1' }, + }); + assert.equal(typeof preview.error, 'string'); + }); +}); diff --git a/src/lib/event-relays/preview.ts b/src/lib/event-relays/preview.ts new file mode 100644 index 000000000..c085cbced --- /dev/null +++ b/src/lib/event-relays/preview.ts @@ -0,0 +1,33 @@ +import { lookupOwnPath } from './path.ts'; +import { renderMessageTemplate } from './template.ts'; + +export type RelayPreview = { + resourceId?: string; + payload?: unknown; + error?: string; +}; + +export function previewEventRelay(options: { + resourceIdPath: string; + messageTemplate: unknown; + samplePayload: unknown; +}): RelayPreview { + try { + const resourceId = lookupOwnPath( + options.samplePayload, + options.resourceIdPath + ); + if (resourceId === undefined) { + return { + error: `Resource ID path '${options.resourceIdPath}' was not found`, + }; + } + const payload = renderMessageTemplate( + options.messageTemplate, + options.samplePayload + ); + return { resourceId: String(resourceId), payload }; + } catch (err) { + return { error: err instanceof Error ? err.message : String(err) }; + } +} diff --git a/src/lib/event-relays/template.ts b/src/lib/event-relays/template.ts new file mode 100644 index 000000000..23143bee0 --- /dev/null +++ b/src/lib/event-relays/template.ts @@ -0,0 +1,65 @@ +import { lookupOwnPath, parseDotPath } from './path.ts'; +import { MAX_TEMPLATE_BYTES } from './path.ts'; + +const PLACEHOLDER = /\{\{\s*payload\.([A-Za-z_][A-Za-z0-9_.]*)\s*\}\}/g; +const EXACT_PLACEHOLDER = /^\{\{\s*payload\.([A-Za-z_][A-Za-z0-9_.]*)\s*\}\}$/; +const MAX_DEPTH = 10; + +export function renderMessageTemplate( + template: unknown, + payload: unknown +): unknown { + const serialized = JSON.stringify(template); + if (!serialized || serialized.length > MAX_TEMPLATE_BYTES) { + throw new Error('Message template is invalid or too large'); + } + return renderValue(template, payload, 0); +} + +function renderValue(value: unknown, payload: unknown, depth: number): unknown { + if (depth > MAX_DEPTH) { + throw new Error('Message template is nested too deeply'); + } + if (typeof value === 'string') { + return interpolateString(value, payload); + } + if (Array.isArray(value)) { + return value.map(item => renderValue(item, payload, depth + 1)); + } + if (value !== null && typeof value === 'object') { + const output: Record = {}; + for (const key of Object.keys(value as Record)) { + output[key] = renderValue( + (value as Record)[key], + payload, + depth + 1 + ); + } + return output; + } + return value; +} + +function interpolateString(value: string, payload: unknown): unknown { + const exact = value.trim().match(EXACT_PLACEHOLDER); + if (exact) { + parseDotPath(exact[1]); + const resolved = lookupOwnPath(payload, exact[1]); + if (resolved === undefined) { + throw new Error(`Placeholder payload.${exact[1]} was not found`); + } + return resolved; + } + + return value.replace(PLACEHOLDER, (_match, path: string) => { + parseDotPath(path); + const resolved = lookupOwnPath(payload, path); + if (resolved === undefined) { + throw new Error(`Placeholder payload.${path} was not found`); + } + if (resolved === null || typeof resolved !== 'object') { + return String(resolved); + } + return JSON.stringify(resolved); + }); +} diff --git a/src/lib/models/Router.ts b/src/lib/models/Router.ts index 9360c8aa3..1aa55cfbb 100644 --- a/src/lib/models/Router.ts +++ b/src/lib/models/Router.ts @@ -86,3 +86,35 @@ export type UpdateSecurityClientRequest = { alias?: string; notes?: string; }; + +export type EventRelay = { + _id: string; + name: string; + notes?: string; + active: boolean; + busEvent: string; + socketEvent: string; + resourceType: string; + resourceIdPath: string; + permission: string; + messageTemplate: unknown; + createdAt: string; + updatedAt: string; +}; + +export type EventRelaysResponse = { + relays: EventRelay[]; + count: number; +}; + +export type EventRelayWriteRequest = { + name: string; + notes?: string; + active?: boolean; + busEvent: string; + socketEvent: string; + resourceType: string; + resourceIdPath: string; + permission: string; + messageTemplate: unknown; +}; From 5dd363c67d1cfdc47df33e768c544313ffcfaa6f Mon Sep 17 00:00:00 2001 From: John Chantzigoulas Date: Wed, 9 Sep 2026 11:05:17 +0300 Subject: [PATCH 2/6] feat(database): add opt-in live document updates in the data explorer Subscribe to MongoDB change streams through a short-lived admin socket ticket so operators can see pending updates without auto-refreshing the grid. --- .cursor/conduit-backend/docker-compose.yml | 26 +++- .cursor/conduit-backend/mongo/init-replica.sh | 20 +++ .cursor/conduit-backend/mongo/keyfile | 1 + next-env.d.ts | 2 +- package.json | 2 + pnpm-lock.yaml | 65 ++++++++ .../(modules)/database/settings/page.tsx | 21 ++- .../database/models/data-explorer/index.tsx | 24 ++- .../data-explorer/live-updates-status.tsx | 92 ++++++++++++ .../database/models/settings-panel/index.tsx | 62 +++++++- src/components/database/settings/settings.tsx | 15 +- .../database/settings/settingsForm.tsx | 69 ++++++++- src/components/database/settings/zod.ts | 5 + src/lib/api/database/index.ts | 8 + src/lib/api/realtime/ticket.ts | 28 ++++ src/lib/hooks/use-database-live-updates.ts | 139 ++++++++++++++++++ src/lib/logic/EnvManager.ts | 7 +- src/lib/models/database/schemas.ts | 6 + src/lib/models/database/settings.ts | 19 +++ src/lib/realtime/adminSocketUrl.test.ts | 33 +++++ src/lib/realtime/adminSocketUrl.ts | 21 +++ src/lib/realtime/change-events.test.ts | 44 ++++++ src/lib/realtime/change-events.ts | 72 +++++++++ 23 files changed, 762 insertions(+), 19 deletions(-) create mode 100644 .cursor/conduit-backend/mongo/init-replica.sh create mode 100644 .cursor/conduit-backend/mongo/keyfile create mode 100644 src/components/database/models/data-explorer/live-updates-status.tsx create mode 100644 src/lib/api/realtime/ticket.ts create mode 100644 src/lib/hooks/use-database-live-updates.ts create mode 100644 src/lib/realtime/adminSocketUrl.test.ts create mode 100644 src/lib/realtime/adminSocketUrl.ts create mode 100644 src/lib/realtime/change-events.test.ts create mode 100644 src/lib/realtime/change-events.ts diff --git a/.cursor/conduit-backend/docker-compose.yml b/.cursor/conduit-backend/docker-compose.yml index e7bf83366..6d2081e09 100644 --- a/.cursor/conduit-backend/docker-compose.yml +++ b/.cursor/conduit-backend/docker-compose.yml @@ -21,8 +21,32 @@ services: MONGO_INITDB_DATABASE: 'conduit' MONGO_INITDB_ROOT_USERNAME: 'conduit' MONGO_INITDB_ROOT_PASSWORD: 'pass' + entrypoint: + - bash + - -c + - | + cp /mongo-keyfile /tmp/keyfile + chmod 400 /tmp/keyfile + chown mongodb:mongodb /tmp/keyfile + exec docker-entrypoint.sh mongod --replSet rs0 --bind_ip_all --keyFile /tmp/keyfile volumes: - mongo:/data/db + - ./mongo/keyfile:/mongo-keyfile:ro + + mongo-init-replica: + container_name: 'conduit-mongo-init' + image: 'docker.io/library/mongo:4.4.15' + restart: on-failure + network_mode: 'host' + depends_on: + - mongodb + environment: + MONGO_HOST: '127.0.0.1' + MONGO_INITDB_ROOT_USERNAME: 'conduit' + MONGO_INITDB_ROOT_PASSWORD: 'pass' + volumes: + - ./mongo/init-replica.sh:/init-replica.sh:ro + command: ['bash', '/init-replica.sh'] redis: container_name: 'conduit-redis' @@ -47,7 +71,7 @@ services: CLIENT_HTTP_PORT: '3000' CLIENT_SOCKET_PORT: '3001' __DEFAULT_HOST_URL: '${ADMIN_DEFAULT_HOST_URL:-http://localhost:3030}' - DB_CONN_URI: 'mongodb://conduit:pass@127.0.0.1:27017/conduit?authSource=admin' + DB_CONN_URI: 'mongodb://conduit:pass@127.0.0.1:27017/conduit?authSource=admin&replicaSet=rs0' volumes: mongo: diff --git a/.cursor/conduit-backend/mongo/init-replica.sh b/.cursor/conduit-backend/mongo/init-replica.sh new file mode 100644 index 000000000..e4bb536d2 --- /dev/null +++ b/.cursor/conduit-backend/mongo/init-replica.sh @@ -0,0 +1,20 @@ +#!/bin/bash +set -euo pipefail +HOST="${MONGO_HOST:-conduit-mongo}" +USER="${MONGO_INITDB_ROOT_USERNAME:-conduit}" +PASS="${MONGO_INITDB_ROOT_PASSWORD:-pass}" + +until mongo --host "$HOST" -u "$USER" -p "$PASS" --authenticationDatabase admin --quiet --eval 'db.adminCommand({ ping: 1 })' >/dev/null 2>&1; do + sleep 2 +done + +mongo --host "$HOST" -u "$USER" -p "$PASS" --authenticationDatabase admin --quiet --eval ' + try { + rs.status(); + } catch (err) { + rs.initiate({ + _id: "rs0", + members: [{ _id: 0, host: "'"$HOST"':27017" }] + }); + } +' diff --git a/.cursor/conduit-backend/mongo/keyfile b/.cursor/conduit-backend/mongo/keyfile new file mode 100644 index 000000000..8d7ffaf78 --- /dev/null +++ b/.cursor/conduit-backend/mongo/keyfile @@ -0,0 +1 @@ +conduit-dev-mongo-replica-keyfile-do-not-use-in-production-replace-me-before-any-real-deployment-0123456789abcdefghijklmnopqrstuvwxyz diff --git a/next-env.d.ts b/next-env.d.ts index c4b7818fb..9edff1c7c 100644 --- a/next-env.d.ts +++ b/next-env.d.ts @@ -1,6 +1,6 @@ /// /// -import "./.next/dev/types/routes.d.ts"; +import "./.next/types/routes.d.ts"; // NOTE: This file should not be edited // see https://nextjs.org/docs/app/api-reference/config/typescript for more information. diff --git a/package.json b/package.json index 32e45a065..033b1c920 100644 --- a/package.json +++ b/package.json @@ -10,6 +10,7 @@ "lint": "eslint .", "test:semantic-colors": "node --experimental-strip-types --test src/lib/semantic-colors.test.ts src/lib/reactflow-edge-colors.test.ts", "test:event-relays": "node --experimental-strip-types --test src/lib/event-relays/preview.test.ts", + "test:realtime": "node --experimental-strip-types --test src/lib/realtime/adminSocketUrl.test.ts src/lib/realtime/change-events.test.ts", "build:docker": "docker build --platform linux/amd64 -t ghcr.io/conduitplatform/conduit-ui:latest .", "prepare": "husky", "release": "standard-version" @@ -80,6 +81,7 @@ "recharts": "^3.8.1", "rehype-prism-plus": "^2.0.2", "server-only": "^0.0.1", + "socket.io-client": "^4.8.3", "tailwind-merge": "^3.4.0", "tailwindcss": "^4.2.2", "tailwindcss-animate": "^1.0.7", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 44a6c2ade..6f27a7453 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -198,6 +198,9 @@ importers: server-only: specifier: ^0.0.1 version: 0.0.1 + socket.io-client: + specifier: ^4.8.3 + version: 4.8.3 tailwind-merge: specifier: ^3.4.0 version: 3.5.0 @@ -4603,6 +4606,12 @@ packages: integrity: sha512-L18DaJsXSUk2+42pv8mLs5jJT2hqFkFE4j21wOmgbUqsZ2hL72NsUU785g9RXgo3s0ZNgVl42TiHp3ZtOv/Vyg==, } + engine.io-client@6.6.6: + resolution: + { + integrity: sha512-iY6QdftLQ9pyiPoX082bpf/u1UewnOaJrtJIF9T0++QB34lZrj0uP+Q/bj8AlUsAxqhnkTV2BS8SBZSxOmoV5Q==, + } + engine.io-parser@5.2.3: resolution: { @@ -8119,6 +8128,13 @@ packages: integrity: sha512-DkkO/dz7MGln0dHn5bmN3pPy+JmywNICWrJqVWiVOyvXjWQFIv9c2h24JrQLLFJ2aQVQf/Cvl1vblnd4r2apLQ==, } + socket.io-client@4.8.3: + resolution: + { + integrity: sha512-uP0bpjWrjQmUt5DTHq9RuoCBdFJF10cdX9X+a368j/Ft0wmaVgxlrjvK3kjvgCODOMMOz9lcaRzxmso0bTWZ/g==, + } + engines: { node: '>=10.0.0' } + socket.io-parser@4.2.6: resolution: { @@ -8971,6 +8987,21 @@ packages: utf-8-validate: optional: true + ws@8.21.3: + resolution: + { + integrity: sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==, + } + engines: { node: '>=10.0.0' } + peerDependencies: + bufferutil: ^4.0.1 + utf-8-validate: '>=5.0.2' + peerDependenciesMeta: + bufferutil: + optional: true + utf-8-validate: + optional: true + wsl-utils@0.1.0: resolution: { @@ -8978,6 +9009,13 @@ packages: } engines: { node: '>=18' } + xmlhttprequest-ssl@2.1.2: + resolution: + { + integrity: sha512-TEU+nJVUUnA4CYJFLvK5X9AOeH4KvDvhIfm0vV1GaQRtchnG0hgK5p8hw/xjv8cunWYCsiPCSDzObPyhEwq3KQ==, + } + engines: { node: '>=0.4.0' } + xtend@4.0.2: resolution: { @@ -11989,6 +12027,18 @@ snapshots: emoji-regex@9.2.2: {} + engine.io-client@6.6.6: + dependencies: + '@socket.io/component-emitter': 3.1.2 + debug: 4.4.3 + engine.io-parser: 5.2.3 + ws: 8.21.3 + xmlhttprequest-ssl: 2.1.2 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + engine.io-parser@5.2.3: {} engine.io@6.6.6: @@ -14371,6 +14421,17 @@ snapshots: - supports-color - utf-8-validate + socket.io-client@4.8.3: + dependencies: + '@socket.io/component-emitter': 3.1.2 + debug: 4.4.3 + engine.io-client: 6.6.6 + socket.io-parser: 4.2.6 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + socket.io-parser@4.2.6: dependencies: '@socket.io/component-emitter': 3.1.2 @@ -14936,10 +14997,14 @@ snapshots: ws@8.20.0: {} + ws@8.21.3: {} + wsl-utils@0.1.0: dependencies: is-wsl: 3.1.0 + xmlhttprequest-ssl@2.1.2: {} + xtend@4.0.2: {} y18n@5.0.8: {} diff --git a/src/app/(dashboard)/(modules)/database/settings/page.tsx b/src/app/(dashboard)/(modules)/database/settings/page.tsx index 20308cca0..7b8b50c19 100644 --- a/src/app/(dashboard)/(modules)/database/settings/page.tsx +++ b/src/app/(dashboard)/(modules)/database/settings/page.tsx @@ -1,14 +1,19 @@ import { Settings } from '@/components/database/settings/settings'; -import { getDatabaseSettings, getDatabaseType } from '@/lib/api/database'; +import { + getDatabaseRealtimeStatus, + getDatabaseSettings, + getDatabaseType, +} from '@/lib/api/database'; export default async function DatabaseSettingsPage() { - const [settingsResult, typeResult] = await Promise.allSettled([ - getDatabaseSettings(), - getDatabaseType(), - ]); + const [settingsResult, typeResult, realtimeResult] = await Promise.allSettled( + [getDatabaseSettings(), getDatabaseType(), getDatabaseRealtimeStatus()] + ); const databaseType = typeResult.status === 'fulfilled' ? typeResult.value.result : 'Unknown'; + const realtimeStatus = + realtimeResult.status === 'fulfilled' ? realtimeResult.value : null; if (settingsResult.status === 'rejected') { return ( @@ -25,6 +30,10 @@ export default async function DatabaseSettingsPage() { } return ( - + ); } diff --git a/src/components/database/models/data-explorer/index.tsx b/src/components/database/models/data-explorer/index.tsx index f87754449..59ceb830d 100644 --- a/src/components/database/models/data-explorer/index.tsx +++ b/src/components/database/models/data-explorer/index.tsx @@ -31,6 +31,8 @@ import { import { Badge } from '@/components/ui/badge'; import { toast } from '@/lib/hooks/use-toast'; import { deleteSchemaDocument } from '@/lib/api/database'; +import { useDatabaseLiveUpdates } from '@/lib/hooks/use-database-live-updates'; +import { LiveUpdatesStatus } from './live-updates-status'; import { analyzeModelDataPermissions, getExtensionFieldNames, @@ -74,6 +76,11 @@ export function DataExplorer({ schema, documents }: DataExplorerProps) { () => analyzeModelDataPermissions(schema), [schema] ); + const liveEnabled = schema.modelOptions?.conduit?.realtime?.enabled === true; + const liveUpdates = useDatabaseLiveUpdates({ + schemaName: schema.name, + enabled: liveEnabled, + }); // Get schema fields for column definitions const schemaFields = React.useMemo(() => { @@ -160,6 +167,7 @@ export function DataExplorer({ schema, documents }: DataExplorerProps) { }; const handleRefresh = () => { + liveUpdates.consumePendingUpdates(); router.refresh(); toast({ title: 'Data refreshed' }); }; @@ -185,6 +193,7 @@ export function DataExplorer({ schema, documents }: DataExplorerProps) { const handleDocumentSaved = () => { setSelectedDocument(null); setIsCreatingNew(false); + liveUpdates.consumePendingUpdates(); router.refresh(); }; @@ -372,9 +381,22 @@ export function DataExplorer({ schema, documents }: DataExplorerProps) { + + {/* Refresh */} - {/* Create New */} diff --git a/src/components/database/models/data-explorer/live-updates-status.tsx b/src/components/database/models/data-explorer/live-updates-status.tsx new file mode 100644 index 000000000..19a2095c5 --- /dev/null +++ b/src/components/database/models/data-explorer/live-updates-status.tsx @@ -0,0 +1,92 @@ +'use client'; + +import * as React from 'react'; +import { Button } from '@/components/ui/button'; +import { + Tooltip, + TooltipContent, + TooltipProvider, + TooltipTrigger, +} from '@/components/ui/tooltip'; +import { cn } from '@/lib/utils'; +import type { LiveConnectionState } from '@/lib/hooks/use-database-live-updates'; + +type LiveUpdatesStatusProps = { + connectionState: LiveConnectionState; + pendingUpdates: number; + errorMessage?: string | null; + onApplyUpdates: () => void; +}; + +const STATE_LABEL: Record, string> = { + connecting: 'Connecting', + live: 'Live', + reconnecting: 'Reconnecting', + error: 'Unavailable', +}; + +export function LiveUpdatesStatus({ + connectionState, + pendingUpdates, + errorMessage, + onApplyUpdates, +}: LiveUpdatesStatusProps) { + if (connectionState === 'idle') return null; + + const statusHint = + connectionState === 'error' + ? errorMessage || 'Live updates are unavailable' + : connectionState === 'live' + ? 'Listening for document changes. The table does not refresh until you apply updates.' + : 'Connecting to live updates.'; + + return ( + +
+ + + + + {STATE_LABEL[connectionState]} + + + {statusHint} + + + {pendingUpdates > 0 && ( + + + + + + Refresh the current filtered page. Open documents stay as they + are. + + + )} +
+ + ); +} diff --git a/src/components/database/models/settings-panel/index.tsx b/src/components/database/models/settings-panel/index.tsx index aeed051f9..ddf158b32 100644 --- a/src/components/database/models/settings-panel/index.tsx +++ b/src/components/database/models/settings-panel/index.tsx @@ -25,6 +25,7 @@ import { } from '@/components/ui/card'; import { ScrollArea } from '@/components/ui/scroll-area'; import { Badge } from '@/components/ui/badge'; +import { Switch } from '@/components/ui/switch'; import { Label } from '@/components/ui/label'; import { Alert, AlertDescription } from '@/components/ui/alert'; import { @@ -45,7 +46,7 @@ import { AlertDialogTitle, AlertDialogTrigger, } from '@/components/ui/alert-dialog'; -import { Save, Trash2, AlertTriangle, Info } from 'lucide-react'; +import { Save, Trash2, AlertTriangle, Info, Radio } from 'lucide-react'; import { toast } from '@/lib/hooks/use-toast'; /** Radix Select rejects empty string values */ @@ -141,6 +142,12 @@ export function SettingsPanel({ const [initialReadPreference, setInitialReadPreference] = React.useState(() => getSchemaReadPreference(schema) ); + const [realtimeEnabled, setRealtimeEnabled] = React.useState( + schema.modelOptions?.conduit?.realtime?.enabled ?? false + ); + const [initialRealtimeEnabled, setInitialRealtimeEnabled] = React.useState( + schema.modelOptions?.conduit?.realtime?.enabled ?? false + ); const isOwnedByDatabase = !schema.ownerModule || schema.ownerModule === 'database'; @@ -150,6 +157,8 @@ export function SettingsPanel({ const nextAuthEnabled = schema.modelOptions?.conduit?.authorization?.enabled ?? false; const nextReadPreference = getSchemaReadPreference(schema); + const nextRealtimeEnabled = + schema.modelOptions?.conduit?.realtime?.enabled ?? false; setCrudOperations(nextCrudOperations); setInitialCrudOperations(nextCrudOperations); @@ -157,6 +166,8 @@ export function SettingsPanel({ setInitialAuthEnabled(nextAuthEnabled); setMongoReadPreference(nextReadPreference); setInitialReadPreference(nextReadPreference); + setRealtimeEnabled(nextRealtimeEnabled); + setInitialRealtimeEnabled(nextRealtimeEnabled); }, [schema]); const loadIndexes = React.useCallback(async () => { @@ -199,6 +210,7 @@ export function SettingsPanel({ JSON.stringify(crudOperations) !== JSON.stringify(initialCrudOperations) || authEnabled !== initialAuthEnabled || mongoReadPreference !== initialReadPreference || + realtimeEnabled !== initialRealtimeEnabled || JSON.stringify(indices.map(getIndexSignature)) !== JSON.stringify(initialIndices.map(getIndexSignature)); @@ -243,6 +255,9 @@ export function SettingsPanel({ }, ...(databaseType === 'MongoDB' ? { + realtime: { + enabled: realtimeEnabled, + }, readPreference: mongoReadPreference !== MONGO_READ_PREF_DEFAULT ? mongoReadPreference @@ -256,6 +271,7 @@ export function SettingsPanel({ setInitialCrudOperations(crudOperations); setInitialAuthEnabled(authEnabled); setInitialReadPreference(mongoReadPreference); + setInitialRealtimeEnabled(realtimeEnabled); toast({ title: 'Settings saved' }); onSave(); } catch (error: any) { @@ -273,6 +289,7 @@ export function SettingsPanel({ loadIndexes, mongoReadPreference, onSave, + realtimeEnabled, schema._id, syncIndexes, ]); @@ -406,6 +423,49 @@ export function SettingsPanel({

+ {databaseType === 'MongoDB' && ( + + + + + Live updates + + + Opt this schema into MongoDB change streams. The Data Explorer + shows an updates-available badge instead of auto-refreshing. + + + +
+
+ +

+ Requires the module-level live updates switch and a + replica set. +

+
+ +
+ {realtimeEnabled && crudOperations.read?.enabled !== true && ( +

+ CMS read is off, so client apps cannot subscribe. Admin Data + Explorer can still listen. +

+ )} + + + )} + {databaseType === 'MongoDB' && ( diff --git a/src/components/database/settings/settings.tsx b/src/components/database/settings/settings.tsx index 85596ca5d..f7e726428 100644 --- a/src/components/database/settings/settings.tsx +++ b/src/components/database/settings/settings.tsx @@ -1,6 +1,6 @@ 'use client'; import { z } from 'zod'; -import { DatabaseConfig } from '@/lib/models/database'; +import { DatabaseConfig, DatabaseRealtimeStatus } from '@/lib/models/database'; import { useState } from 'react'; import { useForm } from 'react-hook-form'; import { rhfZodResolver } from '@/lib/zod-form'; @@ -13,14 +13,18 @@ import { useSettingsSave } from '@/lib/hooks/use-settings-save'; interface Props { data: DatabaseConfig; databaseType: string; + realtimeStatus?: DatabaseRealtimeStatus | null; } -export const Settings = ({ data, databaseType }: Props) => { +export const Settings = ({ data, databaseType, realtimeStatus }: Props) => { const [edit, setEdit] = useState(false); const { save, isSaving } = useSettingsSave('Database'); const form = useForm>({ resolver: rhfZodResolver(DatabaseSettingsSchema), - defaultValues: data, + defaultValues: { + ...data, + realtime: { enabled: data.realtime?.enabled ?? false }, + }, }); const { reset, control, handleSubmit } = form; @@ -37,10 +41,10 @@ export const Settings = ({ data, databaseType }: Props) => {
-

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 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/hooks/use-database-live-updates.ts b/src/lib/hooks/use-database-live-updates.ts new file mode 100644 index 000000000..15b5d92e0 --- /dev/null +++ b/src/lib/hooks/use-database-live-updates.ts @@ -0,0 +1,139 @@ +'use client'; + +import * as React from 'react'; +import { io, type Socket } from 'socket.io-client'; +import { issueAdminRealtimeTicket } from '@/lib/api/realtime/ticket'; +import { + parseDatabaseChangeEvent, + 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 }); + }); + + socket.on('change', (data: unknown) => { + const event = parseDatabaseChangeEvent(data); + if (!event || !shouldCountChange(event, schemaName)) 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', () => { + 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..1c9cf8e2a --- /dev/null +++ b/src/lib/realtime/adminSocketUrl.test.ts @@ -0,0 +1,33 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import { deriveAdminSocketUrl } 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' + ); + }); +}); diff --git a/src/lib/realtime/adminSocketUrl.ts b/src/lib/realtime/adminSocketUrl.ts new file mode 100644 index 000000000..d6d68719e --- /dev/null +++ b/src/lib/realtime/adminSocketUrl.ts @@ -0,0 +1,21 @@ +export function deriveAdminSocketUrl( + apiBaseUrl: string, + explicit?: string | null +): string { + if (explicit && explicit.trim() !== '') { + return trimTrailingSlash(explicit.trim()); + } + + const url = new URL(apiBaseUrl); + if (url.port === '3030') { + url.port = '3031'; + } + 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..d1ca73926 --- /dev/null +++ b/src/lib/realtime/change-events.test.ts @@ -0,0 +1,44 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import { + parseDatabaseChangeEvent, + 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); + }); +}); diff --git a/src/lib/realtime/change-events.ts b/src/lib/realtime/change-events.ts new file mode 100644 index 000000000..715cbdd57 --- /dev/null +++ b/src/lib/realtime/change-events.ts @@ -0,0 +1,72 @@ +export const DATABASE_CHANGE_EVENT_VERSION = 1; + +export 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; +} + +function safeParseJson(value: string): unknown { + try { + return JSON.parse(value); + } catch { + return null; + } +} From 707a7e71640c60b5ac3685489cc2cbe56dc4378a Mon Sep 17 00:00:00 2001 From: John Chantzigoulas Date: Wed, 9 Sep 2026 13:02:54 +0300 Subject: [PATCH 3/6] fix(database): tolerate raw array from schema indexes API The Database module returns the index list itself, so reading result.indexes threw on the model settings tab. --- .../database/models/settings-panel/index.tsx | 4 +- src/lib/api/database/index.ts | 6 +- src/lib/database/schema-indexes.test.ts | 38 +++++++++++++ src/lib/database/schema-indexes.ts | 56 +++++++++++++++++++ 4 files changed, 100 insertions(+), 4 deletions(-) create mode 100644 src/lib/database/schema-indexes.test.ts create mode 100644 src/lib/database/schema-indexes.ts diff --git a/src/components/database/models/settings-panel/index.tsx b/src/components/database/models/settings-panel/index.tsx index ddf158b32..35d3d3b94 100644 --- a/src/components/database/models/settings-panel/index.tsx +++ b/src/components/database/models/settings-panel/index.tsx @@ -99,7 +99,7 @@ function getIndexSignature(index: SchemaIndex) { } function isPrimaryIndex(index: SchemaIndex) { - return index.fields.length === 1 && index.fields[0] === '_id'; + return index.fields?.length === 1 && index.fields[0] === '_id'; } function getSchemaReadPreference(schema: DeclaredSchema) { @@ -174,7 +174,7 @@ export function SettingsPanel({ setIsLoadingIndices(true); try { const result = await getSchemaIndexes(schema._id); - const customIndexes = (result.indexes as SchemaIndex[]).filter( + const customIndexes = result.indexes.filter( index => !isPrimaryIndex(index) ); setIndices(customIndexes); diff --git a/src/lib/api/database/index.ts b/src/lib/api/database/index.ts index 59e65ad5a..effca7244 100644 --- a/src/lib/api/database/index.ts +++ b/src/lib/api/database/index.ts @@ -11,6 +11,7 @@ import { SchemaOptions, } from '@/lib/models/database'; import { CustomEndpoint } from '@/lib/models/database/custom-endpoints'; +import { normalizeSchemaIndexResponse } from '@/lib/database/schema-indexes'; export const getPendingSchemas = async (args: { skip?: number; @@ -342,9 +343,10 @@ export const createSchemaIndexes = async ( }; export const getSchemaIndexes = async (schemaId: string) => { - 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 ( 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); +} From 87c2d7b0fb0570e4f7ae3aac692265aecb2f53c8 Mon Sep 17 00:00:00 2001 From: John Chantzigoulas Date: Wed, 9 Sep 2026 15:06:02 +0300 Subject: [PATCH 4/6] feat(database): add copyable live-update connection snippets Give schema settings JS, Postman, and Admin snippets so developers can subscribe to change events without guessing the handshake. Deduplicate resume tokens so reconnects do not inflate the pending-update count. --- apps/.DS_Store | Bin 6148 -> 0 bytes next-env.d.ts | 2 +- package.json | 2 +- .../database/models/settings-panel/index.tsx | 7 + .../settings-panel/live-update-snippets.tsx | 321 ++++++++++++++++++ src/lib/hooks/use-database-live-updates.ts | 8 +- src/lib/realtime/adminSocketUrl.test.ts | 28 +- src/lib/realtime/adminSocketUrl.ts | 25 +- src/lib/realtime/change-events.test.ts | 13 + src/lib/realtime/change-events.ts | 17 + src/lib/realtime/connection-snippets.test.ts | 70 ++++ src/lib/realtime/connection-snippets.ts | 116 +++++++ 12 files changed, 603 insertions(+), 6 deletions(-) delete mode 100644 apps/.DS_Store create mode 100644 src/components/database/models/settings-panel/live-update-snippets.tsx create mode 100644 src/lib/realtime/connection-snippets.test.ts create mode 100644 src/lib/realtime/connection-snippets.ts diff --git a/apps/.DS_Store b/apps/.DS_Store deleted file mode 100644 index 7d5804ed1d402a1b22a88611db7e85d7dbf0636b..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 6148 zcmeHKyG{c^3>-s>2%40X`vVmG!73yLbwp1Q2_d12gh=YE_%1#zrzGv9fE4&w3fOG5SuOcW)mtYo=e@SkpXgrmNq6HqC=Ah# kiP4U^@pgO@MOoK;>M3CI+4PpcC~o;JV19z+WqH1SGK(umAu6 diff --git a/next-env.d.ts b/next-env.d.ts index 9edff1c7c..c4b7818fb 100644 --- a/next-env.d.ts +++ b/next-env.d.ts @@ -1,6 +1,6 @@ /// /// -import "./.next/types/routes.d.ts"; +import "./.next/dev/types/routes.d.ts"; // NOTE: This file should not be edited // see https://nextjs.org/docs/app/api-reference/config/typescript for more information. diff --git a/package.json b/package.json index 033b1c920..33da150ef 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,7 @@ "lint": "eslint .", "test:semantic-colors": "node --experimental-strip-types --test src/lib/semantic-colors.test.ts src/lib/reactflow-edge-colors.test.ts", "test:event-relays": "node --experimental-strip-types --test src/lib/event-relays/preview.test.ts", - "test:realtime": "node --experimental-strip-types --test src/lib/realtime/adminSocketUrl.test.ts src/lib/realtime/change-events.test.ts", + "test:realtime": "node --experimental-strip-types --test src/lib/realtime/adminSocketUrl.test.ts src/lib/realtime/change-events.test.ts src/lib/realtime/connection-snippets.test.ts", "build:docker": "docker build --platform linux/amd64 -t ghcr.io/conduitplatform/conduit-ui:latest .", "prepare": "husky", "release": "standard-version" diff --git a/src/components/database/models/settings-panel/index.tsx b/src/components/database/models/settings-panel/index.tsx index 35d3d3b94..a1fd7f0fb 100644 --- a/src/components/database/models/settings-panel/index.tsx +++ b/src/components/database/models/settings-panel/index.tsx @@ -15,6 +15,7 @@ import { CrudPermissions } from './crud-permissions'; import { deriveCrudOperationsFromSchema } from './crud-state'; import { AuthSettings } from './auth-settings'; import { IndicesConfig } from './indices-config'; +import { LiveUpdateSnippets } from './live-update-snippets'; import { Button } from '@/components/ui/button'; import { Card, @@ -462,6 +463,12 @@ export function SettingsPanel({ Explorer can still listen.

)} + )} diff --git a/src/components/database/models/settings-panel/live-update-snippets.tsx b/src/components/database/models/settings-panel/live-update-snippets.tsx new file mode 100644 index 000000000..c1d6556a5 --- /dev/null +++ b/src/components/database/models/settings-panel/live-update-snippets.tsx @@ -0,0 +1,321 @@ +'use client'; + +import * as React from 'react'; +import { Check, Copy } from 'lucide-react'; +import { Button } from '@/components/ui/button'; +import { Tabs, TabsContent, TabsList, TabsTrigger } from '@/components/ui/tabs'; +import { + Tooltip, + TooltipContent, + TooltipProvider, + TooltipTrigger, +} from '@/components/ui/tooltip'; +import { toast } from '@/lib/hooks/use-toast'; +import { getAdminSettings } from '@/lib/api/settings'; +import { getRouterSettings } from '@/lib/api/router'; +import { + deriveAdminSocketUrl, + deriveClientSocketUrl, +} from '@/lib/realtime/adminSocketUrl'; +import { + DATABASE_SOCKET_LISTEN_EVENTS, + DATABASE_SOCKET_PATH, + buildAdminJavascriptSnippet, + buildClientJavascriptSnippet, + buildPostmanConnection, + socketEndpoint, +} from '@/lib/realtime/connection-snippets'; + +const DEFAULT_ADMIN_SOCKET = 'http://localhost:3031'; +const DEFAULT_CLIENT_SOCKET = 'http://localhost:3001'; + +type LiveUpdateSnippetsProps = { + schemaName: string; + realtimeEnabled: boolean; + cmsReadEnabled: boolean; + authorizationEnabled: boolean; +}; + +export function LiveUpdateSnippets({ + schemaName, + realtimeEnabled, + cmsReadEnabled, + authorizationEnabled, +}: LiveUpdateSnippetsProps) { + const [adminSocketUrl, setAdminSocketUrl] = + React.useState(DEFAULT_ADMIN_SOCKET); + const [clientSocketUrl, setClientSocketUrl] = React.useState( + DEFAULT_CLIENT_SOCKET + ); + + React.useEffect(() => { + let cancelled = false; + void Promise.all([getAdminSettings(), getRouterSettings()]) + .then(([admin, router]) => { + if (cancelled) return; + setAdminSocketUrl( + safeSocketUrl( + admin?.config?.hostUrl, + deriveAdminSocketUrl, + DEFAULT_ADMIN_SOCKET + ) + ); + setClientSocketUrl( + safeSocketUrl( + router?.config?.hostUrl, + deriveClientSocketUrl, + DEFAULT_CLIENT_SOCKET + ) + ); + }) + .catch(() => { + if (cancelled) return; + setAdminSocketUrl(DEFAULT_ADMIN_SOCKET); + setClientSocketUrl(DEFAULT_CLIENT_SOCKET); + }); + return () => { + cancelled = true; + }; + }, []); + + const clientSnippet = buildClientJavascriptSnippet({ + socketUrl: clientSocketUrl, + schemaName, + documentScoped: authorizationEnabled, + }); + const adminSnippet = buildAdminJavascriptSnippet({ + socketUrl: adminSocketUrl, + schemaName, + }); + const postman = buildPostmanConnection({ + adminSocketUrl, + schemaName, + }); + + return ( + +
+
+

Connect to change events

+

+ Subscribe this schema on the{' '} + /database/ Socket.IO namespace. + Events are metadata only — no document body. +

+
+ + {!realtimeEnabled && ( +

+ Enable live updates and save before this socket will receive events. +

+ )} + +
+ + + + +
+ + + + + JavaScript + + + Postman + + + Admin + + + + + {!cmsReadEnabled && ( +

+ CMS read is off, so client apps cannot subscribe. Turn on CRUD + read, or use the Admin tab. +

+ )} + {authorizationEnabled && ( +

+ Authorization is on — client subscribe must include a document + the user can read. +

+ )} + + + + +
    +
  1. + New → Socket.IO Request. Postman has no Handshake Auth field — + put the token on Headers instead. +
    2. +
  2. + +
    3. +
  3. + Settings: handshake path{' '} + {postman.path}, client + version {postman.clientVersion}. +
    4. +
  4. + Headers: +
    + {postman.headers.map(header => ( + + ))} +
    +
    5. +
  5. + Listen for{' '} + {postman.listenFor.map(event => ( + + {event} + + ))} + , click Connect, then emit: +
    6. +
+ +

+ Event name must be exactly{' '} + {postman.subscribeEvent}, + format JSON. Run Admin login first so{' '} + adminToken is set. +

+ + + +

+ Same handshake as Data Explorer. Use an admin JWT, or a 30-second{' '} + POST /realtime/ticket token in{' '} + auth.token. +

+ + + +
+ + ); +} + +function CopyRow({ + label, + value, + display, +}: { + label: string; + value: string; + display?: string; +}) { + return ( +
+
+

{label}

+

+ {display ?? value} +

+
+ +
+ ); +} + +function SnippetBlock({ + label, + value, + language, +}: { + label: string; + value: string; + language: 'javascript' | 'json'; +}) { + return ( +
+
+

{label}

+ +
+ 
+        {value}
+
+
+ ); +} + +function CopyButton({ value, label }: { value: string; label: string }) { + const [copied, setCopied] = React.useState(false); + + const handleCopy = async () => { + try { + await navigator.clipboard.writeText(value); + setCopied(true); + toast({ title: `${label} copied` }); + window.setTimeout(() => setCopied(false), 1500); + } catch { + toast({ + title: 'Copy failed', + variant: 'destructive', + }); + } + }; + + return ( + + + + + +

{copied ? 'Copied' : `Copy ${label}`}

+ + + ); +} + +function safeSocketUrl( + hostUrl: unknown, + derive: (url: string) => string, + fallback: string +): string { + if (typeof hostUrl !== 'string' || hostUrl.trim() === '') return fallback; + try { + return derive(hostUrl); + } catch { + return fallback; + } +} diff --git a/src/lib/hooks/use-database-live-updates.ts b/src/lib/hooks/use-database-live-updates.ts index 15b5d92e0..a1fd0489d 100644 --- a/src/lib/hooks/use-database-live-updates.ts +++ b/src/lib/hooks/use-database-live-updates.ts @@ -5,6 +5,7 @@ import { io, type Socket } from 'socket.io-client'; import { issueAdminRealtimeTicket } from '@/lib/api/realtime/ticket'; import { parseDatabaseChangeEvent, + rememberResumeToken, shouldCountChange, } from '@/lib/realtime/change-events'; @@ -71,9 +72,11 @@ export function useDatabaseLiveUpdates(options: { 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); }); @@ -89,7 +92,10 @@ export function useDatabaseLiveUpdates(options: { setConnectionState('error'); }); - socket.on('connect_error', () => { + socket.on('connect_error', err => { + setErrorMessage( + err instanceof Error ? err.message : 'Live updates failed' + ); scheduleReconnect(); }); diff --git a/src/lib/realtime/adminSocketUrl.test.ts b/src/lib/realtime/adminSocketUrl.test.ts index 1c9cf8e2a..e74d44f47 100644 --- a/src/lib/realtime/adminSocketUrl.test.ts +++ b/src/lib/realtime/adminSocketUrl.test.ts @@ -1,6 +1,9 @@ import assert from 'node:assert/strict'; import { describe, it } from 'node:test'; -import { deriveAdminSocketUrl } from './adminSocketUrl.ts'; +import { + deriveAdminSocketUrl, + deriveClientSocketUrl, +} from './adminSocketUrl.ts'; describe('deriveAdminSocketUrl', () => { it('uses an explicit socket URL when provided', () => { @@ -31,3 +34,26 @@ describe('deriveAdminSocketUrl', () => { ); }); }); + +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 index d6d68719e..a6a509eb1 100644 --- a/src/lib/realtime/adminSocketUrl.ts +++ b/src/lib/realtime/adminSocketUrl.ts @@ -1,14 +1,35 @@ 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); - if (url.port === '3030') { - url.port = '3031'; + const mapped = restToSocketPort[url.port]; + if (mapped) { + url.port = mapped; } url.pathname = ''; url.search = ''; diff --git a/src/lib/realtime/change-events.test.ts b/src/lib/realtime/change-events.test.ts index d1ca73926..004af6436 100644 --- a/src/lib/realtime/change-events.test.ts +++ b/src/lib/realtime/change-events.test.ts @@ -2,6 +2,7 @@ import assert from 'node:assert/strict'; import { describe, it } from 'node:test'; import { parseDatabaseChangeEvent, + rememberResumeToken, shouldCountChange, } from './change-events.ts'; @@ -42,3 +43,15 @@ describe('shouldCountChange', () => { 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 index 715cbdd57..df485879e 100644 --- a/src/lib/realtime/change-events.ts +++ b/src/lib/realtime/change-events.ts @@ -63,6 +63,23 @@ export function shouldCountChange( 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); 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..fdeb03480 --- /dev/null +++ b/src/lib/realtime/connection-snippets.ts @@ -0,0 +1,116 @@ +export const DATABASE_SOCKET_NAMESPACE = '/database/'; +export const DATABASE_SOCKET_PATH = '/realtime'; +export const DATABASE_SOCKET_LISTEN_EVENTS = [ + 'change', + 'connected', + 'conduit_error', +] as const; + +export 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); +}); +`; +} + +export 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 + ), + }; +} From 2295bc7010fe1825de04206c75f88262f1e030af Mon Sep 17 00:00:00 2001 From: John Chantzigoulas Date: Thu, 10 Sep 2026 14:54:28 +0300 Subject: [PATCH 5/6] feat(router): document when and how to use event relays Give operators an in-page guide so they can tell relays from queues and wire a bus event to permission-scoped socket subscribers. --- .../router/event-relays/event-relay-docs.tsx | 230 ++++++++++++++++++ .../router/event-relays/event-relay-form.tsx | 2 +- .../router/event-relays/event-relay-list.tsx | 51 +++- 3 files changed, 279 insertions(+), 4 deletions(-) create mode 100644 src/components/router/event-relays/event-relay-docs.tsx diff --git a/src/components/router/event-relays/event-relay-docs.tsx b/src/components/router/event-relays/event-relay-docs.tsx new file mode 100644 index 000000000..a1c3ed895 --- /dev/null +++ b/src/components/router/event-relays/event-relay-docs.tsx @@ -0,0 +1,230 @@ +'use client'; + +import { ChevronDown } from 'lucide-react'; +import { Card } from '@/components/ui/card'; +import { + Collapsible, + CollapsibleContent, + CollapsibleTrigger, +} from '@/components/ui/collapsible'; +import { cn } from '@/lib/utils'; + +const CLIENT_SNIPPET = `const socket = io(\`\${SOCKET_URL}/events/\`, { + path: '/realtime', + extraHeaders: { authorization: \`Bearer \${accessToken}\` }, +}); +socket.emit('subscribe', relayId, resourceId); +socket.on('order-updated', payload => {}); +socket.emit('unsubscribe', relayId, resourceId);`; + +const STEPS = [ + { + title: 'Bus event', + body: 'A module publishes JSON on an exact Redis channel — for example database realtime or a custom module.', + }, + { + title: 'Active relay', + body: 'The matching relay reads the resource id from the payload and renders the message template.', + }, + { + title: 'ReBAC subscribe', + body: 'Clients never pick a room name. Subscribe succeeds only if the user has the relay permission on that resource.', + }, + { + title: 'Socket emit', + body: 'Router emits socketEvent to the hashed /events/ room. Delivery is ephemeral — missed events are gone.', + }, +] as const; + +const SCOPE = [ + { + title: 'Use when', + body: 'You already publish JSON on an exact Redis bus channel and need live, per-resource UI updates with a ReBAC check.', + }, + { + title: 'Skip when', + body: 'You need replay, history, guaranteed delivery, wildcard channels, or a broadcast with no permission check. Relays are not a queue.', + }, + { + title: 'Requires', + body: 'Router sockets enabled, the Authorization module available, and something publishing on that exact channel.', + }, +] as const; + +interface EventRelayDocsProps { + open: boolean; + onOpenChange: (open: boolean) => void; +} + +export function EventRelayDocs({ open, onOpenChange }: EventRelayDocsProps) { + return ( + + + svg]:rotate-180' + )} + > + + + How Event Relays work + + + Forward an exact bus event to permission-scoped socket + subscribers. Not a queue, and not a generic websocket broadcast. + + + + + +
+
+

Scope

+
+ {SCOPE.map(item => ( +
+

+ {item.title} +

+

+ {item.body} +

+
+ ))} +
+
+ +
+

+ How a message moves +

+
    + {STEPS.map((step, index) => ( +
  1. + + {index + 1} + +
    +

    + {step.title} +

    +

    + {step.body} +

    +
    +
    2. + ))} +
+
+ +
+

+ Configure a relay +

+

+ Example: notify clients when an Order document changes. +

+
+ + + + + +
+
+ +
+

+ Subscribe from a client +

+

+ Connect to {'/events/'} with{' '} + {'path: /realtime'} and a user bearer token. Then + subscribe with the relay id and resource id. +

+ 
+                {CLIENT_SNIPPET}
+
+
+ +
+

Limits

+
    +
  • + Bus channels must match exactly. Patterns like{' '} + {'database:change:*'} are not supported. +
    • +
  • + Subscribe fails closed if Authorization is unavailable or the + user lacks permission. +
    • +
  • + Turn a relay off with Active to stop forwarding without + deleting it. Deleting drops current subscribers immediately. +
    • +
+
+
+ + + + ); +} + +function Field({ + name, + value, + hint, +}: { + name: string; + value: string; + hint: string; +}) { + return ( +
+
+ {name} +
+
+ {value} +

{hint}

+
+
+ ); +} + +function Code({ children }: { children: string }) { + return ( + + {children} + + ); +} diff --git a/src/components/router/event-relays/event-relay-form.tsx b/src/components/router/event-relays/event-relay-form.tsx index 833aa3db3..277c67480 100644 --- a/src/components/router/event-relays/event-relay-form.tsx +++ b/src/components/router/event-relays/event-relay-form.tsx @@ -111,7 +111,7 @@ export function EventRelayForm({ diff --git a/src/components/router/event-relays/event-relay-list.tsx b/src/components/router/event-relays/event-relay-list.tsx index 162e8a0a5..c64e5a456 100644 --- a/src/components/router/event-relays/event-relay-list.tsx +++ b/src/components/router/event-relays/event-relay-list.tsx @@ -2,7 +2,7 @@ import { useCallback, useMemo, useState } from 'react'; import { ColumnDef } from '@tanstack/react-table'; -import { Plus, Radio, Zap } from 'lucide-react'; +import { BookOpen, Plus, Radio, Zap } from 'lucide-react'; import { Button } from '@/components/ui/button'; import { Badge } from '@/components/ui/badge'; import { DataTable } from '@/components/ui/data-table'; @@ -15,6 +15,12 @@ import { DialogHeader, DialogTitle, } from '@/components/ui/dialog'; +import { + Tooltip, + TooltipContent, + TooltipProvider, + TooltipTrigger, +} from '@/components/ui/tooltip'; import { PageActions, PageDescription, @@ -23,6 +29,7 @@ import { } from '@/components/ui/page-header'; import { SearchInput } from '@/components/ui/form-inputs/SearchInput'; import { DeleteAlert } from '@/components/helpers/delete'; +import { EventRelayDocs } from '@/components/router/event-relays/event-relay-docs'; import { EventRelayForm } from '@/components/router/event-relays/event-relay-form'; import { EventRelay, EventRelayWriteRequest } from '@/lib/models/Router'; import { @@ -47,8 +54,23 @@ export function EventRelayList({ const router = useRouter(); const [isCreateOpen, setIsCreateOpen] = useState(false); const [editing, setEditing] = useState(null); + const [docsOpen, setDocsOpen] = useState(count === 0); const { save, isSaving } = useSettingsSave('Event Relay'); + const toggleDocs = useCallback(() => { + setDocsOpen(open => { + const next = !open; + if (next) { + requestAnimationFrame(() => { + document + .getElementById('event-relay-docs') + ?.scrollIntoView({ behavior: 'smooth', block: 'start' }); + }); + } + return next; + }); + }, []); + const refresh = useCallback(() => { router.refresh(); }, [router]); @@ -166,7 +188,7 @@ export function EventRelayList({ return (
- +
Event Relays @@ -175,6 +197,27 @@ export function EventRelayList({
+ + + + + + + {docsOpen + ? 'Hide how Event Relays work' + : 'Show how Event Relays work'} + + +