# Plan 25: Admin Dashboard API

## Goal

Build a comprehensive set of HTTP admin endpoints that expose full system introspection and operational control for a browser-based admin GUI. An operator should be able to view every namespace, topic, subscriber, consumer group, connection, and segment — and perform operational commands like force-flushing, compacting, purging topics, disconnecting clients, and resetting consumer group offsets — all through the existing `/admin` route group with the system-level `admin_api_key`.

## Motivation

The server currently exposes five admin-adjacent endpoints: two drain endpoints (`POST /admin/drain`, `GET /admin/drain/status`), two log-level endpoints (`PUT/GET /v1/admin/log-level`), and health probes. The Prometheus metrics endpoint provides time-series counters and gauges, but it cannot answer operational questions like:

- "Which subscribers are lagging behind on topic X?"
- "What consumer groups exist, who are their members, and what's their inflight state?"
- "Which connections are idle or slow?"
- "What do the segment files look like for this topic — how many, how big, do they have indexes?"
- "Can I peek at the actual messages at offset 42000?"
- "I need to force-purge a topic that's stuck" or "reset this consumer group back to offset N"

An admin GUI needs structured JSON endpoints for all of this, not just Prometheus counters. The existing `topic_info` and `list_topics` endpoints are namespace-scoped (require a namespace API key) and return limited data. The admin API must provide a cross-namespace, system-level view with deeper detail and write capabilities.

## Prerequisites

- Plans 1–15 complete (core server, storage, fetch protocol)
- Plan 14 complete (graceful deployment — `ServerLifecycle`, `/admin` route group, `admin_api_key`, `require_admin_auth` middleware)

## Design Decisions

### Single admin token, not RBAC

The admin API uses the existing `admin_api_key` from `DeployConfig` — a single system-level credential. No roles, no per-endpoint permissions, no multi-user model. If you have the admin key, you have full access. This matches the "system admin" use case and avoids complexity. Multi-tenant admin access (if ever needed) is a separate future concern.

### Extend existing `/admin` route group

All new endpoints live under `/admin/v1/...` and use the existing `require_admin_auth` middleware. The `/admin` prefix is already outside the `require_initialized` gate, so admin endpoints work during startup/drain (useful for debugging initialization issues). The `/v1` sub-prefix allows future API versioning.

### Extensibility pattern

Every new feature added to the server (future plans) should be able to expose admin endpoints by:
1. Adding route handlers in a new or existing admin module
2. Registering them on the admin router via a `admin_routes()` builder function
3. Adding any new trait methods needed on `NamespaceManager` / `QueueEngine` / `StorageBackend`

The admin router is assembled in `src/server/http.rs` as a nested `Router` with shared `AdminState` — new features merge their routes onto this router. No central "admin controller" that must know about every subsystem.

### Read vs. write endpoint conventions

- **GET** endpoints are read-only queries — safe to call from dashboards, monitoring, scripts
- **POST** endpoints are operational commands that mutate state — the GUI should confirm before calling
- **DELETE** endpoints are destructive operations — the GUI should double-confirm

All write endpoints return a result describing what changed (e.g., `{ "bytes_freed": 1048576, "messages_freed": 500 }`), not just 200 OK.

### Response envelope

All admin responses use a consistent envelope for the GUI to parse uniformly:

```json
{
  "data": { ... },
  "timestamp": "2026-03-21T10:30:00Z"
}
```

Error responses use the existing HTTP error format: `{ "error": { "code": "...", "message": "..." } }`.

### CORS for browser GUI

The admin HTTP server needs CORS headers so a browser-based GUI (served from a different origin or as a static SPA) can call the API. Configurable via `[admin]` config or defaulting to the admin bind address.

## Dependency Graph

```
Phase A (foundation)
    │
    ├──► Phase B (namespace & topic queries)
    │         │
    │         ├──► Phase C (subscriber & connection queries)
    │         │         │
    │         │         └──► Phase D (operational commands)
    │         │
    │         └──► Phase E (storage & segment inspection)
    │                   │
    │                   └──► Phase F (message browser)
    │
    └──► Phase G (config & metrics JSON)

Phase D + Phase E + Phase F + Phase G ──► Phase H (integration tests)
```

**Recommended order**: A → B → C → E → D → F → G → H (C before D because commands need the query infrastructure; E before F because message browsing needs segment awareness)

---

## Phase A: Admin Router Foundation

### A.1 — AdminState and router structure

Create `src/server/http/admin.rs` as the admin endpoint module. Define `AdminState` that holds references to all subsystems an admin endpoint might need:

```rust
pub struct AdminState<M: NamespaceManager, Q: QueueEngine, S: StorageBackend> {
    pub manager: Arc<M>,
    pub engine: Arc<Q>,
    pub storage: Arc<S>,
    pub lifecycle: Arc<ServerLifecycle>,
    pub config: Arc<ServerConfig>,         // immutable startup config snapshot
    pub start_time: Instant,               // for uptime calculation
    pub start_time_wall: u64,              // wall-clock millis at startup
}
```

This is separate from the existing `AppState` (which serves the business API). The admin router is built via a function:

```rust
pub fn admin_router<M, Q, S>(state: AdminState<M, Q, S>) -> Router
```

This returns a `Router` that gets nested under `/admin/v1` in `HttpServer::new()`. The existing drain endpoints stay at `/admin/drain` (no `/v1` prefix) for backward compatibility.

### A.2 — Response envelope types

Define shared response types in `src/server/http/admin.rs`:

```rust
#[derive(Serialize)]
struct AdminResponse<T: Serialize> {
    data: T,
    timestamp: String,  // RFC 3339
}

fn admin_ok<T: Serialize>(data: T) -> Json<AdminResponse<T>> { ... }
```

### A.3 — CORS layer

