From ef426a82186d9d83c54a3457ebe36149bdd4c60e Mon Sep 17 00:00:00 2001
From: t8y2 <1156263951@qq.com>
Date: Mon, 15 Jun 2026 01:40:06 +0800
Subject: [PATCH] feat(redis): add Pub/Sub publish and subscribe with WebSocket
 real-time push

---
 Cargo.lock                                    |  33 +++
 .../src/components/redis/RedisKeyBrowser.vue  |  13 +-
 .../src/components/redis/RedisPubSubPanel.vue | 275 ++++++++++++++++++
 apps/desktop/src/i18n/locales/en.ts           |  21 ++
 apps/desktop/src/i18n/locales/es.ts           |  21 ++
 apps/desktop/src/i18n/locales/it.ts           |  21 ++
 apps/desktop/src/i18n/locales/pt-BR.ts        |  21 ++
 apps/desktop/src/i18n/locales/zh-CN.ts        |  21 ++
 apps/desktop/src/i18n/locales/zh-TW.ts        |  21 ++
 apps/desktop/src/lib/api.ts                   |   6 +
 apps/desktop/src/lib/http.ts                  |   4 +
 apps/desktop/src/lib/tauri.ts                 |   4 +
 apps/desktop/src/main.ts                      |  29 ++
 apps/desktop/vite.config.ts                   |  15 +-
 crates/dbx-core/src/db/redis_driver.rs        |  55 ++++
 crates/dbx-core/src/redis_ops.rs              |  40 +++
 crates/dbx-web/Cargo.toml                     |   3 +-
 crates/dbx-web/src/main.rs                    |   2 +
 crates/dbx-web/src/routes/mod.rs              |   1 +
 crates/dbx-web/src/routes/redis.rs            |  21 ++
 crates/dbx-web/src/routes/redis_pubsub_ws.rs  | 127 ++++++++
 src-tauri/Cargo.toml                          |   1 +
 src-tauri/src/commands/mod.rs                 |   1 +
 src-tauri/src/commands/redis_cmd.rs           |  12 +
 src-tauri/src/commands/redis_pubsub_server.rs | 145 +++++++++
 src-tauri/src/lib.rs                          |   2 +
 26 files changed, 904 insertions(+), 11 deletions(-)
 create mode 100644 apps/desktop/src/components/redis/RedisPubSubPanel.vue
 create mode 100644 crates/dbx-web/src/routes/redis_pubsub_ws.rs
 create mode 100644 src-tauri/src/commands/redis_pubsub_server.rs

