fix(search): assemble batched SSE results (#6186)

src/locales/en-US.ts

@@ -1284,6 +1284,8 @@ export default {
     searching: 'Searching, please wait...',
     noData: 'No Data',
     noResourceFound: 'No resources found',
+    searchStreamTimeout: 'The search connection timed out. Please try again later.',
+    searchStreamDisconnected: 'The search connection was interrupted. Please try again later.',
     aiRecommend: 'AI Recommendation',
     reRecommend: 'Regenerate Recommendation',
     aiRecommendError: 'AI Recommendation Failed',

src/locales/zh-CN.ts

@@ -1276,6 +1276,8 @@ export default {
     searching: '正在搜索，请稍候...',
     noData: '没有数据',
     noResourceFound: '未搜索到任何资源',
+    searchStreamTimeout: '搜索连接长时间无响应，请稍后重试',
+    searchStreamDisconnected: '搜索连接已中断，请稍后重试',
     aiRecommend: '智能推荐',
     reRecommend: '重新生成推荐',
     aiRecommendError: '智能推荐失败',

src/locales/zh-TW.ts

@@ -1274,6 +1274,8 @@ export default {
     searching: '正在搜索，請稍候...',
     noData: '沒有數據',
     noResourceFound: '未搜索到任何資源',
+    searchStreamTimeout: '搜索連線長時間無回應，請稍後重試',
+    searchStreamDisconnected: '搜索連線已中斷，請稍後重試',
     aiRecommend: '智能推薦',
     reRecommend: '重新生成推薦',
     aiRecommendError: '智能推薦失敗',

src/pages/resource.vue

@@ -1,5 +1,4 @@
 <script setup lang="ts">
-import { debounce } from 'lodash-es'
 import type { LocationQuery } from 'vue-router'
 import NoDataFound from '@/components/states/NoDataFound.vue'
 import api from '@/api'
@@ -19,6 +18,7 @@ import { useToast } from 'vue-toastification'
 import { useKeepAliveRefresh } from '@/composables/useKeepAliveRefresh'
 import { useUserStore } from '@/stores'
 import { buildUserPermissionContext, hasPermission } from '@/utils/permission'
+import { SearchReplaceBatchCollector, isSearchReplaceBatchEvent } from '@/utils/searchStream'
 import { getCurrentLocale } from '@/plugins/i18n'
 
 // 国际化
@@ -313,7 +313,8 @@ let searchStreamIdleTimer: ReturnType<typeof setTimeout> | null = null
 const streamPreviewLimit = 24
 const streamUiFlushDelay = 1000
 const streamPreviewBufferLimit = streamPreviewLimit * 4
-const searchStreamIdleTimeout = 90_000
+// 兼容尚未提供心跳的旧后端，同时保留异常半开连接的最终兜底。
+const searchStreamIdleTimeout = 5 * 60_000
 const searchStreamDoneCloseDelay = 1500
 
 const streamTotalCount = ref(0)
@@ -342,6 +343,8 @@ const showResultHeader = computed(() => isRefreshed.value && !progressActive.val
 
 let pendingStreamItems: Array<Context> = []
 let pendingSubtitleStreamItems: Array<SubtitleInfo> = []
+const streamReplaceBatchCollector = new SearchReplaceBatchCollector<Context>()
+const subtitleReplaceBatchCollector = new SearchReplaceBatchCollector<SubtitleInfo>()
 let streamFlushTimer: ReturnType<typeof setTimeout> | null = null
 let streamFinalResultApplied = false
 let pendingProgressText: string | null = null
@@ -393,21 +396,9 @@ function handleRemoveFilter(key: string, value: string) {
   torrentFilter.removeFilter(key, value)
 }
 
-// 添加安全超时，确保进度条不会永远卡住
-const watchProgressValue = watch(
-  progressValue,
-  debounce(async () => {
-    if (progressActive.value && progressValue.value < 100) {
-      console.warn('卡进度超时 关闭进度条')
-      stopLoadingProgress()
-    }
-  }, 60_000),
-)
-
 // 使用SSE监听加载进度
 function startLoadingProgress() {
   clearProgressResetTimer()
-  watchProgressValue.resume()
   progressText.value = t('resource.searching')
   progressValue.value = 0
   progressEnabled.value = true
@@ -416,7 +407,6 @@ function startLoadingProgress() {
 
 // 停止监听加载进度
 function stopLoadingProgress() {
-  watchProgressValue.pause()
   progressActive.value = false
 
   // 确保进度显示100%，然后再渐进清零
@@ -429,6 +419,7 @@ function stopLoadingProgress() {
   }, 1500)
 }
 
+// 清除延迟归零任务，避免新搜索继承上一轮的进度重置。
 function clearProgressResetTimer() {
   if (progressResetTimer) {
     clearTimeout(progressResetTimer)
@@ -451,6 +442,7 @@ function closeSearchEventSource(source?: EventSource) {
   clearSearchStreamIdleTimer()
 }
 
+// 清除搜索流空闲计时器。
 function clearSearchStreamIdleTimer() {
   if (searchStreamIdleTimer) {
     clearTimeout(searchStreamIdleTimer)
@@ -473,6 +465,8 @@ function clearStreamPreviewState(resetFinalState: boolean = false) {
   pendingProgressText = null
   pendingProgressValue = null
   pendingStreamTotalCount = null
+  streamReplaceBatchCollector.reset()
+  subtitleReplaceBatchCollector.reset()
   streamPreviewDataList.value = []
   streamPreviewSubtitleDataList.value = []
   if (resetFinalState) {
@@ -514,6 +508,7 @@ function flushBufferedStreamState() {
   isRefreshed.value = true
 }
 
+// 合并短时间内到达的预览和进度更新。
 function scheduleStreamFlush() {
   if (streamFlushTimer) return
   streamFlushTimer = setTimeout(() => {
@@ -673,6 +668,7 @@ function appendSubtitleStreamResults(items: SubtitleInfo[]) {
   scheduleStreamFlush()
 }
 
+// 完整最终结果到达后原子替换资源列表。
 function applyFinalStreamResults(items: Context[]) {
   streamFinalResultApplied = true
   flushBufferedStreamState()
@@ -708,12 +704,27 @@ function getSubtitleItemKey(item: SubtitleInfo, index: number) {
 
 // 处理搜索流消息
 function handleSearchStreamMessage(eventData: { [key: string]: any }) {
+  if (eventData.type === 'heartbeat') return
+
   if (eventData.type === 'error') {
     updateSearchProgress(eventData, true)
     errorDescription.value = eventData.message_i18n || eventData.message || t('resource.noResourceFound')
     return
   }
 
+  if (isSearchReplaceBatchEvent<Context | SubtitleInfo>(eventData)) {
+    if (isSubtitleSearch.value) {
+      const completedItems = subtitleReplaceBatchCollector.append(eventData)
+      updateSearchProgress(eventData, completedItems !== null)
+      if (completedItems) applyFinalSubtitleStreamResults(completedItems)
+    } else {
+      const completedItems = streamReplaceBatchCollector.append(eventData)
+      updateSearchProgress(eventData, completedItems !== null)
+      if (completedItems) applyFinalStreamResults(completedItems)
+    }
+    return
+  }
+
   if (isSubtitleSearch.value) {
     const subtitleItems = Array.isArray(eventData.items) ? (eventData.items as SubtitleInfo[]) : []
     if (eventData.type === 'append') {
@@ -834,7 +845,8 @@ function searchByStream(params: SearchParams, requestToken?: string) {
     const resetIdleTimeout = () => {
       clearSearchStreamIdleTimer()
       searchStreamIdleTimer = setTimeout(() => {
-        settleSearchStream(() => reject(new Error(t('resource.noResourceFound'))))
+        errorDescription.value = t('resource.searchStreamTimeout')
+        settleSearchStream(() => reject(new Error(errorDescription.value)))
       }, searchStreamIdleTimeout)
     }
 
@@ -874,7 +886,8 @@ function searchByStream(params: SearchParams, requestToken?: string) {
         return
       }
 
-      settleSearchStream(() => reject(new Error(t('resource.noResourceFound'))))
+      errorDescription.value = t('resource.searchStreamDisconnected')
+      settleSearchStream(() => reject(new Error(errorDescription.value)))
     }
   })
 }
@@ -927,8 +940,14 @@ async function fetchData(options: { force?: boolean; params?: SearchParams; sile
       try {
         await searchByStream(currentSearchParams, requestToken)
       } catch (error) {
+        const streamErrorMessage = error instanceof Error ? error.message : t('resource.searchStreamDisconnected')
         console.warn('渐进式搜索连接失败，回退到普通搜索:', error)
+        try {
           await searchByRequest(currentSearchParams, requestToken)
+        } catch (fallbackError) {
+          errorDescription.value = streamErrorMessage
+          throw fallbackError
+        }
       }
       stopLoadingProgress()
       // 搜索完成后移除地址栏参数，避免分享/刷新残留搜索条件

src/utils/__tests__/searchStream.spec.ts

@@ -0,0 +1,42 @@
+import { SearchReplaceBatchCollector, isSearchReplaceBatchEvent } from '@/utils/searchStream'
+import { describe, expect, it } from 'vitest'
+
+/** 构造一个与后端最终结果分批协议一致的测试事件。 */
+function createBatch(batchIndex: number, items: number[]) {
+  return {
+    batch_count: 3,
+    batch_index: batchIndex,
+    items,
+    replace_batch: true,
+    total_items: 5,
+    type: batchIndex === 0 ? 'replace' : 'append',
+  }
+}
+
+describe('SearchReplaceBatchCollector', () => {
+  it('returns one atomic snapshot after every ordered batch arrives', () => {
+    const collector = new SearchReplaceBatchCollector<number>()
+
+    expect(collector.append(createBatch(0, [1, 2]))).toBeNull()
+    expect(collector.append(createBatch(1, [3, 4]))).toBeNull()
+    expect(collector.append(createBatch(2, [5]))).toEqual([1, 2, 3, 4, 5])
+  })
+
+  it('rejects a missing batch and accepts a fresh sequence afterwards', () => {
+    const collector = new SearchReplaceBatchCollector<number>()
+
+    expect(collector.append(createBatch(0, [1, 2]))).toBeNull()
+    expect(() => collector.append(createBatch(2, [5]))).toThrow('搜索结果批次顺序不连续')
+
+    expect(collector.append(createBatch(0, [1, 2]))).toBeNull()
+    expect(collector.append(createBatch(1, [3, 4]))).toBeNull()
+    expect(collector.append(createBatch(2, [5]))).toEqual([1, 2, 3, 4, 5])
+  })
+
+  it('rejects malformed batch metadata', () => {
+    const malformedEvent = { ...createBatch(0, [1, 2]), batch_count: 0 }
+
+    expect(isSearchReplaceBatchEvent(malformedEvent)).toBe(false)
+    expect(() => new SearchReplaceBatchCollector<number>().append(malformedEvent)).toThrow('搜索结果批次格式无效')
+  })
+})

src/utils/searchStream.ts

@@ -0,0 +1,79 @@
+/** 后端最终搜索结果分批事件的传输字段。 */
+export interface SearchReplaceBatchEvent<T> extends Record<string, unknown> {
+  batch_count: number
+  batch_index: number
+  items: T[]
+  replace_batch: true
+  total_items: number
+}
+
+/** 判断搜索流事件是否为结构完整的最终结果批次。 */
+export function isSearchReplaceBatchEvent<T>(event: Record<string, unknown>): event is SearchReplaceBatchEvent<T> {
+  return (
+    event.replace_batch === true &&
+    Number.isInteger(event.batch_index) &&
+    Number(event.batch_index) >= 0 &&
+    Number.isInteger(event.batch_count) &&
+    Number(event.batch_count) > 0 &&
+    Number(event.batch_index) < Number(event.batch_count) &&
+    Number.isInteger(event.total_items) &&
+    Number(event.total_items) >= 0 &&
+    Array.isArray(event.items)
+  )
+}
+
+/**
+ * 按 SSE 顺序收集最终搜索结果批次，仅在全部批次完整到达后返回可提交快照。
+ */
+export class SearchReplaceBatchCollector<T> {
+  private batchCount = 0
+  private expectedTotalItems = 0
+  private items: T[] = []
+  private nextBatchIndex = 0
+
+  /** 清除未完成批次，供新搜索、断流回退和组件卸载时复用。 */
+  reset(): void {
+    this.batchCount = 0
+    this.expectedTotalItems = 0
+    this.items = []
+    this.nextBatchIndex = 0
+  }
+
+  /**
+   * 接收一个最终结果批次；未收齐时返回 null，完整且数量一致时返回结果快照。
+   */
+  append(event: Record<string, unknown>): T[] | null {
+    if (!isSearchReplaceBatchEvent<T>(event)) {
+      this.reset()
+      throw new Error('搜索结果批次格式无效')
+    }
+
+    if (event.batch_index === 0) {
+      this.reset()
+      this.batchCount = event.batch_count
+      this.expectedTotalItems = event.total_items
+    }
+
+    if (
+      event.batch_index !== this.nextBatchIndex ||
+      event.batch_count !== this.batchCount ||
+      event.total_items !== this.expectedTotalItems
+    ) {
+      this.reset()
+      throw new Error('搜索结果批次顺序不连续')
+    }
+
+    this.items.push(...event.items)
+    this.nextBatchIndex += 1
+    if (this.nextBatchIndex < this.batchCount) return null
+
+    if (this.items.length !== this.expectedTotalItems) {
+      this.reset()
+      throw new Error('搜索结果批次数量不完整')
+    }
+
+    const completedItems = this.items
+    this.reset()
+    return completedItems
+  }
+}