From 2fe26052a73a2c1e8a70ffab13689b5e38e885d2 Mon Sep 17 00:00:00 2001
From: Guoyu Su <yss2813483011xxl@gmail.com>
Date: Mon, 31 Aug 2026 23:47:51 +0800
Subject: [PATCH] feat(kafka): add message display ordering

---
 .../src/components/mq/MessageBrowser.vue      | 17 +++++-
 .../mq/__tests__/MessageBrowser.spec.ts       | 53 ++++++++++++++++++-
 apps/desktop/src/i18n/locales/en.ts           |  9 ++--
 apps/desktop/src/i18n/locales/es.ts           |  9 ++--
 apps/desktop/src/i18n/locales/it.ts           |  9 ++--
 apps/desktop/src/i18n/locales/ja.ts           |  9 ++--
 apps/desktop/src/i18n/locales/ko.ts           |  9 ++--
 apps/desktop/src/i18n/locales/pt-BR.ts        |  9 ++--
 apps/desktop/src/i18n/locales/zh-CN.ts        |  9 ++--
 apps/desktop/src/i18n/locales/zh-TW.ts        |  9 ++--
 .../lib/mq/__tests__/kafkaMessageSort.spec.ts | 32 +++++++++++
 apps/desktop/src/lib/mq/kafkaMessageSort.ts   | 23 ++++++++
 12 files changed, 171 insertions(+), 26 deletions(-)
 create mode 100644 apps/desktop/src/lib/mq/__tests__/kafkaMessageSort.spec.ts
 create mode 100644 apps/desktop/src/lib/mq/kafkaMessageSort.ts

diff --git a/apps/desktop/src/components/mq/MessageBrowser.vue b/apps/desktop/src/components/mq/MessageBrowser.vue
index a0a62a2cbe..447674b16e 100644
--- a/apps/desktop/src/components/mq/MessageBrowser.vue
+++ b/apps/desktop/src/components/mq/MessageBrowser.vue
@@ -7,6 +7,7 @@ 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 { sortKafkaMessagesByPublishTime, type KafkaMessageDisplayOrder } from "@/lib/mq/kafkaMessageSort";
 import { parseNonNegativeSafeInteger } from "@/lib/mq/mqPeekFilters";
 import { useToast } from "@/composables/useToast";
 import { Button } from "@/components/ui/button";
@@ -40,6 +41,7 @@ const advancedExpanded = ref(false);
 const messageSearchQuery = ref("");
 type KafkaPeekStartPosition = NonNullable<PeekMessagesOptions["startPosition"]>;
 const kafkaStartPosition = ref<KafkaPeekStartPosition>("latest");
+const kafkaMessageDisplayOrder = ref<KafkaMessageDisplayOrder>("newest");
 let messageRequestVersion = 0;
 
 const isKafka = computed(() => props.mqSystemKind === "kafka");
@@ -57,6 +59,7 @@ const filteredMessages = computed(() => {
   if (!query) return messages.value;
   return searchableMessages.value.filter(({ searchText }) => kafkaMessageSearchTextMatches(searchText, query)).map(({ message }) => message);
 });
