From 0eb8d05ed8eda6ddfcaaa0768fa336b8655eaf9f Mon Sep 17 00:00:00 2001
From: Jz0ojiang <hi@im0o.top>
Date: Tue, 11 Aug 2026 18:49:59 +0800
Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E8=AF=B7=E6=B1=82=E9=A1=B5?=
 =?UTF-8?q?=EF=BC=9A=E5=AE=9E=E6=97=B6=E6=98=BE=E7=A4=BA=E5=86=85=E6=A0=B8?=
 =?UTF-8?q?=E6=97=A5=E5=BF=97=E6=B5=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

流式接口不能走 api.ts 的 request()（它会把整个响应读完），页面自己
fetch 加 stream.ts 按行读。解析和追加上限放在 web/logs.ts 里，因为
「脏行跳过」「超上限丢最旧的」是点名要测的行为，埋在组件里测不到。

收到的行先进缓冲、每 200ms 批量刷进列表；暂停就是不刷但照收，
缓冲也设上限，恢复时补上。级别是建连时定的，切换即重连。
---
 web/logs.test.ts           | 106 ++++++++++++++
 web/logs.ts                |  87 ++++++++++++
 web/views/RequestsPage.vue | 283 +++++++++++++++++++++++++++++++++++++
 3 files changed, 476 insertions(+)
 create mode 100644 web/logs.test.ts
 create mode 100644 web/logs.ts
 create mode 100644 web/views/RequestsPage.vue

