feat(search): whole-topic background message search (UI, REST, CLI, MCP)

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- **Whole-topic message search** (F2): `POST /api/clusters/{clusterId}/topics/{topic}/search` starts a bounded background scan (regex / contains / key / timestamp filters, capped at 1000 results and 2M scanned messages, max 3 concurrent jobs) with progress polling and cancellation (`GET`/`DELETE .../searches/{id}`). Exposed in the UI as a new **Search** tab in the explorer, in the CLI as `kview search` / `kview search-status`, and in MCP as `kview_search_messages` + `kview_search_status` (18 tools now). Read-only like the browser: throwaway group, never commits.
 - **API authentication with roles** (`kview.auth.*`, off by default): `token` mode (static bearer tokens) and `oidc` mode (JWTs from any issuer, configurable role claim), each with an **admin** and a **read-only** role — reads for everyone, mutations (produce, topic/group management, offset resets, connection management) admin-only. The web UI prompts for the token (🔑 in the top bar), the CLI takes `--token`/`KVIEW_TOKEN`, the MCP server takes `KVIEW_TOKEN`. In authenticated modes CORS is closed unless `KVIEW_AUTH_ALLOWED_ORIGINS` lists origins; the static UI and `/actuator/health` stay public. See the new Authentication section in the README.
 - **AI/MCP safety pack**: global read-only mode (`KVIEW_READONLY=true` — every mutation refused for all clients and roles, UI shows a banner), an append-only mutation audit trail (`KVIEW_AUDIT=true` → `<data-dir>/audit.log` JSONL with client, method, path and status; all three clients send an `X-Kview-Client` identifier), a public `GET /api/meta` endpoint, and a per-tool allowlist for the MCP server (`KVIEW_ALLOWED_TOOLS`, e.g. serve only read-only tools to an AI agent).
 

README.md

@@ -42,7 +42,8 @@ consumer group, so it is safe to point at production topics.
 - **Topics**: leaders, replicas, ISR, begin/end offsets, message counts and configs.
   Create and delete topics, increase partitions, edit configs.
 - **Data explorer**: read messages (latest N, exact offsets, from a timestamp, filter by
-  key or value contains), pretty-printed JSON with a copy button, live tail over SSE,
+  key or value contains), whole-topic background search with regex filters and progress,
+  pretty-printed JSON with a copy button, live tail over SSE,
   reproduce a message to any topic with one click.
 - **Produce**: key, value, partition, timestamp and headers, with JSON validation as you type.
 - **Consumer groups**: state, per-partition lag with bars, member assignments, delete a
@@ -116,7 +117,7 @@ scripting. If the server runs with authentication enabled, add `--token <T>` (en
 ### 3. MCP server (AI assistants)
 
 `mcp/kview-mcp.js` is a [Model Context Protocol](https://modelcontextprotocol.io) server
-that exposes Kview as 16 tools, so Claude Desktop, Cursor, ZCode or any MCP client can
+that exposes Kview as 18 tools, so Claude Desktop, Cursor, ZCode or any MCP client can
 inspect and operate your Kafka cluster in natural language: cluster health, topic
 inventory, message browsing with filters, producing test events, consumer-group lag and
 broker distribution. Zero dependencies, speaks newline-delimited JSON-RPC over stdio.

cli/kview.mjs

@@ -64,6 +64,17 @@ function fail(msg) {
   process.exit(1);
 }
 
+/** Starts are async: poll the search until done or the wait budget runs out. */
+async function pollSearch(searchId, waitSec) {
+  const deadline = Date.now() + waitSec * 1000;
+  let status;
+  do {
+    status = await api('GET', clusterPath(`/searches/${encodeURIComponent(searchId)}`));
+    if (status.state === 'RUNNING') await new Promise((r) => setTimeout(r, 500));
+  } while (status.state === 'RUNNING' && Date.now() < deadline);
+  return status;
+}
+
 // ---- output helpers ----------------------------------------------------------
 const oneLine = (s, max = 70) =>
   s === null || s === undefined ? '' :
@@ -111,6 +122,9 @@ Topics:    topics | topic <name> | topic-create <name> [--partitions N] [--rf N]
 Messages:  produce <topic> [-k key] [-v payload] [--file path] [--partition N] [--header k=v...]
            browse <topic> [--start latest|earliest|timestamp] [--limit N] [--partition N]
                   [--key-contains S] [--value-contains S] [--ts 2026-01-01T10:00:00Z]
+           search <topic> [--value-regex RE] [--key-contains S] [--value-contains S]
+                  [--from-ts MS|ISO] [--to-ts MS|ISO] [--limit N] [--partition N] [--wait SEC]
+           search-status <searchId>
            tail <topic> [--from latest|earliest] [--partition N]     (Ctrl+C stops)
 Groups:    groups | lag <group> | reset-offsets <group> --topic T --mode earliest|latest|offset|timestamp
            [--value V] | group-delete <group>
@@ -227,6 +241,34 @@ Examples:  kview overview
       (out.reachedEnd ? ' (end of partition reached)' : ' (older messages may exist — raise --limit)'));
   },
 
