feat(web): chat pinia store + useChatStream (fetch+ReadableStream SSE)

web/src/composables/useChatStream.ts

@@ -0,0 +1,107 @@
+import { ref } from 'vue'
+import { useAuthStore } from '@/stores/auth'
+
+export type ChatStreamEventType =
+  | 'message_meta'
+  | 'text_delta'
+  | 'thinking_delta'
+  | 'usage'
+  | 'done'
+  | 'error'
+
+export interface ChatStreamEvent {
+  id: number
+  event: ChatStreamEventType
+  data: unknown
+}
+
+export interface ChatStreamHandlers {
+  onEvent: (e: ChatStreamEvent) => void
+  onClose?: () => void
+  onError?: (err: Error) => void
+}
+
+export interface ChatStreamHandle {
+  stop: () => void
+}
+
+/**
+ * useChatStream 通过 fetch + ReadableStream 自实现 SSE 解析，
+ * 以支持 Authorization Bearer 头（EventSource 不支持自定义 header）。
+ * start() 立即返回 stop 句柄，读流在后台 fire-and-forget。
+ */
+export function useChatStream() {
+  const auth = useAuthStore()
+  const reading = ref(false)
+
+  function start(
+    streamUrl: string,
+    since: number,
+    h: ChatStreamHandlers,
+  ): ChatStreamHandle {
+    const url = new URL(streamUrl, window.location.origin)
+    url.searchParams.set('since', String(since))
+    const ctrl = new AbortController()
+    reading.value = true
+
+    void (async () => {
+      try {
+        const headers: Record<string, string> = { Accept: 'text/event-stream' }
+        if (auth.token) headers['Authorization'] = `Bearer ${auth.token}`
+        const resp = await fetch(url.toString(), {
+          headers,
+          signal: ctrl.signal,
+        })
+        if (!resp.ok || !resp.body) {
+          throw new Error(`SSE HTTP ${resp.status}`)
+        }
+        const reader = resp.body.getReader()
+        const decoder = new TextDecoder()
+        let buf = ''
+        for (;;) {
+          const { value, done } = await reader.read()
+          if (done) break
+          buf += decoder.decode(value, { stream: true })
+          let idx = buf.indexOf('\n\n')
+          while (idx >= 0) {
+            const raw = buf.slice(0, idx)
+            buf = buf.slice(idx + 2)
+            const ev = parseSSE(raw)
+            if (ev) h.onEvent(ev)
+            idx = buf.indexOf('\n\n')
+          }
+        }
+        h.onClose?.()
+      } catch (err) {
+        const e = err as { name?: string }
+        if (e?.name === 'AbortError') return
+        h.onError?.(err instanceof Error ? err : new Error(String(err)))
+      } finally {
+        reading.value = false
+      }
+    })()
+
+    return { stop: () => ctrl.abort() }
+  }
+
+  return { start, reading }
+}
+
+function parseSSE(raw: string): ChatStreamEvent | null {
+  let id = 0
+  let event = ''
+  const dataLines: string[] = []
+  for (const line of raw.split('\n')) {
+    if (line.startsWith('id:')) id = Number(line.slice(3).trim())
+    else if (line.startsWith('event:')) event = line.slice(6).trim()
+    else if (line.startsWith('data:')) dataLines.push(line.slice(5).trim())
+  }
+  if (!event) return null
+  let data: unknown
+  try {
+    data = JSON.parse(dataLines.join('\n'))
+  } catch {
+    data = dataLines.join('\n')
+  }
+  return { id, event: event as ChatStreamEventType, data }
+}

web/src/stores/chat.ts