diff --git a/Cargo.lock b/Cargo.lock
index b0be95f3e..a847d0f13 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -646,6 +646,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
 dependencies = [
  "axum-core",
+ "base64 0.22.1",
  "bytes",
  "form_urlencoded",
  "futures-util",
@@ -665,8 +666,10 @@ dependencies = [
  "serde_json",
  "serde_path_to_error",
  "serde_urlencoded",
+ "sha1 0.10.6",
  "sync_wrapper",
  "tokio",
+ "tokio-tungstenite",
  "tower",
  "tower-layer",
  "tower-service",
@@ -1844,6 +1847,7 @@ name = "dbx"
 version = "0.5.32"
 dependencies = [
  "anyhow",
+ "axum",
  "base64 0.22.1",
  "calamine",
  "chrono",
@@ -1947,6 +1951,7 @@ dependencies = [
  "futures",
  "log",
  "pbkdf2 0.12.2",
+ "redis",
  "reqwest 0.12.28",
  "rustls 0.23.40",
  "serde",
@@ -8539,6 +8544,18 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-tungstenite"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.18"
@@ -8832,6 +8849,22 @@ version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
 
+[[package]]
+name = "tungstenite"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
+dependencies = [
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "rand 0.9.4",
+ "sha1 0.10.6",
+ "thiserror 2.0.18",
+]
+
 [[package]]
 name = "twox-hash"
 version = "2.1.2"
diff --git a/apps/desktop/src/components/redis/RedisKeyBrowser.vue b/apps/desktop/src/components/redis/RedisKeyBrowser.vue
index 1988462ba..6301867b0 100644
--- a/apps/desktop/src/components/redis/RedisKeyBrowser.vue
+++ b/apps/desktop/src/components/redis/RedisKeyBrowser.vue
@@ -1,7 +1,7 @@
 <script setup lang="ts">
 import { computed, nextTick, ref, onMounted, onUnmounted, onActivated, onDeactivated, watch } from "vue";
 import { useI18n } from "vue-i18n";
-import { Search, RefreshCw, Loader2, ChevronRight, ChevronDown, FolderClosed, FolderOpen, Trash2, Plus, KeyRound, TerminalSquare, Asterisk, History } from "@lucide/vue";
+import { Search, RefreshCw, Loader2, ChevronRight, ChevronDown, FolderClosed, FolderOpen, Trash2, Plus, KeyRound, TerminalSquare, Asterisk, History, Radio } from "@lucide/vue";
 import { RecycleScroller } from "vue-virtual-scroller";
 import "vue-virtual-scroller/dist/vue-virtual-scroller.css";
 import { Splitpanes, Pane } from "splitpanes";
@@ -15,6 +15,7 @@ import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs";
 import { Switch } from "@/components/ui/switch";
 import DangerConfirmDialog from "@/components/editor/DangerConfirmDialog.vue";
 import RedisValueViewer from "./RedisValueViewer.vue";
+import RedisPubSubPanel from "./RedisPubSubPanel.vue";
 import * as api from "@/lib/api";
 import type { RedisKeyInfo, RedisScanResult, HistoryEntry } from "@/lib/api";
 import { uuid } from "@/lib/utils";
@@ -44,7 +45,7 @@ interface CreateKeyEntry {
   field?: string;
   score?: string;
 }
-type RedisSidePanel = "detail" | "command";
+type RedisSidePanel = "detail" | "command" | "pubsub";
 type RedisCommandHistoryEntry = {
   id: number;
   prompt: string;
@@ -942,6 +943,10 @@ defineExpose({ focusSearch });
                   <TerminalSquare class="size-3.5" />
                   {{ t("redis.commandLine") }}
                 </TabsTrigger>
+                <TabsTrigger value="pubsub" class="h-6 flex-none gap-1.5 rounded-md px-2 text-xs">
+                  <Radio class="size-3.5" />
+                  {{ t("redis.pubsub") }}
+                </TabsTrigger>
               </TabsList>
               <Button v-if="activeSidePanel === 'command'" variant="ghost" size="icon" class="h-6 w-6" :title="t('redis.clearHistory')" @click="clearPersistedRedisHistory">
                 <History class="size-3.5" />
@@ -987,6 +992,10 @@ defineExpose({ focusSearch });
                 </form>
               </div>
             </TabsContent>
+
+            <TabsContent value="pubsub" class="m-0 min-h-0 flex-1 flex flex-col">
+              <RedisPubSubPanel :connection-id="connectionId" :db="db" />
+            </TabsContent>
           </Tabs>
         </div>
       </Pane>
diff --git a/apps/desktop/src/components/redis/RedisPubSubPanel.vue b/apps/desktop/src/components/redis/RedisPubSubPanel.vue
new file mode 100644
index 000000000..33c267989
--- /dev/null
+++ b/apps/desktop/src/components/redis/RedisPubSubPanel.vue
@@ -0,0 +1,275 @@
+<script setup lang="ts">
+import { ref, nextTick, onBeforeUnmount } from "vue";
+import { useI18n } from "vue-i18n";
+import * as api from "@/lib/api";
+import { useToast } from "@/composables/useToast";
+
+const props = defineProps<{
+  connectionId: string;
+  db: number;
+}>();
+
+const { t } = useI18n();
+const { toast } = useToast();
+
+// State
+const channels = ref<string[]>([]);
+const patterns = ref<string[]>([]);
+const newChannel = ref("");
+const newPattern = ref("");
+
+const publishChannel = ref("");
+const publishMessage = ref("");
+
+interface PubSubEntry {
+  id: number;
+  channel: string;
+  pattern: string | null;
+  payload: string;
+  timestamp: Date;
+}
+const messages = ref<PubSubEntry[]>([]);
+let nextId = 0;
+
+let ws: WebSocket | null = null;
+const connected = ref(false);
+const connecting = ref(false);
+
+const messagesContainer = ref<HTMLElement | null>(null);
+
+function scrollToBottom() {
+  void nextTick(() => {
+    if (messagesContainer.value) {
+      messagesContainer.value.scrollTop = messagesContainer.value.scrollHeight;
+    }
+  });
+}
+
+function addMessage(channel: string, pattern: string | null, payload: string) {
+  messages.value.push({
+    id: ++nextId,
+    channel,
+    pattern,
+    payload,
+    timestamp: new Date(),
+  });
+  // Keep max 500 messages
+  if (messages.value.length > 500) {
+    messages.value.shift();
+  }
+  scrollToBottom();
+}
+
+// Connect WebSocket
+async function connect() {
+  if (ws && ws.readyState === WebSocket.OPEN) return;
+  connecting.value = true;
+  try {
+    ws = api.redisPubSubConnect(props.connectionId);
+
+    ws.onopen = () => {
+      connected.value = true;
+      connecting.value = false;
+      // Re-subscribe existing channels
+      if (channels.value.length > 0) {
+        ws!.send(JSON.stringify({ type: "subscribe", channels: channels.value }));
+      }
+      if (patterns.value.length > 0) {
+        ws!.send(JSON.stringify({ type: "psubscribe", patterns: patterns.value }));
+      }
+    };
+
+    ws.onmessage = (event) => {
+      try {
+        const data = JSON.parse(event.data as string);
+        if (data.error) {
+          toast(data.error, 5000);
+          return;
+        }
+        addMessage(data.channel, data.pattern ?? null, data.payload);
+      } catch {
+        // Non-JSON message, skip
+      }
+    };
+
+    ws.onclose = () => {
+      connected.value = false;
+      connecting.value = false;
+      ws = null;
+    };
+
+    ws.onerror = () => {
+      connected.value = false;
+      connecting.value = false;
+      ws = null;
+    };
+  } catch (e) {
+    connecting.value = false;
+    toast(t("redis.pubsubWsConnectFailed", { error: e instanceof Error ? e.message : String(e) }), 5000);
+  }
+}
+
+function disconnect() {
+  if (ws) {
+    ws.close();
+    ws = null;
+    connected.value = false;
+  }
+}
+
+function subscribe() {
+  const ch = newChannel.value.trim();
+  if (!ch) return;
+  if (channels.value.includes(ch)) return;
+  channels.value.push(ch);
+  newChannel.value = "";
+  if (ws && ws.readyState === WebSocket.OPEN) {
+    ws.send(JSON.stringify({ type: "subscribe", channels: [ch] }));
+  }
+}
+
+function psubscribe() {
+  const pat = newPattern.value.trim();
+  if (!pat) return;
+  if (patterns.value.includes(pat)) return;
+  patterns.value.push(pat);
+  newPattern.value = "";
+  if (ws && ws.readyState === WebSocket.OPEN) {
+    ws.send(JSON.stringify({ type: "psubscribe", patterns: [pat] }));
+  }
+}
+
+function unsubscribe(channel: string) {
+  channels.value = channels.value.filter((c) => c !== channel);
+  if (ws && ws.readyState === WebSocket.OPEN) {
+    ws.send(JSON.stringify({ type: "unsubscribe", channels: [channel] }));
+  }
+}
+
+function punsubscribe(pattern: string) {
+  patterns.value = patterns.value.filter((p) => p !== pattern);
+  if (ws && ws.readyState === WebSocket.OPEN) {
+    ws.send(JSON.stringify({ type: "punsubscribe", patterns: [pattern] }));
+  }
+}
+
+async function publish() {
+  const ch = publishChannel.value.trim();
+  const msg = publishMessage.value;
+  if (!ch || !msg) return;
+  try {
+    await api.redisPubSubPublish(props.connectionId, props.db, ch, msg);
+    // Echo locally only if subscribed — Redis PubSub doesn't deliver to publisher
+    const isSubscribed = channels.value.includes(ch) || patterns.value.some((p) => new RegExp("^" + p.replace(/\*/g, ".*") + "$").test(ch));
+    if (isSubscribed) {
+      addMessage(ch, null, msg);
+    }
+    publishMessage.value = "";
+  } catch (e) {
+    toast(t("redis.pubsubPublishFailed", { error: e instanceof Error ? e.message : String(e) }), 3000);
+  }
+}
+
+function clearMessages() {
+  messages.value = [];
+}
+
+// Auto-connect with a short delay for server readiness
+setTimeout(connect, 500);
+
+onBeforeUnmount(() => {
+  disconnect();
+});
+</script>
+
+<template>
+  <div class="flex flex-col h-full">
+    <!-- Connection bar -->
+    <div class="flex items-center gap-2 px-3 py-1.5 border-b bg-muted/30">
+      <span class="text-xs font-medium">{{ t("redis.pubsub") }}</span>
+      <span class="flex-1"></span>
+      <span class="inline-flex items-center gap-1 text-xs" :class="connected ? 'text-green-600' : 'text-muted-foreground'">
+        <span class="inline-block w-2 h-2 rounded-full" :class="connected ? 'bg-green-500' : 'bg-gray-400'"></span>
+        {{ connected ? t("redis.pubsubConnected") : connecting ? t("redis.pubsubConnecting") : t("redis.pubsubDisconnected") }}
+      </span>
+      <button v-if="!connected" class="text-xs px-2 py-0.5 rounded bg-primary text-primary-foreground hover:bg-primary/90" :disabled="connecting" @click="connect">
+        {{ t("redis.pubsubConnect") }}
+      </button>
+      <button v-else class="text-xs px-2 py-0.5 rounded border hover:bg-muted" @click="disconnect">
+        {{ t("redis.pubsubDisconnect") }}
+      </button>
+    </div>
+
+    <!-- Subscriptions -->
+    <div class="px-3 py-2 border-b space-y-2">
+      <!-- Channel subscriptions -->
+      <div>
+        <div class="text-xs font-medium mb-1">{{ t("redis.pubsubChannels") }}</div>
+        <div class="flex gap-1 mb-1">
+          <input v-model="newChannel" type="text" autocomplete="off" autocapitalize="off" autocorrect="off" spellcheck="false" class="flex-1 h-7 px-2 text-xs border rounded bg-background" :placeholder="t('redis.pubsubChannelPlaceholder')" @keydown.enter="subscribe" />
+          <button class="text-xs px-2 py-0.5 rounded bg-primary text-primary-foreground hover:bg-primary/90" @click="subscribe">
+            {{ t("redis.pubsubSubscribe") }}
+          </button>
+        </div>
+        <div v-if="channels.length > 0" class="flex flex-wrap gap-1">
+          <span v-for="ch in channels" :key="ch" class="inline-flex items-center gap-1 text-xs px-1.5 py-0.5 rounded bg-blue-100 text-blue-800 dark:bg-blue-900 dark:text-blue-200">
+            {{ ch }}
+            <button class="hover:text-red-500" @click="unsubscribe(ch)">×</button>
+          </span>
+        </div>
+      </div>
+
+      <!-- Pattern subscriptions -->
+      <div>
+        <div class="text-xs font-medium mb-1">{{ t("redis.pubsubPatterns") }}</div>
+        <div class="flex gap-1 mb-1">
+          <input v-model="newPattern" type="text" autocomplete="off" autocapitalize="off" autocorrect="off" spellcheck="false" class="flex-1 h-7 px-2 text-xs border rounded bg-background" :placeholder="t('redis.pubsubPatternPlaceholder')" @keydown.enter="psubscribe" />
+          <button class="text-xs px-2 py-0.5 rounded bg-primary text-primary-foreground hover:bg-primary/90" @click="psubscribe">
+            {{ t("redis.pubsubPsubscribe") }}
+          </button>
+        </div>
+        <div v-if="patterns.length > 0" class="flex flex-wrap gap-1">
+          <span v-for="pat in patterns" :key="pat" class="inline-flex items-center gap-1 text-xs px-1.5 py-0.5 rounded bg-purple-100 text-purple-800 dark:bg-purple-900 dark:text-purple-200">
+            {{ pat }}
+            <button class="hover:text-red-500" @click="punsubscribe(pat)">×</button>
+          </span>
+        </div>
+      </div>
+    </div>
+
+    <!-- Messages -->
+    <div class="flex-1 flex flex-col min-h-0">
+      <div class="flex items-center gap-2 px-3 py-1 border-b">
+        <span class="text-xs font-medium">{{ t("redis.pubsubMessages") }}</span>
+        <span class="text-xs text-muted-foreground">({{ messages.length }})</span>
+        <span class="flex-1"></span>
+        <button class="text-xs px-2 py-0.5 rounded border hover:bg-muted" @click="clearMessages">
+          {{ t("redis.pubsubClear") }}
+        </button>
+      </div>
+      <div ref="messagesContainer" class="flex-1 overflow-auto px-3 py-1 font-mono text-xs">
+        <div v-if="messages.length === 0" class="text-muted-foreground py-4 text-center">
+          {{ t("redis.pubsubEmpty") }}
+        </div>
+        <div v-for="msg in messages" :key="msg.id" class="py-0.5 border-b border-dashed border-border/50">
+          <span class="text-muted-foreground">{{ msg.timestamp.toLocaleTimeString() }}</span>
+          <span class="ml-2 font-semibold text-blue-600 dark:text-blue-400">{{ msg.channel }}</span>
+          <span v-if="msg.pattern" class="ml-1 text-purple-500">({{ msg.pattern }})</span>
+          <span class="ml-2">{{ msg.payload }}</span>
+        </div>
+      </div>
+    </div>
+
+    <!-- Publish -->
+    <div class="px-3 py-2 border-t space-y-1.5">
+      <div class="text-xs font-medium">{{ t("redis.pubsubPublish") }}</div>
+      <input v-model="publishChannel" type="text" class="w-full h-7 px-2 text-xs border rounded bg-background" :placeholder="t('redis.pubsubPublishChannel')" autocomplete="off" autocapitalize="off" autocorrect="off" spellcheck="false" />
+      <div class="flex gap-1">
+        <input v-model="publishMessage" type="text" autocomplete="off" autocapitalize="off" autocorrect="off" spellcheck="false" class="flex-1 h-7 px-2 text-xs border rounded bg-background" :placeholder="t('redis.pubsubPublishMessage')" @keydown.enter="publish" />
+        <button class="text-xs px-3 py-0.5 rounded bg-primary text-primary-foreground hover:bg-primary/90" @click="publish">
+          {{ t("redis.pubsubSend") }}
+        </button>
+      </div>
+    </div>
+  </div>
+</template>
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index eaea75b19..2b89e5ae8 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -1513,6 +1513,27 @@
     clearHistory: "Clear command history",
     historyCleared: "Redis command history cleared",
     blockedCommand: "Command {command} is blocked for safety. Disable the shield icon in the toolbar to allow.",
+    pubsub: "Pub/Sub",
+    pubsubConnected: "Connected",
+    pubsubConnecting: "Connecting...",
+    pubsubDisconnected: "Disconnected",
+    pubsubConnect: "Connect",
+    pubsubDisconnect: "Disconnect",
+    pubsubChannels: "Channels",
+    pubsubChannelPlaceholder: "Channel name",
+    pubsubSubscribe: "Subscribe",
+    pubsubPatterns: "Patterns",
+    pubsubPatternPlaceholder: "Pattern (e.g. news:*)",
+    pubsubPsubscribe: "PSubscribe",
+    pubsubMessages: "Messages",
+    pubsubClear: "Clear",
+    pubsubEmpty: "No messages yet. Subscribe to a channel to receive messages.",
+    pubsubPublish: "Publish",
+    pubsubPublishChannel: "Channel",
+    pubsubPublishMessage: "Message",
+    pubsubSend: "Send",
+    pubsubPublishFailed: "Publish failed: {error}",
+    pubsubWsConnectFailed: "WebSocket connection failed: {error}",
   },
   mongo: {
     documents: "{count} documents",
diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts
index a9fe82009..587db099c 100644
--- a/apps/desktop/src/i18n/locales/es.ts
+++ b/apps/desktop/src/i18n/locales/es.ts
@@ -1263,6 +1263,27 @@
     clearHistory: "Borrar historial de comandos",
     historyCleared: "Historial de comandos Redis borrado",
     blockedCommand: "El comando {command} está bloqueado por seguridad. Desactiva el icono de escudo en la barra de herramientas para permitirlo.",
+    pubsub: "Pub/Sub",
+    pubsubConnected: "Conectado",
+    pubsubConnecting: "Conectando...",
+    pubsubDisconnected: "Desconectado",
+    pubsubConnect: "Conectar",
+    pubsubDisconnect: "Desconectar",
+    pubsubChannels: "Canales",
+    pubsubChannelPlaceholder: "Nombre del canal",
+    pubsubSubscribe: "Suscribirse",
+    pubsubPatterns: "Patrones",
+    pubsubPatternPlaceholder: "Patrón (ej. news:*)",
+    pubsubPsubscribe: "PSuscribirse",
+    pubsubMessages: "Mensajes",
+    pubsubClear: "Limpiar",
+    pubsubEmpty: "Sin mensajes. Suscríbete a un canal para recibir mensajes.",
+    pubsubPublish: "Publicar",
+    pubsubPublishChannel: "Canal",
+    pubsubPublishMessage: "Mensaje",
+    pubsubSend: "Enviar",
+    pubsubPublishFailed: "Publicación fallida: {error}",
+    pubsubWsConnectFailed: "Conexión WebSocket fallida: {error}",
   },
   mongo: {
     documents: "{count} documentos",
diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts
index ccc075af6..c4aabd999 100644
--- a/apps/desktop/src/i18n/locales/it.ts
+++ b/apps/desktop/src/i18n/locales/it.ts
@@ -1375,6 +1375,27 @@
     clearHistory: "Cancella cronologia comandi",
     historyCleared: "Cronologia comandi Redis cancellata",
     blockedCommand: "Il comando {command} è bloccato per sicurezza. Disattiva l'icona scudo nella barra degli strumenti per consentirlo.",
+    pubsub: "Pub/Sub",
+    pubsubConnected: "Connesso",
+    pubsubConnecting: "Connessione in corso...",
+    pubsubDisconnected: "Disconnesso",
+    pubsubConnect: "Connetti",
+    pubsubDisconnect: "Disconnetti",
+    pubsubChannels: "Canali",
+    pubsubChannelPlaceholder: "Nome canale",
+    pubsubSubscribe: "Sottoscrivi",
+    pubsubPatterns: "Pattern",
+    pubsubPatternPlaceholder: "Pattern (es. news:*)",
+    pubsubPsubscribe: "PSottoscrivi",
+    pubsubMessages: "Messaggi",
+    pubsubClear: "Cancella",
+    pubsubEmpty: "Nessun messaggio. Sottoscrivi un canale per ricevere messaggi.",
+    pubsubPublish: "Pubblica",
+    pubsubPublishChannel: "Canale",
+    pubsubPublishMessage: "Messaggio",
+    pubsubSend: "Invia",
+    pubsubPublishFailed: "Pubblicazione fallita: {error}",
+    pubsubWsConnectFailed: "Connessione WebSocket fallita: {error}",
   },
   mongo: {
     documents: "{count} documenti",
diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts
index 66f35e016..fd3f26039 100644
--- a/apps/desktop/src/i18n/locales/pt-BR.ts
+++ b/apps/desktop/src/i18n/locales/pt-BR.ts
@@ -1386,6 +1386,27 @@
     clearHistory: "Limpar histórico de comandos",
     historyCleared: "Histórico de comandos Redis limpo",
     blockedCommand: "O comando {command} está bloqueado por segurança. Desative o ícone de escudo na barra de ferramentas para permiti-lo.",
+    pubsub: "Pub/Sub",
+    pubsubConnected: "Conectado",
+    pubsubConnecting: "Conectando...",
+    pubsubDisconnected: "Desconectado",
+    pubsubConnect: "Conectar",
+    pubsubDisconnect: "Desconectar",
+    pubsubChannels: "Canais",
+    pubsubChannelPlaceholder: "Nome do canal",
+    pubsubSubscribe: "Inscrever-se",
+    pubsubPatterns: "Padrões",
+    pubsubPatternPlaceholder: "Padrão (ex. news:*)",
+    pubsubPsubscribe: "PInscrever-se",
+    pubsubMessages: "Mensagens",
+    pubsubClear: "Limpar",
+    pubsubEmpty: "Nenhuma mensagem. Inscreva-se em um canal para receber mensagens.",
+    pubsubPublish: "Publicar",
+    pubsubPublishChannel: "Canal",
+    pubsubPublishMessage: "Mensagem",
+    pubsubSend: "Enviar",
+    pubsubPublishFailed: "Publicação falhou: {error}",
+    pubsubWsConnectFailed: "Conexão WebSocket falhou: {error}",
   },
   mongo: {
     documents: "{count} documentos",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index 54b1617bb..9b71282f1 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -1512,6 +1512,27 @@
     clearHistory: "清除命令历史",
     historyCleared: "Redis 命令历史已清除",
     blockedCommand: "命令 {command} 因安全原因已被拦截。点击工具栏盾牌图标可关闭拦截。",
+    pubsub: "发布/订阅",
+    pubsubConnected: "已连接",
+    pubsubConnecting: "连接中...",
+    pubsubDisconnected: "未连接",
+    pubsubConnect: "连接",
+    pubsubDisconnect: "断开",
+    pubsubChannels: "频道",
+    pubsubChannelPlaceholder: "频道名称",
+    pubsubSubscribe: "订阅",
+    pubsubPatterns: "模式匹配",
+    pubsubPatternPlaceholder: "模式 (如 news:*)",
+    pubsubPsubscribe: "模式订阅",
+    pubsubMessages: "消息",
+    pubsubClear: "清空",
+    pubsubEmpty: "暂无消息，订阅一个频道来接收消息。",
+    pubsubPublish: "发布消息",
+    pubsubPublishChannel: "频道",
+    pubsubPublishMessage: "消息内容",
+    pubsubSend: "发送",
+    pubsubPublishFailed: "发布失败: {error}",
+    pubsubWsConnectFailed: "WebSocket 连接失败: {error}",
   },
   mongo: {
     documents: "{count} 个文档",
diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts
index 3a945e199..6e27b78cc 100644
--- a/apps/desktop/src/i18n/locales/zh-TW.ts
+++ b/apps/desktop/src/i18n/locales/zh-TW.ts
@@ -1363,6 +1363,27 @@
     clearHistory: "清除命令歷史",
     historyCleared: "Redis 命令歷史已清除",
     blockedCommand: "命令 {command} 因安全原因已被攔截。點擊工具列盾牌圖示可關閉攔截。",
+    pubsub: "發布/訂閱",
+    pubsubConnected: "已連線",
+    pubsubConnecting: "連線中...",
+    pubsubDisconnected: "未連線",
+    pubsubConnect: "連線",
+    pubsubDisconnect: "斷開",
+    pubsubChannels: "頻道",
+    pubsubChannelPlaceholder: "頻道名稱",
+    pubsubSubscribe: "訂閱",
+    pubsubPatterns: "模式匹配",
+    pubsubPatternPlaceholder: "模式 (如 news:*)",
+    pubsubPsubscribe: "模式訂閱",
+    pubsubMessages: "訊息",
+    pubsubClear: "清空",
+    pubsubEmpty: "暫無訊息，訂閱一個頻道來接收訊息。",
+    pubsubPublish: "發布訊息",
+    pubsubPublishChannel: "頻道",
+    pubsubPublishMessage: "訊息內容",
+    pubsubSend: "傳送",
+    pubsubPublishFailed: "發布失敗: {error}",
+    pubsubWsConnectFailed: "WebSocket 連線失敗: {error}",
   },
   mongo: {
     documents: "{count} 個文件",
diff --git a/apps/desktop/src/lib/api.ts b/apps/desktop/src/lib/api.ts
index ea555dbaf..58122d7c0 100644
--- a/apps/desktop/src/lib/api.ts
+++ b/apps/desktop/src/lib/api.ts
@@ -261,6 +261,12 @@ export const redisDeleteKeys = forward("redisDeleteKeys");
 export const redisFlushDb = forward("redisFlushDb");
 export const redisExecuteCommand = forward("redisExecuteCommand");
 export const redisLoadMore = forward("redisLoadMore");
+export const redisPubSubPublish = forward("redisPubSubPublish");
+
+export function redisPubSubConnect(connectionId: string): WebSocket {
+  const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
+  return new WebSocket(`${protocol}//${window.location.host}/api/redis/pubsub/ws?connectionId=${encodeURIComponent(connectionId)}`);
+}
 
 // etcd
 export const etcdListPrefix = forward("etcdListPrefix");
diff --git a/apps/desktop/src/lib/http.ts b/apps/desktop/src/lib/http.ts
index b755c382a..7efcd5331 100644
--- a/apps/desktop/src/lib/http.ts
+++ b/apps/desktop/src/lib/http.ts
@@ -1354,6 +1354,10 @@ export async function redisLoadMore(connectionId: string, db: number, keyRaw: st
   return post("/api/redis/load-more", { connectionId, db, keyRaw, keyType, cursor, count });
 }
 
+export async function redisPubSubPublish(connectionId: string, db: number, channel: string, message: string): Promise<{ subscribers: number }> {
+  return post("/api/redis/pubsub/publish", { connectionId, db, channel, message });
+}
+
 // ---------------------------------------------------------------------------
 // etcd
 // ---------------------------------------------------------------------------
diff --git a/apps/desktop/src/lib/tauri.ts b/apps/desktop/src/lib/tauri.ts
index ee5c1008d..a2297955b 100644
--- a/apps/desktop/src/lib/tauri.ts
+++ b/apps/desktop/src/lib/tauri.ts
@@ -1192,6 +1192,10 @@ export async function redisLoadMore(connectionId: string, db: number, keyRaw: st
   return invoke("redis_load_more", { connectionId, db, keyRaw, keyType, cursor, count });
 }
 
+export async function redisPubSubPublish(connectionId: string, db: number, channel: string, message: string): Promise<{ subscribers: number }> {
+  return invoke("redis_pubsub_publish", { connectionId, db, channel, message });
+}
+
 // --- etcd ---
 export type KvValueEncoding = "utf8" | "base64";
 
diff --git a/apps/desktop/src/main.ts b/apps/desktop/src/main.ts
index 0a1ed8469..c40f4b99c 100644
--- a/apps/desktop/src/main.ts
+++ b/apps/desktop/src/main.ts
@@ -45,6 +45,33 @@ function installStartupErrorHandlers() {
   });
 }
 
+function installGlobalInputAttrs() {
+  const ATTRS: [string, string][] = [
+    ["autocomplete", "off"],
+    ["autocapitalize", "off"],
+    ["autocorrect", "off"],
+    ["spellcheck", "false"],
+  ];
+  const MARKER = "data-input-attrs-set";
+  const apply = (el: Element) => {
+    if ((el.tagName === "INPUT" || el.tagName === "TEXTAREA") && !el.hasAttribute(MARKER)) {
+      for (const [k, v] of ATTRS) el.setAttribute(k, v);
+      el.setAttribute(MARKER, "");
+    }
+  };
+  document.querySelectorAll("input, textarea").forEach(apply);
+  new MutationObserver((mutations) => {
+    for (const m of mutations) {
+      for (const node of m.addedNodes) {
+        if (node instanceof Element) {
+          apply(node);
+          node.querySelectorAll("input, textarea").forEach(apply);
+        }
+      }
+    }
+  }).observe(document.body, { childList: true, subtree: true });
+}
+
 async function bootstrap() {
   console.log("[STARTUP] frontend bootstrap begin");
   const [{ default: i18n, loadSavedLocale }, { default: App }] = await Promise.all([import("./i18n"), import("./App.vue")]);
@@ -58,6 +85,8 @@ async function bootstrap() {
   app.use(VueVirtualScroller);
   app.mount("#root");
   console.log("[STARTUP] vue mounted");
+
+  installGlobalInputAttrs();
 }
 
 installDebugLogCapture();
diff --git a/apps/desktop/vite.config.ts b/apps/desktop/vite.config.ts
index 268390ac2..6aaf76078 100644
--- a/apps/desktop/vite.config.ts
+++ b/apps/desktop/vite.config.ts
@@ -75,14 +75,13 @@ export default defineConfig(async () => ({
           port: 1421,
         }
       : undefined,
-    proxy: isTauri
-      ? undefined
-      : {
-          "/api": {
-            target: "http://localhost:4224",
-            changeOrigin: true,
-          },
-        },
+    proxy: {
+      "/api": {
+        target: "http://localhost:4224",
+        changeOrigin: true,
+        ws: true,
+      },
+    },
     watch: {
       ignored: ["**/src-tauri/**"],
     },
diff --git a/crates/dbx-core/src/db/redis_driver.rs b/crates/dbx-core/src/db/redis_driver.rs
index 1329e9fe3..19e767712 100644
--- a/crates/dbx-core/src/db/redis_driver.rs
+++ b/crates/dbx-core/src/db/redis_driver.rs
@@ -68,6 +68,13 @@ pub struct RedisCommandResult {
     pub value: serde_json::Value,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PubSubMessage {
+    pub channel: String,
+    pub pattern: Option<String>,
+    pub payload: String,
+}
+
 pub enum RedisConnection {
     Direct(Mutex<redis::aio::MultiplexedConnection>),
     Cluster(RedisClusterPool),
@@ -456,6 +463,54 @@ pub async fn connect_direct_node(
     connect_client(client).await
 }
 
+pub async fn connect_pubsub(
+    config: &ConnectionConfig,
+    host: &str,
+    port: u16,
+    timeout: std::time::Duration,
+) -> Result<redis::aio::PubSub, String> {
+    let mut last_error = None;
+    for auth in redis_auth_candidates(&config.username, &config.password) {
+        let client = redis::Client::open(connection_info(
+            host,
+            port,
+            config.ssl,
+            config.redis_tls_insecure(),
+            &auth.username,
+            &auth.password,
+            redis_database_index(config),
+        ))
+        .map_err(|e| format!("Redis connection failed: {e}"))?;
+        match tokio::time::timeout(timeout, client.get_async_pubsub()).await {
+            Ok(Ok(pubsub)) => return Ok(pubsub),
+            Ok(Err(err)) => {
+                let err_str = err.to_string();
+                if last_error.is_none() || is_redis_auth_error(&err_str) {
+                    let should_retry = is_redis_auth_error(&err_str);
+                    last_error = Some(err_str);
+                    if !should_retry {
+                        break;
+                    }
+                } else {
+                    return Err(err_str);
+                }
+            }
+            Err(_) => {
+                last_error = Some(format!("Redis PubSub connection timed out ({}s)", timeout.as_secs()));
+                break;
+            }
+        }
+    }
+    Err(last_error.unwrap_or_else(|| "Redis PubSub connection failed".to_string()))
+}
+
+pub async fn publish_message<C>(con: &mut C, channel: &str, message: &str) -> Result<u64, String>
+where
+    C: ConnectionLike + Send + Sync + Unpin,
+{
+    redis::cmd("PUBLISH").arg(channel).arg(message).query_async(con).await.map_err(|e| e.to_string())
+}
+
 pub async fn list_databases<C>(con: &mut C) -> Result<Vec<RedisDatabaseInfo>, String>
 where
     C: ConnectionLike + Send + Sync + Unpin,
diff --git a/crates/dbx-core/src/redis_ops.rs b/crates/dbx-core/src/redis_ops.rs
index 712e53e52..7a0a577ab 100644
--- a/crates/dbx-core/src/redis_ops.rs
+++ b/crates/dbx-core/src/redis_ops.rs
@@ -739,3 +739,43 @@ pub async fn redis_load_more_in_db_core(
         _ => Err("Not a Redis connection".to_string()),
     }
 }
+
+pub async fn redis_publish_core(
+    state: &AppState,
+    connection_id: &str,
+    db: u32,
+    channel: &str,
+    message: &str,
+) -> Result<u64, String> {
+    ensure_redis_pool(state, connection_id).await?;
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::Redis(redis) => match redis {
+            RedisConnection::Direct(con) => {
+                let mut con = con.lock().await;
+                redis_driver::select_db(&mut *con, db).await?;
+                redis_driver::publish_message(&mut *con, channel, message).await
+            }
+            RedisConnection::Cluster(cluster) => {
+                redis_driver::ensure_cluster_db(db)?;
+                let mut con = cluster.connection.lock().await;
+                redis_driver::publish_message(&mut *con, channel, message).await
+            }
+        },
+        _ => Err("Not a Redis connection".to_string()),
+    }
+}
+
+pub async fn redis_create_pubsub_core(state: &AppState, connection_id: &str) -> Result<redis::aio::PubSub, String> {
+    let configs = state.configs.read().await;
+    let config = configs.get(connection_id).ok_or("Connection config not found")?.clone();
+    drop(configs);
+
+    if config.db_type != crate::models::connection::DatabaseType::Redis {
+        return Err("Not a Redis connection".to_string());
+    }
+
+    let (host, port) = state.connection_host_port(connection_id, &config).await?;
+    let timeout = std::time::Duration::from_secs(config.effective_connect_timeout_secs());
+    redis_driver::connect_pubsub(&config, &host, port, timeout).await
+}
diff --git a/crates/dbx-web/Cargo.toml b/crates/dbx-web/Cargo.toml
index f2bd3c164..21c81dbf4 100644
--- a/crates/dbx-web/Cargo.toml
+++ b/crates/dbx-web/Cargo.toml
@@ -10,7 +10,8 @@ path = "src/main.rs"
 
 [dependencies]
 dbx-core = { path = "../dbx-core" }
-axum = { version = "0.8", features = ["multipart"] }
+redis = { version = "0.32", features = ["tokio-comp"] }
+axum = { version = "0.8", features = ["multipart", "ws"] }
 tower-http = { version = "0.6", features = ["cors", "fs", "compression-gzip", "trace"] }
 tokio = { version = "1", features = ["full"] }
 serde = { version = "1.0", features = ["derive"] }
diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs
index 7eafe763f..82c1f748e 100644
--- a/crates/dbx-web/src/main.rs
+++ b/crates/dbx-web/src/main.rs
@@ -258,6 +258,8 @@ async fn main() {
         .route("/redis/delete-keys", post(routes::redis::delete_keys))
         .route("/redis/flush-db", post(routes::redis::flush_db))
         .route("/redis/execute-command", post(routes::redis::execute_command))
+        .route("/redis/pubsub/publish", post(routes::redis::publish_message))
+        .route("/redis/pubsub/ws", get(routes::redis_pubsub_ws::ws_handler))
         // etcd
         .route("/etcd/list-prefix", post(routes::etcd::list_prefix))
         .route("/etcd/get", post(routes::etcd::get))
diff --git a/crates/dbx-web/src/routes/mod.rs b/crates/dbx-web/src/routes/mod.rs
index b1db531cd..68de749c3 100644
--- a/crates/dbx-web/src/routes/mod.rs
+++ b/crates/dbx-web/src/routes/mod.rs
@@ -12,6 +12,7 @@ pub mod mongo;
 pub mod plugins;
 pub mod query;
 pub mod redis;
+pub mod redis_pubsub_ws;
 pub mod saved_sql;
 pub mod schema;
 pub mod schema_cache;
diff --git a/crates/dbx-web/src/routes/redis.rs b/crates/dbx-web/src/routes/redis.rs
index 64c6685d2..30af50818 100644
--- a/crates/dbx-web/src/routes/redis.rs
+++ b/crates/dbx-web/src/routes/redis.rs
@@ -156,6 +156,15 @@ pub struct RedisCommandRequest {
     pub skip_safety_check: Option<bool>,
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RedisPubSubPublishRequest {
+    pub connection_id: String,
+    pub db: u32,
+    pub channel: String,
+    pub message: String,
+}
+
 pub async fn list_databases(
     State(state): State<Arc<WebState>>,
     Json(req): Json<RedisConnectionRequest>,
@@ -452,3 +461,15 @@ pub async fn execute_command(
     .map_err(AppError)?;
     Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?))
 }
+
+pub async fn publish_message(
+    State(state): State<Arc<WebState>>,
+    Json(req): Json<RedisPubSubPublishRequest>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    ensure_writable(&state.app, &req.connection_id, "PUBLISH").await?;
+    let count =
+        dbx_core::redis_ops::redis_publish_core(&state.app, &req.connection_id, req.db, &req.channel, &req.message)
+            .await
+            .map_err(AppError)?;
+    Ok(Json(serde_json::json!({ "subscribers": count })))
+}
diff --git a/crates/dbx-web/src/routes/redis_pubsub_ws.rs b/crates/dbx-web/src/routes/redis_pubsub_ws.rs
new file mode 100644
index 000000000..22e13260d
--- /dev/null
+++ b/crates/dbx-web/src/routes/redis_pubsub_ws.rs
@@ -0,0 +1,127 @@
+use std::sync::Arc;
+
+use axum::extract::ws::{Message, WebSocket};
+use axum::extract::State;
+use axum::extract::{Query, WebSocketUpgrade};
+use axum::response::IntoResponse;
+use futures::{SinkExt, StreamExt};
+use serde::Deserialize;
+
+use crate::state::WebState;
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PubSubWsParams {
+    pub connection_id: String,
+}
+
+pub async fn ws_handler(
+    ws: WebSocketUpgrade,
+    Query(params): Query<PubSubWsParams>,
+    State(state): State<Arc<WebState>>,
+) -> impl IntoResponse {
+    let connection_id = params.connection_id;
+    ws.on_upgrade(move |socket| handle_pubsub_socket(socket, state, connection_id))
+}
+
+async fn handle_pubsub_socket(socket: WebSocket, state: Arc<WebState>, connection_id: String) {
+    // Create PubSub connection
+    let pubsub = match dbx_core::redis_ops::redis_create_pubsub_core(&state.app, &connection_id).await {
+        Ok(p) => p,
+        Err(e) => {
+            let (mut sender, _) = socket.split();
+            let _ = sender.send(Message::Text(format!(r#"{{"error":"{e}"}}"#).into())).await;
+            return;
+        }
+    };
+
+    let (mut sink, mut stream) = pubsub.split();
+    let (mut ws_sender, mut ws_receiver) = socket.split();
+
+    // Channel for WebSocket commands -> PubSub sink
+    let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+
+    // Task: Read WebSocket commands
+    let ws_read = tokio::spawn(async move {
+        while let Some(Ok(msg)) = ws_receiver.next().await {
+            match msg {
+                Message::Text(text) => {
+                    if cmd_tx.send(text.to_string()).is_err() {
+                        break;
+                    }
+                }
+                Message::Close(_) => break,
+                _ => {}
+            }
+        }
+    });
+
+    // Task: Apply commands to PubSub sink
+    let sink_handle = tokio::spawn(async move {
+        while let Some(text) = cmd_rx.recv().await {
+            if let Err(e) = handle_pubsub_command(&mut sink, &text).await {
+                log::warn!("PubSub command error: {e}");
+            }
+        }
+    });
+
+    // Forward Redis messages to WebSocket (uses ws_sender, no mutex)
+    while let Some(msg) = stream.next().await {
+        let payload: String = msg.get_payload().unwrap_or_default();
+        let channel = msg.get_channel_name().to_string();
+        let pattern: Option<String> = msg.get_pattern().ok();
+        let json = serde_json::json!({
+            "channel": channel,
+            "pattern": pattern,
+            "payload": payload,
+        });
+        let text = serde_json::to_string(&json).unwrap_or_default();
+        if ws_sender.send(Message::Text(text.into())).await.is_err() {
+            break;
+        }
+    }
+
+    ws_read.abort();
+    sink_handle.abort();
+}
+
+#[derive(Deserialize)]
+#[serde(tag = "type")]
+enum PubSubCommand {
+    #[serde(rename = "subscribe")]
+    Subscribe { channels: Vec<String> },
+    #[serde(rename = "psubscribe")]
+    Psubscribe { patterns: Vec<String> },
+    #[serde(rename = "unsubscribe")]
+    Unsubscribe { channels: Vec<String> },
+    #[serde(rename = "punsubscribe")]
+    Punsubscribe { patterns: Vec<String> },
+}
+
+async fn handle_pubsub_command(sink: &mut redis::aio::PubSubSink, text: &str) -> Result<(), String> {
+    let cmd: PubSubCommand = serde_json::from_str(text).map_err(|e| format!("Invalid PubSub command: {e}"))?;
+
+    match cmd {
+        PubSubCommand::Subscribe { channels } => {
+            for ch in &channels {
+                sink.subscribe(ch).await.map_err(|e| format!("Subscribe error: {e}"))?;
+            }
+        }
+        PubSubCommand::Psubscribe { patterns } => {
+            for pat in &patterns {
+                sink.psubscribe(pat).await.map_err(|e| format!("PSubscribe error: {e}"))?;
+            }
+        }
+        PubSubCommand::Unsubscribe { channels } => {
+            for ch in &channels {
+                sink.unsubscribe(ch).await.map_err(|e| format!("Unsubscribe error: {e}"))?;
+            }
+        }
+        PubSubCommand::Punsubscribe { patterns } => {
+            for pat in &patterns {
+                sink.punsubscribe(pat).await.map_err(|e| format!("PUnsubscribe error: {e}"))?;
+            }
+        }
+    }
+    Ok(())
+}
diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml
index f7268ad45..6fc53db0c 100644
--- a/src-tauri/Cargo.toml
+++ b/src-tauri/Cargo.toml
@@ -56,5 +56,6 @@ csv = "1.4.0"
 calamine = "0.30.1"
 zip = "4.6.1"
 dbx-core = { path = "../crates/dbx-core", default-features = false }
+axum = { version = "0.8", features = ["ws"] }
 font-kit = "0.14.3"
 tauri-plugin-clipboard-manager = "2.3.2"
diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs
index 215c9b3a9..e03052b97 100644
--- a/src-tauri/src/commands/mod.rs
+++ b/src-tauri/src/commands/mod.rs
@@ -22,6 +22,7 @@ pub mod plugins;
 pub mod query;
 pub mod query_cancel;
 pub mod redis_cmd;
+pub mod redis_pubsub_server;
 pub mod saved_sql;
 pub mod schema;
 pub mod schema_cache;
diff --git a/src-tauri/src/commands/redis_cmd.rs b/src-tauri/src/commands/redis_cmd.rs
index 385ef04ac..4bfbdf843 100644
--- a/src-tauri/src/commands/redis_cmd.rs
+++ b/src-tauri/src/commands/redis_cmd.rs
@@ -303,3 +303,15 @@ pub async fn redis_load_more(
     dbx_core::redis_ops::redis_load_more_in_db_core(&state, &connection_id, db, &key_raw, &key_type, cursor, count)
         .await
 }
+
+#[tauri::command]
+pub async fn redis_pubsub_publish(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    db: u32,
+    channel: String,
+    message: String,
+) -> Result<u64, String> {
+    ensure_connection_writable(&state, &connection_id, "PUBLISH").await?;
+    dbx_core::redis_ops::redis_publish_core(&state, &connection_id, db, &channel, &message).await
+}
diff --git a/src-tauri/src/commands/redis_pubsub_server.rs b/src-tauri/src/commands/redis_pubsub_server.rs
new file mode 100644
index 000000000..ec1adbcd2
--- /dev/null
+++ b/src-tauri/src/commands/redis_pubsub_server.rs
@@ -0,0 +1,145 @@
+use std::sync::Arc;
+
+use axum::extract::ws::{Message, WebSocket};
+use axum::extract::{Query, State, WebSocketUpgrade};
+use axum::response::IntoResponse;
+use axum::routing::get;
+use axum::Router;
+use futures::{SinkExt, StreamExt};
+use serde::Deserialize;
+
+use dbx_core::connection::AppState;
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct PubSubWsParams {
+    connection_id: String,
+}
+
+pub fn build_pubsub_router(state: Arc<AppState>) -> Router {
+    Router::new().route("/api/redis/pubsub/ws", get(ws_handler)).with_state(state)
+}
+
+async fn ws_handler(
+    ws: WebSocketUpgrade,
+    Query(params): Query<PubSubWsParams>,
+    State(state): State<Arc<AppState>>,
+) -> impl IntoResponse {
+    let connection_id = params.connection_id;
+    ws.on_upgrade(move |socket| handle_socket(socket, state, connection_id))
+}
+
+async fn handle_socket(socket: WebSocket, state: Arc<AppState>, connection_id: String) {
+    // Create PubSub connection
+    let pubsub = match dbx_core::redis_ops::redis_create_pubsub_core(&state, &connection_id).await {
+        Ok(p) => p,
+        Err(e) => {
+            let (mut sender, _) = socket.split();
+            let _ = sender.send(Message::Text(format!(r#"{{"error":"{e}"}}"#).into())).await;
+            return;
+        }
+    };
+
+    let (mut sink, mut stream) = pubsub.split();
+    let (mut ws_sender, mut ws_receiver) = socket.split();
+
+    // Channel for WebSocket commands -> PubSub sink
+    let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+
+    // Task: Read WebSocket commands
+    let ws_read = tokio::spawn(async move {
+        while let Some(Ok(msg)) = ws_receiver.next().await {
+            match msg {
+                Message::Text(text) => {
+                    if cmd_tx.send(text.to_string()).is_err() {
+                        break;
+                    }
+                }
+                Message::Close(_) => break,
+                _ => {}
+            }
+        }
+    });
+
+    // Task: Apply commands to PubSub sink
+    let sink_handle = tokio::spawn(async move {
+        while let Some(text) = cmd_rx.recv().await {
+            if let Err(e) = handle_command(&mut sink, &text).await {
+                log::warn!("PubSub command error: {e}");
+            }
+        }
+    });
+
+    // Forward Redis messages to WebSocket (uses ws_sender, no mutex contention)
+    while let Some(msg) = stream.next().await {
+        let payload: String = msg.get_payload().unwrap_or_default();
+        let channel = msg.get_channel_name().to_string();
+        let pattern: Option<String> = msg.get_pattern().ok();
+        let json = serde_json::json!({
+            "channel": channel,
+            "pattern": pattern,
+            "payload": payload,
+        });
+        let text = serde_json::to_string(&json).unwrap_or_default();
+        if ws_sender.send(Message::Text(text.into())).await.is_err() {
+            break;
+        }
+    }
+
+    ws_read.abort();
+    sink_handle.abort();
+}
+
+#[derive(Deserialize)]
+#[serde(tag = "type")]
+enum PubSubCommand {
+    #[serde(rename = "subscribe")]
+    Subscribe { channels: Vec<String> },
+    #[serde(rename = "psubscribe")]
+    Psubscribe { patterns: Vec<String> },
+    #[serde(rename = "unsubscribe")]
+    Unsubscribe { channels: Vec<String> },
+    #[serde(rename = "punsubscribe")]
+    Punsubscribe { patterns: Vec<String> },
+}
+
+async fn handle_command(sink: &mut redis::aio::PubSubSink, text: &str) -> Result<(), String> {
+    let cmd: PubSubCommand = serde_json::from_str(text).map_err(|e| format!("Invalid PubSub command: {e}"))?;
+
+    match cmd {
+        PubSubCommand::Subscribe { channels } => {
+            for ch in &channels {
+                sink.subscribe(ch).await.map_err(|e| format!("Subscribe error: {e}"))?;
+            }
+        }
+        PubSubCommand::Psubscribe { patterns } => {
+            for pat in &patterns {
+                sink.psubscribe(pat).await.map_err(|e| format!("PSubscribe error: {e}"))?;
+            }
+        }
+        PubSubCommand::Unsubscribe { channels } => {
+            for ch in &channels {
+                sink.unsubscribe(ch).await.map_err(|e| format!("Unsubscribe error: {e}"))?;
+            }
+        }
+        PubSubCommand::Punsubscribe { patterns } => {
+            for pat in &patterns {
+                sink.punsubscribe(pat).await.map_err(|e| format!("PUnsubscribe error: {e}"))?;
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Start the embedded web server for PubSub WebSocket support.
+/// Runs on a background task using the shared AppState.
+pub fn start_pubsub_server(state: Arc<AppState>) {
+    let router = build_pubsub_router(state);
+    tauri::async_runtime::spawn(async move {
+        let port: u16 = std::env::var("DBX_PORT").ok().and_then(|p| p.parse().ok()).unwrap_or(4224);
+        let addr = std::net::SocketAddr::from(([0, 0, 0, 0], port));
+        let listener = tokio::net::TcpListener::bind(addr).await.expect("Failed to bind PubSub server");
+        log::info!("PubSub WebSocket server listening on {addr}");
+        axum::serve(listener, router).await.expect("PubSub server error");
+    });
+}
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index fc26b77a8..9aae747ab 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -286,6 +286,7 @@ pub fn run() {
                 Arc::new(AppState::new_with_plugin_dir_and_app_version(storage, plugin_dir, env!("CARGO_PKG_VERSION")))
             };
             app.manage(state.clone());
+            commands::redis_pubsub_server::start_pubsub_server(state.clone());
             app.manage(commands::saved_sql::SavedSqlStorageState { data_dir: data_dir.clone() });
             app.manage(commands::external_sql::ExternalSqlOpenState::default());
             app.manage(commands::external_db::ExternalDbOpenState::default());
@@ -484,6 +485,7 @@ pub fn run() {
             commands::redis_cmd::redis_flush_db,
             commands::redis_cmd::redis_execute_command,
             commands::redis_cmd::redis_load_more,
+            commands::redis_cmd::redis_pubsub_publish,
             commands::etcd_cmd::etcd_list_prefix,
             commands::etcd_cmd::etcd_get,
             commands::etcd_cmd::etcd_put,