feat(mq): implement Apache Pulsar message queue admin

* feat(mq): implement Apache Pulsar message queue admin

- Add complete MQ admin adapter layer with version detection
  * Pulsar adapter with multi-version API support
  * OAuth2 token caching and authentication
  * JWT token signing for Pulsar clients
  * Transport layer (SSH/Proxy) integration

- Implement comprehensive admin operations
  * Tenant, namespace, topic management
  * Subscription and consumer/producer monitoring
  * Rate limits, backlog quotas, retention policies
  * Permissions and role-based access control
  * Raw admin API for advanced operations

- Add desktop UI components
  * MQ admin console with tabbed interface
  * Tenant, namespace, topic panels
  * Subscription management and cursor operations
  * Monitoring panel with stats visualization
  * Policy configuration UI

- Add web routes for MQ admin operations
- Add secret store support for MQ auth credentials
- Add comprehensive test coverage

* fix: del local test files

* fix(mq): mark message queue as unsupported mcpMode in driver manifest

MQ admin connections do not use the MCP query bridge - they use a
separate REST admin API. Mark mcpMode as 'unsupported' to align with
the test expectations and reflect the actual architecture.

Fixes CI test: driver-manifest.test.ts

* fix(mq): fix TypeScript errors in MQ components

- Fix typo in SubscriptionsPanel.vue: messagformatError(e)Id -> message.messageId
- Use formatError() helper in RawApiPanel.vue instead of accessing e?.message
- Fix undefined 'tab' reference in ConnectionTree.vue: use config.db_type instead

Fixes frontend CI TypeScript check errors.

* fix(mcp): add 'mq' to MCP server connection type description

The MCP server's connection type description must include all database
types from the driver manifest. Add 'mq' to the list to fix the test:
'add connection database type description includes every manifest database type'

Fixes: packages/mcp-server/tests/driver-manifest.test.ts

* fix(mq): add missing ConnectionConfig fields for Rust compilation

- Add gbase_server and keepalive_interval_secs fields to all ConnectionConfig
  test constructors in mq/config.rs, mq/service.rs, and storage.rs
- Import uses_bare_mysql_pool in connection.rs tests module

Fixes compilation errors from upstream ConnectionConfig changes.

* chore(ui): simplify Pulsar icon to logo-only version

Remove text from Pulsar icon (127x30 -> 35x30), keeping only
the wave logo symbol for a cleaner, more consistent appearance
with other database icons.

* fix(mq): return removed pools from drain_connection_pools

---------

Co-authored-by: t8y2 <1156263951@qq.com>

src-tauri/src/commands/connection.rs

@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::sync::Arc;
 use tauri::State;
 
