feat(kafka): filter loaded messages

apps/desktop/src/components/mq/MessageBrowser.vue

@@ -6,10 +6,12 @@ import type { MqSystemKind, PeekedMessage, PeekMessagesOptions, TopicRef } from
 import { mqPeekMessages } from "@/lib/backend/api";
 import { formatError } from "@/lib/backend/errorUtils";
 import { copyToClipboard } from "@/lib/common/clipboard";
+import { buildKafkaMessageSearchText, kafkaMessageSearchTextMatches, normalizeKafkaMessageSearchQuery } from "@/lib/mq/kafkaMessageSearch";
 import { parseNonNegativeSafeInteger } from "@/lib/mq/mqPeekFilters";
 import { useToast } from "@/composables/useToast";
 import { Button } from "@/components/ui/button";
 import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
+import MqSearchInput from "@/components/mq/shared/MqSearchInput.vue";
 
 type MessageBrowserAppearance = "form" | "monitoring";
 
@@ -35,6 +37,7 @@ const partition = ref<string | number>("");
 const offset = ref<string | number>("");
 const count = ref(20);
 const advancedExpanded = ref(false);
+const messageSearchQuery = ref("");
 type KafkaPeekStartPosition = NonNullable<PeekMessagesOptions["startPosition"]>;
 const kafkaStartPosition = ref<KafkaPeekStartPosition>("latest");
 let messageRequestVersion = 0;
@@ -42,6 +45,18 @@ let messageRequestVersion = 0;
 const isKafka = computed(() => props.mqSystemKind === "kafka");
 const isKafkaOffsetMode = computed(() => kafkaStartPosition.value === "offset");
 const isMonitoring = computed(() => props.appearance === "monitoring");