Add `tower-http` CORS middleware to the admin router. Default: allow `*` origin (admin is behind a firewall or VPN). Configurable via `DeployConfig` if needed later.

### A.4 — Wire AdminState in main.rs

In `main.rs`, construct `AdminState` from the existing `Arc`s (manager, engine, storage, lifecycle) and pass it to `admin_router()`. Merge the resulting router into the HTTP server alongside the existing `/admin/drain` routes and `/v1` business routes.

### Files changed
- `src/server/http/admin.rs` — new file (admin router, AdminState, response types)
- `src/server/http.rs` — register admin module, merge admin router
- `src/main.rs` — construct AdminState, pass to HTTP server
- `Cargo.toml` — `tower-http` features for CORS (if not already present)

---

## Phase B: Namespace & Topic Query Endpoints

### B.1 — System overview

**`GET /admin/v1/overview`**

Dashboard landing page data. Returns global system health in one call.

```json
{
  "uptime_secs": 86400,
  "started_at": "2026-03-20T10:30:00Z",
  "server_state": "running",
  "tcp": { "bind_address": "127.0.0.1", "port": 9876 },
  "http": { "bind_address": "127.0.0.1", "port": 8080 },
  "memory": {
    "used_bytes": 52428800,
    "max_bytes": 4294967296,
    "usage_pct": 1.2,
    "warning_fired": false
  },
  "disk": {
    "data_dir": "./data",
    "used_pct": 45,
    "available_bytes": 107374182400,
    "disk_backed": false,
    "warning_fired": false
  },
  "totals": {
    "namespaces": 3,
    "topics": 42,
    "active_connections": 17
  }
}
```

**Implementation**: Read from `engine.total_bytes()`, `lifecycle.state()`, config struct, `AtomicBool` warning flags. Namespace/topic/connection counts: iterate `NamespaceState` atomics. `disk_usage()` from `storage/disk.rs`.

### B.2 — List namespaces

**`GET /admin/v1/namespaces`**

```json
{
  "namespaces": [
    {
      "name": "default",
      "active_connections": 12,
      "topic_count": 8,
      "message_count": 500000,
      "disk_bytes": 1073741824,
      "quotas": {
        "max_total_messages": null,
        "max_topics": null,
        "max_connections": null,
        "max_disk_bytes": null
      },
      "retention": { "max_age_secs": 604800, "max_bytes": null, "max_segments": 64 }
    }
  ]
}
```

**New trait method needed**: `NamespaceManager::list_namespaces() -> Vec<NamespaceId>` — return all configured namespace IDs. Currently the manager has namespace state but no method to list them all. Add to trait + `DefaultNamespaceManager`.

**Implementation**: Iterate configured namespaces, read `NamespaceState` atomics for counts, `engine.namespace_disk_bytes(ns)` for disk, `NamespaceQuota` for quotas, `NamespaceConfig.retention` from config.

### B.3 — Namespace detail

**`GET /admin/v1/namespaces/{ns}`**

Same as list entry but includes topics list and permissions:

```json
{
  "name": "default",
  "active_connections": 12,
  "topic_count": 8,
  "message_count": 500000,
  "disk_bytes": 1073741824,
  "permissions": ["Subscribe", "Publish", "Query", "Fetch"],
  "quotas": { ... },
  "retention": { ... },
  "topics": [
    {
      "name": "events",
      "depth": 5000,
      "bytes_used": 1048576,
      "base_offset": 450000,
      "head_offset": 455000,
      "subscriber_count": 3,
      "consumer_group_count": 1
    }
  ]
}
```

**Implementation**: `manager.list_topics(ns)` for topic names, `engine.topic_stats(ns, topic)` per topic, subscriber/group counts from `TopicDeliveryState`.

### B.4 — Topic detail

**`GET /admin/v1/namespaces/{ns}/topics/{topic}`**

Deep topic view with subscriber summary and segment summary:

```json
{
  "namespace": "default",
  "name": "events",
  "depth": 5000,
  "bytes_used": 1048576,
  "base_offset": 450000,
  "head_offset": 455000,
  "oldest_timestamp_millis": 1700000000000,
  "newest_timestamp_millis": 1700086400000,
  "dirty_bytes": 32768,
  "min_live_offset": 449800,
  "subscribers": {
    "individual_count": 2,
    "group_count": 1,
    "total_inflight": 17
  },
  "storage": {
    "segment_count": 12,
    "total_disk_bytes": 805306368
  }
}
```

**New trait method needed**: `QueueEngine::topic_dirty_bytes(ns, topic) -> u64` — expose `Topic.dirty_bytes()` through the engine. Currently `dirty_bytes` is only accessible inside the topic lock during flush.

**New method needed**: `StorageBackend::segment_count(ns, topic) -> usize` — count segments without listing full metadata (lightweight). Or reuse `scan_segments()` from retention module which is already `pub(crate)`.

### Files changed
- `src/server/http/admin.rs` — route handlers for B.1–B.4
- `src/namespace.rs` — add `list_namespaces()` to `NamespaceManager` trait
- `src/namespace/manager.rs` — implement `list_namespaces()`
- `src/queue.rs` — add `topic_dirty_bytes()` to `QueueEngine` trait (optional)
- `src/queue/engine.rs` — implement `topic_dirty_bytes()`

---

## Phase C: Subscriber & Connection Inspection

### C.1 — New trait methods for subscriber inspection

Add to `NamespaceManager` trait:

```rust
async fn topic_subscribers(&self, ns: &NamespaceId, topic: &TopicName) -> Result<TopicSubscriberInfo>;
async fn list_connections(&self) -> Vec<ConnectionInfo>;
async fn connection_detail(&self, client: &ClientId) -> Option<ConnectionDetail>;
```

Define response structs in `src/types.rs`:

