feat(redis): support MONITOR live command log

apps/desktop/src/lib/__tests__/app/openTabsPersistence.spec.ts

@@ -81,6 +81,15 @@ describe("openTabsPersistence originalSql round-trip", () => {
     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" })]);
 

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);
+  });
+});

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,

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<WebSocket> {
-  return new WebSocket(apiWebSocketUrl(`/api/redis/pubsub/ws?connectionId=${encodeURIComponent(connectionId)}`));
+export async function redisPubSubConnect(connectionId: string, monitor = false): Promise<WebSocket> {
+  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<RedisSlowlogEntry[]> {

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<WebSocket> {
+export async function redisPubSubConnect(connectionId: string, monitor = false): Promise<WebSocket> {
   const port = await invoke<number>("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<RedisSlowlogEntry[]> {

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";

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<WebSocket>, 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<void>((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() };
+}