diff --git a/apps/desktop/src/components/layout/ContentArea.vue b/apps/desktop/src/components/layout/ContentArea.vue index 8e8cf7758d..10428ece01 100644 --- a/apps/desktop/src/components/layout/ContentArea.vue +++ b/apps/desktop/src/components/layout/ContentArea.vue @@ -483,7 +483,7 @@ const resultArchiveExporting = ref(false); const canExportResultArchive = computed(() => props.activeTab.mode === "query" && (!!props.activeTab.result || !!props.activeTab.results?.length || !!props.activeTab.resultRuns?.length)); const resultAutoSave = computed(() => props.activeTab.resultAutoSave === true); const activeResultRunItem = computed(() => resultRuns.value.find((run) => run.active)); -const activeResultIsLoading = computed(() => isActiveResultLoading(props.activeTab)); +const activeResultIsLoading = computed(() => !props.activeTab.redisMonitorActive && isActiveResultLoading(props.activeTab)); const showResultRunTabs = computed(() => resultRuns.value.length > 0 && resultRunDisplayMode.value === "tabs"); const showResultRunSelector = computed(() => resultRuns.value.length > 0 && resultRunDisplayMode.value === "list"); const canCloseQueryResult = computed(() => props.activeTab.mode === "query" && !props.activeTab.isExecuting && !props.activeTab.activeResultRunId && (!!props.activeTab.result || !!props.activeTab.results?.length || props.activeTab.resultEvicted === true)); @@ -1717,6 +1717,11 @@ defineExpose({ /> +
+ {{ t("redis.monitorListening") }} + +
+ { expect(restored.isExecuting).toBe(false); }); + it("does not resume a MONITOR stream when restoring tabs", () => { + const tab = queryTab({ sql: "MONITOR", redisMonitorActive: true, isExecuting: true }); + const [saved] = serializeOpenTabs([tab]); + const [restored] = roundTrip([tab]); + expect(saved).not.toHaveProperty("redisMonitorActive"); + expect(restored.redisMonitorActive).toBe(false); + expect(restored.isExecuting).toBe(false); + }); + it("preserves an external Doris catalog across tab restore", () => { const [restored] = roundTrip([queryTab({ database: "dbx_catalog_completion", catalog: "dbx_mysql_catalog" })]); diff --git a/apps/desktop/src/lib/__tests__/redis/redisMonitor.spec.ts b/apps/desktop/src/lib/__tests__/redis/redisMonitor.spec.ts new file mode 100644 index 0000000000..c0bb603ea6 --- /dev/null +++ b/apps/desktop/src/lib/__tests__/redis/redisMonitor.spec.ts @@ -0,0 +1,102 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { REDIS_MONITOR_MAX_ROWS, startRedisMonitor } from "../../redis/redisMonitor"; +import { classifyRedisCommandSafety } from "../../redis/redisCommandSafety"; + +function createSocket() { + return { close: vi.fn(), onmessage: null, onerror: null, onclose: null } as unknown as WebSocket; +} + +function receive(socket: WebSocket, data: object) { + socket.onmessage?.(new MessageEvent("message", { data: JSON.stringify(data) })); +} + +afterEach(() => vi.useRealTimers()); + +describe("Redis MONITOR stream", () => { + it("allows standalone MONITOR without permitting arbitrary commands", () => { + expect(classifyRedisCommandSafety("monitor")).toBe("allowed"); + expect(classifyRedisCommandSafety(" MONITOR; ")).toBe("allowed"); + expect(classifyRedisCommandSafety("MONITOR extra")).toBe("blocked"); + expect(classifyRedisCommandSafety("MONITOR; FLUSHALL")).toBe("blocked"); + }); + it("streams successive batches and bounds retained rows", async () => { + vi.useFakeTimers(); + const socket = createSocket(); + const onRows = vi.fn(); + const monitor = startRedisMonitor(async () => socket, onRows); + await Promise.resolve(); + receive(socket, { ready: true }); + expect(onRows).toHaveBeenLastCalledWith([]); + receive(socket, { message: "first" }); + vi.advanceTimersByTime(100); + expect(onRows).toHaveBeenLastCalledWith(["first"]); + for (let index = 0; index < REDIS_MONITOR_MAX_ROWS + 10; index++) receive(socket, { message: String(index) }); + vi.advanceTimersByTime(100); + const rows = onRows.mock.lastCall![0]; + expect(rows).toHaveLength(REDIS_MONITOR_MAX_ROWS); + expect(rows[0]).toBe("10"); + monitor.stop(); + await monitor.done; + expect(socket.close).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("closes a socket that arrives after cancellation", async () => { + const socket = createSocket(); + let resolveConnection!: (socket: WebSocket) => void; + const monitor = startRedisMonitor( + () => + new Promise((resolve) => { + resolveConnection = resolve; + }), + vi.fn(), + ); + monitor.stop(); + resolveConnection(socket); + await monitor.done; + expect(socket.close).toHaveBeenCalledOnce(); + }); + + it("reports server errors and cleans up", async () => { + const socket = createSocket(); + const monitor = startRedisMonitor(async () => socket, vi.fn()); + const failure = expect(monitor.done).rejects.toThrow("NOPERM"); + await Promise.resolve(); + receive(socket, { error: "NOPERM" }); + await failure; + expect(socket.close).toHaveBeenCalledOnce(); + }); + + it("reports unexpected disconnects and ignores events after stop", async () => { + vi.useFakeTimers(); + const socket = createSocket(); + const onRows = vi.fn(); + const monitor = startRedisMonitor(async () => socket, onRows); + const failure = expect(monitor.done).rejects.toThrow("connection closed"); + await Promise.resolve(); + receive(socket, { ready: true }); + socket.onclose?.(new Event("close") as CloseEvent); + await failure; + receive(socket, { message: "late" }); + vi.advanceTimersByTime(1000); + expect(onRows).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + + it("times out setup without timing out an idle ready stream", async () => { + vi.useFakeTimers(); + const socket = createSocket(); + const monitor = startRedisMonitor(async () => socket, vi.fn()); + await Promise.resolve(); + receive(socket, { ready: true }); + vi.advanceTimersByTime(60000); + expect(socket.close).not.toHaveBeenCalled(); + monitor.stop(); + await monitor.done; + const pending = startRedisMonitor(() => new Promise(() => {}), vi.fn()); + const failure = expect(pending.done).rejects.toThrow("timed out"); + vi.advanceTimersByTime(30000); + await failure; + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/apps/desktop/src/lib/app/openTabsPersistence.ts b/apps/desktop/src/lib/app/openTabsPersistence.ts index 33a8d1ae5b..46c54dee4c 100644 --- a/apps/desktop/src/lib/app/openTabsPersistence.ts +++ b/apps/desktop/src/lib/app/openTabsPersistence.ts @@ -233,6 +233,7 @@ function restoreOpenTabsArray(parsed: unknown, rawActiveTabId: string | null, op mode, sql: typeof tab.sql === "string" ? tab.sql : "", isExecuting: false, + redisMonitorActive: false, isCancelling: false, queryExecutionStartedAt: undefined, executingResultRunId: undefined, diff --git a/apps/desktop/src/lib/backend/http.ts b/apps/desktop/src/lib/backend/http.ts index 278b21b322..f6bba47bd3 100644 --- a/apps/desktop/src/lib/backend/http.ts +++ b/apps/desktop/src/lib/backend/http.ts @@ -2989,8 +2989,8 @@ export async function redisPubSubPublish(connectionId: string, db: number, chann }); } -export async function redisPubSubConnect(connectionId: string): Promise { - return new WebSocket(apiWebSocketUrl(`/api/redis/pubsub/ws?connectionId=${encodeURIComponent(connectionId)}`)); +export async function redisPubSubConnect(connectionId: string, monitor = false): Promise { + return new WebSocket(apiWebSocketUrl(`/api/redis/pubsub/ws?connectionId=${encodeURIComponent(connectionId)}&monitor=${monitor}`)); } export async function redisSlowlogGet(connectionId: string, count: number, nodeHost?: string, nodePort?: number): Promise { diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index f9e696df98..9495e9294b 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -2926,9 +2926,9 @@ export async function redisPubSubPublish(connectionId: string, db: number, chann return invoke("redis_pubsub_publish", { connectionId, db, channel, message }); } -export async function redisPubSubConnect(connectionId: string): Promise { +export async function redisPubSubConnect(connectionId: string, monitor = false): Promise { const port = await invoke("redis_pubsub_server_port"); - return new WebSocket(`ws://127.0.0.1:${port}/api/redis/pubsub/ws?connectionId=${encodeURIComponent(connectionId)}`); + return new WebSocket(`ws://127.0.0.1:${port}/api/redis/pubsub/ws?connectionId=${encodeURIComponent(connectionId)}&monitor=${monitor}`); } export async function redisSlowlogGet(connectionId: string, count: number, nodeHost?: string, nodePort?: number): Promise { diff --git a/apps/desktop/src/lib/redis/redisCommandSafety.ts b/apps/desktop/src/lib/redis/redisCommandSafety.ts index 2ef9aab814..67fc907632 100644 --- a/apps/desktop/src/lib/redis/redisCommandSafety.ts +++ b/apps/desktop/src/lib/redis/redisCommandSafety.ts @@ -1,3 +1,5 @@ +import { isRedisMonitorCommand } from "./redisMonitor"; + export type RedisCommandSafety = "allowed" | "write" | "confirm" | "blocked"; const BLOCKED_COMMANDS = new Set(["KEYS", "FLUSHALL", "SHUTDOWN", "CONFIG", "SAVE", "BGSAVE", "SLAVEOF", "REPLICAOF", "MIGRATE", "MODULE", "SCRIPT", "EVAL", "EVALSHA"]); @@ -273,6 +275,7 @@ export function firstRedisCommandToken(command: string): string { } export function classifyRedisCommandSafety(command: string): RedisCommandSafety { + if (isRedisMonitorCommand(command)) return "allowed"; const token = firstRedisCommandToken(command); if (BLOCKED_COMMANDS.has(token)) return "blocked"; if (CONFIRM_COMMANDS.has(token)) return "confirm"; diff --git a/apps/desktop/src/lib/redis/redisMonitor.ts b/apps/desktop/src/lib/redis/redisMonitor.ts new file mode 100644 index 0000000000..83d49d451a --- /dev/null +++ b/apps/desktop/src/lib/redis/redisMonitor.ts @@ -0,0 +1,66 @@ +export const REDIS_MONITOR_MAX_ROWS = 1000; + +export function isRedisMonitorCommand(command: string): boolean { + return /^MONITOR\s*;?$/i.test(command.trim()); +} + +export function startRedisMonitor(connect: () => Promise, onRows: (rows: string[]) => void) { + let socket: WebSocket | undefined; + let finished = false; + let rows: string[] = []; + let dirty = false; + let resolveDone!: () => void; + let rejectDone!: (error: Error) => void; + const done = new Promise((resolve, reject) => { + resolveDone = resolve; + rejectDone = reject; + }); + const flush = () => { + if (dirty) { + dirty = false; + onRows(rows.slice()); + } + }; + const timer = setInterval(flush, 100); + const timeout = setTimeout(() => finish(new Error("Redis MONITOR connection timed out")), 30000); + function finish(error?: Error) { + if (finished) return; + finished = true; + clearInterval(timer); + clearTimeout(timeout); + flush(); + socket?.close(); + if (error) rejectDone(error); + else resolveDone(); + } + void connect() + .then((connected) => { + socket = connected; + if (finished) { + socket.close(); + return; + } + socket.onmessage = (event) => { + if (finished) return; + try { + const data = JSON.parse(event.data); + if (typeof data.error === "string") { + finish(new Error(data.error)); + } else if (data.ready === true) { + clearTimeout(timeout); + onRows([]); + } else if (typeof data.message === "string") { + rows.push(data.message.slice(0, 16384)); + if (rows.length > REDIS_MONITOR_MAX_ROWS) rows = rows.slice(-REDIS_MONITOR_MAX_ROWS); + dirty = true; + } + } catch { + finish(new Error("Invalid Redis MONITOR response")); + } + }; + socket.onerror = () => finish(new Error("Redis MONITOR connection failed")); + socket.onclose = () => finish(new Error("Redis MONITOR connection closed")); + }) + .catch((error) => finish(error instanceof Error ? error : new Error(String(error)))); + return { done, stop: () => finish() }; +} diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index 9abf4431c3..cc9b3c605c 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -1,5 +1,6 @@ import { UPDATE_RESTORE_KEY, assertUpdateAllowsInteraction } from "@/lib/app/updatePreparation"; import { defineStore } from "pinia"; +import { isRedisMonitorCommand, startRedisMonitor } from "@/lib/redis/redisMonitor"; import { uuid } from "@/lib/common/utils"; import { computed, markRaw, nextTick, onScopeDispose, reactive, ref, watch } from "vue"; import { useI18n } from "vue-i18n"; @@ -789,6 +790,7 @@ function getI18nT() { } export const useQueryStore = defineStore("query", () => { + const redisMonitors = new Map void>(); const t = getI18nT(); const settingsStore = useSettingsStore(); const tabs = ref([]); @@ -5769,6 +5771,43 @@ export const useQueryStore = defineStore("query", () => { .map((line) => line.trim()) .filter((line) => line.length > 0); if (commands.length === 0) return false; + if (commands.length === 1 && isRedisMonitorCommand(commands[0])) { + const monitor = startRedisMonitor( + () => api.redisPubSubConnect(executionConnectionId, true), + (rows) => { + const current = findExecutionTab(id); + if (current?.executionId !== executionId) return; + current.result = markQueryResultRowsRaw(annotateQueryResultSource({ columns: ["MONITOR"], rows: rows.map((message) => [message]), affected_rows: 0, execution_time_ms: performance.now() - startedAt }, "MONITOR")); + current.results = undefined; + current.activeResultIndex = undefined; + current.queryEditabilityReason = undefined; + current.queryWriteTargets = undefined; + current.tableMeta = undefined; + current.resultBaseSql = "MONITOR"; + current.redisMonitorActive = true; + current.queryAnalysis = undefined; + current.querySourceColumns = undefined; + current.resultColumnComments = undefined; + current.queryDisplaySourceColumns = undefined; + current.mongoEditTarget = undefined; + if (!producedResult) { + publishResultGeneration(current, "execute"); + syncDisplayedResultRun(current, "MONITOR", captureResultRun); + } + producedResult = true; + touchResult(current); + }, + ); + redisMonitors.set(executionId, monitor.stop); + try { + await monitor.done; + } finally { + redisMonitors.delete(executionId); + const current = findExecutionTab(id); + if (current?.executionId === executionId) current.redisMonitorActive = false; + } + return producedResult; + } queryExecutionLog("info", "redis:start", { traceId, db: currentDb, commandCount: commands.length, sqlLength: sql.length }); const allResults: QueryResult[] = []; @@ -7205,6 +7244,13 @@ export const useQueryStore = defineStore("query", () => { const executionId = tab.executionId; if (!executionId) return false; + const stopMonitor = redisMonitors.get(executionId); + if (stopMonitor) { + tab.isCancelling = true; + tab.cancelRequestCount = (tab.cancelRequestCount ?? 0) + 1; + stopMonitor(); + return true; + } tab.isCancelling = true; // 单调递增、不随取消结果回退：导航流程据此判断"执行期间用户请求过停止" // （isCancelling 在取消失败或查询先完成时会被清掉，无法承担这个语义） diff --git a/apps/desktop/src/types/database.ts b/apps/desktop/src/types/database.ts index 373f724198..92129448ec 100644 --- a/apps/desktop/src/types/database.ts +++ b/apps/desktop/src/types/database.ts @@ -1224,6 +1224,7 @@ export interface QueryTab { explainTableSql?: string; lastExplainedSql?: string; isExecuting: boolean; + redisMonitorActive?: boolean; isCancelling?: boolean; queryExecutionStartedAt?: number; /** Ephemeral per-statement progress for the latest multi-statement execution. */ diff --git a/crates/dbx-core/src/db/redis_driver.rs b/crates/dbx-core/src/db/redis_driver.rs index a847fa361c..45efb84262 100644 --- a/crates/dbx-core/src/db/redis_driver.rs +++ b/crates/dbx-core/src/db/redis_driver.rs @@ -1121,6 +1121,35 @@ pub async fn connect_direct_node( connect_client(client).await } +pub async fn connect_monitor( + config: &ConnectionConfig, + host: &str, + port: u16, + timeout: std::time::Duration, +) -> Result { + let mut last_error = String::new(); + for info in standalone_connection_infos(config, host, port) { + let client = redis::Client::open(info).map_err(|error| error.to_string())?; + let connect = async { + let mut validation = client.get_multiplexed_async_connection().await?; + redis::cmd("MONITOR").query_async::<()>(&mut validation).await?; + drop(validation); + client.get_async_monitor().await + }; + match tokio::time::timeout(timeout, connect).await { + Ok(Ok(monitor)) => return Ok(monitor), + Ok(Err(error)) => { + last_error = error.to_string(); + if !is_redis_auth_error(&last_error) { + break; + } + } + Err(_) => return Err("Redis MONITOR connection timed out".to_string()), + } + } + Err(last_error) +} + pub async fn connect_pubsub( config: &ConnectionConfig, host: &str, @@ -2218,6 +2247,11 @@ where { let argv = parse_command_argv(command_text)?; let command = argv[0].to_ascii_uppercase(); + if command == "MONITOR" { + return Err( + "MONITOR requires a dedicated streaming connection; run MONITOR alone in the query editor".to_string() + ); + } let safety = classify_command(&command); if !skip_safety_check && safety == RedisCommandSafety::Blocked { return Err(format!("Redis command is blocked for safety: {command}")); diff --git a/crates/dbx-core/src/redis_ops.rs b/crates/dbx-core/src/redis_ops.rs index 2de097c73b..c4bb3e46ef 100644 --- a/crates/dbx-core/src/redis_ops.rs +++ b/crates/dbx-core/src/redis_ops.rs @@ -1141,6 +1141,28 @@ pub async fn redis_publish_core( } } +pub async fn redis_create_monitor_core(state: &AppState, connection_id: &str) -> Result { + let config = state.configs.read().await.get(connection_id).ok_or("Connection config not found")?.clone(); + if config.db_type != crate::models::connection::DatabaseType::Redis { + return Err("Not a Redis connection".to_string()); + } + if config.uses_redis_sentinel() || config.uses_redis_cluster() { + return Err( + "MONITOR requires a direct Redis node connection; connect to the node you want to monitor".to_string() + ); + } + let (host, port) = state.connection_host_port(connection_id, &config).await?; + let mut runtime_config = config.clone(); + state.apply_session_credential(&config, &mut runtime_config, connection_id); + redis_driver::connect_monitor( + &runtime_config, + &host, + port, + std::time::Duration::from_secs(config.effective_connect_timeout_secs()), + ) + .await +} + pub async fn redis_create_pubsub_core(state: &AppState, connection_id: &str) -> Result { let configs = state.configs.read().await; let config = configs.get(connection_id).ok_or("Connection config not found")?.clone(); diff --git a/crates/dbx-web/src/routes/redis_pubsub_ws.rs b/crates/dbx-web/src/routes/redis_pubsub_ws.rs index 22e13260db..da0388361a 100644 --- a/crates/dbx-web/src/routes/redis_pubsub_ws.rs +++ b/crates/dbx-web/src/routes/redis_pubsub_ws.rs @@ -13,6 +13,8 @@ use crate::state::WebState; #[serde(rename_all = "camelCase")] pub struct PubSubWsParams { pub connection_id: String, + #[serde(default)] + monitor: bool, } pub async fn ws_handler( @@ -21,7 +23,61 @@ pub async fn ws_handler( State(state): State>, ) -> impl IntoResponse { let connection_id = params.connection_id; - ws.on_upgrade(move |socket| handle_pubsub_socket(socket, state, connection_id)) + let owner = dbx_core::session_credentials::current_credential_owner(); + ws.on_upgrade(move |socket| async move { + if params.monitor { + dbx_core::session_credentials::with_credential_owner( + owner, + handle_monitor_socket(socket, state, connection_id), + ) + .await; + } else { + handle_pubsub_socket(socket, state, connection_id).await; + } + }) +} + +async fn handle_monitor_socket(mut socket: WebSocket, state: Arc, connection_id: String) { + let monitor = tokio::select! { + result = dbx_core::redis_ops::redis_create_monitor_core(&state.app, &connection_id) => result, + _ = socket.recv() => return, + }; + let monitor = match monitor { + Ok(monitor) => monitor, + Err(error) => { + let _ = socket.send(Message::Text(serde_json::json!({ "error": error }).to_string().into())).await; + return; + } + }; + if socket.send(Message::Text(serde_json::json!({ "ready": true }).to_string().into())).await.is_err() { + return; + } + let mut messages = monitor.into_on_message::(); + let (mut sender, mut receiver) = socket.split(); + loop { + tokio::select! { + incoming = receiver.next() => { + match incoming { + Some(Ok(Message::Ping(payload))) => { + if sender.send(Message::Pong(payload)).await.is_err() { break; } + } + Some(Ok(Message::Pong(_))) => {} + _ => break, + } + } + message = messages.next() => { + let Some(message) = message else { + let _ = sender.send(Message::Text(serde_json::json!({ "error": "Redis MONITOR connection closed" }).to_string().into())).await; + break; + }; + let event = Message::Text(serde_json::json!({ "message": message }).to_string().into()); + match tokio::time::timeout(std::time::Duration::from_secs(5), sender.send(event)).await { + Ok(Ok(())) => {} + _ => break, + } + } + } + } } async fn handle_pubsub_socket(socket: WebSocket, state: Arc, connection_id: String) { diff --git a/docs/content/docs/redis.cn.mdx b/docs/content/docs/redis.cn.mdx index 3d9a5f5fd9..16eedfb3b6 100644 --- a/docs/content/docs/redis.cn.mdx +++ b/docs/content/docs/redis.cn.mdx @@ -114,6 +114,16 @@ Key 统计和 Slowlog 会额外调用 Redis 命令。大型实例或集群中应 清空 DB、批量删除和 `FLUSH*` 无法撤销。只读连接会拒绝写入；生产环境保护和危险命令规则会继续生效。 +## 实时 MONITOR + +![Redis MONITOR 实时日志](/screenshots/redis-monitor.png) + +在 Redis 查询编辑器中单独执行 `MONITOR`，结果区会持续显示当前节点的命令日志，而不是只返回初始 `OK`。监听范围包含该节点的所有 DB，不受编辑器当前选择的 DB 限制。 + +点击“停止监听”或执行工具栏的停止按钮结束监听；关闭页签也会释放专用连接。界面保留最近 1000 条记录，单条超长记录最多显示 16384 个字符。监听结束后保留已显示的结果。 + +目前支持直连 Redis 节点及其 TLS、隧道配置。Sentinel 和 Cluster 连接请另建目标节点的直连连接；不会自动汇总集群日志。Redis 账号需要具备执行 `MONITOR` 的权限，持续监听会增加服务端开销。 + ## 安全与性能建议 - 生产 Redis 使用只读账号或 ACL，并在 DBX 启用只读/生产保护 diff --git a/docs/content/docs/redis.mdx b/docs/content/docs/redis.mdx index 272c02c641..40e16c6949 100644 --- a/docs/content/docs/redis.mdx +++ b/docs/content/docs/redis.mdx @@ -114,6 +114,16 @@ Key statistics and Slowlog require additional Redis commands. Limit scope and en Flush, bulk delete, and `FLUSH*` operations cannot be undone. Read-only connections reject writes, while production and dangerous-command checks remain active. +## Live MONITOR + +![Live Redis MONITOR logs](/screenshots/redis-monitor.png) + +Run `MONITOR` on its own in the Redis query editor to receive live command logs instead of only the initial `OK`. Monitoring covers all databases on the connected node, regardless of the database selected in the editor. + +Use **Stop listening**, the execution toolbar's stop button, or close the tab to release the dedicated connection. The result retains the latest 1,000 entries, with each entry limited to 16,384 characters. Stopping preserves the displayed results. + +Direct Redis node connections, including their TLS and tunnel settings, are supported. For Sentinel or Cluster, create a direct connection to the node to inspect; logs are not aggregated across nodes. The Redis account needs permission to execute `MONITOR`, which adds server overhead while active. + ## Safety and Performance Guidance - Use a read-only account or ACL for production and enable DBX read-only/production protection diff --git a/docs/public/screenshots/redis-monitor.png b/docs/public/screenshots/redis-monitor.png new file mode 100644 index 0000000000..ea95b5e752 Binary files /dev/null and b/docs/public/screenshots/redis-monitor.png differ diff --git a/src-tauri/src/commands/redis_pubsub_server.rs b/src-tauri/src/commands/redis_pubsub_server.rs index 295ee84bd0..2973d8cfe3 100644 --- a/src-tauri/src/commands/redis_pubsub_server.rs +++ b/src-tauri/src/commands/redis_pubsub_server.rs @@ -56,6 +56,8 @@ impl Drop for PubSubServerState { #[serde(rename_all = "camelCase")] struct PubSubWsParams { connection_id: String, + #[serde(default)] + monitor: bool, } pub fn build_pubsub_router(state: Arc) -> Router { @@ -93,7 +95,56 @@ async fn ws_handler( State(state): State>, ) -> impl IntoResponse { let connection_id = params.connection_id; - ws.on_upgrade(move |socket| handle_socket(socket, state, connection_id)) + ws.on_upgrade(move |socket| async move { + if params.monitor { + handle_monitor_socket(socket, state, connection_id).await; + } else { + handle_socket(socket, state, connection_id).await; + } + }) +} + +async fn handle_monitor_socket(mut socket: WebSocket, state: Arc, connection_id: String) { + let monitor = tokio::select! { + result = dbx_core::redis_ops::redis_create_monitor_core(&state, &connection_id) => result, + _ = socket.recv() => return, + }; + let monitor = match monitor { + Ok(monitor) => monitor, + Err(error) => { + let _ = socket.send(Message::Text(serde_json::json!({ "error": error }).to_string().into())).await; + return; + } + }; + if socket.send(Message::Text(serde_json::json!({ "ready": true }).to_string().into())).await.is_err() { + return; + } + let mut messages = monitor.into_on_message::(); + let (mut sender, mut receiver) = socket.split(); + loop { + tokio::select! { + incoming = receiver.next() => { + match incoming { + Some(Ok(Message::Ping(payload))) => { + if sender.send(Message::Pong(payload)).await.is_err() { break; } + } + Some(Ok(Message::Pong(_))) => {} + _ => break, + } + } + message = messages.next() => { + let Some(message) = message else { + let _ = sender.send(Message::Text(serde_json::json!({ "error": "Redis MONITOR connection closed" }).to_string().into())).await; + break; + }; + let event = Message::Text(serde_json::json!({ "message": message }).to_string().into()); + match tokio::time::timeout(std::time::Duration::from_secs(5), sender.send(event)).await { + Ok(Ok(())) => {} + _ => break, + } + } + } + } } async fn handle_socket(socket: WebSocket, state: Arc, connection_id: String) {