```rust
pub struct TopicSubscriberInfo {
    pub individual: Vec<SubscriberInfo>,
    pub groups: Vec<ConsumerGroupInfo>,
}

pub struct SubscriberInfo {
    pub client_id: ClientId,
    pub next_offset: TopicOffset,
    pub lag: u64,                    // head_offset - next_offset
    pub inflight_count: usize,
    pub is_slow: bool,
    pub total_drops: u64,
    pub consecutive_failures: u64,
    pub last_activity_ago_ms: u64,   // millis since last activity
}

pub struct ConsumerGroupInfo {
    pub name: GroupName,
    pub member_count: usize,
    pub members: Vec<ClientId>,
    pub next_offset: TopicOffset,
    pub lag: u64,
    pub inflight_count: usize,
    pub is_slow: bool,
    pub total_drops: u64,
    pub emptied_at_ago_ms: Option<u64>,
}

pub struct ConnectionInfo {
    pub client_id: ClientId,
    pub namespace: NamespaceId,
    pub subscription_count: usize,
}

pub struct ConnectionDetail {
    pub client_id: ClientId,
    pub namespace: NamespaceId,
    pub subscriptions: Vec<SubscriptionEntry>,
}

pub struct SubscriptionEntry {
    pub topic: TopicName,
    pub group: Option<GroupName>,
}
```

**Implementation**: `topic_subscribers()` locks `TopicDeliveryState`, iterates `individual` and `groups` maps, snapshots each entry. `list_connections()` reads `client_namespaces` + `client_subscriptions` reverse indexes. `connection_detail()` reads reverse indexes for a specific client.

### C.2 — Subscriber endpoints

**`GET /admin/v1/namespaces/{ns}/topics/{topic}/subscribers`**

Returns full subscriber and consumer group detail for a topic:

```json
{
  "topic": "events",
  "head_offset": 455000,
  "individual": [
    {
      "client_id": 42,
      "next_offset": 454990,
      "lag": 10,
      "inflight_count": 5,
      "is_slow": false,
      "total_drops": 0,
      "consecutive_failures": 0,
      "last_activity_ago_ms": 2000
    }
  ],
  "groups": [
    {
      "name": "workers",
      "member_count": 3,
      "members": [101, 102, 103],
      "next_offset": 454800,
      "lag": 200,
      "inflight_count": 12,
      "is_slow": false,
      "total_drops": 0,
      "emptied_at_ago_ms": null
    }
  ]
}
```

### C.3 — Consumer group detail

**`GET /admin/v1/namespaces/{ns}/topics/{topic}/groups/{group}`**

Returns per-member detail for a specific consumer group:

```json
{
  "name": "workers",
  "topic": "events",
  "next_offset": 454800,
  "lag": 200,
  "is_slow": false,
  "total_drops": 0,
  "inflight": [
    {
      "offset": 454800,
      "message_id": 982341,
      "delivered_to": 101,
      "delivered_ago_ms": 1500,
      "delivery_count": 1
    }
  ],
  "members": [101, 102, 103]
}
```

**New trait method needed**: `NamespaceManager::group_detail(ns, topic, group) -> Result<ConsumerGroupDetail>` — returns group state including inflight entries. Inflight entries are small (bounded by `MAX_INFLIGHT` and timeout) so serializing them is safe.

### C.4 — Connection endpoints

**`GET /admin/v1/connections`**

List all active TCP connections across all namespaces:

```json
{
  "total": 17,
  "connections": [
    {
      "client_id": 42,
      "namespace": "default",
      "subscription_count": 2
    }
  ]
}
```

**`GET /admin/v1/connections/{client_id}`**

```json
{
  "client_id": 42,
  "namespace": "default",
  "subscriptions": [
    { "topic": "events", "group": null },
    { "topic": "commands", "group": "workers" }
  ]
}
```

### C.5 — Slow consumer overview

**`GET /admin/v1/slow-consumers`**

Cross-namespace aggregation of all subscribers/groups currently marked slow:

```json
{
  "slow_consumers": [
    {
      "client_id": 42,
      "namespace": "default",
      "topic": "events",
      "group": null,
      "type": "individual",
      "consecutive_failures": 150,
      "total_drops": 340,
      "next_offset": 440000,
      "lag": 15000
    }
  ]
}
```

**New trait method needed**: `NamespaceManager::slow_consumers() -> Vec<SlowConsumerInfo>` — scans all topics across all namespaces, returns entries where `is_slow == true`. This is an expensive scan but only called on admin request, not on hot path.

### Files changed
- `src/types.rs` — new info structs (TopicSubscriberInfo, SubscriberInfo, ConsumerGroupInfo, etc.)
- `src/namespace.rs` — add trait methods
- `src/namespace/manager.rs` — implement trait methods
- `src/server/http/admin.rs` — route handlers for C.2–C.5

---

## Phase D: Operational Commands

### D.1 — Force disconnect

**`POST /admin/v1/connections/{client_id}/disconnect`**

Force-close a specific TCP connection. The server sends an `Error { InternalError, "disconnected by admin" }` frame before closing.

**Implementation**: Add a `disconnect_signal` mechanism. The simplest approach: `NamespaceManager::force_disconnect(client_id)` drops all of the client's subscription senders (`mpsc::Sender`), causing the forwarders to detect `ChannelClosed` and exit. Then calls the existing `disconnect()` cleanup. The TCP connection loop detects the dead delivery channel and exits.

This does NOT require a per-connection watch channel — it works through the existing channel-closure detection path. The `disconnect()` method already handles all cleanup (remove from subscriber maps, decrement counters, reassign inflight).

Response:

```json
{
  "disconnected": true,
  "client_id": 42,
  "namespace": "default",
  "subscriptions_removed": 2
}
```

Returns 404 if `client_id` not found.

### D.2 — Force flush

**`POST /admin/v1/flush`**