diff --git a/web/logs.test.ts b/web/logs.test.ts
new file mode 100644
index 0000000..0e72122
--- /dev/null
+++ b/web/logs.test.ts
@@ -0,0 +1,106 @@
+import { afterEach, describe, expect, it, mock } from 'bun:test'
+import { LOG_CAP, parseLogLine, pushCapped, streamLogs } from './logs.ts'
+import { LogStreamError } from './logs.ts'
+
+describe('解析日志行', () => {
+  it('形状对的行解出级别和内容', () => {
+    expect(parseLogLine('{"type":"info","payload":"[TCP] dial example.com"}')).toEqual({
+      type: 'info',
+      payload: '[TCP] dial example.com',
+    })
+  })
+
+  it('不是 JSON 的脏行返回 null 而不是抛出去', () => {
+    expect(parseLogLine('not json at all')).toBeNull()
+    expect(parseLogLine('{"type":"info","payl')).toBeNull()
+  })
+
+  it('是 JSON 但形状不对的也跳过', () => {
+    expect(parseLogLine('null')).toBeNull()
+    expect(parseLogLine('[1,2]')).toBeNull()
+    expect(parseLogLine('{"type":"info"}')).toBeNull()
+    expect(parseLogLine('{"type":"info","payload":42}')).toBeNull()
+  })
+})
+
+describe('成批追加上限', () => {
+  it('超出上限丢最旧的，留下最新的', () => {
+    const list = [1, 2, 3]
+    pushCapped(list, [4, 5, 6], 4)
+    expect(list).toEqual([3, 4, 5, 6])
+  })
+
+  it('一批比上限还大时只留下限条', () => {
+    const list: number[] = []
+    pushCapped(list, [1, 2, 3, 4, 5], 3)
+    expect(list).toEqual([3, 4, 5])
+  })
+})
+
+function streamOf(chunks: string[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder()
+  return new ReadableStream({
+    start(controller) {
+      for (const chunk of chunks) controller.enqueue(encoder.encode(chunk))
+      controller.close()
+    },
+  })
+}
+
+const realFetch = globalThis.fetch
+
+// bun 的测试环境没有 localStorage，而 streamLogs 要从里面取令牌
+if (typeof globalThis.localStorage === 'undefined') {
+  ;(globalThis as Record<string, unknown>).localStorage = {
+    getItem: () => '',
+    setItem: () => {},
+    removeItem: () => {},
+  }
+}
+
+afterEach(() => {
+  globalThis.fetch = realFetch
+})
+
+describe('读日志流', () => {
+  it('脏行被跳过，不会中断后面的行', async () => {
+    globalThis.fetch = mock(() =>
+      Promise.resolve(
+        new Response(
+          streamOf([
+            '{"type":"info","payload":"第一条"}\n',
+            '这不是 JSON\n',
+            '{"type":"warning","payload":"第',
+            '二条"}\n',
+          ]),
+        ),
+      ),
+    ) as unknown as typeof fetch
+
+    const lines: string[] = []
+    for await (const line of streamLogs('info', new AbortController().signal)) {
+      lines.push(`${line.type}: ${line.payload}`)
+    }
+    expect(lines).toEqual(['info: 第一条', 'warning: 第二条'])
+  })
+
+  it('接口报错时用服务端给的 message 抛出', async () => {
+    globalThis.fetch = mock(() =>
+      Promise.resolve(
+        new Response(JSON.stringify({ message: '内核未连接' }), { status: 502 }),
+      ),
+    ) as unknown as typeof fetch
+
+    const run = async () => {
+      for await (const _ of streamLogs('info', new AbortController().signal)) {
+        // 不该有产出
+      }
+    }
+    await expect(run()).rejects.toThrow(new LogStreamError('内核未连接'))
+  })
+
+  it('LOG_CAP 是个有限值，无限流不会把内存吃光', () => {
+    expect(LOG_CAP).toBeGreaterThan(0)
+    expect(LOG_CAP).toBeLessThanOrEqual(10000)
+  })
+})
diff --git a/web/logs.ts b/web/logs.ts
new file mode 100644
index 0000000..9a6ef74
--- /dev/null
+++ b/web/logs.ts
@@ -0,0 +1,87 @@
+/**
+ * 内核日志流。
+ *
+ * 接口是 `GET /api/kernel/logs?level=...`，一条永不结束的流，一行一个 JSON，
+ * 形如 `{ "type": "info", "payload": "..." }`。所以不能走 api.ts 的 request()——
+ * 那个会把整个响应读完，对流等于挂死。这里自己 fetch，按行读用 stream.ts。
+ *
+ * 解析和成批追加放在组件外面，是因为「脏行跳过」「超上限丢最旧的」这两条
+ * 是验收里点名要测的行为，埋在 .vue 里测不到。
+ */
+import { getToken } from './api'
+import { readLines } from './stream'
+
+/** 内核接受的日志级别。连接建立时定死，中途改不了，所以切换等于重连。 */
+export const LEVELS = ['debug', 'info', 'warning', 'error', 'silent'] as const
+export type LogLevel = (typeof LEVELS)[number]
+
+/**
+ * 页面上保留的最近条数。日志是无限流，不设上限内存会一直涨，
+ * 而目标机器只有 1-2G。超出就丢最旧的。
+ */
+export const LOG_CAP = 1000
+
+export interface LogLine {
+  type: string
+  payload: string
+}
+
+/** 解析一行日志。脏行（不是 JSON、形状不对）返回 null，由调用方跳过，不能抛。 */
+export function parseLogLine(line: string): LogLine | null {
+  let value: unknown
+  try {
+    value = JSON.parse(line)
+  } catch {
+    return null
+  }
+  if (typeof value !== 'object' || value === null || Array.isArray(value)) return null
+  const { type, payload } = value as Record<string, unknown>
+  if (typeof type !== 'string' || typeof payload !== 'string') return null
+  return { type, payload }
+}
+
+/** 把一批新条目追加进列表，超出上限从头部丢最旧的。 */
+export function pushCapped<T>(list: T[], items: T[], cap: number): void {
+  list.push(...items)
+  if (list.length > cap) list.splice(0, list.length - cap)
+}
+
+export class LogStreamError extends Error {
+  constructor(message: string) {
+    super(message)
+    this.name = 'LogStreamError'
+  }
+}
+
+/**
+ * 连上内核的日志流，逐条产出。停止的方式是中止 signal。
+ *
+ * 流正常结束（for await 走完）意味着内核那边断了连接，比如内核重启——
+ * 调用方要把它和「主动中止」区分开，前者该告诉用户。
+ */
+export async function* streamLogs(
+  level: LogLevel,
+  signal: AbortSignal,
+): AsyncGenerator<LogLine> {
+  const res = await fetch(`/api/kernel/logs?level=${level}`, {
+    headers: { Authorization: `Bearer ${getToken()}` },
+    signal,
+  })
+  if (!res.ok) {
+    // 失败时后端回的是一次性的 { message, hint? }，不是流
+    let message = `日志接口请求失败（HTTP ${String(res.status)}）`
+    try {
+      const payload = JSON.parse(await res.text()) as Record<string, unknown>
+      if (typeof payload.message === 'string') message = payload.message
+    } catch {
+      // 保留默认消息
+    }
+    throw new LogStreamError(message)
+  }
+  if (!res.body) throw new LogStreamError('日志接口没有返回数据流')
+
+  for await (const line of readLines(res.body)) {
+    const entry = parseLogLine(line)
+    if (entry) yield entry
+  }
+}
diff --git a/web/views/RequestsPage.vue b/web/views/RequestsPage.vue
new file mode 100644
index 0000000..0a6efad
--- /dev/null
+++ b/web/views/RequestsPage.vue
@@ -0,0 +1,283 @@
+<script setup lang="ts">
+/**
+ * 请求页：内核的日志流，一条一条「这次请求匹配了哪条规则、走了哪个出站」。
+ *
+ * 编排只有三件事：
+ *   1. 流的生命周期——进页面开始读，离开页面中止；切换级别断开重连，
+ *      因为内核的级别是建立连接时定的，中途改不了
+ *   2. 渲染节奏——收到的行先进缓冲，定时一批批刷进列表；暂停就是不刷，
+ *      缓冲照收（也设上限），恢复时补上
+ *   3. 滚动——默认跟着最新条目，用户往上滚就停，滚回底部恢复
+ */
+import { computed, nextTick, onBeforeUnmount, onMounted, ref, watch } from 'vue'
+import { kernel } from '../runtime'
+import { LEVELS, LOG_CAP, pushCapped, streamLogs } from '../logs'
+import type { LogLevel, LogLine } from '../logs'
+import IconPause from '~icons/ri/pause-line'
+import IconPlay from '~icons/ri/play-line'
+
+interface Row extends LogLine {
+  id: number
+  /** 收到那一刻的时间。日志行本身不带时间戳 */
+  time: string
+}
+
+const level = ref<LogLevel>('info')
+const query = ref('')
+const paused = ref(false)
+const rows = ref<Row[]>([])
+const failure = ref('')
+/** 内核那边把流断了（比如重启）。和主动中止区分开，这个要告诉用户 */
+const ended = ref(false)
+
+/** 级别标签的着色。只染级别那一小块，整行染色在低密度的设计里很吵 */
+const LEVEL_CLASS: Record<string, string> = {
+  debug: 'lv-debug',
+  info: 'lv-info',
+  warning: 'lv-warn',
+  error: 'lv-error',
+}
+
+let pending: Row[] = []
+let nextId = 0
+let aborter: AbortController | null = null
+let timer = 0
+
+const clock = (d: Date): string =>
+  [d.getHours(), d.getMinutes(), d.getSeconds()]
+    .map((n) => String(n).padStart(2, '0'))
+    .join(':')
+
+const listEl = ref<HTMLElement | null>(null)
+/** 是否跟随最新条目。用户手动上滚时关掉，滚回底部再开 */
+let follow = true
+
+function onScroll() {
+  const el = listEl.value
+  if (!el) return
+  follow = el.scrollTop + el.clientHeight >= el.scrollHeight - 24
+}
+
+async function scrollToBottom() {
+  if (!follow) return
+  await nextTick()
+  const el = listEl.value
+  if (el) el.scrollTop = el.scrollHeight
+}
+
+function flush() {
+  if (paused.value) {
+    // 暂停时继续接收但不渲染。缓冲也设上限，暂停再久内存也不涨
+    if (pending.length > LOG_CAP) pending.splice(0, pending.length - LOG_CAP)
+    return
+  }
+  if (pending.length === 0) return
+  pushCapped(rows.value, pending, LOG_CAP)
+  pending = []
+  void scrollToBottom()
+}
+
+async function pump() {
+  // 切换级别或内核恢复在线都从这里重连。旧流先停，不然两条流一起往缓冲里倒
+  aborter?.abort()
+  const controller = new AbortController()
+  aborter = controller
+  failure.value = ''
+  ended.value = false
+  try {
+    for await (const line of streamLogs(level.value, controller.signal)) {
+      pending.push({ ...line, id: nextId++, time: clock(new Date()) })
+    }
+    if (!controller.signal.aborted) ended.value = true
+  } catch (error) {
+    if (controller.signal.aborted) return
+    failure.value = error instanceof Error ? error.message : String(error)
+  }
+}
+
+onMounted(() => {
+  timer = window.setInterval(flush, 200)
+  if (kernel.value.online) void pump()
+})
+
+// 挂载时内核不在线就不连；等它上线再开始。反方向（流到一半内核掉了）
+// 由流自己报错或结束来呈现
+watch(
+  () => kernel.value.online,
+  (online) => {
+    if (online) void pump()
+  },
+)
+
+watch(level, () => {
+  if (kernel.value.online) void pump()
+})
+
+onBeforeUnmount(() => {
+  aborter?.abort()
+  window.clearInterval(timer)
+})
+
+const visible = computed(() => {
+  const q = query.value.trim().toLowerCase()
+  if (!q) return rows.value
+  return rows.value.filter((row) => row.payload.toLowerCase().includes(q))
+})
+</script>
+
+<template>
+  <div class="page">
+    <header class="head">
+      <select v-model="level" class="ctl" aria-label="日志级别">
+        <option v-for="l in LEVELS" :key="l" :value="l">{{ l }}</option>
+      </select>
+
+      <input v-model="query" class="ctl search" type="search" placeholder="筛选" />
+
+      <button class="ctl pause" type="button" @click="paused = !paused">
+        <IconPlay v-if="paused" />
+        <IconPause v-else />
+        <span>{{ paused ? '继续' : '暂停' }}</span>
+      </button>
+    </header>
+
+    <div v-if="failure" class="failure">
+      <strong>{{ failure }}</strong>
+    </div>
+
+    <div ref="listEl" class="log" @scroll="onScroll">
+      <p v-if="!kernel.online" class="state">内核未连接，无法读取日志。</p>
+
+      <template v-else>
+        <div v-for="row in visible" :key="row.id" class="row">
+          <span class="time mono">{{ row.time }}</span>
+          <span :class="['lv', LEVEL_CLASS[row.type] ?? 'lv-info']">{{ row.type }}</span>
+          <span class="msg mono">{{ row.payload }}</span>
+        </div>
+
+        <p v-if="rows.length === 0 && !failure && !ended" class="state">
+          暂无日志。没有流量经过内核时不会产生请求记录。
+        </p>
+        <p v-else-if="visible.length === 0 && rows.length > 0" class="state">
+          没有包含「{{ query.trim() }}」的日志。
+        </p>
+      </template>
+    </div>
+
+    <p v-if="ended && !failure" class="notice">日志流已中断。切换级别可重新连接。</p>
+  </div>
+</template>
+
+<style scoped>
+/* 日志区自己滚动、撑满视口剩余高度：自动跟随需要它有自己的滚动容器 */
+.page {
+  display: flex;
+  flex-direction: column;
+  height: calc(100vh - var(--s-6) - var(--s-7));
+}
+
+.head {
+  display: flex;
+  align-items: center;
+  gap: var(--s-3);
+  margin-bottom: var(--s-4);
+}
+
+.ctl {
+  padding: var(--s-2) var(--s-3);
+  border: 1px solid var(--c-border-strong);
+  border-radius: var(--r-sm);
+  background: var(--c-surface);
+}
+
+.search {
+  flex: 1;
+  max-width: 320px;
+}
+
+.pause {
+  display: flex;
+  align-items: center;
+  gap: var(--s-2);
+  margin-left: auto;
+  cursor: pointer;
+}
+.pause:hover {
+  border-color: var(--c-accent);
+  color: var(--c-accent-text);
+}
+
+.failure {
+  border: 1px solid color-mix(in srgb, var(--c-danger) 35%, var(--c-border));
+  border-radius: var(--r-md);
+  background: var(--c-danger-soft);
+  padding: var(--s-3) var(--s-4);
+  margin-bottom: var(--s-4);
+  font-size: var(--t-sm);
+}
+
+.log {
+  flex: 1;
+  min-height: 0;
+  overflow-y: auto;
+  border: 1px solid var(--c-border);
+  border-radius: var(--r-md);
+  background: var(--c-surface);
+  padding: var(--s-3) var(--s-4);
+}
+
+.row {
+  display: grid;
+  grid-template-columns: auto auto minmax(0, 1fr);
+  gap: var(--s-3);
+  align-items: baseline;
+  padding: 2px 0;
+  font-size: var(--t-sm);
+  line-height: 1.6;
+}
+
+.time {
+  color: var(--c-text-faint);
+}
+
+.lv {
+  padding: 0 var(--s-2);
+  border-radius: var(--r-sm);
+  font-size: var(--t-xs);
+  text-align: center;
+}
+
+.lv-debug {
+  background: var(--c-surface-sunken);
+  color: var(--c-text-faint);
+}
+.lv-info {
+  background: var(--c-accent-soft);
+  color: var(--c-accent-text);
+}
+.lv-warn {
+  background: var(--c-warn-soft);
+  color: var(--c-warn);
+}
+.lv-error {
+  background: var(--c-danger-soft);
+  color: var(--c-danger);
+}
+
+.msg {
+  word-break: break-all;
+  white-space: pre-wrap;
+}
+
+.state {
+  color: var(--c-text-faint);
+  font-size: var(--t-sm);
+  padding: var(--s-3) 0;
+}
+
+.notice {
+  color: var(--c-text-soft);
+  font-size: var(--t-sm);
+  margin-top: var(--s-3);
+}
+</style>