+  async search() {
+    const topic = positional[1] ?? fail('usage: kview search <topic> [--value-regex RE] [--key-contains S] [--value-contains S] [--from-ts MS|ISO] [--to-ts MS|ISO] [--limit N] [--partition N] [--wait SEC]');
+    const toMs = (v) => (v === undefined ? undefined : (/^-?\d{13,}$/.test(v) ? Number(v) : Date.parse(v)));
+    const body = {
+      keyContains: flags['key-contains'],
+      valueContains: flags['value-contains'],
+      valueRegex: flags['value-regex'],
+      fromTs: toMs(flags['from-ts']),
+      toTs: toMs(flags['to-ts']),
+      limit: num('limit'),
+      partition: num('partition'),
+    };
+    const { searchId } = await api('POST', clusterPath(`/topics/${encodeURIComponent(topic)}/search`), body);
+    const status = await pollSearch(searchId, num('wait') ?? 60);
+    if (JSON_OUT) { console.log(JSON.stringify(status, null, 2)); return; }
+    for (const m of status.results ?? []) console.log(msgLine(m) + decodedSuffix(m));
+    console.error(`-- ${status.matched} matched · scanned ${status.scanned} · state ${status.state}`
+      + (status.state === 'RUNNING' ? ` — still running, poll with: kview search-status ${searchId}` : '')
+      + (status.limitReached ? ' · limit reached' : ''));
+  },
+
+  async 'search-status'() {
+    const searchId = positional[1] ?? fail('usage: kview search-status <searchId>');
+    const status = await api('GET', clusterPath(`/searches/${encodeURIComponent(searchId)}`));
+    console.log(JSON_OUT ? JSON.stringify(status, null, 2)
+      : `search ${status.searchId}  state=${status.state}  matched=${status.matched}  scanned=${status.scanned}  partitions=${status.partitionsDone}/${status.partitionsTotal}`);
+  },
+
   async tail() {
     const topic = positional[1] ?? fail('usage: kview tail <topic> [--from latest|earliest] [--partition N]');
     const qs = new URLSearchParams({ from: flags.from ?? 'latest' });

mcp/kview-mcp.js

@@ -174,6 +174,37 @@ const impl = {
     return api(`/api/clusters/${cluster(p)}/groups`);
   },
 
+  async search(p) {
+    const topic = requireArg(p, 'topic');
+    const body = {};
+    for (const k of ['keyContains', 'valueContains', 'valueRegex']) {
+      if (p[k] !== undefined) body[k] = String(p[k]);
+    }
+    for (const k of ['fromTs', 'toTs']) {
+      if (p[k] !== undefined) body[k] = Number(p[k]);
+    }
+    if (p.limit !== undefined) body.limit = Math.min(Number(p.limit) || 100, 1000);
+    if (p.partition !== undefined) body.partition = Number(p.partition);
+    const { searchId } = await api(`/api/clusters/${cluster(p)}/topics/${encodeURIComponent(topic)}/search`,
+      { method: 'POST', body: JSON.stringify(body) });
+    const maxWaitMs = Math.min(Number(p.maxWaitMs) || 15000, 60000);
+    const deadline = Date.now() + maxWaitMs;
+    let status;
+    do {
+      await new Promise((r) => setTimeout(r, 400));
+      status = await api(`/api/clusters/${cluster(p)}/searches/${searchId}`);
+    } while (status.state === 'RUNNING' && Date.now() < deadline);
+    if (status.state === 'RUNNING') {
+      status.note = `Still running (scanned ${status.scanned} so far) — poll with kview_search_status(searchId='${searchId}')`;
+    }
+    return status;
+  },
+
+  async searchStatus(p) {
+    const searchId = requireArg(p, 'searchId');
+    return api(`/api/clusters/${cluster(p)}/searches/${searchId}`);
+  },
+
   async groupDetail(p) {
     const group = requireArg(p, 'group');
     return api(`/api/clusters/${cluster(p)}/groups/${encodeURIComponent(group)}`);
@@ -310,6 +341,32 @@ const TOOLS = [
     }, ['topic']),
     run: impl.browse,
   },