Force-flush all dirty topics to disk immediately.

**Implementation**: Call `engine.force_flush()` which already exists. Return stats:

```json
{
  "flushed": true
}
```

### D.3 — Force compact

**`POST /admin/v1/namespaces/{ns}/topics/{topic}/compact`**

Force in-memory compaction to a specific offset.

Request body:
```json
{ "to_offset": 450000 }
```

**Implementation**: Call `engine.compact(ns, topic, to_offset)`. This bypasses `min_live_offset` — it's an admin override. The admin GUI should warn that compacting below `min_live_offset` will cause subscribers to skip messages.

Response:
```json
{
  "bytes_freed": 1048576,
  "messages_freed": 500
}
```

### D.4 — Force retention sweep

**`POST /admin/v1/retention/sweep`**

Trigger an immediate retention sweep across all namespaces.

**Implementation**: Add a `mpsc::Sender<oneshot::Sender<SweepResult>>` channel to the retention task (same pattern as drain trigger). The admin handler sends a request, the retention task runs `sweep()`, and sends back the result.

Response:
```json
{
  "segments_deleted": 12,
  "namespaces_swept": 3,
  "topics_swept": 42
}
```

### D.5 — Purge topic

**`DELETE /admin/v1/namespaces/{ns}/topics/{topic}`**

Delete all messages from a topic — both in-memory buffer and all segment files on disk. Active subscribers are NOT disconnected; they simply won't receive any more old messages (their `next_offset` will be ahead of `base_offset`, or they'll get `OffsetNotAvailable` on resubscribe).

**New trait method**: `QueueEngine::purge_topic(ns, topic) -> Result<PurgeResult>`
- Acquires topic lock, calls `buffer.clear()`, resets `base_offset = head_offset`, resets `bytes_used = 0`
- Calls `storage.delete_all_segments(ns, topic)` — new storage method that deletes the entire segment directory
- Updates `total_bytes` atomic
- Returns `PurgeResult { messages_purged, bytes_freed, segments_deleted }`

**New trait method**: `StorageBackend::delete_all_segments(ns, topic) -> Result<usize>` — removes entire segment directory contents (`.seg` + `.idx` files), evicts all cached indexes, resets namespace disk bytes.

Response:
```json
{
  "messages_purged": 50000,
  "bytes_freed": 104857600,
  "segments_deleted": 12
}
```

### D.6 — Reset consumer group offset

**`POST /admin/v1/namespaces/{ns}/topics/{topic}/groups/{group}/reset`**

Reset a consumer group's cursor to a specific offset or timestamp. Clears all inflight entries for the group (they'll be redelivered from the new offset).

Request body:
```json
{ "to_offset": 450000 }
```
or:
```json
{ "to_timestamp": 1700000000000 }
```

**New trait method**: `NamespaceManager::reset_group_offset(ns, topic, group, offset) -> Result<ResetResult>`
- Locks `TopicDeliveryState`, finds group in `groups` map
- Clears `inflight` map
- Sets `next_offset = offset`
- Resets `consecutive_send_failures = 0`, `total_drops = 0`, `is_slow = false`
- Notifies delivery task (group delivery task will pick up from new offset)
- Returns `ResetResult { previous_offset, new_offset, inflight_cleared }`

If `to_timestamp` is provided, resolve via `engine.offset_at_timestamp(ns, topic, ts)` first.

Response:
```json
{
  "previous_offset": 454800,
  "new_offset": 450000,
  "inflight_cleared": 12
}
```

### D.7 — Admin publish

**`POST /admin/v1/namespaces/{ns}/topics/{topic}/publish`**

Publish a message as admin — bypasses namespace API key auth (uses admin key instead). Useful for injecting test messages or replaying data.

Request body: raw bytes (same as existing publish endpoint).

Response: same as existing publish response `{ offset, message_id, timestamp_millis }`.

**Implementation**: Reuse existing publish logic but skip namespace permission check. The admin key is already validated by `require_admin_auth`.

### Files changed
- `src/server/http/admin.rs` — route handlers for D.1–D.7
- `src/namespace.rs` — add `force_disconnect()`, `reset_group_offset()` to trait
- `src/namespace/manager.rs` — implement new trait methods
- `src/queue.rs` — add `purge_topic()` to `QueueEngine` trait
- `src/queue/engine.rs` — implement `purge_topic()`
- `src/storage.rs` — add `delete_all_segments()` to `StorageBackend` trait
- `src/storage/fs.rs` — implement `delete_all_segments()`
- `src/retention.rs` — add trigger channel for on-demand sweep
- `src/main.rs` — wire retention trigger channel

---

## Phase E: Storage & Segment Inspection

### E.1 — List segments

**`GET /admin/v1/namespaces/{ns}/topics/{topic}/segments`**

```json
{
  "namespace": "default",
  "topic": "events",
  "segment_count": 12,
  "total_disk_bytes": 805306368,
  "segments": [
    {
      "base_offset": 0,
      "filename": "00000000000000000000.seg",
      "size_bytes": 67108864,
      "has_index": true,
      "index_size_bytes": 1416,
      "mtime_millis": 1700000000000,
      "is_active": true
    }
  ]
}
```

**Implementation**: Use `retention::scan_segments(seg_dir)` to get `SegmentInfo` entries. Check for `.idx` file existence and size. Determine `is_active` by comparing to the LRU writer cache's current segment offset.

**New method needed**: `StorageBackend::list_segments_detail(ns, topic) -> Result<Vec<SegmentDetail>>` — richer than `scan_segments()`, includes index presence/size and active flag.

### E.2 — Segment detail

**`GET /admin/v1/namespaces/{ns}/topics/{topic}/segments/{base_offset}`**

Single segment with index entries:

```json
{
  "base_offset": 400000,
  "filename": "00000000000000400000.seg",
  "size_bytes": 67108864,
  "mtime_millis": 1700000000000,
  "is_active": false,
  "index": {
    "exists": true,
    "entry_count": 59,
    "size_bytes": 1416,
    "interval": 256,
    "entries": [
      { "offset": 400000, "byte_position": 0, "timestamp_millis": 1700000000000 },
      { "offset": 400256, "byte_position": 65536, "timestamp_millis": 1700000256000 }
    ]
  }
}
```

**Implementation**: `SegmentIndex::load(path)` returns entries. Cap entries at query param `?max_index_entries=100` (default 100) to prevent enormous responses on large indexes.

### E.3 — Rebuild segment index

**`POST /admin/v1/namespaces/{ns}/topics/{topic}/segments/{base_offset}/rebuild-index`**

Force rebuild the `.idx` file for a segment by scanning the `.seg` file with `RecordStream`.

**Implementation**: Call existing `rebuild_index(seg_path, index_interval)`. Evict old index from `index_cache`. Return the new index stats.

Response:
```json
{
  "rebuilt": true,
  "entry_count": 59,
  "size_bytes": 1416
}
```

### E.4 — Delete specific segments

**`POST /admin/v1/namespaces/{ns}/topics/{topic}/segments/delete`**

Delete specific segments by offset list. Admin override — does NOT check `min_live_offset`.

Request body:
```json
{ "base_offsets": [0, 100000, 200000] }
```

**Implementation**: For each offset, delete the `.seg` and `.idx` files, update segment_index, evict from index_cache, update namespace_disk_bytes. Skip the active writer segment (return error for that one).

Response:
```json
{
  "deleted": 3,
  "skipped": 0,
  "bytes_freed": 201326592
}
```

### Files changed
- `src/server/http/admin.rs` — route handlers for E.1–E.4
- `src/storage.rs` — add `list_segments_detail()` to trait
- `src/storage/fs.rs` — implement `list_segments_detail()`, expose `rebuild_index()` through trait

---

## Phase F: Message Browser

### F.1 — Browse messages by offset

**`GET /admin/v1/namespaces/{ns}/topics/{topic}/messages?offset={u64}&limit={u32}`**

Peek at messages starting from an offset. Payloads are base64-encoded and truncated for preview.

Query params:
- `offset` (required) — starting offset
- `limit` (optional, default 50, max 500) — number of messages
- `payload_limit` (optional, default 1024) — max bytes of payload to include per message (0 = omit payloads)

```json
{
  "topic": "events",
  "messages": [
    {
      "offset": 450000,
      "message_id": 982341,
      "timestamp_millis": 1700000000000,
      "payload_size": 256,
      "payload_base64": "eyJldmVudCI6InVzZXIu...",
      "payload_truncated": false,
      "source": "memory"
    }
  ],
  "returned": 50,
  "has_more": true,
  "base_offset": 445000,
  "head_offset": 455000
}
```

**Implementation**: Use `engine.consume_from(ns, topic, offset)` for in-memory messages (offset >= base_offset). For cold messages (offset < base_offset), use `storage.read_cold_batch(ns, topic, offset, limit)`. Tag each message with `source: "memory"` or `source: "disk"`. Base64-encode payload, truncate to `payload_limit` bytes before encoding.

### F.2 — Browse messages by timestamp

**`GET /admin/v1/namespaces/{ns}/topics/{topic}/messages?timestamp={u64}&limit={u32}`**

Same as F.1 but starts from the first message at or after the given timestamp.

**Implementation**: Resolve timestamp to offset via `engine.offset_at_timestamp(ns, topic, ts)`, then delegate to offset-based browsing.

### F.3 — Single message detail

**`GET /admin/v1/namespaces/{ns}/topics/{topic}/messages/{offset}`**

Full message detail with complete payload (no truncation). Query param `?encoding=base64|hex|utf8` controls payload representation (default base64).

```json
{
  "offset": 450000,
  "message_id": 982341,
  "timestamp_millis": 1700000000000,
  "payload_size": 256,
  "payload_base64": "...",
  "source": "memory"
}
```

**Implementation**: Single `engine.consume_from()` call. Falls back to `storage.read_at()` for cold messages.

### Files changed
- `src/server/http/admin.rs` — route handlers for F.1–F.3

---

## Phase G: Config & Metrics JSON

### G.1 — Running config viewer

**`GET /admin/v1/config`**

Returns the running configuration with API keys redacted.

```json
{
  "tcp": { "bind_address": "127.0.0.1", "port": 9876, "max_message_size": 1048576, ... },
  "http": { "bind_address": "127.0.0.1", "port": 8080 },
  "storage": { "data_dir": "./data", "fsync_strategy": "every_batch", ... },
  "queue": { "max_depth": 100000, "backpressure_policy": "DropOldest", ... },
  "deploy": { "drain_timeout_secs": 30, "shutdown_timeout_secs": 10, ... },
  "metrics": { "enabled": true },
  "namespaces": {
    "default": {
      "api_key": "****",
      "permissions": ["Subscribe", "Publish", "Query", "Fetch"],
      "max_total_messages": null,
      ...
    }
  }
}
```

**Implementation**: Serialize `ServerConfig` to JSON. Write a custom serializer or post-process to redact `api_key` fields (replace with `"****"`). Also redact `admin_api_key`.

### G.2 — Metrics JSON snapshot

**`GET /admin/v1/metrics`**

Returns key metrics as structured JSON (not Prometheus text format). This is purpose-built for the dashboard — pre-aggregated, labeled, ready to render.

```json
{
  "counters": {
    "messages_published_total": [
      { "namespace": "default", "topic": "events", "value": 500000 }
    ],
    "messages_delivered_total": [
      { "namespace": "default", "topic": "events", "mode": "individual", "value": 1500000 }
    ]
  },
  "gauges": {
    "connections_active": [ { "namespace": "default", "value": 12 } ],
    "queue_depth": [ { "namespace": "default", "topic": "events", "value": 5000 } ],
    "global_memory_bytes": 52428800,
    "disk_used_pct": 45
  }
}
```

