This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 4142cb96 feat(messages): expose persisted query history (#2113)
4142cb96 is described below
commit 4142cb96da28c4e5190738832053a4cf69448275
Author: lizhimins <[email protected]>
AuthorDate: Thu Aug 13 19:17:10 2026 +0800
feat(messages): expose persisted query history (#2113)
Add server-side message and trace query history with pagination, search
and summaries, plus a searchable history drawer on the message page.
Search terms are LIKE-escaped so wildcards match literally. Rebased onto
current rocketmq-studio (InstanceSelect / topicError coexist).
Co-authored-by: Yue Wang <[email protected]>
---
.../instance/message/MessageQueryHistoryVO.java | 29 ++++
.../instance/message/QueryHistoryController.java | 60 ++++++++
.../instance/message/QueryHistoryService.java | 100 +++++++++++++
.../instance/message/QueryHistorySummaryVO.java | 20 +++
.../instance/message/TraceQueryHistoryVO.java | 25 ++++
.../message/QueryHistoryControllerTest.java | 76 ++++++++++
.../instance/message/QueryHistoryServiceTest.java | 49 +++++++
web/src/api/messageHistory.test.ts | 57 ++++++++
web/src/api/messageHistory.ts | 80 +++++++++++
web/src/components/MessageQueryHistoryDrawer.tsx | 158 +++++++++++++++++++++
.../__tests__/MessageQueryHistoryDrawer.test.tsx | 91 ++++++++++++
web/src/pages/instance/message.tsx | 10 ++
12 files changed, 755 insertions(+)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageQueryHistoryVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageQueryHistoryVO.java
new file mode 100644
index 00000000..931239c1
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageQueryHistoryVO.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+@Data
+@Builder
+public class MessageQueryHistoryVO {
+ private Long id;
+ private String queryType;
+ private String topic;
+ private String msgId;
+ private String tag;
+ private String messageKey;
+ private Long startTime;
+ private Long endTime;
+ private int resultCount;
+ private String clusterId;
+ private String queriedBy;
+ private LocalDateTime queriedAt;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryController.java
new file mode 100644
index 00000000..d5ed62d2
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryController.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.apache.rocketmq.studio.common.domain.Result;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/api/query-history")
+@RequiredArgsConstructor
+public class QueryHistoryController {
+ private static final int MAX_PAGE_SIZE = 100;
+ private final QueryHistoryService queryHistoryService;
+
+ @GetMapping("/messages")
+ public Result<PageResult<MessageQueryHistoryVO>> listMessageQueries(
+ @RequestParam(required = false) String clusterId,
+ @RequestParam(required = false) String queryType,
+ @RequestParam(required = false) String search,
+ @RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "20") int pageSize) {
+ validatePage(page, pageSize);
+ return Result.ok(queryHistoryService.listMessageQueries(
+ clusterId, queryType, search, page, pageSize));
+ }
+
+ @GetMapping("/traces")
+ public Result<PageResult<TraceQueryHistoryVO>> listTraceQueries(
+ @RequestParam(required = false) String clusterId,
+ @RequestParam(required = false) String search,
+ @RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "20") int pageSize) {
+ validatePage(page, pageSize);
+ return Result.ok(queryHistoryService.listTraceQueries(clusterId,
search, page, pageSize));
+ }
+
+ @GetMapping("/summary")
+ public Result<QueryHistorySummaryVO> summary(@RequestParam(required =
false) String clusterId) {
+ return Result.ok(queryHistoryService.summarize(clusterId));
+ }
+
+ private void validatePage(int page, int pageSize) {
+ if (page < 1) {
+ throw new BusinessException(400, "page must be at least 1");
+ }
+ if (pageSize < 1 || pageSize > MAX_PAGE_SIZE) {
+ throw new BusinessException(400, "pageSize must be between 1 and "
+ MAX_PAGE_SIZE);
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
index 68b56d1c..0db51126 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
@@ -17,6 +17,8 @@
package org.apache.rocketmq.studio.instance.message;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
@@ -24,11 +26,14 @@ import
org.apache.rocketmq.studio.persistence.entity.RmqMessageQuery;
import org.apache.rocketmq.studio.persistence.entity.RmqTraceQuery;
import org.apache.rocketmq.studio.persistence.mapper.RmqMessageQueryMapper;
import org.apache.rocketmq.studio.persistence.mapper.RmqTraceQueryMapper;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
import java.time.Clock;
import java.time.LocalDateTime;
+import java.util.List;
@Slf4j
@Service
@@ -88,6 +93,65 @@ public class QueryHistoryService {
log.debug("Trace query recorded: clusterId={} msgId={} topic={}",
clusterId, msgId, topic);
}
+ public PageResult<MessageQueryHistoryVO> listMessageQueries(String
clusterId, String queryType,
+ String
search, int page, int pageSize) {
+ String pattern = escapeLike(search);
+ QueryWrapper<RmqMessageQuery> query = new
QueryWrapper<RmqMessageQuery>()
+ .eq(StringUtils.hasText(clusterId), "cluster_id", clusterId)
+ .eq(StringUtils.hasText(queryType), "query_type", queryType)
+ .and(StringUtils.hasText(search), nested -> nested
+ .like("topic", pattern)
+ .or().like("msg_id", pattern)
+ .or().like("message_key", pattern)
+ .or().like("queried_by", pattern))
+ .orderByDesc("queried_at", "id");
+ Page<RmqMessageQuery> result = messageQueryMapper.selectPage(new
Page<>(page, pageSize), query);
+ List<MessageQueryHistoryVO> items = result.getRecords().stream()
+ .map(QueryHistoryService::toMessageHistory).toList();
+ return PageResult.of(items, result.getTotal(), page, pageSize);
+ }
+
+ public PageResult<TraceQueryHistoryVO> listTraceQueries(String clusterId,
String search,
+ int page, int
pageSize) {
+ String pattern = escapeLike(search);
+ QueryWrapper<RmqTraceQuery> query = new QueryWrapper<RmqTraceQuery>()
+ .eq(StringUtils.hasText(clusterId), "cluster_id", clusterId)
+ .and(StringUtils.hasText(search), nested -> nested
+ .like("topic", pattern)
+ .or().like("msg_id", pattern)
+ .or().like("queried_by", pattern))
+ .orderByDesc("queried_at", "id");
+ Page<RmqTraceQuery> result = traceQueryMapper.selectPage(new
Page<>(page, pageSize), query);
+ List<TraceQueryHistoryVO> items = result.getRecords().stream()
+ .map(QueryHistoryService::toTraceHistory).toList();
+ return PageResult.of(items, result.getTotal(), page, pageSize);
+ }
+
+ public QueryHistorySummaryVO summarize(String clusterId) {
+ QueryWrapper<RmqMessageQuery> messageFilter = new
QueryWrapper<RmqMessageQuery>()
+ .eq(StringUtils.hasText(clusterId), "cluster_id", clusterId);
+ QueryWrapper<RmqTraceQuery> traceFilter = new
QueryWrapper<RmqTraceQuery>()
+ .eq(StringUtils.hasText(clusterId), "cluster_id", clusterId);
+ long messageCount = messageQueryMapper.selectCount(messageFilter);
+ long traceCount = traceQueryMapper.selectCount(traceFilter);
+ RmqMessageQuery latestMessage = messageQueryMapper.selectOne(
+ new QueryWrapper<RmqMessageQuery>()
+ .eq(StringUtils.hasText(clusterId), "cluster_id",
clusterId)
+ .orderByDesc("queried_at", "id").last("LIMIT 1"));
+ RmqTraceQuery latestTrace = traceQueryMapper.selectOne(
+ new QueryWrapper<RmqTraceQuery>()
+ .eq(StringUtils.hasText(clusterId), "cluster_id",
clusterId)
+ .orderByDesc("queried_at", "id").last("LIMIT 1"));
+ LocalDateTime latest = latestOf(
+ latestMessage == null ? null : latestMessage.getQueriedAt(),
+ latestTrace == null ? null : latestTrace.getQueriedAt());
+ return QueryHistorySummaryVO.builder()
+ .messageQueries(messageCount)
+ .traceQueries(traceCount)
+ .latestQueryAt(latest)
+ .build();
+ }
+
@Scheduled(fixedDelayString =
"${studio.query-history.cleanup-interval:PT24H}")
public void purgeExpiredQueries() {
int retentionDays = properties.getRetentionDays();
@@ -119,4 +183,40 @@ public class QueryHistoryService {
log.warn("Failed to purge expired trace query records: {}",
e.getMessage());
}
}
+
+ private static LocalDateTime latestOf(LocalDateTime left, LocalDateTime
right) {
+ if (left == null) return right;
+ if (right == null) return left;
+ return left.isAfter(right) ? left : right;
+ }
+
+ private static MessageQueryHistoryVO toMessageHistory(RmqMessageQuery
query) {
+ return MessageQueryHistoryVO.builder()
+
.id(query.getId()).queryType(query.getQueryType()).topic(query.getTopic())
+
.msgId(query.getMsgId()).tag(query.getTag()).messageKey(query.getMessageKey())
+ .startTime(query.getStartTime()).endTime(query.getEndTime())
+ .resultCount(query.getResultCount() == null ? 0 :
query.getResultCount())
+
.clusterId(query.getClusterId()).queriedBy(query.getQueriedBy())
+ .queriedAt(query.getQueriedAt()).build();
+ }
+
+ private static TraceQueryHistoryVO toTraceHistory(RmqTraceQuery query) {
+ return TraceQueryHistoryVO.builder()
+
.id(query.getId()).msgId(query.getMsgId()).topic(query.getTopic())
+ .nodeCount(query.getNodeCount() == null ? 0 :
query.getNodeCount())
+ .consumerCount(query.getConsumerCount() == null ? 0 :
query.getConsumerCount())
+
.clusterId(query.getClusterId()).queriedBy(query.getQueriedBy())
+ .queriedAt(query.getQueriedAt()).build();
+ }
+
+ /**
+ * Escapes LIKE wildcards so user-supplied search terms match literally
instead of being
+ * interpreted as {@code %}/{@code _} patterns.
+ */
+ private static String escapeLike(String search) {
+ if (!StringUtils.hasText(search)) {
+ return search;
+ }
+ return search.replace("\\", "\\\\").replace("%", "\\%").replace("_",
"\\_");
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySummaryVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySummaryVO.java
new file mode 100644
index 00000000..5a27c3c0
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySummaryVO.java
@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+@Data
+@Builder
+public class QueryHistorySummaryVO {
+ private long messageQueries;
+ private long traceQueries;
+ private LocalDateTime latestQueryAt;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/TraceQueryHistoryVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/TraceQueryHistoryVO.java
new file mode 100644
index 00000000..68413f6e
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/TraceQueryHistoryVO.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+@Data
+@Builder
+public class TraceQueryHistoryVO {
+ private Long id;
+ private String msgId;
+ private String topic;
+ private int nodeCount;
+ private int consumerCount;
+ private String clusterId;
+ private String queriedBy;
+ private LocalDateTime queriedAt;
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryControllerTest.java
new file mode 100644
index 00000000..f9fb7ae3
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryControllerTest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(QueryHistoryController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class QueryHistoryControllerTest {
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private QueryHistoryService queryHistoryService;
+
+ @Test
+ void listsFilteredMessageHistory() throws Exception {
+ MessageQueryHistoryVO item = MessageQueryHistoryVO.builder()
+ .id(1L).queryType("TOPIC").topic("orders").resultCount(2)
+ .clusterId("instance-a").queriedBy("alice")
+ .queriedAt(LocalDateTime.of(2026, 8, 5, 12, 0)).build();
+ when(queryHistoryService.listMessageQueries("instance-a", "TOPIC",
"orders", 2, 10))
+ .thenReturn(PageResult.of(List.of(item), 11, 2, 10));
+
+ mockMvc.perform(get("/api/query-history/messages")
+ .param("clusterId", "instance-a")
+ .param("queryType", "TOPIC")
+ .param("search", "orders")
+ .param("page", "2")
+ .param("pageSize", "10"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.total").value(11))
+ .andExpect(jsonPath("$.data.items[0].topic").value("orders"));
+
+ verify(queryHistoryService).listMessageQueries("instance-a", "TOPIC",
"orders", 2, 10);
+ }
+
+ @Test
+ void rejectsOversizedHistoryPages() throws Exception {
+ mockMvc.perform(get("/api/query-history/traces").param("pageSize",
"101"))
+ .andExpect(status().isBadRequest())
+ .andExpect(jsonPath("$.message").value("pageSize must be
between 1 and 100"));
+ }
+
+ @Test
+ void returnsHistorySummary() throws Exception {
+ when(queryHistoryService.summarize("instance-a"))
+ .thenReturn(QueryHistorySummaryVO.builder()
+ .messageQueries(7).traceQueries(3)
+ .latestQueryAt(LocalDateTime.of(2026, 8, 5, 12,
0)).build());
+
+ mockMvc.perform(get("/api/query-history/summary").param("clusterId",
"instance-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.messageQueries").value(7))
+ .andExpect(jsonPath("$.data.traceQueries").value(3));
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
index b9108bc4..8d8e821e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
@@ -17,6 +17,8 @@
package org.apache.rocketmq.studio.instance.message;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
import org.apache.rocketmq.studio.persistence.entity.RmqMessageQuery;
import org.apache.rocketmq.studio.persistence.entity.RmqTraceQuery;
@@ -102,4 +104,51 @@ class QueryHistoryServiceTest {
verify(messageQueryMapper, never()).delete(any());
verify(traceQueryMapper, never()).delete(any());
}
+
+ @Test
+ void listsMessageHistoryAsNewestFirstPage() {
+ RmqMessageQuery entity = new RmqMessageQuery();
+ entity.setId(9L);
+ entity.setQueryType("KEY");
+ entity.setTopic("orders");
+ entity.setMessageKey("order-1");
+ entity.setResultCount(3);
+ entity.setClusterId("cluster-a");
+ entity.setQueriedBy("alice");
+ entity.setQueriedAt(LocalDateTime.of(2026, 8, 5, 12, 0));
+ when(messageQueryMapper.selectPage(any(Page.class),
any(Wrapper.class)))
+ .thenAnswer(invocation -> {
+ Page<RmqMessageQuery> result = invocation.getArgument(0);
+ result.setRecords(java.util.List.of(entity));
+ result.setTotal(1);
+ return result;
+ });
+
+ PageResult<MessageQueryHistoryVO> result = service.listMessageQueries(
+ "cluster-a", "KEY", "order", 1, 20);
+
+ assertThat(result.getTotal()).isEqualTo(1);
+ assertThat(result.getItems()).singleElement().satisfies(item -> {
+ assertThat(item.getMessageKey()).isEqualTo("order-1");
+ assertThat(item.getQueriedBy()).isEqualTo("alice");
+ });
+ }
+
+ @Test
+ void summarizesBothHistoryStreams() {
+ RmqMessageQuery message = new RmqMessageQuery();
+ message.setQueriedAt(LocalDateTime.of(2026, 8, 5, 10, 0));
+ RmqTraceQuery trace = new RmqTraceQuery();
+ trace.setQueriedAt(LocalDateTime.of(2026, 8, 5, 12, 0));
+ when(messageQueryMapper.selectCount(any())).thenReturn(7L);
+ when(traceQueryMapper.selectCount(any())).thenReturn(4L);
+ when(messageQueryMapper.selectOne(any())).thenReturn(message);
+ when(traceQueryMapper.selectOne(any())).thenReturn(trace);
+
+ QueryHistorySummaryVO summary = service.summarize("cluster-a");
+
+ assertThat(summary.getMessageQueries()).isEqualTo(7);
+ assertThat(summary.getTraceQueries()).isEqualTo(4);
+
assertThat(summary.getLatestQueryAt()).isEqualTo(LocalDateTime.of(2026, 8, 5,
12, 0));
+ }
}
diff --git a/web/src/api/messageHistory.test.ts
b/web/src/api/messageHistory.test.ts
new file mode 100644
index 00000000..540c7b0c
--- /dev/null
+++ b/web/src/api/messageHistory.test.ts
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+import MockAdapter from 'axios-mock-adapter';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import client from './client';
+import {
+ getQueryHistorySummary,
+ listMessageQueryHistory,
+ listTraceQueryHistory,
+} from './messageHistory';
+
+const mock = new MockAdapter(client);
+
+describe('message query history API', () => {
+ beforeEach(() => mock.reset());
+ afterEach(() => mock.reset());
+
+ it('forwards filters and pagination for message history', async () => {
+ mock.onGet('/query-history/messages').reply((config) => {
+ expect(config.params).toEqual({
+ clusterId: 'instance-a',
+ search: 'orders',
+ page: 2,
+ pageSize: 20,
+ });
+ return [200, { code: 200, data: { items: [], total: 0, page: 2, size: 20
} }];
+ });
+
+ const result = await listMessageQueryHistory({
+ clusterId: 'instance-a',
+ search: 'orders',
+ page: 2,
+ pageSize: 20,
+ });
+ expect(result.page).toBe(2);
+ });
+
+ it('loads trace history and summary for an instance', async () => {
+ mock.onGet('/query-history/traces').reply(200, {
+ code: 200,
+ data: { items: [{ id: 1, msgId: 'msg-1' }], total: 1, page: 1, size: 20
},
+ });
+ mock.onGet('/query-history/summary').reply((config) => {
+ expect(config.params).toEqual({ clusterId: 'instance-a' });
+ return [200, { code: 200, data: { messageQueries: 3, traceQueries: 1 }
}];
+ });
+
+ await expect(listTraceQueryHistory({ clusterId: 'instance-a'
})).resolves.toMatchObject({
+ total: 1,
+ });
+ await
expect(getQueryHistorySummary('instance-a')).resolves.toMatchObject({
traceQueries: 1 });
+ });
+});
diff --git a/web/src/api/messageHistory.ts b/web/src/api/messageHistory.ts
new file mode 100644
index 00000000..a48abe5e
--- /dev/null
+++ b/web/src/api/messageHistory.ts
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+import client from './client';
+
+export interface HistoryPage<T> {
+ items: T[];
+ total: number;
+ page: number;
+ size: number;
+}
+
+export interface MessageQueryHistory {
+ id: number;
+ queryType: 'TOPIC' | 'KEY' | 'MSG_ID';
+ topic?: string;
+ msgId?: string;
+ tag?: string;
+ messageKey?: string;
+ startTime?: number;
+ endTime?: number;
+ resultCount: number;
+ clusterId?: string;
+ queriedBy?: string;
+ queriedAt: string;
+}
+
+export interface TraceQueryHistory {
+ id: number;
+ msgId: string;
+ topic?: string;
+ nodeCount: number;
+ consumerCount: number;
+ clusterId?: string;
+ queriedBy?: string;
+ queriedAt: string;
+}
+
+export interface QueryHistorySummary {
+ messageQueries: number;
+ traceQueries: number;
+ latestQueryAt?: string;
+}
+
+export async function listMessageQueryHistory(params: {
+ clusterId?: string;
+ queryType?: string;
+ search?: string;
+ page?: number;
+ pageSize?: number;
+}) {
+ const response = await client.get<{ data: HistoryPage<MessageQueryHistory>
}>(
+ '/query-history/messages',
+ { params },
+ );
+ return response.data.data;
+}
+
+export async function listTraceQueryHistory(params: {
+ clusterId?: string;
+ search?: string;
+ page?: number;
+ pageSize?: number;
+}) {
+ const response = await client.get<{ data: HistoryPage<TraceQueryHistory> }>(
+ '/query-history/traces',
+ { params },
+ );
+ return response.data.data;
+}
+
+export async function getQueryHistorySummary(clusterId?: string) {
+ const response = await client.get<{ data: QueryHistorySummary
}>('/query-history/summary', {
+ params: clusterId ? { clusterId } : undefined,
+ });
+ return response.data.data;
+}
diff --git a/web/src/components/MessageQueryHistoryDrawer.tsx
b/web/src/components/MessageQueryHistoryDrawer.tsx
new file mode 100644
index 00000000..ca5e2f5b
--- /dev/null
+++ b/web/src/components/MessageQueryHistoryDrawer.tsx
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+import { useCallback, useEffect, useRef, useState } from 'react';
+import { Alert, Drawer, Flex, Input, Statistic, Table, Tabs, Tag } from 'antd';
+import type { ColumnsType } from 'antd/es/table';
+import {
+ getQueryHistorySummary,
+ listMessageQueryHistory,
+ listTraceQueryHistory,
+ type MessageQueryHistory,
+ type QueryHistorySummary,
+ type TraceQueryHistory,
+} from '../api/messageHistory';
+
+interface Props {
+ open: boolean;
+ clusterId?: string;
+ onClose: () => void;
+}
+
+const PAGE_SIZE = 20;
+const formatTime = (value?: string) => (value ? new
Date(value).toLocaleString() : '-');
+
+const MessageQueryHistoryDrawer = ({ open, clusterId, onClose }: Props) => {
+ const [tab, setTab] = useState<'messages' | 'traces'>('messages');
+ const [search, setSearch] = useState('');
+ const [page, setPage] = useState(1);
+ const [loading, setLoading] = useState(false);
+ const [error, setError] = useState('');
+ const [summary, setSummary] = useState<QueryHistorySummary>();
+ const [messageRows, setMessageRows] = useState<MessageQueryHistory[]>([]);
+ const [traceRows, setTraceRows] = useState<TraceQueryHistory[]>([]);
+ const [total, setTotal] = useState(0);
+ const requestId = useRef(0);
+
+ const load = useCallback(async () => {
+ if (!open) return;
+ const id = ++requestId.current;
+ setLoading(true);
+ setError('');
+ try {
+ const [nextSummary, result] = await Promise.all([
+ getQueryHistorySummary(clusterId),
+ tab === 'messages'
+ ? listMessageQueryHistory({
+ clusterId,
+ search: search || undefined,
+ page,
+ pageSize: PAGE_SIZE,
+ })
+ : listTraceQueryHistory({
+ clusterId,
+ search: search || undefined,
+ page,
+ pageSize: PAGE_SIZE,
+ }),
+ ]);
+ if (id !== requestId.current) return;
+ setSummary(nextSummary);
+ setTotal(result.total);
+ if (tab === 'messages') setMessageRows(result.items as
MessageQueryHistory[]);
+ else setTraceRows(result.items as TraceQueryHistory[]);
+ } catch (loadError) {
+ if (id === requestId.current) {
+ setError(loadError instanceof Error ? loadError.message : '查询历史加载失败');
+ }
+ } finally {
+ if (id === requestId.current) setLoading(false);
+ }
+ }, [clusterId, open, page, search, tab]);
+
+ useEffect(() => {
+ // Loading is asynchronous; state updates happen after the history API
resolves.
+ // eslint-disable-next-line react-hooks/set-state-in-effect
+ void load();
+ return () => {
+ requestId.current += 1;
+ };
+ }, [load]);
+
+ const messageColumns: ColumnsType<MessageQueryHistory> = [
+ { title: '类型', dataIndex: 'queryType', width: 90, render: (value) =>
<Tag>{value}</Tag> },
+ { title: 'Topic', dataIndex: 'topic', ellipsis: true },
+ { title: 'Message ID / Key', render: (_, row) => row.msgId ||
row.messageKey || '-' },
+ { title: '结果数', dataIndex: 'resultCount', width: 80 },
+ { title: '操作者', dataIndex: 'queriedBy', width: 110 },
+ { title: '查询时间', dataIndex: 'queriedAt', width: 180, render: formatTime },
+ ];
+ const traceColumns: ColumnsType<TraceQueryHistory> = [
+ { title: 'Message ID', dataIndex: 'msgId', ellipsis: true },
+ { title: 'Topic', dataIndex: 'topic', ellipsis: true },
+ { title: '轨迹节点', dataIndex: 'nodeCount', width: 90 },
+ { title: '消费者', dataIndex: 'consumerCount', width: 90 },
+ { title: '操作者', dataIndex: 'queriedBy', width: 110 },
+ { title: '查询时间', dataIndex: 'queriedAt', width: 180, render: formatTime },
+ ];
+
+ return (
+ <Drawer title="服务端查询历史" width={900} open={open} onClose={onClose}
destroyOnHidden>
+ <Flex gap={32} style={{ marginBottom: 16 }}>
+ <Statistic title="消息查询" value={summary?.messageQueries ?? 0} />
+ <Statistic title="轨迹查询" value={summary?.traceQueries ?? 0} />
+ <Statistic title="最近查询" value={formatTime(summary?.latestQueryAt)} />
+ </Flex>
+ <Input.Search
+ allowClear
+ placeholder="搜索 Topic、Message ID、Key 或操作者"
+ onSearch={(value) => {
+ setPage(1);
+ setSearch(value.trim());
+ }}
+ style={{ marginBottom: 12, width: 420 }}
+ />
+ {error && <Alert type="error" showIcon message={error} style={{
marginBottom: 12 }} />}
+ <Tabs
+ activeKey={tab}
+ onChange={(key) => {
+ setPage(1);
+ setTab(key as 'messages' | 'traces');
+ }}
+ items={[
+ {
+ key: 'messages',
+ label: '消息查询',
+ children: (
+ <Table
+ rowKey="id"
+ loading={loading}
+ columns={messageColumns}
+ dataSource={messageRows}
+ pagination={{ current: page, pageSize: PAGE_SIZE, total,
onChange: setPage }}
+ />
+ ),
+ },
+ {
+ key: 'traces',
+ label: '轨迹查询',
+ children: (
+ <Table
+ rowKey="id"
+ loading={loading}
+ columns={traceColumns}
+ dataSource={traceRows}
+ pagination={{ current: page, pageSize: PAGE_SIZE, total,
onChange: setPage }}
+ />
+ ),
+ },
+ ]}
+ />
+ </Drawer>
+ );
+};
+
+export default MessageQueryHistoryDrawer;
diff --git a/web/src/components/__tests__/MessageQueryHistoryDrawer.test.tsx
b/web/src/components/__tests__/MessageQueryHistoryDrawer.test.tsx
new file mode 100644
index 00000000..5d6cfa0a
--- /dev/null
+++ b/web/src/components/__tests__/MessageQueryHistoryDrawer.test.tsx
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { render, screen, waitFor } from '@testing-library/react';
+import userEvent from '@testing-library/user-event';
+import { App } from 'antd';
+import MessageQueryHistoryDrawer from '../MessageQueryHistoryDrawer';
+import {
+ getQueryHistorySummary,
+ listMessageQueryHistory,
+ listTraceQueryHistory,
+} from '../../api/messageHistory';
+
+vi.mock('../../api/messageHistory', () => ({
+ getQueryHistorySummary: vi.fn(),
+ listMessageQueryHistory: vi.fn(),
+ listTraceQueryHistory: vi.fn(),
+}));
+
+beforeAll(() => {
+ Object.defineProperty(window, 'matchMedia', {
+ value: vi.fn().mockImplementation(() => ({
+ matches: false,
+ addListener: vi.fn(),
+ removeListener: vi.fn(),
+ addEventListener: vi.fn(),
+ removeEventListener: vi.fn(),
+ dispatchEvent: vi.fn(),
+ })),
+ });
+});
+
+describe('MessageQueryHistoryDrawer', () => {
+ beforeEach(() => {
+ vi.clearAllMocks();
+ vi.mocked(getQueryHistorySummary).mockResolvedValue({ messageQueries: 4,
traceQueries: 2 });
+ vi.mocked(listMessageQueryHistory).mockResolvedValue({
+ items: [
+ {
+ id: 1,
+ queryType: 'KEY',
+ topic: 'orders',
+ messageKey: 'order-1',
+ resultCount: 2,
+ queriedBy: 'alice',
+ queriedAt: '2026-08-05T12:00:00Z',
+ },
+ ],
+ total: 1,
+ page: 1,
+ size: 20,
+ });
+ vi.mocked(listTraceQueryHistory).mockResolvedValue({
+ items: [
+ {
+ id: 2,
+ msgId: 'msg-1',
+ topic: 'orders',
+ nodeCount: 3,
+ consumerCount: 1,
+ queriedBy: 'bob',
+ queriedAt: '2026-08-05T12:00:00Z',
+ },
+ ],
+ total: 1,
+ page: 1,
+ size: 20,
+ });
+ });
+
+ it('loads persisted message and trace history by instance', async () => {
+ const user = userEvent.setup();
+ render(
+ <App>
+ <MessageQueryHistoryDrawer open clusterId="instance-a"
onClose={vi.fn()} />
+ </App>,
+ );
+
+ expect(await screen.findByText('order-1')).toBeInTheDocument();
+ expect(listMessageQueryHistory).toHaveBeenCalledWith(
+ expect.objectContaining({ clusterId: 'instance-a' }),
+ );
+ await user.click(screen.getByRole('tab', { name: '轨迹查询' }));
+ expect(await screen.findByText('msg-1')).toBeInTheDocument();
+ await waitFor(() => expect(listTraceQueryHistory).toHaveBeenCalled());
+ });
+});
diff --git a/web/src/pages/instance/message.tsx
b/web/src/pages/instance/message.tsx
index 857f7ce2..02212af8 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -53,6 +53,7 @@ import dayjs from 'dayjs';
import type { Dayjs } from 'dayjs';
import PageHeader from '../../components/PageHeader';
import { InstanceSelect } from '../../components/InstanceSelect';
+import MessageQueryHistoryDrawer from
'../../components/MessageQueryHistoryDrawer';
import { useLang } from '../../i18n/LangContext';
import type { MessageQuery, MessageRecord, TraceRecord } from
'../../api/message';
import { getMessageTrace, queryMessages } from '../../services/messageService';
@@ -307,6 +308,7 @@ const MessagePageContent = ({
const [queryError, setQueryError] = useState<string | null>(null);
const [traceError, setTraceError] = useState<string | null>(null);
const [recentQueries, setRecentQueries] =
useState<RecentQuery[]>(loadRecentQueries);
+ const [historyDrawerOpen, setHistoryDrawerOpen] = useState(false);
const queryGenerationRef = useRef(0);
const traceGenerationRef = useRef(0);
@@ -869,6 +871,9 @@ const MessagePageContent = ({
<Button icon={<ReloadOutlined />} onClick={handleReset}>
重置
</Button>
+ <Button icon={<HistoryOutlined />} onClick={() =>
setHistoryDrawerOpen(true)}>
+ 服务端历史
+ </Button>
</Space>
</Space>
</Card>
@@ -876,6 +881,11 @@ const MessagePageContent = ({
{topicError && (
<Alert showIcon type="error" message={topicError} style={{
marginBottom: 16 }} />
)}
+ <MessageQueryHistoryDrawer
+ open={historyDrawerOpen}
+ clusterId={selectedInstanceId || undefined}
+ onClose={() => setHistoryDrawerOpen(false)}
+ />
{queryError && (
<Alert showIcon type="warning" message={queryError} style={{
marginBottom: 16 }} />