+const displayedMessages = computed(() => (isKafka.value ? sortKafkaMessagesByPublishTime(filteredMessages.value, kafkaMessageDisplayOrder.value) : filteredMessages.value));
 
 function peekGroupName(): string {
   if (props.mqSystemKind === "rocketmq") return "__dbx_rocketmq_viewer__";
@@ -223,6 +226,18 @@ watch(kafkaStartPosition, () => {
           </SelectContent>
         </Select>
       </label>
+      <label v-if="isKafka">
+        <span>{{ t("mqMessages.timeOrder") }}</span>
+        <Select v-model="kafkaMessageDisplayOrder">
+          <SelectTrigger data-testid="kafka-message-display-order" class="message-browser-start-position">
+            <SelectValue />
+          </SelectTrigger>
+          <SelectContent position="popper" class="message-browser-start-position-content">
+            <SelectItem value="newest">{{ t("mqMessages.newestFirst") }}</SelectItem>
+            <SelectItem value="oldest">{{ t("mqMessages.oldestFirst") }}</SelectItem>
+          </SelectContent>
+        </Select>
+      </label>
       <label v-if="isKafka">
         <span>{{ t("mqMessages.partition") }}</span>
         <input v-model="partition" data-testid="kafka-peek-partition" type="number" min="0" :placeholder="t('mqMessages.partitionPlaceholderAll')" :disabled="loading" />
@@ -263,7 +278,7 @@ watch(kafkaStartPosition, () => {
     <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 filteredMessages" :key="`${message.properties?.partition ?? 'p'}-${message.messageId || message.position}`" class="message-row">
+      <article v-for="message in displayedMessages" :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>
diff --git a/apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts b/apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts
index 7082074bbe..13ad96690c 100644
--- a/apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts
+++ b/apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts
@@ -74,13 +74,14 @@ async function setSelectValue(trigger: HTMLElement, value: string) {
   await flushUi();
 }
 
-async function mountBrowser(mqSystemKind: "kafka" | "rabbitmq" = "kafka") {
+async function mountBrowser(mqSystemKind: "kafka" | "rabbitmq" = "kafka", appearance: "form" | "monitoring" = "form") {
   root = document.createElement("div");
   document.body.appendChild(root);
   app = createApp(MessageBrowser, {
     connectionId: "mq-1",
     topic: { ...TOPIC, tenant: mqSystemKind === "rabbitmq" ? "_rabbitmq" : "_kafka" },
     mqSystemKind,
+    appearance,
   });
   app.mount(root);
   await flushUi();
@@ -121,6 +122,10 @@ async function loadMessages(container: ParentNode) {
   await flushUi();
 }
 
+function displayedPayloads(container: ParentNode): string[] {
+  return [...container.querySelectorAll<HTMLElement>(".message-payload")].map((element) => element.textContent || "");
+}
+
 beforeEach(() => {
   backend.mqPeekMessages.mockReset();
   clipboard.copyToClipboard.mockReset();
@@ -229,6 +234,51 @@ describe("MessageBrowser", () => {
     expect(backend.mqPeekMessages).toHaveBeenCalledWith("mq-1", expect.objectContaining({ topic: "events" }), "__dbx_kafka_viewer__", 20, { startPosition: "latest" });
   });
 
+  it("orders loaded Kafka messages newest first by default and changes presentation without reloading", async () => {
+    backend.mqPeekMessages.mockResolvedValueOnce([
+      { position: 1, messageId: "1", publishTime: "1000", payloadBase64: "", payloadText: "oldest", properties: {}, headers: {} },
+      { position: 2, messageId: "2", publishTime: "3000", payloadBase64: "", payloadText: "newest", properties: {}, headers: {} },
+      { position: 3, messageId: "3", publishTime: "2000", payloadBase64: "", payloadText: "middle", properties: {}, headers: {} },
+    ]);
+    const browser = await mountBrowser();
+
+    await loadMessages(browser);
+    expect(displayedPayloads(browser)).toEqual(["newest", "middle", "oldest"]);
+
+    const displayOrder = browser.querySelector<HTMLElement>('[data-testid="kafka-message-display-order"]');
+    if (!displayOrder) throw new Error("Kafka message display order select not found");
+    await setSelectValue(displayOrder, "oldest");
+
+    expect(displayedPayloads(browser)).toEqual(["oldest", "middle", "newest"]);
+    expect(backend.mqPeekMessages).toHaveBeenCalledTimes(1);
+  });
+
+  it("keeps the selected Kafka display order when filtering loaded messages", async () => {
+    backend.mqPeekMessages.mockResolvedValueOnce([
+      { position: 1, messageId: "1", publishTime: "1000", payloadBase64: "", payloadText: "matching oldest", properties: {}, headers: {} },
+      { position: 2, messageId: "2", publishTime: "3000", payloadBase64: "", payloadText: "matching newest", properties: {}, headers: {} },
+      { position: 3, messageId: "3", publishTime: "2000", payloadBase64: "", payloadText: "excluded", properties: {}, headers: {} },
+    ]);
+    const browser = await mountBrowser();
+
+    await loadMessages(browser);
+    const displayOrder = browser.querySelector<HTMLElement>('[data-testid="kafka-message-display-order"]');
+    const filter = browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]');
+    if (!displayOrder || !filter) throw new Error("Kafka message presentation controls not found");
+    await setSelectValue(displayOrder, "oldest");
+    await setInputValue(filter, "matching");
+
+    expect(displayedPayloads(browser)).toEqual(["matching oldest", "matching newest"]);
+    expect(backend.mqPeekMessages).toHaveBeenCalledTimes(1);
+  });
+
+  it("shows Kafka display ordering in the monitoring appearance", async () => {
+    const browser = await mountBrowser("kafka", "monitoring");
+
+    expect(browser.querySelector('[data-testid="kafka-message-display-order"]')).not.toBeNull();
+    expect(browser.querySelector('[data-testid="kafka-peek-start-position"]')).toBeNull();
+  });
+
   it("normalizes a decimal count before sending the request", async () => {
     const browser = await mountBrowser();
     const countInput = browser.querySelector<HTMLInputElement>('[data-testid="peek-count"]');
@@ -383,6 +433,7 @@ describe("MessageBrowser", () => {
     await loadMessages(browser);
 
     expect(browser.querySelector('[data-testid="kafka-message-filter"]')).toBeNull();
+    expect(browser.querySelector('[data-testid="kafka-message-display-order"]')).toBeNull();
   });
 
   it("sends explicit earliest and offset read positions", async () => {
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index 0a34da4901..3909d0af34 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -7847,9 +7847,12 @@ export default {
     kafkaLatest: "Recent N messages (default)",
     kafkaEarliest: "Earliest messages",
     kafkaOffset: "From offset",
-    kafkaLatestHint: "Reads up to N candidates from each partition, then returns the newest N by timestamp across the topic.",
-    kafkaEarliestHint: "Reads history forward from each selected partition.",
-    kafkaOffsetHint: "Reads forward from the specified offset in the chosen partition, or every partition when left empty.",
+    timeOrder: "Time order",
+    newestFirst: "Newest first",
+    oldestFirst: "Oldest first",
+    kafkaLatestHint: "Reads up to N candidates from each partition, then selects the newest N by timestamp across the topic.",
+    kafkaEarliestHint: "Reads from the earliest available message in each selected partition.",
+    kafkaOffsetHint: "Reads from the specified offset in the chosen partition, or every partition when left empty.",
     messagesLoading: "Loading messages...",
     noMessages: "No messages",
     filterLoadedPlaceholder: "Filter loaded messages",
diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts
index 0de808e33e..93aa8009e9 100644
--- a/apps/desktop/src/i18n/locales/es.ts
+++ b/apps/desktop/src/i18n/locales/es.ts
@@ -7748,9 +7748,12 @@ export default withEnglishFallback({
     kafkaLatest: "Últimos N mensajes (predeterminado)",
     kafkaEarliest: "Mensajes más antiguos",
     kafkaOffset: "Desde offset",
-    kafkaLatestHint: "Lee hasta N candidatos por partición y devuelve los N más recientes del tema por marca de tiempo.",
-    kafkaEarliestHint: "Lee el historial hacia adelante desde cada partición seleccionada.",
-    kafkaOffsetHint: "Lee hacia adelante desde el offset indicado en la partición elegida, o en todas si se deja vacía.",
+    timeOrder: "Orden temporal",
+    newestFirst: "Más recientes primero",
+    oldestFirst: "Más antiguos primero",
+    kafkaLatestHint: "Lee hasta N candidatos por partición y selecciona los N más recientes del tema por marca de tiempo.",
+    kafkaEarliestHint: "Lee desde el primer mensaje disponible de cada partición seleccionada.",
+    kafkaOffsetHint: "Lee 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",
diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts
index 4c660d569a..4e4f9731d4 100644
--- a/apps/desktop/src/i18n/locales/it.ts
+++ b/apps/desktop/src/i18n/locales/it.ts
@@ -7749,9 +7749,12 @@ export default withEnglishFallback({
     kafkaLatest: "Ultimi N messaggi (predefinito)",
     kafkaEarliest: "Primi messaggi",
     kafkaOffset: "Da offset",
-    kafkaLatestHint: "Legge fino a N candidati per partizione e restituisce i N messaggi più recenti del topic in base al timestamp.",
-    kafkaEarliestHint: "Legge la cronologia in avanti da ogni partizione selezionata.",
-    kafkaOffsetHint: "Legge in avanti dall'offset indicato nella partizione scelta, o in tutte se lasciata vuota.",
+    timeOrder: "Ordine temporale",
+    newestFirst: "Più recenti prima",
+    oldestFirst: "Meno recenti prima",
+    kafkaLatestHint: "Legge fino a N candidati per partizione e seleziona i N messaggi più recenti del topic in base al timestamp.",
+    kafkaEarliestHint: "Legge dal primo messaggio disponibile di ogni partizione selezionata.",
+    kafkaOffsetHint: "Legge dall'offset indicato nella partizione scelta, o in tutte se lasciata vuota.",
     messagesLoading: "Caricamento messaggi...",
     noMessages: "Nessun messaggio",
     filterLoadedPlaceholder: "Filtra i messaggi caricati",
diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts
index 4503593cdd..f4a2d5945b 100644
--- a/apps/desktop/src/i18n/locales/ja.ts
+++ b/apps/desktop/src/i18n/locales/ja.ts
@@ -7800,9 +7800,12 @@ export default withEnglishFallback({
     kafkaLatest: "最新 N 件（デフォルト）",
     kafkaEarliest: "最も古いメッセージ",
     kafkaOffset: "Offset から読み取る",
-    kafkaLatestHint: "各パーティションから最大 N 件の候補を読み取り、トピック全体でタイムスタンプが新しい N 件を返します。",
-    kafkaEarliestHint: "選択した各パーティションの先頭から昇順に読み取ります。",
-    kafkaOffsetHint: "指定 Offset から、選択したパーティション（空欄ならすべて）を昇順に読み取ります。",
+    timeOrder: "時刻順",
+    newestFirst: "新しい順",
+    oldestFirst: "古い順",
+    kafkaLatestHint: "各パーティションから最大 N 件の候補を読み取り、トピック全体でタイムスタンプが新しい N 件を選択します。",
+    kafkaEarliestHint: "選択した各パーティションで最も古い利用可能なメッセージから読み取ります。",
+    kafkaOffsetHint: "指定 Offset から、選択したパーティション（空欄ならすべて）を読み取ります。",
     messagesLoading: "メッセージを読み込み中...",
     noMessages: "メッセージがありません",
     filterLoadedPlaceholder: "読み込み済みメッセージを絞り込む",
diff --git a/apps/desktop/src/i18n/locales/ko.ts b/apps/desktop/src/i18n/locales/ko.ts
index 8682a328e1..54e403f7f1 100644
--- a/apps/desktop/src/i18n/locales/ko.ts
+++ b/apps/desktop/src/i18n/locales/ko.ts
@@ -7204,9 +7204,12 @@ export default withEnglishFallback({
     kafkaLatest: "최근 N개 메시지(기본값)",
     kafkaEarliest: "가장 이른 메시지",
     kafkaOffset: "오프셋부터 읽기",
-    kafkaLatestHint: "파티션마다 최대 N개의 후보를 읽은 뒤 토픽 전체에서 타임스탬프가 가장 최신인 N개를 반환합니다.",
-    kafkaEarliestHint: "선택한 각 파티션의 처음부터 오름차순으로 읽습니다.",
-    kafkaOffsetHint: "지정한 오프셋부터 선택한 파티션(비워 두면 전체)을 오름차순으로 읽습니다.",
+    timeOrder: "시간 순서",
+    newestFirst: "최신순",
+    oldestFirst: "오래된순",
+    kafkaLatestHint: "파티션마다 최대 N개의 후보를 읽은 뒤 토픽 전체에서 타임스탬프가 가장 최신인 N개를 선택합니다.",
+    kafkaEarliestHint: "선택한 각 파티션에서 가장 이른 메시지부터 읽습니다.",
+    kafkaOffsetHint: "지정한 오프셋부터 선택한 파티션(비워 두면 전체)을 읽습니다.",
     messagesLoading: "메시지를 불러오는 중...",
     noMessages: "메시지가 없습니다",
     filterLoadedPlaceholder: "불러온 메시지 필터링",
diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts
index ac1b78cf38..8fb0065317 100644
--- a/apps/desktop/src/i18n/locales/pt-BR.ts
+++ b/apps/desktop/src/i18n/locales/pt-BR.ts
@@ -7750,9 +7750,12 @@ export default withEnglishFallback({
     kafkaLatest: "Últimas N mensagens (padrão)",
     kafkaEarliest: "Mensagens mais antigas",
     kafkaOffset: "A partir do offset",
-    kafkaLatestHint: "Lê até N candidatos por partição e retorna as N mensagens mais recentes do tópico por data e hora.",
-    kafkaEarliestHint: "Lê o histórico em ordem crescente de cada partição selecionada.",
-    kafkaOffsetHint: "Lê em ordem crescente a partir do offset informado na partição escolhida, ou em todas se deixada vazia.",
+    timeOrder: "Ordem temporal",
+    newestFirst: "Mais recentes primeiro",
+    oldestFirst: "Mais antigas primeiro",
+    kafkaLatestHint: "Lê até N candidatos por partição e seleciona as N mensagens mais recentes do tópico por data e hora.",
+    kafkaEarliestHint: "Lê a partir da primeira mensagem disponível de cada partição selecionada.",
+    kafkaOffsetHint: "Lê 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",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index 3c25b7885b..f85e6617a0 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -7829,9 +7829,12 @@ export default withEnglishFallback({
     kafkaLatest: "最近 N 条（默认）",
     kafkaEarliest: "最早消息",
     kafkaOffset: "从指定 Offset 读取",
-    kafkaLatestHint: "每个分区最多读取 N 条候选消息，再按时间全局倒序返回最新 N 条。",
-    kafkaEarliestHint: "从每个选定分区的最早可读消息开始正序读取。",
-    kafkaOffsetHint: "从指定 Offset 开始正序读取；填写分区时仅读该分区，留空则对全部分区分别读取。",
+    timeOrder: "时间顺序",
+    newestFirst: "最新优先",
+    oldestFirst: "最早优先",
+    kafkaLatestHint: "每个分区最多读取 N 条候选消息，再按时间选出全局最新 N 条。",
+    kafkaEarliestHint: "从每个选定分区的最早可读消息开始读取。",
+    kafkaOffsetHint: "从指定 Offset 开始读取；填写分区时仅读该分区，留空则对全部分区分别读取。",
     messagesLoading: "消息加载中...",
     noMessages: "暂无消息",
     filterLoadedPlaceholder: "筛选已加载消息",
diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts
index 4d0906b27f..05bbfc6cdd 100644
--- a/apps/desktop/src/i18n/locales/zh-TW.ts
+++ b/apps/desktop/src/i18n/locales/zh-TW.ts
@@ -7738,9 +7738,12 @@ export default withEnglishFallback({
     kafkaLatest: "最近 N 則（預設）",
     kafkaEarliest: "最早訊息",
     kafkaOffset: "從指定 Offset 讀取",
-    kafkaLatestHint: "每個分區最多讀取 N 筆候選訊息，再依時間全域倒序回傳最新 N 筆。",
-    kafkaEarliestHint: "從每個選定分區最早可讀的訊息開始正序讀取。",
-    kafkaOffsetHint: "從指定 Offset 開始正序讀取；填寫分區時只讀該分區，留空則對全部分區分別讀取。",
+    timeOrder: "時間順序",
+    newestFirst: "最新優先",
+    oldestFirst: "最早優先",
+    kafkaLatestHint: "每個分區最多讀取 N 筆候選訊息，再依時間選出全域最新 N 筆。",
+    kafkaEarliestHint: "從每個選定分區最早可讀的訊息開始讀取。",
+    kafkaOffsetHint: "從指定 Offset 開始讀取；填寫分區時只讀該分區，留空則對全部分區分別讀取。",
     messagesLoading: "訊息載入中...",
     noMessages: "暫無訊息",
     filterLoadedPlaceholder: "篩選已載入訊息",
diff --git a/apps/desktop/src/lib/mq/__tests__/kafkaMessageSort.spec.ts b/apps/desktop/src/lib/mq/__tests__/kafkaMessageSort.spec.ts
new file mode 100644
index 0000000000..d037d9c335
--- /dev/null
+++ b/apps/desktop/src/lib/mq/__tests__/kafkaMessageSort.spec.ts
@@ -0,0 +1,32 @@
+import { describe, expect, it } from "vitest";
+import type { PeekedMessage } from "@/types/mq";
+import { sortKafkaMessagesByPublishTime } from "@/lib/mq/kafkaMessageSort";
+
+function message(position: number, publishTime?: string): PeekedMessage {
+  return {
+    position,
+    messageId: String(position),
+    publishTime,
+    payloadBase64: "",
+    payloadText: `message ${position}`,
+    properties: {},
+    headers: {},
+  };
+}
+
+describe("Kafka message display sorting", () => {
+  it("sorts valid timestamps in either direction without mutating the input", () => {
+    const messages = [message(1, "1000"), message(2, "3000"), message(3, "2000")];
+
+    expect(sortKafkaMessagesByPublishTime(messages, "newest").map(({ position }) => position)).toEqual([2, 3, 1]);
+    expect(sortKafkaMessagesByPublishTime(messages, "oldest").map(({ position }) => position)).toEqual([1, 3, 2]);
+    expect(messages.map(({ position }) => position)).toEqual([1, 2, 3]);
+  });
+
+  it("keeps equal and unavailable timestamps stable after valid timestamps", () => {
+    const messages = [message(1), message(2, "2000"), message(3, "invalid"), message(4, "2000"), message(5, "")];
+
+    expect(sortKafkaMessagesByPublishTime(messages, "newest").map(({ position }) => position)).toEqual([2, 4, 1, 3, 5]);
+    expect(sortKafkaMessagesByPublishTime(messages, "oldest").map(({ position }) => position)).toEqual([2, 4, 1, 3, 5]);
+  });
+});
diff --git a/apps/desktop/src/lib/mq/kafkaMessageSort.ts b/apps/desktop/src/lib/mq/kafkaMessageSort.ts
new file mode 100644
index 0000000000..2cf0548773
--- /dev/null
+++ b/apps/desktop/src/lib/mq/kafkaMessageSort.ts
@@ -0,0 +1,23 @@
+import type { PeekedMessage } from "@/types/mq";
+
+export type KafkaMessageDisplayOrder = "newest" | "oldest";
+
+function parsePublishTime(value?: string): number | null {
+  if (value == null || value.trim() === "") return null;
+  const timestamp = Number(value);
+  return Number.isFinite(timestamp) ? timestamp : null;
+}
+
+export function sortKafkaMessagesByPublishTime(messages: readonly PeekedMessage[], order: KafkaMessageDisplayOrder): PeekedMessage[] {
+  return messages
+    .map((message, index) => ({ message, index, timestamp: parsePublishTime(message.publishTime) }))
+    .sort((left, right) => {
+      if (left.timestamp == null && right.timestamp != null) return 1;
+      if (left.timestamp != null && right.timestamp == null) return -1;
+      if (left.timestamp != null && right.timestamp != null && left.timestamp !== right.timestamp) {
+        return order === "newest" ? right.timestamp - left.timestamp : left.timestamp - right.timestamp;
+      }
+      return left.index - right.index;
+    })
+    .map(({ message }) => message);
+}