**Implementation**: Parse the Prometheus text output from `PrometheusHandle::render()` into structured JSON. Alternatively, maintain a parallel set of atomic counters on `AdminState` that are cheaper to read — but parsing the Prometheus output is simpler and keeps metrics as the single source of truth.

### Files changed
- `src/server/http/admin.rs` — route handlers for G.1–G.2
- `src/config.rs` — add `Serialize` derive to config structs (if not already present), add redaction helper

---

## Phase H: Integration Tests

### H.1 — Test infrastructure

Create `tests/admin_integration.rs`. Use the existing `DeployTestServer` from `tests/common/mod.rs` — it provides a full FsStorage-backed server with TCP + HTTP.

- Admin endpoints use the `admin_api_key` from `DeployConfig`
- Tests use `reqwest` (already a dev dependency) for HTTP calls
- Each test gets its own `TempDir` for isolation

### H.2 — Query endpoint tests

- `test_admin_overview` — start server, publish some messages, verify overview stats
- `test_admin_list_namespaces` — verify namespace listing with quotas
- `test_admin_namespace_detail` — verify topics appear after publish
- `test_admin_topic_detail` — verify depth, offsets, subscriber counts
- `test_admin_subscribers` — connect TCP clients, subscribe, verify subscriber listing shows correct offsets/lag
- `test_admin_consumer_group` — subscribe with groups, verify member list and group offset
- `test_admin_connections` — verify connection listing matches connected clients
- `test_admin_slow_consumers` — trigger slow consumer detection, verify slow-consumers endpoint
- `test_admin_segments` — publish enough data to create multiple segments, verify segment listing
- `test_admin_messages` — publish messages, browse by offset and timestamp, verify payload preview

### H.3 — Command endpoint tests

- `test_admin_force_disconnect` — connect client, force disconnect via admin, verify cleanup
- `test_admin_force_flush` — publish messages, force flush, verify data on disk
- `test_admin_force_compact` — publish, compact to offset, verify reduced depth
- `test_admin_purge_topic` — publish, purge, verify topic is empty and segments deleted
- `test_admin_reset_group` — subscribe with group, consume some, reset offset, verify group resumes from new offset
- `test_admin_retention_sweep` — configure tight retention, trigger sweep, verify segments deleted
- `test_admin_rebuild_index` — delete index file, rebuild via admin, verify index restored

### H.4 — Auth and error tests

- `test_admin_no_key` — verify 403 when admin_api_key not configured
- `test_admin_wrong_key` — verify 403 with incorrect key
- `test_admin_namespace_not_found` — verify 404 for nonexistent namespace
- `test_admin_topic_not_found` — verify 404 for nonexistent topic
- `test_admin_connection_not_found` — verify 404 for nonexistent client_id

### Files changed
- `tests/admin_integration.rs` — new test file
- `tests/common/mod.rs` — add admin test helpers (HTTP client with admin auth header)

---

## Complete Endpoint Reference

### Query Endpoints (GET)

| Endpoint | Phase | Purpose |
|---|---|---|
| `GET /admin/v1/overview` | B.1 | System-wide health snapshot |
| `GET /admin/v1/namespaces` | B.2 | All namespaces with stats |
| `GET /admin/v1/namespaces/{ns}` | B.3 | Namespace detail with topics |
| `GET /admin/v1/namespaces/{ns}/topics/{topic}` | B.4 | Deep topic stats |
| `GET /admin/v1/namespaces/{ns}/topics/{topic}/subscribers` | C.2 | Subscribers and groups for topic |
| `GET /admin/v1/namespaces/{ns}/topics/{topic}/groups/{group}` | C.3 | Consumer group detail with inflight |
| `GET /admin/v1/connections` | C.4 | All active TCP connections |
| `GET /admin/v1/connections/{client_id}` | C.4 | Connection detail with subscriptions |
| `GET /admin/v1/slow-consumers` | C.5 | Cross-namespace slow consumer list |
| `GET /admin/v1/namespaces/{ns}/topics/{topic}/segments` | E.1 | Segment files with metadata |
| `GET /admin/v1/namespaces/{ns}/topics/{topic}/segments/{base_offset}` | E.2 | Segment detail with index entries |
| `GET /admin/v1/namespaces/{ns}/topics/{topic}/messages` | F.1/F.2 | Browse messages by offset or timestamp |
| `GET /admin/v1/namespaces/{ns}/topics/{topic}/messages/{offset}` | F.3 | Single message with full payload |
| `GET /admin/v1/config` | G.1 | Running config (keys redacted) |
| `GET /admin/v1/metrics` | G.2 | Metrics as structured JSON |

### Key Management Endpoints

| Endpoint | Phase | Purpose |
|---|---|---|
| `GET /admin/v1/namespaces/{ns}/keys` | I.1 | List all keys in namespace (redacted) |
| `POST /admin/v1/namespaces/{ns}/keys` | I.2 | Create a named key |
| `GET /admin/v1/namespaces/{ns}/keys/{name}` | I.3 | Get key details (redacted) |
| `PUT /admin/v1/namespaces/{ns}/keys/{name}` | I.4 | Update key permissions |
| `DELETE /admin/v1/namespaces/{ns}/keys/{name}` | I.5 | Revoke a runtime key |

### Command Endpoints (POST/DELETE)

