Managing Pulsar Queues with DBX: A Complete Guide to Apache Pulsar Administration

DBX provides a unified interface for managing Apache Pulsar queues through a Rust-based backend and Vue frontend, supporting tenants, namespaces, topics, and subscriptions with built-in security features.

The DBX repository (t8y2/dbx) extends beyond traditional database exploration to offer comprehensive Message Queue (MQ) administration, with Apache Pulsar as its first supported system. Managing Pulsar queues with DBX involves a multi-layered architecture that bridges TypeScript API calls to Rust-implemented Pulsar Admin REST API clients. This guide covers the technical implementation, configuration options, and security measures implemented across the codebase.

Architecture and Data Flow

Layered Architecture Overview

DBX implements a clean architecture separating UI concerns from backend operations. The frontend Vue components (such as MqAdminConsole, TenantsPanel, and TopicsPanel) communicate through a unified API layer defined in apps/desktop/src/lib/mq-api.ts. This abstraction allows the same code to execute via Tauri commands (mq-tauri.ts) for desktop builds or HTTP endpoints (mq-http.ts) for web deployments.

The backend processes requests through two entry points: src-tauri/src/commands/mq_cmd.rs for desktop and crates/dbx-web/src/routes/mq.rs for web APIs. Both funnel into the core service layer at crates/dbx-core/src/mq/service.rs, which orchestrates adapter caching and business logic. The MqAdminRegistry maintains cached instances of MessageQueueAdmin trait implementations, with PulsarAdapter (defined in crates/dbx-core/src/mq/adapters/pulsar.rs) serving as the concrete client for Pulsar-specific operations.

Key Components and Source Files

The repository organizes MQ functionality across several critical files:

• crates/dbx-core/src/mq/port.rs – Defines the MessageQueueAdmin trait interface used by all MQ adapters
• crates/dbx-core/src/mq/service.rs – Contains core business logic functions (prefixed with mq_*_core) that handle connection resolution and error handling
• crates/dbx-core/src/mq/adapters/pulsar.rs – Implements PulsarAdapter with methods like list_tenants and create_tenant (lines 1400–1500)
• crates/dbx-core/src/mq/adapters/pulsar_version.rs – Handles version detection against /admin/v2/brokers/version to select appropriate API profiles
• src-tauri/src/commands/mq_cmd.rs – Exposes Rust functions to the frontend via tauri::generate_handler! with safety checks

Connection Configuration

DBX stores Pulsar connection parameters in the external_config JSON field. A minimal configuration requires specifying the system type, admin URL, and authentication method:

{
  "systemKind": "pulsar",
  "adminUrl": "https://pulsar.example.com:8443",
  "auth": {
    "kind": "token",
    "token": "eyJhbGciOi... (JWT)"
  },
  "tlsSkipVerify": false
}

The UI enforces mq as the database type and bypasses generic host/port fields. This configuration deserializes into MqAdminConfig (defined in crates/dbx-core/src/mq/config.rs) before instantiation by the PulsarAdapter.

Core Operations for Managing Pulsar Queues

Tenants and Namespaces

DBX exposes tenant management through the mqListTenants and mqCreateTenant functions. Listing tenants calls PulsarAdapter::list_tenants, while creation accepts parameters for adminRoles and allowedClusters. Namespace operations follow a similar pattern with mqListNamespaces and mqCreateNamespace, mapping directly to Pulsar's /admin/v2/namespaces/{tenant}/{namespace} endpoints.

Topics and Subscriptions

Topic management supports both partitioned and non-partitioned queues. The mqCreateTopic function accepts a TopicRef struct and CreateTopicOpts containing an optional partition count. The adapter optimizes performance using concurrent requests controlled by DETAIL_REQUEST_CONCURRENCY and PARTITION_METADATA_CONCURRENCY constants when fetching stats.

Subscription operations include:

• List: mqListSubscriptions(connectionId, topicRef)
• Create: mqCreateSubscription(connectionId, topicRef, name, { kind })
• Reset cursor: mqResetCursor(connectionId, topicRef, name, { kind })
• Clear backlog: mqClearBacklog(connectionId, topicRef, name)

All mutating operations invoke ensure_connection_writable in mq_cmd.rs to respect read-only connection flags.

Policies and Permissions

Fine-grained control over Pulsar queues utilizes mqSetPublishRate and mqSetRetention for throttling and storage policies. Permission management uses mqGrantPermission, which maps to the MessageQueueAdmin::grant_permission trait method implemented in the Pulsar adapter.

Raw Admin API Access

For operations not yet covered by the UI, the Raw API panel allows direct REST calls:

const resp = await mqRawRequest(connectionId, {
  method: 'GET',
  path: '/admin/v2/clusters',
  query: {},
  body: null
});

The backend validates paths to prevent Server-Side Request Forgery (SSRF), rejecting any path containing .., scheme identifiers, or failing to start with /.

Authentication and Security

Supported Authentication Strategies

