feat(ui): synchronize management views with native Bun websockets

README.md

@@ -137,6 +137,14 @@ and TCP/UDP port mappings, alongside the backup: these settings are not stored i
 
 Development setup and checks are in [`CONTRIBUTING.md`](./CONTRIBUTING.md).
 
+### Live administration
+
+The management UI shares one native Bun WebSocket connection per visible browser tab. Changes to hosts, users, profile pictures, permissions, and settings invalidate the affected visible views. Application events are also distributed through the existing Redis connection setup. Expired or revoked sessions are revalidated through the same authorization services as HTTP requests.
+
+Pages subscribe only to the data they display. Access logs stop streaming when their page is left, a historical page is selected, or the tab becomes hidden. Returning opens a fresh subscription. Controller-backed log and runtime changes are sampled on the server only while subscribed; the browser does not poll them. The Overview includes the actual WebSocket connection status.
+
+Realtime code lives in `web/src/websockets`, separated into `Client`, `Server`, `Helpers`, and `Types`. Production upgrades share the web server port. Development proxies the same `/api/live` URL to a native Bun listener on a temporary loopback port; no additional public port or realtime credentials are needed.
+
 ### Certificate renewal and retries
 
 ACME certificates become due after two thirds of their actual certificate lifetime. The

controller/src/runtime/access_logs/cache.rs