@@ -0,0 +1,212 @@
+import { defineStore } from 'pinia'
+import { computed, ref } from 'vue'
+
+import {
+  chatApi,
+  type ChatMessage,
+  type Conversation,
+  type ListConversationsParams,
+} from '@/api/chat'
+import {
+  useChatStream,
+  type ChatStreamEvent,
+  type ChatStreamHandle,
+} from '@/composables/useChatStream'
+
+interface UsagePayload {
+  PromptTokens?: number
+  CompletionTokens?: number
+  ThinkingTokens?: number
+}
+
+interface TextDeltaPayload {
+  text?: string
+}
+
+interface ErrorPayload {
+  code?: string
+  message?: string
+}
+
+interface MetaPayload {
+  id?: number
+  role?: string
+  status?: ChatMessage['status']
+}
+
+export const useChatStore = defineStore('chat', () => {
+  const conversations = ref<Conversation[]>([])
+  const currentConversation = ref<Conversation | null>(null)
+  const messages = ref<ChatMessage[]>([])
+  const streamingMessageId = ref<number | null>(null)
+  const streamHandle = ref<ChatStreamHandle | null>(null)
+
+  const isStreaming = computed(() => streamingMessageId.value !== null)
+
+  async function loadConversations(filter: ListConversationsParams = {}) {
+    conversations.value = await chatApi.listConversations(filter)
+  }
+
+  async function openConversation(id: string) {
+    detachStream()
+    const detail = await chatApi.getConversation(id)
+    currentConversation.value = detail.conversation
+    messages.value = detail.messages
+    const pending = detail.messages.find((m) => m.status === 'pending')
+    if (pending) {
+      streamingMessageId.value = pending.id
+      attachStream(`/api/v1/conversations/${id}/stream`, pending.id - 1)
+    }
+  }
+
+  async function sendMessage(content: string, attachmentIds: string[] = []) {
+    if (!currentConversation.value) return
+    const conv = currentConversation.value
+    const resp = await chatApi.sendMessage(conv.id, {
+      content,
+      attachment_ids: attachmentIds,
+    })
+    const now = new Date().toISOString()
+    messages.value.push({
+      id: resp.user_message_id,
+      conversation_id: conv.id,
+      role: 'user',
+      content,
+      status: 'ok',
+      created_at: now,
+      updated_at: now,
+    })
+    messages.value.push({
+      id: resp.assistant_message_id,
+      conversation_id: conv.id,
+      role: 'assistant',
+      content: '',
+      status: 'pending',
+      created_at: now,
+      updated_at: now,
+    })
+    streamingMessageId.value = resp.assistant_message_id
+    attachStream(resp.stream_url, resp.user_message_id)
+  }
+
+  async function cancelMessage() {
+    const id = streamingMessageId.value
+    if (id == null) return
+    await chatApi.cancelMessage(id)
+    detachStream()
+  }
+
+  async function retryMessage(messageId: number) {
+    if (!currentConversation.value) return
+    const resp = await chatApi.retryMessage(messageId)
+    const idx = messages.value.findIndex((m) => m.id === messageId)
+    if (idx >= 0) {
+      const old = messages.value[idx]
+      messages.value[idx] = {
+        ...old,
+        id: resp.assistant_message_id,
+        content: '',
+        thinking: null,
+        status: 'pending',
+        error_message: null,
+      }
+    }
+    streamingMessageId.value = resp.assistant_message_id
+    attachStream(resp.stream_url, resp.assistant_message_id - 1)
+  }
+
+  async function deleteConversation(id: string) {
+    await chatApi.deleteConversation(id)
+    conversations.value = conversations.value.filter((c) => c.id !== id)
+    if (currentConversation.value?.id === id) {
+      currentConversation.value = null
+      messages.value = []
+      detachStream()
+    }
+  }
+
+  function attachStream(streamUrl: string, sinceMessageID: number) {
+    detachStream()
+    const stream = useChatStream()
+    streamHandle.value = stream.start(streamUrl, sinceMessageID, {
+      onEvent: (ev) => applyStreamEvent(ev),
+      onClose: () => {
+        streamHandle.value = null
+      },
+      onError: (err) => {
+        console.error('chat stream error', err)
+        const id = streamingMessageId.value
+        if (id != null) {
+          const msg = messages.value.find((m) => m.id === id)
+          if (msg && msg.status === 'pending') {
+            msg.status = 'error'
+            msg.error_message = err.message
+          }
+        }
+        streamingMessageId.value = null
+        streamHandle.value = null
+      },
+    })
+  }
+
+  function detachStream() {
+    streamHandle.value?.stop()
+    streamHandle.value = null
+  }
+
+  function applyStreamEvent(ev: ChatStreamEvent) {
+    const targetId = streamingMessageId.value
+    if (targetId == null) return
+    const msg = messages.value.find((m) => m.id === targetId)
+    if (!msg) return
+    switch (ev.event) {
+      case 'message_meta': {
+        const d = ev.data as MetaPayload
+        if (d.status) msg.status = d.status
+        break
+      }
+      case 'text_delta': {
+        const d = ev.data as TextDeltaPayload
+        if (d.text) msg.content += d.text
+        break
+      }
+      case 'thinking_delta': {
+        const d = ev.data as TextDeltaPayload
+        if (d.text) msg.thinking = (msg.thinking ?? '') + d.text
+        break
+      }
+      case 'usage': {
+        const u = ev.data as UsagePayload
+        msg.prompt_tokens = u.PromptTokens ?? msg.prompt_tokens
+        msg.completion_tokens = u.CompletionTokens ?? msg.completion_tokens
+        msg.thinking_tokens = u.ThinkingTokens ?? msg.thinking_tokens
+        break
+      }
+      case 'done':
+        msg.status = 'ok'
+        streamingMessageId.value = null
+        break
+      case 'error': {
+        const d = ev.data as ErrorPayload
+        msg.status = 'error'
+        msg.error_message = d.message ?? d.code ?? 'stream error'
+        streamingMessageId.value = null
+        break
+      }
+    }
+  }
+
+  return {
+    conversations,
+    currentConversation,
+    messages,
+    streamingMessageId,
+    isStreaming,
+    loadConversations,
+    openConversation,
+    sendMessage,
+    cancelMessage,
+    retryMessage,
+    deleteConversation,
+  }
+})