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

> Master Apache Pulsar queue management with DBX. This guide covers tenants, namespaces, topics, and subscriptions using a powerful Rust backend and Vue frontend for seamless administration.

- Repository: [skyler/dbx](https://github.com/t8y2/dbx)
- Tags: how-to-guide
- Published: 2026-07-02

---

**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`](https://github.com/t8y2/dbx/blob/main/apps/desktop/src/lib/mq-api.ts). This abstraction allows the same code to execute via **Tauri commands** ([`mq-tauri.ts`](https://github.com/t8y2/dbx/blob/main/mq-tauri.ts)) for desktop builds or **HTTP endpoints** ([`mq-http.ts`](https://github.com/t8y2/dbx/blob/main/mq-http.ts)) for web deployments.

The backend processes requests through two entry points: [`src-tauri/src/commands/mq_cmd.rs`](https://github.com/t8y2/dbx/blob/main/src-tauri/src/commands/mq_cmd.rs) for desktop and [`crates/dbx-web/src/routes/mq.rs`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/crates/dbx-core/src/mq/port.rs)** – Defines the `MessageQueueAdmin` trait interface used by all MQ adapters
- **[`crates/dbx-core/src/mq/service.rs`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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:

```json
{
  "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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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:

```typescript
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`](https://github.com/t8y2/dbx/blob/main/crates/dbx-core/src/mq/auth.rs) and [`crates/dbx-core/src/mq/token.rs`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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:

```bash
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:

```typescript
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:

```rust
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:

```typescript
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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/crates/dbx-core/src/mq/adapters/pulsar.rs), the architecture defines a `MessageQueueAdmin` trait in [`port.rs`](https://github.com/t8y2/dbx/blob/main/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`](https://github.com/t8y2/dbx/blob/main/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.