DBX implements five authentication strategies in crates/dbx-core/src/mq/auth.rs and crates/dbx-core/src/mq/token.rs:

1. Token – JWT passed as Authorization: Bearer … headers
2. Basic – HTTP Basic Auth encoding username and password
3. OAuth2 – Supports issuerUrl, clientId, clientSecret, and audience with OAuth2TokenCache maintaining tokens for 60 seconds
4. API Key – Custom header injection (e.g., X-API-Key)
5. None – No authentication headers sent

Token signing for Pulsar-specific JWTs uses the sign_pulsar_token function in mq/token.rs.

Security Measures and SSRF Protection

Security implementations span multiple layers:

• Read-only enforcement – ensure_connection_writable in src-tauri/src/commands/mq_cmd.rs rejects mutating operations on connections marked read_only
• SSRF protection – Raw request validation (lines 120–133 in mq_cmd.rs) blocks directory traversal and absolute URLs
• TLS verification – Configurable via tlsSkipVerify with strict verification as the default

Version Compatibility and Feature Detection

When initializing a connection, PulsarAdapter::new probes the /admin/v2/brokers/version endpoint (implementation in pulsar_version.rs). The detected version selects a PulsarApiProfile that enables or disables specific API calls, ensuring graceful degradation when managing older Pulsar clusters.

To compile without MQ support (reducing binary size), disable the default mq-admin feature:

cargo build --release --no-default-features --features duckdb-bundled

Implementation Examples

Listing Tenants via TypeScript

The frontend consumes MQ operations through the unified API layer:

import { mqListTenants } from '@/lib/mq-api'

async function showTenants(connId: string) {
  const tenants = await mqListTenants(connId)
  console.table(tenants)
}

This traverses: mqListTenants → mq_cmd::mq_list_tenants → mq_test_connection_core → adapter.list_tenants.

Creating Partitioned Topics in Rust

For server-side implementations or custom tooling:

use dbx_core::mq::service::mq_create_topic_core;
use dbx_core::mq::types::{TopicRef, CreateTopicOpts};

let topic = TopicRef {
    tenant: "my-tenant".into(),
    namespace: "my-namespace".into(),
    topic: "orders".into(),
    persistent: true,
};

let opts = CreateTopicOpts { 
    partitions: Some(4), 
    ..Default::default() 
};

let result = mq_create_topic_core(&state, "conn-id", &topic, &opts).await?;

Executing Raw Admin Requests

Advanced administration requires direct API access:

import { mqRawRequest } from '@/lib/mq-api'

const resp = await mqRawRequest('conn-id', {
  method: 'POST',
  path: '/admin/v2/namespaces/my-tenant/my-namespace/policies',
  query: {},
  body: { retention_time_in_minutes: 1440 }
})

Summary

• DBX provides comprehensive Pulsar queue management through a layered architecture separating Vue frontend, TypeScript API wrappers, Tauri commands, and Rust service implementations
• Core operations include tenant/namespace administration, partitioned topic creation, subscription management, and policy configuration via the MessageQueueAdmin trait
• Security features encompass read-only connection enforcement, OAuth2 token caching, SSRF protection on raw requests, and configurable TLS verification
• Version detection automatically adapts API behavior to target Pulsar cluster capabilities
• Extensibility is built into the architecture via the MessageQueueAdmin trait, with placeholders for future Kafka and RocketMQ adapters

Frequently Asked Questions

How does DBX handle Pulsar authentication?

DBX supports five authentication strategies defined in crates/dbx-core/src/mq/auth.rs: Token (JWT), Basic (HTTP Basic Auth), OAuth2 (with 60-second token caching), API Key (custom headers), and None. The PulsarAdapter applies these credentials to all requests to the Pulsar Admin REST API, with OAuth2 tokens maintained in OAuth2TokenCache to avoid repeated authentication calls.

What security measures protect against SSRF when using the Raw API?

The Raw API implementation in src-tauri/src/commands/mq_cmd.rs (lines 120–133) validates that request paths start with / and do not contain .. sequences or URL schemes. This prevents attackers from using the Raw API endpoint to access internal network resources or local files. Additionally, all connections respect the read_only flag enforced by ensure_connection_writable.

Can DBX manage multiple message queue systems?

Yes. While the current implementation focuses on Apache Pulsar via PulsarAdapter in crates/dbx-core/src/mq/adapters/pulsar.rs, the architecture defines a MessageQueueAdmin trait in port.rs that abstracts all MQ operations. Adding support for Kafka or RocketMQ requires implementing this trait for the new system and registering the adapter in MqAdminRegistry::build_adapter.

How does DBX detect Pulsar version compatibility?

Upon first connection, PulsarAdapter::new queries /admin/v2/brokers/version using the version detection logic in pulsar_version.rs. The response selects a PulsarApiProfile that determines available features and API endpoints, ensuring DBX gracefully handles older Pulsar versions that may lack newer administrative capabilities.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}