@@ -167,8 +168,10 @@ async fn connect_agent_pool(
 
 #[cfg(test)]
 mod tests {
-    use super::mongo_legacy_connect_params;
+    use super::{load_connection_configs, mongo_legacy_connect_params, save_connection_configs};
+    use dbx_core::connection::{AppState, PoolKind};
     use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
+    use dbx_core::storage::Storage;
 
     fn mongodb_config() -> ConnectionConfig {
         ConnectionConfig {
@@ -218,6 +221,29 @@ mod tests {
         }
     }
 
+    fn mq_config(id: &str, admin_url: &str) -> ConnectionConfig {
+        let mut config = mongodb_config();
+        config.id = id.to_string();
+        config.name = "Pulsar".to_string();
+        config.db_type = DatabaseType::MessageQueue;
+        config.driver_profile = None;
+        config.driver_label = None;
+        config.url_params = None;
+        config.host = String::new();
+        config.port = 0;
+        config.username = String::new();
+        config.password = String::new();
+        config.database = None;
+        config.connection_string = None;
+        config.external_config = Some(serde_json::json!({
+            "systemKind": "pulsar",
+            "adminUrl": admin_url,
+            "auth": { "kind": "none" },
+            "pinnedVersion": "3.1"
+        }));
+        config
+    }
+
     #[test]
     fn mongo_legacy_connect_params_preserve_auth_options() {
         let config = mongodb_config();
@@ -231,21 +257,210 @@ mod tests {
             "mongodb://mongouser:secret@172.22.4.42:27017/RestCloud_V45PUB_Gateway?authSource=admin"
         );
     }
+
+    #[cfg(feature = "mq-admin")]
+    #[tokio::test]
+    async fn save_connection_configs_updates_runtime_cache_and_drops_mq_adapter() {
+        let dir = std::env::temp_dir().join(format!("dbx-tauri-conn-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+        let state = AppState::new_with_plugin_dir(storage, dir.join("plugins"));
+        let initial = mq_config("mq-conn", "http://127.0.0.1:8080");
+        state.configs.write().await.insert(initial.id.clone(), initial.clone());
+        state.connections.write().await.insert(initial.id.clone(), PoolKind::MessageQueue);
+        let first = state.mq_registry.get_or_build(&initial).await.unwrap();
+
+        let updated = mq_config("mq-conn", "http://127.0.0.1:8081");
+        save_connection_configs(&state, &[updated.clone()]).await.unwrap();
+
+        let cached_admin_url = state
+            .configs
+            .read()
+            .await
+            .get("mq-conn")
+            .and_then(|config| config.external_config.as_ref())
+            .and_then(|external| external.get("adminUrl"))
+            .and_then(serde_json::Value::as_str)
+            .map(str::to_string);
+        assert_eq!(cached_admin_url.as_deref(), Some("http://127.0.0.1:8081"));
+
+        let second = state.mq_registry.get_or_build(&updated).await.unwrap();
+        assert!(!std::sync::Arc::ptr_eq(&first, &second));
+        assert!(!state.connections.read().await.contains_key(&initial.id));
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
+    #[cfg(feature = "mq-admin")]
+    #[tokio::test]
+    async fn load_connection_configs_syncs_runtime_cache_and_drops_stale_pool() {
+        let dir = std::env::temp_dir().join(format!("dbx-tauri-conn-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+        let state = AppState::new_with_plugin_dir(storage, dir.join("plugins"));
+        let initial = mq_config("mq-conn", "http://127.0.0.1:8080");
+        let updated = mq_config("mq-conn", "http://127.0.0.1:8081");
+        state.storage.save_connections(&[updated.clone()]).await.unwrap();
+        state.configs.write().await.insert(initial.id.clone(), initial.clone());
+        state.connections.write().await.insert(initial.id.clone(), PoolKind::MessageQueue);
+
+        let loaded = load_connection_configs(&state).await.unwrap();
+
+        assert_eq!(loaded.len(), 1);
+        let cached_admin_url = state
+            .configs
+            .read()
+            .await
+            .get("mq-conn")
+            .and_then(|config| config.external_config.as_ref())
+            .and_then(|external| external.get("adminUrl"))
+            .and_then(serde_json::Value::as_str)
+            .map(str::to_string);
+        assert_eq!(cached_admin_url.as_deref(), Some("http://127.0.0.1:8081"));
+        assert!(!state.connections.read().await.contains_key(&initial.id));
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
+    #[cfg(feature = "mq-admin")]
+    #[tokio::test]
+    async fn save_connection_configs_removes_deleted_runtime_config_and_mq_adapter() {
+        let dir = std::env::temp_dir().join(format!("dbx-tauri-conn-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+        let state = AppState::new_with_plugin_dir(storage, dir.join("plugins"));
+        let kept = mongodb_config();
+        let removed = mq_config("removed-mq", "http://127.0.0.1:8080");
+        {
+            let mut configs = state.configs.write().await;
+            configs.insert(kept.id.clone(), kept.clone());
+            configs.insert(removed.id.clone(), removed.clone());
+        }
+        let stale = state.mq_registry.get_or_build(&removed).await.unwrap();
+
+        save_connection_configs(&state, &[kept.clone()]).await.unwrap();
+
+        let configs = state.configs.read().await;
+        assert!(configs.contains_key(&kept.id));
+        assert!(!configs.contains_key("removed-mq"));
+        drop(configs);
+
+        let rebuilt = state.mq_registry.get_or_build(&removed).await.unwrap();
+        assert!(!std::sync::Arc::ptr_eq(&stale, &rebuilt));
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
+    #[cfg(feature = "mq-admin")]
+    #[tokio::test]
+    async fn save_connection_configs_removes_deleted_connection_pools() {
+        let dir = std::env::temp_dir().join(format!("dbx-tauri-conn-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+        let state = AppState::new_with_plugin_dir(storage, dir.join("plugins"));
+        let kept = mongodb_config();
+        let removed = mq_config("removed-mq", "http://127.0.0.1:8080");
+        {
+            let mut configs = state.configs.write().await;
+            configs.insert(kept.id.clone(), kept.clone());
+            configs.insert(removed.id.clone(), removed.clone());
+        }
+        state.connections.write().await.insert(removed.id.clone(), PoolKind::MessageQueue);
+
+        save_connection_configs(&state, &[kept.clone()]).await.unwrap();
+
+        assert!(!state.connections.read().await.contains_key(&removed.id));
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
 }
 
 #[tauri::command]
 pub async fn save_connections(state: State<'_, Arc<AppState>>, configs: Vec<ConnectionConfig>) -> Result<(), String> {
     let configs: Vec<ConnectionConfig> = configs.into_iter().map(|config| config.canonicalized()).collect();
-    state.storage.save_connections(&configs).await
+    save_connection_configs(state.inner(), &configs).await
+}
+
+async fn save_connection_configs(state: &AppState, configs: &[ConnectionConfig]) -> Result<(), String> {
+    state.storage.save_connections(configs).await?;
+    let sync = sync_connection_configs(state, configs).await;
+    remove_connection_pools_for_connection_ids(state, &sync.connection_pool_ids_to_drop).await;
+    drop_mq_adapters_for_connection_ids(state, &sync.mq_adapter_ids_to_drop).await;
+    Ok(())
+}
+
+struct ConnectionConfigSync {
+    mq_adapter_ids_to_drop: Vec<String>,
+    connection_pool_ids_to_drop: Vec<String>,
+}
+
+async fn sync_connection_configs(state: &AppState, configs: &[ConnectionConfig]) -> ConnectionConfigSync {
+    let saved_ids: HashSet<&str> = configs.iter().map(|config| config.id.as_str()).collect();
+    let mut mq_adapter_ids_to_drop = HashSet::new();
+    let mut connection_pool_ids_to_drop = HashSet::new();
+    let mut runtime_configs = state.configs.write().await;
+    runtime_configs.retain(|id, existing| {
+        if saved_ids.contains(id.as_str()) || is_transient_runtime_config_id(id) {
+            true
+        } else {
+            connection_pool_ids_to_drop.insert(id.clone());
+            if existing.db_type == DatabaseType::MessageQueue {
+                mq_adapter_ids_to_drop.insert(id.clone());
+            }
+            false
+        }
+    });
+    for config in configs {
+        if config.db_type == DatabaseType::MessageQueue {
+            mq_adapter_ids_to_drop.insert(config.id.clone());
+        }
+        if let Some(previous) = runtime_configs.insert(config.id.clone(), config.clone()) {
+            if previous.db_type == DatabaseType::MessageQueue {
+                mq_adapter_ids_to_drop.insert(config.id.clone());
+            }
+            if &previous != config {
+                connection_pool_ids_to_drop.insert(config.id.clone());
+            }
+        }
+    }
+    ConnectionConfigSync {
+        mq_adapter_ids_to_drop: mq_adapter_ids_to_drop.into_iter().collect(),
+        connection_pool_ids_to_drop: connection_pool_ids_to_drop.into_iter().collect(),
+    }
+}
+
+fn is_transient_runtime_config_id(id: &str) -> bool {
+    id.starts_with("__test_") || id.starts_with("__visible_draft_")
+}
+
+#[cfg(feature = "mq-admin")]
+async fn drop_mq_adapters_for_connection_ids(state: &AppState, connection_ids: &[String]) {
+    for connection_id in connection_ids {
+        state.mq_registry.drop_connection(connection_id).await;
+    }
+}
+
+#[cfg(not(feature = "mq-admin"))]
+async fn drop_mq_adapters_for_connection_ids(_state: &AppState, _connection_ids: &[String]) {}
+
+async fn remove_connection_pools_for_connection_ids(state: &AppState, connection_ids: &[String]) {
+    for connection_id in connection_ids {
+        state.remove_connection_pools(connection_id).await;
+    }
 }
 
 #[tauri::command]
 pub async fn load_connections(state: State<'_, Arc<AppState>>) -> Result<Vec<ConnectionConfig>, String> {
-    state
-        .storage
-        .load_connections()
-        .await
-        .map(|configs| configs.into_iter().map(|config| config.canonicalized()).collect())
+    load_connection_configs(state.inner()).await
+}
+
+async fn load_connection_configs(state: &AppState) -> Result<Vec<ConnectionConfig>, String> {
+    let configs: Vec<ConnectionConfig> =
+        state.storage.load_connections().await?.into_iter().map(|config| config.canonicalized()).collect();
+    let sync = sync_connection_configs(state, &configs).await;
+    remove_connection_pools_for_connection_ids(state, &sync.connection_pool_ids_to_drop).await;
+    drop_mq_adapters_for_connection_ids(state, &sync.mq_adapter_ids_to_drop).await;
+    Ok(configs)
 }
 
 #[tauri::command]
@@ -481,6 +696,18 @@ pub async fn test_connection(state: State<'_, Arc<AppState>>, config: Connection
                     .await
                     .map(|_| "Connection successful".to_string())
             }
+            #[cfg(feature = "mq-admin")]
+            DatabaseType::MessageQueue => {
+                let mqc = state.mq_admin_config_for_connection(connection_id, &config).await?;
+                let adapter = state.mq_registry.build_transient_config(mqc).await?;
+                adapter.test_connection().await?;
+                Ok("Connection successful".to_string())
+            }
+            #[cfg(not(feature = "mq-admin"))]
+            DatabaseType::MessageQueue => {
+                Err("Message queue admin support is not compiled in this build. Rebuild with the 'mq-admin' feature."
+                    .to_string())
+            }
             db_type if database_capabilities::is_agent_type(&db_type) => {
                 test_agent_connection(state.inner(), &config, &host, port).await
             }
@@ -707,6 +934,20 @@ pub async fn connect_db(state: State<'_, Arc<AppState>>, config: ConnectionConfi
             db::influxdb_driver::test_connection(&client, connect_timeout).await?;
             PoolKind::InfluxDb(client)
         }
+        #[cfg(feature = "mq-admin")]
+        DatabaseType::MessageQueue => {
+            let mqc = state.mq_admin_config_for_connection(&id, &config).await?;
+            let adapter = state.mq_registry.build_transient_config(mqc).await?;
+            adapter.test_connection().await?;
+            PoolKind::MessageQueue
+        }
+        #[cfg(not(feature = "mq-admin"))]
+        DatabaseType::MessageQueue => {
+            return Err(
+                "Message queue admin support is not compiled in this build. Rebuild with the 'mq-admin' feature."
+                    .to_string(),
+            );
+        }
         db_type if database_capabilities::is_agent_type(&db_type) => {
             connect_agent_pool(state.inner(), &db_config, &host, port).await?
         }
@@ -741,6 +982,7 @@ pub async fn connection_final_proxy_port(
 #[tauri::command]
 pub async fn disconnect_db(state: State<'_, Arc<AppState>>, connection_id: String) -> Result<(), String> {
     state.remove_connection_pools_detached(&connection_id).await;
+    drop_mq_adapters_for_connection_ids(state.inner(), std::slice::from_ref(&connection_id)).await;
     state.reset_connection_transport(&connection_id).await;
     if connection_id.starts_with("__visible_draft_") {
         state.configs.write().await.remove(&connection_id);

src-tauri/src/commands/mcp.rs

@@ -1,3 +1,4 @@
+#[cfg(not(windows))]
 use std::env;
 use std::path::Path;
 use std::process::Command;
@@ -276,6 +277,7 @@ fn default_user_shell() -> String {
     }
 }
 
+#[cfg(not(windows))]
 fn shell_command_script(command: &str, args: &[&str]) -> String {
     let mut words = Vec::with_capacity(args.len() + 1);
     words.push(shell_quote(command));
@@ -283,6 +285,7 @@ fn shell_command_script(command: &str, args: &[&str]) -> String {
     format!("printf '%s\\n' {}; {}", shell_quote(SHELL_COMMAND_MARKER), words.join(" "))
 }
 
+#[cfg(not(windows))]
 fn shell_quote(value: &str) -> String {
     if value.is_empty() {
         return "''".to_string();
@@ -299,10 +302,13 @@ fn stdout_after_shell_marker(stdout: &str) -> String {
 
 #[cfg(test)]
 mod tests {
-    use super::{
-        bash_login_script, shell_command_script, shell_quote, stdout_after_shell_marker, SHELL_COMMAND_MARKER,
-    };
+    #[cfg(not(windows))]
+    use super::bash_login_script;
+    #[cfg(not(windows))]
+    use super::{shell_command_script, shell_quote};
+    use super::{stdout_after_shell_marker, SHELL_COMMAND_MARKER};
 
+    #[cfg(not(windows))]
     #[test]
     fn shell_quote_handles_empty_and_single_quotes() {
         assert_eq!(shell_quote(""), "''");
@@ -310,6 +316,7 @@ mod tests {
         assert_eq!(shell_quote("can't"), "'can'\"'\"'t'");
     }
 
+    #[cfg(not(windows))]
     #[test]
     fn shell_command_script_marks_command_output_after_startup_noise() {
         let script = shell_command_script("npm", &["list", "-g", "@dbx-app/mcp-server", "--json"]);
@@ -318,6 +325,7 @@ mod tests {
         assert!(script.contains("'@dbx-app/mcp-server'"));
     }
 
+    #[cfg(not(windows))]
     #[test]
     fn bash_login_script_sources_profile_and_rc_files() {
         let script = bash_login_script("node --version");

src-tauri/src/commands/mod.rs

@@ -18,6 +18,8 @@ pub mod keychain;
 pub mod mcp;
 pub mod mcp_bridge;
 pub mod mongo_cmd;
+#[cfg(feature = "mq-admin")]
+pub mod mq_cmd;
 pub mod plugins;
 pub mod query;
 pub mod query_cancel;

src-tauri/src/commands/mq_cmd.rs

@@ -0,0 +1,448 @@
+//! Desktop (Tauri) commands for message queue admin operations.
+//!
+//! These are thin wrappers around `dbx_core::mq::service::*_core` functions,
+//! with read-only protection (`ensure_connection_writable`) for mutating calls.
+
+use std::sync::Arc;
+
+use tauri::State;
+
+use crate::commands::connection::{ensure_connection_writable, AppState};
+
+// ---- Test connection ----
+
+#[tauri::command]
+pub async fn mq_test_connection(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+) -> Result<dbx_core::mq::MqClusterInfo, String> {
+    dbx_core::mq::service::mq_test_connection_core(&state, &connection_id).await
+}
+
+// ---- Tenants ----
+
+#[tauri::command]
+pub async fn mq_list_tenants(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+) -> Result<Vec<dbx_core::mq::TenantInfo>, String> {
+    dbx_core::mq::service::mq_list_tenants_core(&state, &connection_id).await
+}
+
+#[tauri::command]
+pub async fn mq_get_tenant(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    name: String,
+) -> Result<dbx_core::mq::TenantInfo, String> {
+    dbx_core::mq::service::mq_get_tenant_core(&state, &connection_id, &name).await
+}
+
+#[tauri::command]
+pub async fn mq_create_tenant(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    name: String,
+    config: dbx_core::mq::TenantConfig,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Create tenant").await?;
+    dbx_core::mq::service::mq_create_tenant_core(&state, &connection_id, &name, config).await
+}
+
+#[tauri::command]
+pub async fn mq_update_tenant(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    name: String,
+    config: dbx_core::mq::TenantConfig,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Update tenant").await?;
+    dbx_core::mq::service::mq_update_tenant_core(&state, &connection_id, &name, config).await
+}
+
+#[tauri::command]
+pub async fn mq_delete_tenant(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    name: String,
+    force: bool,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Delete tenant").await?;
+    dbx_core::mq::service::mq_delete_tenant_core(&state, &connection_id, &name, force).await
+}
+
+// ---- Namespaces ----
+
+#[tauri::command]
+pub async fn mq_list_namespaces(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    tenant: String,
+) -> Result<Vec<dbx_core::mq::NamespaceInfo>, String> {
+    dbx_core::mq::service::mq_list_namespaces_core(&state, &connection_id, &tenant).await
+}
+
+#[tauri::command]
+pub async fn mq_create_namespace(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    ns: dbx_core::mq::NamespaceRef,
+    config: dbx_core::mq::NamespaceConfig,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Create namespace").await?;
+    dbx_core::mq::service::mq_create_namespace_core(&state, &connection_id, ns, config).await
+}
+
+#[tauri::command]
+pub async fn mq_delete_namespace(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    ns: dbx_core::mq::NamespaceRef,
+    force: bool,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Delete namespace").await?;
+    dbx_core::mq::service::mq_delete_namespace_core(&state, &connection_id, ns, force).await
+}
+
+#[tauri::command]
+pub async fn mq_get_namespace_policies(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    ns: dbx_core::mq::NamespaceRef,
+) -> Result<serde_json::Value, String> {
+    dbx_core::mq::service::mq_get_namespace_policies_core(&state, &connection_id, ns).await
+}
+
+// ---- Topics ----
+
+#[tauri::command]
+pub async fn mq_list_topics(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    ns: dbx_core::mq::NamespaceRef,
+    opts: dbx_core::mq::ListTopicsOpts,
+) -> Result<Vec<dbx_core::mq::TopicInfo>, String> {
+    dbx_core::mq::service::mq_list_topics_core(&state, &connection_id, ns, opts).await
+}
+
+#[tauri::command]
+pub async fn mq_create_topic(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    partitions: Option<u32>,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Create topic").await?;
+    dbx_core::mq::service::mq_create_topic_core(&state, &connection_id, topic, partitions).await
+}
+
+#[tauri::command]
+pub async fn mq_delete_topic(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    force: bool,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Delete topic").await?;
+    dbx_core::mq::service::mq_delete_topic_core(&state, &connection_id, topic, force).await
+}
+
+#[tauri::command]
+pub async fn mq_update_partitions(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    partitions: u32,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Update partitions").await?;
+    dbx_core::mq::service::mq_update_partitions_core(&state, &connection_id, topic, partitions).await
+}
+
+#[tauri::command]
+pub async fn mq_get_topic_stats(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+) -> Result<dbx_core::mq::TopicStats, String> {
+    dbx_core::mq::service::mq_get_topic_stats_core(&state, &connection_id, topic).await
+}
+
+#[tauri::command]
+pub async fn mq_get_topic_internal_stats(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+) -> Result<serde_json::Value, String> {
+    dbx_core::mq::service::mq_get_topic_internal_stats_core(&state, &connection_id, topic).await
+}
+
+// ---- Subscriptions ----
+
+#[tauri::command]
+pub async fn mq_list_subscriptions(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+) -> Result<Vec<dbx_core::mq::SubscriptionInfo>, String> {
+    dbx_core::mq::service::mq_list_subscriptions_core(&state, &connection_id, topic).await
+}
+
+#[tauri::command]
+pub async fn mq_create_subscription(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+    pos: dbx_core::mq::ResetPosition,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Create subscription").await?;
+    dbx_core::mq::service::mq_create_subscription_core(&state, &connection_id, topic, sub, pos).await
+}
+
+#[tauri::command]
+pub async fn mq_delete_subscription(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+    force: bool,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Delete subscription").await?;
+    dbx_core::mq::service::mq_delete_subscription_core(&state, &connection_id, topic, sub, force).await
+}
+
+#[tauri::command]
+pub async fn mq_skip_messages(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+    count: dbx_core::mq::SkipCount,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Skip messages").await?;
+    dbx_core::mq::service::mq_skip_messages_core(&state, &connection_id, topic, sub, count).await
+}
+
+#[tauri::command]
+pub async fn mq_reset_cursor(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+    pos: dbx_core::mq::ResetPosition,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Reset cursor").await?;
+    dbx_core::mq::service::mq_reset_cursor_core(&state, &connection_id, topic, sub, pos).await
+}
+
+#[tauri::command]
+pub async fn mq_clear_backlog(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Clear backlog").await?;
+    dbx_core::mq::service::mq_clear_backlog_core(&state, &connection_id, topic, sub).await
+}
+
+#[tauri::command]
+pub async fn mq_peek_messages(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+    count: u32,
+) -> Result<Vec<dbx_core::mq::PeekedMessage>, String> {
+    dbx_core::mq::service::mq_peek_messages_core(&state, &connection_id, topic, sub, count).await
+}
+
+#[tauri::command]
+pub async fn mq_expire_messages(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+    expire_seconds: i64,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Expire messages").await?;
+    dbx_core::mq::service::mq_expire_messages_core(&state, &connection_id, topic, sub, expire_seconds).await
+}
+
+// ---- Producers / consumers ----
+
+#[tauri::command]
+pub async fn mq_list_producers(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+) -> Result<Vec<dbx_core::mq::ProducerInfo>, String> {
+    dbx_core::mq::service::mq_list_producers_core(&state, &connection_id, topic).await
+}
+
+#[tauri::command]
+pub async fn mq_list_consumers(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: String,
+) -> Result<Vec<dbx_core::mq::ConsumerInfo>, String> {
+    dbx_core::mq::service::mq_list_consumers_core(&state, &connection_id, topic, sub).await
+}
+
+#[tauri::command]
+pub async fn mq_unload_topic(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Unload topic").await?;
+    dbx_core::mq::service::mq_unload_topic_core(&state, &connection_id, topic).await
+}
+
+// ---- Rate limits / quotas / retention ----
+
+#[tauri::command]
+pub async fn mq_set_publish_rate(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+    rate: dbx_core::mq::PublishRate,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Set publish rate").await?;
+    dbx_core::mq::service::mq_set_publish_rate_core(&state, &connection_id, scope, rate).await
+}
+
+#[tauri::command]
+pub async fn mq_set_dispatch_rate(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+    rate: dbx_core::mq::DispatchRate,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Set dispatch rate").await?;
+    dbx_core::mq::service::mq_set_dispatch_rate_core(&state, &connection_id, scope, rate).await
+}
+
+#[tauri::command]
+pub async fn mq_set_subscribe_rate(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+    rate: dbx_core::mq::SubscribeRate,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Set subscribe rate").await?;
+    dbx_core::mq::service::mq_set_subscribe_rate_core(&state, &connection_id, scope, rate).await
+}
+
+#[tauri::command]
+pub async fn mq_set_backlog_quota(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+    quota: dbx_core::mq::BacklogQuota,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Set backlog quota").await?;
+    dbx_core::mq::service::mq_set_backlog_quota_core(&state, &connection_id, scope, quota).await
+}
+
+#[tauri::command]
+pub async fn mq_set_retention(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+    retention: dbx_core::mq::RetentionPolicy,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Set retention").await?;
+    dbx_core::mq::service::mq_set_retention_core(&state, &connection_id, scope, retention).await
+}
+
+#[tauri::command]
+pub async fn mq_get_effective_policies(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+) -> Result<serde_json::Value, String> {
+    dbx_core::mq::service::mq_get_effective_policies_core(&state, &connection_id, scope).await
+}
+
+// ---- Permissions ----
+
+#[tauri::command]
+pub async fn mq_grant_permission(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+    role: String,
+    actions: Vec<dbx_core::mq::AuthAction>,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Grant permission").await?;
+    dbx_core::mq::service::mq_grant_permission_core(&state, &connection_id, scope, role, actions).await
+}
+
+#[tauri::command]
+pub async fn mq_revoke_permission(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+    role: String,
+) -> Result<(), String> {
+    ensure_connection_writable(&state, &connection_id, "Revoke permission").await?;
+    dbx_core::mq::service::mq_revoke_permission_core(&state, &connection_id, scope, role).await
+}
+
+#[tauri::command]
+pub async fn mq_list_permissions(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    scope: dbx_core::mq::PolicyScope,
+) -> Result<dbx_core::mq::PermissionMap, String> {
+    dbx_core::mq::service::mq_list_permissions_core(&state, &connection_id, scope).await
+}
+
+// ---- Client tokens ----
+
+#[tauri::command]
+pub async fn mq_issue_token(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    req: dbx_core::mq::MqTokenIssueRequest,
+) -> Result<dbx_core::mq::MqIssuedToken, String> {
+    ensure_connection_writable(&state, &connection_id, "Issue MQ token").await?;
+    dbx_core::mq::service::mq_issue_token_core(&state, &connection_id, req).await
+}
+
+#[tauri::command]
+pub async fn mq_list_token_records(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    subject: Option<String>,
+) -> Result<Vec<dbx_core::mq::MqTokenRecord>, String> {
+    dbx_core::mq::service::mq_list_token_records_core(&state, &connection_id, subject).await
+}
+
+// ---- Monitoring ----
+
+#[tauri::command]
+pub async fn mq_get_backlog(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    topic: dbx_core::mq::TopicRef,
+    sub: Option<String>,
+) -> Result<dbx_core::mq::BacklogStats, String> {
+    dbx_core::mq::service::mq_get_backlog_core(&state, &connection_id, topic, sub).await
+}
+
+// ---- Raw request (escape hatch) ----
+
+#[tauri::command]
+pub async fn mq_raw_request(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    req: dbx_core::mq::MqRawRequest,
+) -> Result<dbx_core::mq::MqRawResponse, String> {
+    if req.is_mutating() {
+        ensure_connection_writable(&state, &connection_id, "MQ admin write").await?;
+    }
+    dbx_core::mq::service::mq_raw_request_core(&state, &connection_id, req).await
+}