@@ -83,6 +83,23 @@ impl SnapshotCache {
             .map_err(|_| ReadError::Failed)?;
         let mut state = self.state.lock().map_err(|_| ReadError::Failed)?;
         remove_expired(&mut state, now);
+        let capture_snapshot_query = SnapshotQuery::from_query(capture_query);
+        if let Some(snapshot_id) = state
+            .snapshots
+            .iter()
+            .find(|(_, snapshot)| {
+                snapshot.query == capture_snapshot_query
+                    && snapshot.entries == capture.entries
+                    && snapshot.truncated == capture.truncated
+                    && snapshot.available_hosts == capture.available_hosts
+                    && snapshot.available_statuses == capture.available_statuses
+            })
+            .map(|(snapshot_id, _)| snapshot_id.clone())
+        {
+            touch_snapshot(&mut state, &snapshot_id);
+            let snapshot = state.snapshots.get(&snapshot_id).ok_or(ReadError::Failed)?;
+            return Ok(page_response(&snapshot_id, snapshot, page_query, reset));
+        }
         let mut snapshot_id = new_snapshot_id().ok_or(ReadError::Failed)?;
         while state.snapshots.contains_key(&snapshot_id) {
             snapshot_id = new_snapshot_id().ok_or(ReadError::Failed)?;
@@ -111,7 +128,7 @@ impl SnapshotCache {
         state.bytes = state.bytes.saturating_add(bytes);
         state.lru.push_back(snapshot_id.clone());
         let snapshot = CachedSnapshot {
-            query: SnapshotQuery::from_query(capture_query),
+            query: capture_snapshot_query,
             total: capture.entries.len(),
             entries: capture.entries,
             truncated: capture.truncated,
@@ -315,6 +332,101 @@ mod tests {
         }
     }
 
+    fn entry(path: &str) -> AccessLogEntry {
+        AccessLogEntry {
+            timestamp: "2026-09-13T00:00:00Z".into(),
+            host: "example.com".into(),
+            method: "GET".into(),
+            path: path.into(),
+            status: 200,
+            duration_ms: 1,
+            client_ip: "192.0.2.1".into(),
+            upstream: None,
+            bytes: 0,
+            protocol: "HTTP/2".into(),
+        }
+    }
+
+    fn capture(path: &str) -> CapturedLogs {
+        CapturedLogs {
+            entries: vec![entry(path)],
+            truncated: false,
+            available_hosts: vec!["example.com".into()],
+            available_statuses: vec![200],
+        }
+    }
+
+    #[test]
+    fn identical_captures_reuse_unexpired_snapshot_without_evicting_history() {
+        let cache = SnapshotCache::new();
+        let query = query();
+        let now = Instant::now();
+        let historical = cache
+            .insert(&query, &query, capture("/historical"), false, now)
+            .unwrap();
+        let repeated = cache
+            .insert(&query, &query, capture("/current"), false, now)
+            .unwrap();
+        for _ in 0..MAX_SNAPSHOTS {
+            let response = cache
+                .insert(&query, &query, capture("/current"), false, now)
+                .unwrap();
+            assert_eq!(response.snapshot, repeated.snapshot);
+        }
+        assert_eq!(cache.len(), 2);
+        let mut historical_query = query.clone();
+        historical_query.snapshot = Some(historical.snapshot.clone());
+        assert_eq!(
+            cache
+                .lookup(&historical.snapshot, &historical_query, now)
+                .unwrap()
+                .unwrap()
+                .entries[0]
+                .path,
+            "/historical"
+        );
+    }
+
+    #[test]
+    fn changed_capture_gets_a_new_immutable_snapshot() {
+        let cache = SnapshotCache::new();
+        let query = query();
+        let now = Instant::now();
+        let first = cache
+            .insert(&query, &query, capture("/first"), false, now)
+            .unwrap();
+        let changed = cache
+            .insert(&query, &query, capture("/changed"), false, now)
+            .unwrap();
+        assert_ne!(first.snapshot, changed.snapshot);
+        let mut continuation = query.clone();
+        continuation.snapshot = Some(first.snapshot.clone());
+        assert_eq!(
+            cache
+                .lookup(&first.snapshot, &continuation, now)
+                .unwrap()
+                .unwrap()
+                .entries[0]
+                .path,
+            "/first"
+        );
+    }
+
+    #[test]
+    fn expired_identical_capture_is_replaced() {
+        let cache = SnapshotCache::new();
+        let query = query();
+        let now = Instant::now();
+        let first = cache
+            .insert(&query, &query, capture("/same"), false, now)
+            .unwrap();
+        let replacement = cache
+            .insert(&query, &query, capture("/same"), false, now + SNAPSHOT_TTL)
+            .unwrap();
+        assert_ne!(first.snapshot, replacement.snapshot);
+        assert_eq!(cache.len(), 1);
+    }
+
     #[test]
     fn vector_spare_capacity_is_counted_and_released_before_caching() {
         let query = query();
@@ -388,7 +500,7 @@ mod tests {
     fn aggregate_memory_bound_evicts_before_the_count_limit() {
         let cache = SnapshotCache::new();
         let query = query();
-        for _ in 0..8 {
+        for index in 0..8 {
             let entries = Vec::with_capacity(3 * 1024 * 1024 / size_of::<AccessLogEntry>());
             cache
                 .insert(
@@ -397,7 +509,7 @@ mod tests {
                     CapturedLogs {
                         entries,
                         truncated: false,
-                        available_hosts: Vec::new(),
+                        available_hosts: vec![format!("host-{index}")],
                         available_statuses: Vec::new(),
                     },
                     false,

controller/src/runtime/access_logs/tests.rs

@@ -393,7 +393,7 @@ fn snapshot_cache_expiry_is_fixed_and_evicts_lru_entries() {
     );
 
     let mut ids = Vec::new();
-    for _ in 0..MAX_SNAPSHOTS {
+    for index in 0..MAX_SNAPSHOTS {
         ids.push(
             cache
                 .insert(
@@ -402,7 +402,7 @@ fn snapshot_cache_expiry_is_fixed_and_evicts_lru_entries() {
                     CapturedLogs {
                         entries: Vec::new(),
                         truncated: false,
-                        available_hosts: Vec::new(),
+                        available_hosts: vec![format!("host-{index}")],
                         available_statuses: Vec::new(),
                     },
                     false,

controller/src/runtime/access_logs/tests/snapshots.rs

@@ -124,7 +124,7 @@ async fn equal_timestamps_keep_field_order_after_archive_rename_and_line_reorder
         .map(|entry| entry.path.as_str())
         .collect::<Vec<_>>();
     assert!(!second.snapshot_reset);
-    assert_ne!(second.snapshot, first.snapshot);
+    assert_eq!(second.snapshot, first.snapshot);
     assert_eq!(second_paths, first_paths);
 }
 

docker/production/Dockerfile

@@ -35,6 +35,7 @@ COPY --from=web-build /build/migrate.js ./migrate.js
 COPY --from=web-build /app/docker/web/bootstrap-secrets.mjs ./docker/web/bootstrap-secrets.mjs
 COPY --from=web-build /app/docker/web/request-context.ts ./docker/web/request-context.ts
 COPY --from=web-build /app/docker/web/static-assets.ts ./docker/web/static-assets.ts
+COPY --from=web-build /app/web/src/websockets ./web/src/websockets
 COPY --from=web-build /app/docker/web/serve.mjs ./docker/web/serve.mjs
 COPY --from=web-build /app/docker/web/healthcheck.mjs ./docker/web/healthcheck.mjs
 

docker/web/Dockerfile

@@ -37,6 +37,7 @@ COPY --from=build /build/migrate.js ./migrate.js
 COPY --from=build /app/docker/web/bootstrap-secrets.mjs ./docker/web/bootstrap-secrets.mjs
 COPY --from=build /app/docker/web/request-context.ts ./docker/web/request-context.ts
 COPY --from=build /app/docker/web/static-assets.ts ./docker/web/static-assets.ts
+COPY --from=build /app/web/src/websockets ./web/src/websockets
 COPY --from=build /app/docker/web/serve.mjs ./docker/web/serve.mjs
 COPY --from=build /app/docker/web/healthcheck.mjs ./docker/web/healthcheck.mjs
 

docker/web/serve.mjs

@@ -1,5 +1,6 @@
 import { createRuntimeFetch } from './request-context.ts'
 import { createStaticAssetFetch } from './static-assets.ts'
+import { createLiveWebSocketRuntime } from '../../web/src/websockets/Server/realtimeWebSocket.ts'
 import { fileURLToPath } from 'node:url'
 
 const { default: application } = await import('../../web/dist/server/server.js')
@@ -13,12 +14,22 @@ if (!Number.isInteger(port) || port < 1 || port > 65_535 || !application?.fetch)
 }
 
 const runtimeFetch = createRuntimeFetch(application)
+const live = createLiveWebSocketRuntime({
+    allowedOrigin: process.env.RENTNERPROXY_PUBLIC_ORIGIN ?? 'http://localhost:5173',
+    readSnapshot: (request) => application.fetch(request),
+})
+const staticAssetFetch = createStaticAssetFetch(clientRoot, runtimeFetch)
 const server = Bun.serve({
     hostname,
     port,
 
     maxRequestBodySize: 12 * 1024 * 1024,
-    fetch: createStaticAssetFetch(clientRoot, runtimeFetch),
+    websocket: live.websocket,
+    fetch: async (request, bunServer) => {
+        const result = await live.handle(request, bunServer)
+        if (result.handled) return result.response
+        return staticAssetFetch(request, bunServer)
+    },
 })
 
 let shuttingDown = false
@@ -31,6 +42,7 @@ async function shutdown() {
     try {
         const pending = []
         process.emit('rentnerproxy:shutdown', pending)
+        pending.push(live.shutdown())
         await Promise.all([server.stop(false), ...pending])
         process.exitCode = 0
     } finally {

web/src/db/index.ts

@@ -4,5 +4,14 @@ import { validateDatabaseEnvironment } from '../server/env.server'
 import * as schema from './schema'
 
 const { DATABASE_URL } = validateDatabaseEnvironment()
-const client = new SQL(DATABASE_URL)
+const developmentGlobal = globalThis as typeof globalThis & {
+    rentnerproxyDatabaseClient?: { url: string; client: SQL }
+}
+const cached =
+    process.env.NODE_ENV === 'production' ? undefined : developmentGlobal.rentnerproxyDatabaseClient
+if (cached && cached.url !== DATABASE_URL) void cached.client.close().catch(() => undefined)
+const client = cached?.url === DATABASE_URL ? cached.client : new SQL(DATABASE_URL)
+if (process.env.NODE_ENV !== 'production') {
+    developmentGlobal.rentnerproxyDatabaseClient = { url: DATABASE_URL, client }
+}
 export const db = drizzle(client, { schema })

web/src/features/Admin/AccessPolicyManagement/Hooks/useAccessPolicyManagementLogic.ts

@@ -2,6 +2,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
 import { useCallback, useMemo, useState } from 'react'
 
 import { PERMISSIONS } from '../../../../config/permissions.config'
+import useLiveInvalidation from '../../../../shared/Live/useLiveInvalidation'
 import useToast from '../../../../shared/Toast/Hooks/useToast'
 import type { AccessPolicySummary } from '../../../../shared/Types/access-policies.types'
 import { accessPolicyManagementQueryKeys } from '../queryKeys'
@@ -27,6 +28,7 @@ export default function useAccessPolicyManagementLogic({
     const toast = useToast()
     const queryClient = useQueryClient()
     const permissionSet = useMemo(() => new Set(permissions), [permissions])
+    const canView = permissionSet.has(PERMISSIONS.ACCESS_POLICIES_VIEW)
     const [showCreate, setShowCreate] = useState(false)
     const [selectedPolicy, setSelectedPolicy] = useState<AccessPolicySummary | null>(null)
     const [deleteTarget, setDeleteTarget] = useState<AccessPolicySummary | null>(null)
@@ -35,12 +37,21 @@ export default function useAccessPolicyManagementLogic({
     const policiesQuery = useQuery({
         queryKey: accessPolicyManagementQueryKeys.all,
         queryFn: () => getAccessPoliciesHandler(),
+        enabled: canView,
     })
     const runtimeStatusQuery = useQuery({
         queryKey: accessPolicyManagementQueryKeys.runtimeStatus,
         queryFn: () => getAccessPolicyRuntimeStatusHandler(),
-        refetchInterval: 15_000,
-        refetchIntervalInBackground: false,
+        enabled: canView,
+    })
+    useLiveInvalidation({
+        topic: 'access-policies',
+        query: {},
+        enabled: canView,
+        queryKeys: [
+            accessPolicyManagementQueryKeys.all,
+            accessPolicyManagementQueryKeys.runtimeStatus,
+        ],
     })
     const invalidate = useCallback(async () => {
         await Promise.all([

web/src/features/Admin/AuditLogs/Components/AuditLogsPageView.tsx

@@ -40,23 +40,21 @@ export default function AuditLogsPageView({ logic: { state, handler } }: AuditLo
                 />
             ) : (
                 <AuditLogsTable
+                    actorOptions={state.actorOptions}
                     events={state.events}
                     expandedEventId={state.expandedEventId}
                     formatTimestamp={state.formatTimestamp}
                     filters={state.filters}
                     filterErrors={state.filterErrors}
                     hasMore={state.hasMore}
                     isLoading={state.isLoading}
-                    isRefreshing={state.isRefreshing}
                     pageNumber={state.pageNumber}
                     onActorChange={handler.onActorChange}
                     onActionChange={handler.onActionChange}
                     onResourceChange={handler.onResourceChange}
                     onFromChange={handler.onFromChange}
                     onToChange={handler.onToChange}
-                    onApplyFilters={handler.applyFilters}
                     onResetFilters={handler.resetFilters}
-                    onRefresh={handler.refresh}
                     onPreviousPage={handler.previousPage}
                     onNextPage={handler.nextPage}
                     onToggleDetails={handler.onToggleDetails}