+const normalizedMessageSearchQuery = computed(() => normalizeKafkaMessageSearchQuery(messageSearchQuery.value));
+const searchableMessages = computed(() =>
+  messages.value.map((message) => ({
+    message,
+    searchText: buildKafkaMessageSearchText(message, formatMessageTimestamp(message.publishTime)),
+  })),
+);
+const filteredMessages = computed(() => {
+  const query = normalizedMessageSearchQuery.value;
+  if (!query) return messages.value;
+  return searchableMessages.value.filter(({ searchText }) => kafkaMessageSearchTextMatches(searchText, query)).map(({ message }) => message);
+});
 
 function peekGroupName(): string {
   if (props.mqSystemKind === "rocketmq") return "__dbx_rocketmq_viewer__";
@@ -145,6 +160,7 @@ function formatMessageTimestamp(value?: string): string {
 }
 
 watch([() => props.connectionId, () => props.mqSystemKind], () => {
+  messageSearchQuery.value = "";
   invalidateMessageRequest();
 });
 
@@ -153,6 +169,7 @@ watch(
   () => {
     partition.value = "";
     offset.value = "";
+    messageSearchQuery.value = "";
     invalidateMessageRequest();
   },
 );
@@ -234,11 +251,19 @@ watch(kafkaStartPosition, () => {
       </div>
     </template>
 
+    <div v-if="isKafka && messages.length" class="message-filter-row" data-testid="kafka-message-filter">
+      <MqSearchInput v-model="messageSearchQuery" :placeholder="t('mqMessages.filterLoadedPlaceholder')" :aria-label="t('mqMessages.filterLoadedPlaceholder')" :disabled="loading" data-testid="kafka-message-filter-input" />
+      <span class="mq-result-count" data-testid="kafka-message-filter-count" aria-live="polite">
+        {{ t("mqMessages.filterLoadedCount", { matched: filteredMessages.length, loaded: messages.length }) }}
+      </span>
+    </div>
+
     <div v-if="error" class="panel-error">{{ error }}</div>
     <div v-else-if="loading" class="message-empty">{{ t("mqMessages.messagesLoading") }}</div>
     <div v-else-if="!messages.length" class="message-empty">{{ t("mqMessages.noMessages") }}</div>
+    <div v-else-if="!filteredMessages.length" class="message-empty" data-testid="kafka-message-filter-empty">{{ t("mqMessages.noMatchingMessages") }}</div>
     <div v-else class="message-list">
-      <article v-for="message in messages" :key="`${message.properties?.partition ?? 'p'}-${message.messageId || message.position}`" class="message-row">
+      <article v-for="message in filteredMessages" :key="`${message.properties?.partition ?? 'p'}-${message.messageId || message.position}`" class="message-row">
         <div class="message-meta">
           <span>#{{ message.position }}</span>
           <span v-if="message.properties?.partition != null">{{ t("mqMessages.metaPartition", { partition: message.properties.partition }) }}</span>
@@ -375,6 +400,18 @@ watch(kafkaStartPosition, () => {
   margin-top: 6px;
 }
 
+.message-filter-row {
+  display: flex;
+  align-items: center;
+  gap: 10px;
+  margin-bottom: 12px;
+}
+
+.message-filter-row :deep(.mq-search-input) {
+  width: min(420px, 100%);
+  flex: 1;
+}
+
 .collapse-toggle {
   display: flex;
   align-items: center;

apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts

@@ -263,6 +263,128 @@ describe("MessageBrowser", () => {
     expect(browser.textContent).toContain("partial message");
   });
 
+  it("filters loaded Kafka messages and shows the matching count", async () => {
+    backend.mqPeekMessages.mockResolvedValueOnce([
+      {
+        position: 1,
+        messageId: "11",
+        key: "order-alpha",
+        payloadBase64: "",
+        payloadText: "first payload",
+        properties: { partition: "0" },
+        headers: {},
+      },
+      {
+        position: 2,
+        messageId: "12",
+        key: "order-beta",
+        payloadBase64: "",
+        payloadText: "second payload",
+        properties: { partition: "1" },
+        headers: { TraceId: "REQUEST-99" },
+      },
+    ]);
+    const browser = await mountBrowser();
+
+    await loadMessages(browser);
+    const filter = browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]');
+    if (!filter) throw new Error("Kafka message filter input not found");
+    expect(filter.getAttribute("aria-label")).toBe("mqMessages.filterLoadedPlaceholder");
+    expect(browser.querySelector('[data-testid="kafka-message-filter-count"]')?.textContent).toContain('mqMessages.filterLoadedCount:{"matched":2,"loaded":2}');
+
+    await setInputValue(filter, "request-99");
+
+    expect(browser.textContent).not.toContain("first payload");
+    expect(browser.textContent).toContain("second payload");
+    expect(browser.querySelector('[data-testid="kafka-message-filter-count"]')?.textContent).toContain('mqMessages.filterLoadedCount:{"matched":1,"loaded":2}');
+  });
+
+  it("shows a dedicated empty state without hiding an incomplete warning", async () => {
+    backend.mqPeekMessages.mockResolvedValueOnce({
+      messages: [
+        {
+          position: 1,
+          messageId: "partial",
+          payloadBase64: "",
+          payloadText: "partial message",
+          properties: {},
+          headers: {},
+        },
+      ],
+      incomplete: true,
+    });
+    const browser = await mountBrowser();
+
+    await loadMessages(browser);
+    const filter = browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]');
+    if (!filter) throw new Error("Kafka message filter input not found");
+    await setInputValue(filter, "missing value");
+
+    expect(browser.querySelector('[data-testid="peek-incomplete"]')).not.toBeNull();
+    expect(browser.querySelector('[data-testid="kafka-message-filter-empty"]')?.textContent).toContain("mqMessages.noMatchingMessages");
+    expect(browser.textContent).not.toContain("partial message");
+  });
+
+  it("keeps a filter across reloads and Kafka start-position changes", async () => {
+    backend.mqPeekMessages.mockResolvedValue([
+      {
+        position: 1,
+        messageId: "17",
+        payloadBase64: "",
+        payloadText: "keep this message",
+        properties: {},
+        headers: {},
+      },
+    ]);
+    const browser = await mountBrowser();
+
+    await loadMessages(browser);
+    const filter = browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]');
+    const startPosition = browser.querySelector<HTMLElement>('[data-testid="kafka-peek-start-position"]');
+    if (!filter || !startPosition) throw new Error("Kafka message filter controls not found");
+    await setInputValue(filter, "keep");
+    await loadMessages(browser);
+    expect(browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]')?.value).toBe("keep");
+
+    await setSelectValue(startPosition, "earliest");
+    await loadMessages(browser);
+
+    expect(browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]')?.value).toBe("keep");
+    expect(browser.textContent).toContain("keep this message");
+  });
+
+  it("clears the filter when the topic or connection context changes", async () => {
+    const { browser, topic, connectionId } = await mountBrowserWithMutableTopic();
+
+    await loadMessages(browser);
+    const filter = browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]');
+    if (!filter) throw new Error("Kafka message filter input not found");
+    await setInputValue(filter, "existing");
+
+    topic.value = { ...TOPIC, topic: "payments" };
+    await flushUi();
+    await loadMessages(browser);
+
+    expect(browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]')?.value).toBe("");
+
+    const reloadedFilter = browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]');
+    if (!reloadedFilter) throw new Error("Reloaded Kafka message filter input not found");
+    await setInputValue(reloadedFilter, "existing");
+    connectionId.value = "mq-2";
+    await flushUi();
+    await loadMessages(browser);
+
+    expect(browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]')?.value).toBe("");
+  });
+
+  it("does not expose the loaded-message filter for non-Kafka browsers", async () => {
+    const browser = await mountBrowser("rabbitmq");
+
+    await loadMessages(browser);
+
+    expect(browser.querySelector('[data-testid="kafka-message-filter"]')).toBeNull();
+  });
+
   it("sends explicit earliest and offset read positions", async () => {
     const browser = await mountBrowser();
     const startPosition = browser.querySelector<HTMLElement>('[data-testid="kafka-peek-start-position"]');

apps/desktop/src/i18n/locales/en.ts

@@ -7420,6 +7420,9 @@ export default {
     kafkaOffsetHint: "Reads forward from the specified offset in the chosen partition, or every partition when left empty.",
     messagesLoading: "Loading messages...",
     noMessages: "No messages",
+    filterLoadedPlaceholder: "Filter loaded messages",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "No loaded messages match this filter",
     peekIncomplete: "The message results are incomplete because the broker did not finish the snapshot.",
     selectTopicBeforeLoad: "Select a topic before loading messages",
     partitionMustBeNonNegativeInt: "Partition must be an integer ≥ 0; leave empty for all partitions",

apps/desktop/src/i18n/locales/es.ts

@@ -7355,6 +7355,9 @@ export default withEnglishFallback({
     kafkaOffsetHint: "Lee hacia adelante desde el offset indicado en la partición elegida, o en todas si se deja vacía.",
     messagesLoading: "Cargando mensajes...",
     noMessages: "No hay mensajes",
+    filterLoadedPlaceholder: "Filtrar mensajes cargados",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "Ningún mensaje cargado coincide con este filtro",
     peekIncomplete: "Los resultados de mensajes están incompletos porque el broker no terminó la instantánea.",
     selectTopicBeforeLoad: "Seleccione un tema antes de cargar mensajes",
     partitionMustBeNonNegativeInt: "La partición debe ser un entero ≥ 0; déjelo vacío para todas las particiones",

apps/desktop/src/i18n/locales/it.ts

@@ -7356,6 +7356,9 @@ export default withEnglishFallback({
     kafkaOffsetHint: "Legge in avanti dall'offset indicato nella partizione scelta, o in tutte se lasciata vuota.",
     messagesLoading: "Caricamento messaggi...",
     noMessages: "Nessun messaggio",
+    filterLoadedPlaceholder: "Filtra i messaggi caricati",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "Nessun messaggio caricato corrisponde a questo filtro",
     peekIncomplete: "I risultati dei messaggi sono incompleti perché il broker non ha completato lo snapshot.",
     selectTopicBeforeLoad: "Seleziona un topic prima di caricare i messaggi",
     partitionMustBeNonNegativeInt: "La partizione deve essere un intero ≥ 0; lascia vuoto per tutte le partizioni",

apps/desktop/src/i18n/locales/ja.ts

@@ -7407,6 +7407,9 @@ export default withEnglishFallback({
     kafkaOffsetHint: "指定 Offset から、選択したパーティション（空欄ならすべて）を昇順に読み取ります。",
     messagesLoading: "メッセージを読み込み中...",
     noMessages: "メッセージがありません",
+    filterLoadedPlaceholder: "読み込み済みメッセージを絞り込む",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "このフィルターに一致する読み込み済みメッセージはありません",
     peekIncomplete: "ブローカーがスナップショットの読み取りを完了できなかったため、メッセージ結果は不完全です。",
     selectTopicBeforeLoad: "メッセージを読み込む前にトピックを選択してください",
     partitionMustBeNonNegativeInt: "パーティションは 0 以上の整数である必要があります。空欄の場合はすべてのパーティション",

apps/desktop/src/i18n/locales/ko.ts

@@ -6782,6 +6782,9 @@ export default withEnglishFallback({
     kafkaOffsetHint: "지정한 오프셋부터 선택한 파티션(비워 두면 전체)을 오름차순으로 읽습니다.",
     messagesLoading: "메시지를 불러오는 중...",
     noMessages: "메시지가 없습니다",
+    filterLoadedPlaceholder: "불러온 메시지 필터링",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "이 필터와 일치하는 불러온 메시지가 없습니다",
     peekIncomplete: "브로커가 스냅샷 읽기를 완료하지 못해 메시지 결과가 불완전합니다.",
     selectTopicBeforeLoad: "메시지를 불러오기 전에 토픽을 선택하세요",
     partitionMustBeNonNegativeInt: "파티션은 0 이상의 정수여야 하며, 모든 파티션은 비워 두세요",

apps/desktop/src/i18n/locales/pt-BR.ts

@@ -7357,6 +7357,9 @@ export default withEnglishFallback({
     kafkaOffsetHint: "Lê em ordem crescente a partir do offset informado na partição escolhida, ou em todas se deixada vazia.",
     messagesLoading: "Carregando mensagens...",
     noMessages: "Nenhuma mensagem",
+    filterLoadedPlaceholder: "Filtrar mensagens carregadas",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "Nenhuma mensagem carregada corresponde a este filtro",
     peekIncomplete: "Os resultados das mensagens estão incompletos porque o broker não concluiu o snapshot.",
     selectTopicBeforeLoad: "Selecione um tópico antes de carregar mensagens",
     partitionMustBeNonNegativeInt: "A partição deve ser um inteiro ≥ 0; deixe vazio para todas as partições",

apps/desktop/src/i18n/locales/zh-CN.ts

@@ -7404,6 +7404,9 @@ export default withEnglishFallback({
     kafkaOffsetHint: "从指定 Offset 开始正序读取；填写分区时仅读该分区，留空则对全部分区分别读取。",
     messagesLoading: "消息加载中...",
     noMessages: "暂无消息",
+    filterLoadedPlaceholder: "筛选已加载消息",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "没有已加载消息符合当前筛选条件",
     peekIncomplete: "消息结果不完整：Broker 未能在限制内完成快照读取。",
     selectTopicBeforeLoad: "请先选择主题再加载消息",
     partitionMustBeNonNegativeInt: "分区必须是大于等于 0 的整数；留空表示查询全部分区",

apps/desktop/src/i18n/locales/zh-TW.ts

@@ -7347,6 +7347,9 @@ export default withEnglishFallback({
     kafkaOffsetHint: "從指定 Offset 開始正序讀取；填寫分區時只讀該分區，留空則對全部分區分別讀取。",
     messagesLoading: "訊息載入中...",
     noMessages: "暫無訊息",
+    filterLoadedPlaceholder: "篩選已載入訊息",
+    filterLoadedCount: "{matched} / {loaded}",
+    noMatchingMessages: "沒有已載入訊息符合目前篩選條件",
     peekIncomplete: "訊息結果不完整：Broker 未能在限制內完成快照讀取。",
     selectTopicBeforeLoad: "請先選取主題再載入訊息",
     partitionMustBeNonNegativeInt: "分區必須是大於等於 0 的整數；留空表示查詢全部分區",

apps/desktop/src/lib/mq/__tests__/kafkaMessageSearch.spec.ts

@@ -0,0 +1,45 @@
+import { describe, expect, it } from "vitest";
+import type { PeekedMessage } from "@/types/mq";
+import { buildKafkaMessageSearchText, kafkaMessageSearchTextMatches, normalizeKafkaMessageSearchQuery } from "@/lib/mq/kafkaMessageSearch";
+
+function message(overrides: Partial<PeekedMessage> = {}): PeekedMessage {
+  return {
+    position: 7,
+    messageId: "42",
+    key: "Order-ABC",
+    publishTime: "1724457600000",
+    properties: { partition: "3" },
+    headers: { TraceId: "REQUEST-99", source: "billing" },
+    payloadBase64: "",
+    payloadText: '{"customer":"Alice"}',
+    ...overrides,
+  };
+}
+
+describe("Kafka loaded message search", () => {
+  it("builds a case-insensitive index from every visible message field", () => {
+    const searchText = buildKafkaMessageSearchText(message(), "8/24/2024, 8:00:00 AM");
+
+    for (const query of ["7", "42", "order-abc", "3", "1724457600000", "8:00:00 am", "traceid", "request-99", "billing", "alice"]) {
+      expect(kafkaMessageSearchTextMatches(searchText, normalizeKafkaMessageSearchQuery(query)), query).toBe(true);
+    }
+  });
+
+  it("uses the displayed Base64 payload when text is unavailable", () => {
+    const searchText = buildKafkaMessageSearchText(
+      message({
+        payloadText: undefined,
+        payloadBase64: "AAECAw==",
+      }),
+    );
+
+    expect(kafkaMessageSearchTextMatches(searchText, normalizeKafkaMessageSearchQuery("aecaw=="))).toBe(true);
+  });
+
+  it("treats whitespace-only queries as empty and rejects missing text", () => {
+    const searchText = buildKafkaMessageSearchText(message());
+
+    expect(kafkaMessageSearchTextMatches(searchText, normalizeKafkaMessageSearchQuery("   "))).toBe(true);
+    expect(kafkaMessageSearchTextMatches(searchText, normalizeKafkaMessageSearchQuery("not-present"))).toBe(false);
+  });
+});

apps/desktop/src/lib/mq/kafkaMessageSearch.ts

@@ -0,0 +1,30 @@
+import type { PeekedMessage } from "@/types/mq";
+
+function appendSearchValue(values: string[], value: unknown) {
+  if (value == null || value === "") return;
+  values.push(String(value));
+}
+
+export function normalizeKafkaMessageSearchQuery(query: string): string {
+  return query.trim().toLowerCase();
+}
+
+export function buildKafkaMessageSearchText(message: PeekedMessage, formattedPublishTime?: string): string {
+  const values: string[] = [];
+  appendSearchValue(values, message.position);
+  appendSearchValue(values, message.messageId);
+  appendSearchValue(values, message.key);
+  appendSearchValue(values, message.properties?.partition);
+  appendSearchValue(values, message.publishTime);
+  appendSearchValue(values, formattedPublishTime);
+  for (const [key, value] of Object.entries(message.headers || {})) {
+    appendSearchValue(values, key);
+    appendSearchValue(values, value);
+  }
+  appendSearchValue(values, message.payloadText ?? message.payloadBase64);
+  return values.join("\n").toLowerCase();
+}
+
+export function kafkaMessageSearchTextMatches(searchText: string, normalizedQuery: string): boolean {
+  return normalizedQuery === "" || searchText.includes(normalizedQuery);
+}