diff --git a/apps/desktop/src/components/mq/MessageBrowser.vue b/apps/desktop/src/components/mq/MessageBrowser.vue index 7b77909d04..a0a62a2cbe 100644 --- a/apps/desktop/src/components/mq/MessageBrowser.vue +++ b/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(""); const offset = ref(""); const count = ref(20); const advancedExpanded = ref(false); +const messageSearchQuery = ref(""); type KafkaPeekStartPosition = NonNullable; const kafkaStartPosition = ref("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, () => { +
+ + + {{ t("mqMessages.filterLoadedCount", { matched: filteredMessages.length, loaded: messages.length }) }} + +
+
{{ error }}
{{ t("mqMessages.messagesLoading") }}
{{ t("mqMessages.noMessages") }}
+
{{ t("mqMessages.noMatchingMessages") }}
-
+
#{{ message.position }} {{ t("mqMessages.metaPartition", { partition: message.properties.partition }) }} @@ -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; diff --git a/apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts b/apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts index cee25f65b6..7082074bbe 100644 --- a/apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts +++ b/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('[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('[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('[data-testid="kafka-message-filter-input"]'); + const startPosition = browser.querySelector('[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('[data-testid="kafka-message-filter-input"]')?.value).toBe("keep"); + + await setSelectValue(startPosition, "earliest"); + await loadMessages(browser); + + expect(browser.querySelector('[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('[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('[data-testid="kafka-message-filter-input"]')?.value).toBe(""); + + const reloadedFilter = browser.querySelector('[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('[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('[data-testid="kafka-peek-start-position"]'); diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index cefffc0d42..fbe2d635c8 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/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", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index 72d021627a..3baa65cf35 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/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", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index 68167b6b7f..e08d5c2ef9 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/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", diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index 476af7fa04..bf6f19eaea 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/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 以上の整数である必要があります。空欄の場合はすべてのパーティション", diff --git a/apps/desktop/src/i18n/locales/ko.ts b/apps/desktop/src/i18n/locales/ko.ts index afd1410a97..16bd2248ca 100644 --- a/apps/desktop/src/i18n/locales/ko.ts +++ b/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 이상의 정수여야 하며, 모든 파티션은 비워 두세요", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index d58716087f..f37afadccb 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/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", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 6d9d098168..8f396e1251 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/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 的整数；留空表示查询全部分区", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 7515e9c254..3d46af4931 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/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 的整數；留空表示查詢全部分區", diff --git a/apps/desktop/src/lib/mq/__tests__/kafkaMessageSearch.spec.ts b/apps/desktop/src/lib/mq/__tests__/kafkaMessageSearch.spec.ts new file mode 100644 index 0000000000..5444b12407 --- /dev/null +++ b/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 { + 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); + }); +}); diff --git a/apps/desktop/src/lib/mq/kafkaMessageSearch.ts b/apps/desktop/src/lib/mq/kafkaMessageSearch.ts new file mode 100644 index 0000000000..a2f3cf8bb1 --- /dev/null +++ b/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); +}