diff --git a/README.md b/README.md index 85e0e9c..167532d 100644 --- a/README.md +++ b/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 diff --git a/controller/src/runtime/access_logs/cache.rs b/controller/src/runtime/access_logs/cache.rs index 59ce7a7..71a53d7 100644 --- a/controller/src/runtime/access_logs/cache.rs +++ b/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::()); 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, diff --git a/controller/src/runtime/access_logs/tests.rs b/controller/src/runtime/access_logs/tests.rs index b61497d..692aad6 100644 --- a/controller/src/runtime/access_logs/tests.rs +++ b/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, diff --git a/controller/src/runtime/access_logs/tests/snapshots.rs b/controller/src/runtime/access_logs/tests/snapshots.rs index 9e837ff..38ae0dc 100644 --- a/controller/src/runtime/access_logs/tests/snapshots.rs +++ b/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::>(); assert!(!second.snapshot_reset); - assert_ne!(second.snapshot, first.snapshot); + assert_eq!(second.snapshot, first.snapshot); assert_eq!(second_paths, first_paths); } diff --git a/docker/production/Dockerfile b/docker/production/Dockerfile index bf93bdd..be0ef60 100644 --- a/docker/production/Dockerfile +++ b/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 diff --git a/docker/web/Dockerfile b/docker/web/Dockerfile index e410258..64b5b21 100644 --- a/docker/web/Dockerfile +++ b/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 diff --git a/docker/web/serve.mjs b/docker/web/serve.mjs index 247299c..67dd930 100644 --- a/docker/web/serve.mjs +++ b/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 { diff --git a/web/src/db/index.ts b/web/src/db/index.ts index 6e0e6ba..e077265 100644 --- a/web/src/db/index.ts +++ b/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 }) diff --git a/web/src/features/Admin/AccessPolicyManagement/Hooks/useAccessPolicyManagementLogic.ts b/web/src/features/Admin/AccessPolicyManagement/Hooks/useAccessPolicyManagementLogic.ts index a58fac7..5889a43 100644 --- a/web/src/features/Admin/AccessPolicyManagement/Hooks/useAccessPolicyManagementLogic.ts +++ b/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(null) const [deleteTarget, setDeleteTarget] = useState(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([ diff --git a/web/src/features/Admin/AuditLogs/Components/AuditLogsPageView.tsx b/web/src/features/Admin/AuditLogs/Components/AuditLogsPageView.tsx index 82efa7e..ee9a877 100644 --- a/web/src/features/Admin/AuditLogs/Components/AuditLogsPageView.tsx +++ b/web/src/features/Admin/AuditLogs/Components/AuditLogsPageView.tsx @@ -40,6 +40,7 @@ export default function AuditLogsPageView({ logic: { state, handler } }: AuditLo /> ) : ( -