+  {
+    name: 'kview_search_messages',
+    description: 'Search a whole topic (all partitions, background job) for messages matching text or regex filters, optionally bounded by timestamps. Waits up to maxWaitMs for completion and returns matches with progress. Read-only.',
+    inputSchema: object({
+      topic: str('Topic name'),
+      valueRegex: str('Regular expression the value must match, e.g. "orderId"\\s*:\\s*"ORD-[0-9]+" (optional)'),
+      valueContains: str('Only messages whose value contains this text (optional)'),
+      keyContains: str('Only messages whose key contains this text (optional)'),
+      fromTs: num('Epoch-ms lower timestamp bound (inclusive, optional)'),
+      toTs: num('Epoch-ms upper timestamp bound (exclusive, optional)'),
+      limit: num('Max matches to collect (default 100, max 1000)'),
+      partition: num('Restrict to one partition (optional)'),
+      maxWaitMs: num('How long to wait for completion, ms (default 15000, max 60000)'),
+      cluster: str('Kview connection id (optional)'),
+    }, ['topic']),
+    run: impl.search,
+  },
+  {
+    name: 'kview_search_status',
+    description: 'Progress and results of a background search started with kview_search_messages (searchId is in its output).',
+    inputSchema: object({
+      searchId: str('Search id from kview_search_messages'),
+      cluster: str('Kview connection id (optional)'),
+    }, ['searchId']),
+    run: impl.searchStatus,
+  },
   {
     name: 'kview_consumer_groups',
     description: 'List consumer groups with state, committed partition count and total lag.',

src/main/java/com/ritikbansod/kafkawrapper/auth/MutationRequests.java

@@ -18,7 +18,8 @@ final class MutationRequests implements RequestMatcher {
     static final MutationRequests INSTANCE = new MutationRequests();
 
     private static final PathMatcher ANT = new AntPathMatcher();
-    private static final List<String> READ_SHAPED_POSTS = List.of("/**/browse", "/**/decode", "/**/encode", "/**/test");
+    private static final List<String> READ_SHAPED_POSTS =
+            List.of("/**/browse", "/**/decode", "/**/encode", "/**/test", "/**/search");
 
     private MutationRequests() { }
 

src/main/java/com/ritikbansod/kafkawrapper/browse/MessageBrowserService.java

@@ -158,7 +158,8 @@ private List<TopicPartition> resolvePartitions(KafkaConsumer<byte[], byte[]> con
         return result;
     }
 
-    private BrowserMessage toView(String clusterId, ConsumerRecord<byte[], byte[]> record) {
+    /** Maps a raw record to the API view (schema decode included) — shared with the background search. */
+    public BrowserMessage toView(String clusterId, ConsumerRecord<byte[], byte[]> record) {
         byte[] keyBytes = record.key();
         byte[] valueBytes = record.value();
         Map<String, String> headers = new LinkedHashMap<>();
@@ -178,7 +179,7 @@ private BrowserMessage toView(String clusterId, ConsumerRecord<byte[], byte[]> r
                 schema);
     }
 
-    static String asText(byte[] bytes) {
+    public static String asText(byte[] bytes) {
         return bytes == null ? null : new String(bytes, StandardCharsets.UTF_8);
     }
 

src/main/java/com/ritikbansod/kafkawrapper/search/SearchController.java

@@ -0,0 +1,44 @@
+package com.ritikbansod.kafkawrapper.search;
+
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Map;
+
+/**
+ * Whole-topic background search: start (POST — a read-shaped mutation), poll
+ * progress + results, cancel. The search itself is read-only: never commits,
+ * never joins a real group.
+ */
+@RestController
+@RequestMapping("/api/clusters/{clusterId}")
+public class SearchController {
+
+    private final SearchService searchService;
+
+    public SearchController(SearchService searchService) {
+        this.searchService = searchService;
+    }
+
+    @PostMapping("/topics/{topic}/search")
+    public Map<String, String> start(@PathVariable String clusterId, @PathVariable String topic,
+                                     @RequestBody SearchService.SearchRequest request) {
+        return Map.of("searchId", searchService.start(clusterId, topic, request));
+    }
+
+    @GetMapping("/searches/{searchId}")
+    public SearchService.SearchStatus status(@PathVariable String clusterId, @PathVariable String searchId) {
+        return searchService.status(clusterId, searchId);
+    }
+
+    @DeleteMapping("/searches/{searchId}")
+    public Map<String, String> cancel(@PathVariable String clusterId, @PathVariable String searchId) {
+        searchService.cancel(clusterId, searchId);
+        return Map.of("searchId", searchId, "status", "cancelling");
+    }
+}