| Endpoint | Phase | Purpose |
|---|---|---|
| `POST /admin/v1/connections/{client_id}/disconnect` | D.1 | Force-disconnect a client |
| `POST /admin/v1/flush` | D.2 | Force flush all dirty topics |
| `POST /admin/v1/namespaces/{ns}/topics/{topic}/compact` | D.3 | Force compact to offset |
| `POST /admin/v1/retention/sweep` | D.4 | Trigger immediate retention sweep |
| `DELETE /admin/v1/namespaces/{ns}/topics/{topic}` | D.5 | Purge all messages from topic |
| `POST /admin/v1/namespaces/{ns}/topics/{topic}/groups/{group}/reset` | D.6 | Reset group offset |
| `POST /admin/v1/namespaces/{ns}/topics/{topic}/publish` | D.7 | Admin publish (bypass namespace auth) |
| `POST /admin/v1/namespaces/{ns}/topics/{topic}/segments/{base_offset}/rebuild-index` | E.3 | Rebuild segment index |
| `POST /admin/v1/namespaces/{ns}/topics/{topic}/segments/delete` | E.4 | Delete specific segments |

### Pre-existing Admin Endpoints (unchanged)

| Endpoint | Phase | Purpose |
|---|---|---|
| `POST /admin/drain` | Plan 14 | Trigger graceful drain |
| `GET /admin/drain/status` | Plan 14 | Drain progress |
| `PUT /v1/admin/log-level` | Plan 11 | Change log level at runtime |
| `GET /v1/admin/log-level` | Plan 11 | Current log level |

---

## New Trait Methods Summary

### NamespaceManager trait additions

| Method | Phase | Purpose |
|---|---|---|
| `list_namespaces()` | B.2 | Return all configured namespace IDs |
| `topic_subscribers(ns, topic)` | C.1 | Snapshot of all subscribers + groups |
| `list_connections()` | C.1 | All active connections with namespace |
| `connection_detail(client_id)` | C.1 | Subscriptions for a specific client |
| `slow_consumers()` | C.5 | Cross-namespace slow consumer scan |
| `group_detail(ns, topic, group)` | C.3 | Group state including inflight entries |
| `force_disconnect(client_id)` | D.1 | Close client's subscription channels |
| `reset_group_offset(ns, topic, group, offset)` | D.6 | Reset group cursor and clear inflight |

### QueueEngine trait additions

| Method | Phase | Purpose |
|---|---|---|
| `topic_dirty_bytes(ns, topic)` | B.4 | Unflushed bytes for a topic |
| `purge_topic(ns, topic)` | D.5 | Clear buffer, delete all segments |

### StorageBackend trait additions

| Method | Phase | Purpose |
|---|---|---|
| `list_segments_detail(ns, topic)` | E.1 | Segment metadata with index info |
| `delete_all_segments(ns, topic)` | D.5 | Remove entire segment directory |
| `rebuild_segment_index(ns, topic, base_offset)` | E.3 | Rebuild `.idx` for a segment |

---

## Phase I: API Key Management Endpoints

### Goal

Add CRUD endpoints for programmatic API key management under the existing admin router. Operators can create, list, inspect, update, and revoke named API keys per namespace — without editing config files or restarting the server.

### Prerequisites

- Phase A complete (admin router foundation, `AdminState`, response envelope)
- Plan 4 Phase 4.1 complete (`ApiKeyStore` with `ManagedKey`, `KeySource`, mutation methods, key store file)

### I.1 — List keys in namespace

**`GET /admin/v1/namespaces/{ns}/keys`**

Returns all keys (config + runtime) with secrets redacted.

```json
{
  "data": {
    "namespace": "default",
    "keys": [
      {
        "name": "_seed:default",
        "api_key": "chan...tion",
        "permissions": ["Subscribe", "Publish", "Query", "Fetch"],
        "source": "config",
        "created_at": "1970-01-01T00:00:00Z",
        "expires_at": null
      },
      {
        "name": "ci-publisher",
        "api_key": "gene...0000",
        "permissions": ["Publish"],
        "source": "runtime",
        "created_at": "2026-03-29T10:15:30Z",
        "expires_at": "2026-12-31T23:59:59Z"
      }
    ]
  },
  "timestamp": "2026-03-29T12:00:00Z"
}
```

**Implementation**: Call `ApiKeyStore::list_keys(namespace)`. Returns `Vec<KeyInfo>` with redacted secrets (first 4 + `...` + last 4 chars). Returns 404 if namespace not found.

### I.2 — Create a named key

**`POST /admin/v1/namespaces/{ns}/keys`**

Request body:
```json
{
  "name": "ci-publisher",
  "permissions": ["Publish"],
  "api_key": "optional-caller-provided-key-00",
  "expires_at": "2026-12-31T23:59:59Z"
}
```

- `name` (required): `[a-zA-Z0-9_-]{1,64}`, must not start with `_seed:`
- `permissions` (required): non-empty array of `Permission` variants
- `api_key` (optional): if omitted, server generates a cryptographically random 32-char key. If provided, must be ≥24 chars and globally unique.
- `expires_at` (optional): ISO 8601 timestamp. If omitted, key never expires. Stored but not enforced — expiration enforcement is a future feature.

Response (the **only** time the full secret is returned):
```json
{
  "data": {
    "name": "ci-publisher",
    "namespace": "default",
    "api_key": "the-full-secret-only-shown-once",
    "permissions": ["Publish"],
    "source": "runtime",
    "created_at": "2026-03-29T10:15:30Z",
    "expires_at": "2026-12-31T23:59:59Z"
  },
  "timestamp": "2026-03-29T10:15:30Z"
}
```

**Implementation**: Build `ManagedKey`, call `ApiKeyStore::create_key()`. On success, return 201 with full secret. Errors:
- 400 — invalid name format, empty permissions, API key too short
- 409 — name already exists in namespace, or API key string already in use
- 404 — namespace not found (checked via `NamespaceManager::list_namespaces()`)

### I.3 — Get key details

**`GET /admin/v1/namespaces/{ns}/keys/{name}`**

Returns a single key with redacted secret. Same shape as a list entry.

```json
{
  "data": {
    "name": "ci-publisher",
    "namespace": "default",
    "api_key": "gene...0000",
    "permissions": ["Publish"],
    "source": "runtime",
    "created_at": "2026-03-29T10:15:30Z"
  },
  "timestamp": "2026-03-29T10:15:30Z"
}
```

**Implementation**: Call `ApiKeyStore::get_key(namespace, name)`. Returns 404 if key or namespace not found.

### I.4 — Update key permissions

**`PUT /admin/v1/namespaces/{ns}/keys/{name}`**

Request body:
```json
{
  "permissions": ["Subscribe", "Publish", "Query"]
}
```

Response:
```json
{
  "data": {
    "name": "ci-publisher",
    "namespace": "default",
    "api_key": "gene...0000",
    "permissions": ["Subscribe", "Publish", "Query"],
    "source": "runtime",
    "created_at": "2026-03-29T10:15:30Z"
  },
  "timestamp": "2026-03-29T10:15:31Z"
}
```

**Implementation**: Call `ApiKeyStore::update_key_permissions()`. Errors:
- 400 — empty permissions, config key immutable (`source: config`)
- 404 — key or namespace not found

### I.5 — Revoke a key

**`DELETE /admin/v1/namespaces/{ns}/keys/{name}`**

Response:
```json
{
  "data": {
    "revoked": true,
    "name": "ci-publisher",
    "namespace": "default"
  },
  "timestamp": "2026-03-29T10:16:00Z"
}
```

**Implementation**: Call `ApiKeyStore::revoke_key()`. Existing TCP connections authenticated with this key remain active (session caches namespace, not key). New connections with the revoked key will fail auth. Errors:
- 400 — config key immutable (`source: config`)
- 404 — key or namespace not found

### I.6 — Request/response types

Define in `src/server/http/admin.rs`:

```rust
#[derive(Deserialize)]
pub struct CreateKeyRequest {
    pub name: String,
    pub permissions: Vec<Permission>,
    pub api_key: Option<String>,
    pub expires_at: Option<String>,  // ISO 8601 timestamp, stored but not enforced
}

#[derive(Deserialize)]
pub struct UpdateKeyRequest {
    pub permissions: Vec<Permission>,
}

#[derive(Serialize)]
pub struct CreateKeyResponse {
    pub name: String,
    pub namespace: String,
    pub api_key: String,           // full secret — only on create
    pub permissions: Vec<Permission>,
    pub source: String,
    pub created_at: String,
    pub expires_at: Option<String>,
}

#[derive(Serialize)]
pub struct RevokeKeyResponse {
    pub revoked: bool,
    pub name: String,
    pub namespace: String,
}
```

### Files changed
- `src/server/http/admin.rs` — route handlers for I.1–I.5, request/response types
- `src/server/http.rs` — register key management routes on admin router

### Tasks

- [ ] Define `CreateKeyRequest`, `UpdateKeyRequest` request types
- [ ] Define `CreateKeyResponse`, `RevokeKeyResponse` response types
- [ ] Implement `GET /admin/v1/namespaces/{ns}/keys` handler — list all keys (redacted)
- [ ] Implement `POST /admin/v1/namespaces/{ns}/keys` handler — create key, return full secret
- [ ] Implement `GET /admin/v1/namespaces/{ns}/keys/{name}` handler — get single key (redacted)
- [ ] Implement `PUT /admin/v1/namespaces/{ns}/keys/{name}` handler — update permissions
- [ ] Implement `DELETE /admin/v1/namespaces/{ns}/keys/{name}` handler — revoke key
- [ ] Register all 5 routes on admin router
- [ ] Map `ApiKeyStore` errors to HTTP status codes (400, 404, 409)
- [ ] Validate namespace exists via `NamespaceManager::list_namespaces()` before key operations (404 if not — using the manager is authoritative since it handles additions/removals via reload)
- [ ] Integration test: create key → list keys → verify key appears with redacted secret
- [ ] Integration test: create key → authenticate with new key via TCP → publish succeeds
- [ ] Integration test: create key with caller-provided secret → authenticate with that secret
- [ ] Integration test: create key with duplicate name → 409
- [ ] Integration test: create key with duplicate API key string → 409
- [ ] Integration test: create key with invalid name format → 400
- [ ] Integration test: update runtime key permissions → verify new permissions enforced
- [ ] Integration test: update config key permissions → 400 (immutable)
- [ ] Integration test: revoke runtime key → new auth with that key fails
- [ ] Integration test: revoke config key → 400 (immutable)
- [ ] Integration test: revoke key → existing TCP connection still works (session cached)
- [ ] Integration test: get nonexistent key → 404
- [ ] Integration test: operations on nonexistent namespace → 404
- [ ] Integration test: create key without admin auth → 403

**Verification:** All 5 key management endpoints work correctly. Config keys are visible but immutable. Runtime keys are fully manageable. Secrets are only exposed on create. Error cases return correct HTTP status codes. All existing admin tests pass unchanged.

---

## Future Extensibility

This plan establishes the pattern for future features to add admin endpoints:

1. **Plan 17 (config hot-reload)**: Will add `POST /admin/v1/config/reload` and `GET /admin/v1/config/diff` to the admin router
2. **Plan 18 (dead letter queues)**: Will add `GET /admin/v1/namespaces/{ns}/topics/{topic}/dlq` for DLQ inspection
3. **Plan 20 (wildcard subscriptions)**: Will add subscription pattern info to the subscriber endpoints
4. **Plan 24 (clustering)**: Will add `GET /admin/v1/cluster/nodes` and cluster-wide admin views

Each of these just adds handlers to `admin_router()` and (if needed) new trait methods. No changes to the admin auth, routing, or response envelope infrastructure.
