fix issue 210 slow source search reliability

lib/api/http-utils.ts

@@ -24,12 +24,12 @@ export async function fetchWithTimeout(
 
     // If an external signal is provided, propagate its abort
     const externalSignal = options.signal;
+    let onAbort: (() => void) | undefined;
     if (externalSignal) {
         if (externalSignal.aborted) {
-            clearTimeout(timeoutId);
             controller.abort();
         } else {
-            const onAbort = () => controller.abort();
+            onAbort = () => controller.abort();
             externalSignal.addEventListener('abort', onAbort, { once: true });
         }
     }
@@ -39,11 +39,12 @@ export async function fetchWithTimeout(
             ...options,
             signal: controller.signal,
         });
-        clearTimeout(timeoutId);
         return response;
-    } catch (error) {
+    } finally {
         clearTimeout(timeoutId);
-        throw error;
+        if (externalSignal && onAbort) {
+            externalSignal.removeEventListener('abort', onAbort);
+        }
     }
 }
 
@@ -62,6 +63,10 @@ export async function withRetry<T>(
         } catch (error) {
             lastError = error as Error;
 
+            if (lastError?.name === 'AbortError') {
+                throw lastError;
+            }
+
             if (i < retries) {
                 await new Promise(resolve => setTimeout(resolve, RETRY_DELAY * (i + 1)));
             }

lib/api/source-latency.ts

@@ -0,0 +1,69 @@
+export interface SourceLatencyResult {
+  latency: number;
+  success: boolean;
+  timeout: boolean;
+  method: 'HEAD' | 'GET';
+}
+
+type ProbeAttemptResult = SourceLatencyResult;
+
+interface ProbeSourceLatencyOptions {
+  fetcher?: typeof fetch;
+  timeoutMs?: number;
+  now?: () => number;
+}
+
+async function probeAttempt(
+  url: string,
+  method: 'HEAD' | 'GET',
+  fetcher: typeof fetch,
+  timeoutMs: number,
+  now: () => number,
+): Promise<ProbeAttemptResult> {
+  const controller = new AbortController();
+  let timedOut = false;
+  const timeoutId = setTimeout(() => {
+    timedOut = true;
+    controller.abort();
+  }, timeoutMs);
+  const startedAt = now();
+
+  try {
+    await fetcher(url, {
+      method,
+      signal: controller.signal,
+      redirect: 'follow',
+      ...(method === 'GET' ? { headers: { Range: 'bytes=0-0' } } : {}),
+    });
+
+    return {
+      latency: Math.max(0, Math.round(now() - startedAt)),
+      success: true,
+      timeout: false,
+      method,
+    };
+  } catch (error) {
+    return {
+      latency: Math.max(0, Math.round(now() - startedAt)),
+      success: false,
+      timeout: timedOut || (error instanceof Error && error.name === 'AbortError'),
+      method,
+    };
+  } finally {
+    clearTimeout(timeoutId);
+  }
+}
+
+export async function probeSourceLatency(
+  url: string,
+  options: ProbeSourceLatencyOptions = {},
+): Promise<SourceLatencyResult> {
+  const fetcher = options.fetcher ?? fetch;
+  const timeoutMs = options.timeoutMs ?? 5000;
+  const now = options.now ?? (() => performance.now());
+
+  const headResult = await probeAttempt(url, 'HEAD', fetcher, timeoutMs, now);
+  if (headResult.success) return headResult;
+
+  return probeAttempt(url, 'GET', fetcher, timeoutMs, now);
+}

lib/hooks/useLatencyPing.ts

@@ -5,6 +5,7 @@
 
 import { useState, useEffect, useCallback, useRef, useMemo } from 'react';
 import { settingsStore } from '@/lib/store/settings-store';
+import { probeLatencyTargets } from '@/lib/utils/latency';
 
 interface LatencyState {
     [sourceId: string]: number;
@@ -23,17 +24,18 @@ export function useLatencyPing({
 }: UseLatencyPingOptions) {
     const [latencies, setLatencies] = useState<LatencyState>({});
     const [isLoading, setIsLoading] = useState(false);
-    const intervalRef = useRef<NodeJS.Timeout | null>(null);
+    const intervalRef = useRef<ReturnType<typeof setTimeout> | null>(null);
+    const roundInFlightRef = useRef(false);
     const mountedRef = useRef(true);
 
     // Check if real-time latency is enabled in settings
     const [realtimeEnabled, setRealtimeEnabled] = useState(false);
 
     // Stabilize sourceUrls to prevent unnecessary effect re-runs if parent passes new array
-    const stableSourceUrls = useMemo(() => sourceUrls, [
-        // Create a unique key for the sources array
-        sourceUrls.map(s => `${s.id}|${s.baseUrl}`).join(',')
-    ]);
+    const stableSourceUrls = useMemo(
+        () => sourceUrls.map((source) => ({ ...source })),
+        [sourceUrls],
+    );
 
     useEffect(() => {
         const settings = settingsStore.getSettings();
@@ -50,47 +52,64 @@ export function useLatencyPing({
         };
     }, []);
 
-    const pingSource = useCallback(async (sourceId: string, baseUrl: string): Promise<number | null> => {
+    const pingSource = useCallback(async (baseUrl: string): Promise<number | null> => {
+        const controller = new AbortController();
+        const timeoutId = setTimeout(() => controller.abort(), 12000);
+
         try {
             const response = await fetch('/api/ping', {
                 method: 'POST',
                 headers: { 'Content-Type': 'application/json' },
                 body: JSON.stringify({ url: baseUrl }),
+                signal: controller.signal,
             });
 
             if (response.ok) {
                 const data = await response.json();
-                return data.latency || null;
+                return data.success && typeof data.latency === 'number'
+                    ? data.latency
+                    : null;
             }
             return null;
         } catch {
             return null;
+        } finally {
+            clearTimeout(timeoutId);
         }
     }, []);
 
     const pingAllSources = useCallback(async () => {
-        if (!mountedRef.current || stableSourceUrls.length === 0) return;
+        if (
+            !mountedRef.current ||
+            stableSourceUrls.length === 0 ||
+            roundInFlightRef.current
+        ) return;
 
+        roundInFlightRef.current = true;
         setIsLoading(true);
 
-        const results = await Promise.all(
-            stableSourceUrls.map(async ({ id, baseUrl }) => {
-                const latency = await pingSource(id, baseUrl);
-                return { id, latency };
-            })
-        );
+        try {
+            const results = await probeLatencyTargets(
+                stableSourceUrls,
+                ({ baseUrl }) => pingSource(baseUrl),
+            );
 
-        if (mountedRef.current) {
-            setLatencies(prev => {
-                const newState = { ...prev };
-                results.forEach(({ id, latency }) => {
-                    if (latency !== null) {
-                        newState[id] = latency;
-                    }
+            if (mountedRef.current) {
+                setLatencies(prev => {
+                    const newState = { ...prev };
+                    results.forEach(({ id, latency }) => {
+                        if (latency !== null) {
+                            newState[id] = latency;
+                        }
+                    });
+                    return newState;
                 });
-                return newState;
-            });
-            setIsLoading(false);
+            }
+        } finally {
+            roundInFlightRef.current = false;
+            if (mountedRef.current) {
+                setIsLoading(false);
+            }
         }
     }, [stableSourceUrls, pingSource]);
 
@@ -101,17 +120,30 @@ export function useLatencyPing({
         const shouldPoll = enabled && realtimeEnabled && stableSourceUrls.length > 0;
 
         if (shouldPoll) {
-            // Initial ping
-            pingAllSources();
+            let cancelled = false;
+            const poll = async () => {
+                await pingAllSources();
+                if (!cancelled && mountedRef.current) {
+                    intervalRef.current = setTimeout(poll, intervalMs);
+                }
+            };
 
-            // Set up interval
-            intervalRef.current = setInterval(pingAllSources, intervalMs);
+            void poll();
+
+            return () => {
+                cancelled = true;
+                mountedRef.current = false;
+                if (intervalRef.current) {
+                    clearTimeout(intervalRef.current);
+                    intervalRef.current = null;
+                }
+            };
         }
 
         return () => {
             mountedRef.current = false;
             if (intervalRef.current) {
-                clearInterval(intervalRef.current);
+                clearTimeout(intervalRef.current);
                 intervalRef.current = null;
             }
         };
@@ -120,7 +152,7 @@ export function useLatencyPing({
     const refreshLatency = useCallback((sourceId: string) => {
         const source = stableSourceUrls.find(s => s.id === sourceId);
         if (source) {
-            pingSource(sourceId, source.baseUrl).then(latency => {
+            pingSource(source.baseUrl).then(latency => {
                 if (latency !== null && mountedRef.current) {
                     setLatencies(prev => ({ ...prev, [sourceId]: latency }));
                 }

lib/hooks/useSearchAction.ts

@@ -1,14 +1,13 @@
 import { useRef, useCallback } from 'react';
-import { SOURCE_IDS } from '@/lib/utils/source-names';
 import { sortVideos } from '@/lib/utils/sort';
 import { binaryInsertVideos } from '@/lib/utils/sorted-insert';
 import { processSearchStream } from '@/lib/utils/search-stream';
 import type { SortOption } from '@/lib/store/settings-store';
 import { settingsStore } from '@/lib/store/settings-store';
-import type { Video } from '@/lib/types';
 import { useSearchState } from './useSearchState';
 
 type SearchState = ReturnType<typeof useSearchState>;
+type SearchSourceConfig = { id: string; baseUrl?: string };
 
 interface UseSearchActionProps {
     state: SearchState;
@@ -77,7 +76,10 @@ export function useSearchAction({ state, onCacheUpdate, onUrlUpdate }: UseSearch
             const reader = response.body?.getReader();
             if (!reader) throw new Error('No response stream');
 
-            const sourcesMap = new Map<string, { count: number; name: string }>();
+            const sourceConfigs = new Map<string, SearchSourceConfig>(
+                targetSources.map((source: SearchSourceConfig) => [source.id, source])
+            );
+            const sourcesMap = new Map<string, { count: number; name: string; baseUrl?: string }>();
 
             await processSearchStream({
                 reader,
@@ -95,6 +97,7 @@ export function useSearchAction({ state, onCacheUpdate, onUrlUpdate }: UseSearch
                         sourcesMap.set(sourceId, {
                             count: newVideos.length,
                             name: newVideos[0]?.sourceName || sourceId,
+                            baseUrl: sourceConfigs.get(sourceId)?.baseUrl,
                         });
                     }
                 },
@@ -113,6 +116,7 @@ export function useSearchAction({ state, onCacheUpdate, onUrlUpdate }: UseSearch
                         id: id,
                         name: info.name,
                         count: info.count,
+                        ...(info.baseUrl ? { baseUrl: info.baseUrl } : {}),
                     }));
                     setAvailableSources(sources);
 

lib/types/index.ts

@@ -51,6 +51,7 @@ export interface SourceBadge {
   id: string;
   name: string;
   count: number;
+  baseUrl?: string;
   typeName?: string;
 }
 

lib/utils/latency.ts

@@ -10,6 +10,46 @@ interface LatencyInfo {
   level: 'excellent' | 'good' | 'fair' | 'slow';
 }
 
+export interface LatencyProbeTarget {
+  id: string;
+  baseUrl: string;
+}
+
+export interface LatencyProbeResult {
+  id: string;
+  latency: number | null;
+}
+
+export async function probeLatencyTargets(
+  targets: LatencyProbeTarget[],
+  probe: (target: LatencyProbeTarget) => Promise<number | null>,
+  concurrency: number = 4,
+): Promise<LatencyProbeResult[]> {
+  if (targets.length === 0) return [];
+
+  const results = new Array<LatencyProbeResult>(targets.length);
+  const workerCount = Math.min(
+    targets.length,
+    Math.max(1, Math.floor(concurrency)),
+  );
+  let nextIndex = 0;
+
+  const workers = Array.from({ length: workerCount }, async () => {
+    while (nextIndex < targets.length) {
+      const index = nextIndex;
+      nextIndex += 1;
+      const target = targets[index];
+      results[index] = {
+        id: target.id,
+        latency: await probe(target),
+      };
+    }
+  });
+
+  await Promise.all(workers);
+  return results;
+}
+
 /**
  * Get latency information with color coding
  * @param latency - Response time in milliseconds

lib/utils/search-stream.ts

@@ -35,80 +35,65 @@ export async function processSearchStream({
 }: StreamHandlerParams) {
     const decoder = new TextDecoder();
     let buffer = '';
-    let timeoutId: NodeJS.Timeout | null = null;
     let isCompleted = false;
 
-    // Auto-complete if no progress for 3 seconds
-    const resetTimeout = () => {
-        if (timeoutId) clearTimeout(timeoutId);
+    const blockedCategories = settingsStore.getSettings().blockedCategories;
 
-        timeoutId = setTimeout(() => {
+    while (true) {
+        const { done, value } = await reader.read();
+        if (done) {
             if (!isCompleted) {
-
                 isCompleted = true;
                 onComplete();
             }
-        }, 3000);
-    };
+            break;
+        }
 
-    try {
-        resetTimeout(); // Start initial timeout
-        const blockedCategories = settingsStore.getSettings().blockedCategories;
+        buffer += decoder.decode(value, { stream: true });
+        const lines = buffer.split('\n');
+        buffer = lines.pop() || '';
 
-        while (true) {
-            const { done, value } = await reader.read();
-            if (done) break;
+        for (const line of lines) {
+            if (!line.startsWith('data: ')) continue;
 
-            buffer += decoder.decode(value, { stream: true });
-            const lines = buffer.split('\n');
-            buffer = lines.pop() || '';
+            try {
+                const data = JSON.parse(line.slice(6));
 
-            for (const line of lines) {
-                if (!line.startsWith('data: ')) continue;
-
-                try {
-                    const data = JSON.parse(line.slice(6));
-
-                    if (data.type === 'start') {
-                        onStart(data.totalSources);
-                        resetTimeout();
-                    } else if (data.type === 'videos') {
-                        const newVideos: Video[] = data.videos
-                            .filter((video: any) => hasMinimumMatch(video.vod_name, currentQuery))
-                            .filter((video: any) => !isCategoryBlocked(video, blockedCategories))
-                            .map((video: any) => ({
-                                ...video,
-                                sourceName: video.sourceDisplayName || getSourceName(video.source),
-                                isNew: true,
-                                relevanceScore: calculateRelevanceScore(video, currentQuery),
-                            }));
-                        onVideos(newVideos, data.source);
-                        if (data.pagecount && onPageInfo) {
-                            onPageInfo(data.pagecount);
-                        }
-                        resetTimeout();
-                    } else if (data.type === 'progress') {
-                        onProgress(data.completedSources, data.totalVideosFound);
-                        resetTimeout();
-                    } else if (data.type === 'complete') {
-                        if (timeoutId) clearTimeout(timeoutId);
+                if (data.type === 'start') {
+                    onStart(data.totalSources);
+                } else if (data.type === 'videos') {
+                    const newVideos: Video[] = data.videos
+                        .filter((video: any) => hasMinimumMatch(video.vod_name, currentQuery))
+                        .filter((video: any) => !isCategoryBlocked(video, blockedCategories))
+                        .map((video: any) => ({
+                            ...video,
+                            sourceName: video.sourceDisplayName || getSourceName(video.source),
+                            isNew: true,
+                            relevanceScore: calculateRelevanceScore(video, currentQuery),
+                        }));
+                    onVideos(newVideos, data.source);
+                    if (data.pagecount && onPageInfo) {
+                        onPageInfo(data.pagecount);
+                    }
+                } else if (data.type === 'progress') {
+                    onProgress(data.completedSources, data.totalVideosFound);
+                } else if (data.type === 'complete') {
+                    if (!isCompleted) {
                         isCompleted = true;
                         if (data.maxPageCount && onPageInfo) {
                             onPageInfo(data.maxPageCount);
                         }
                         onComplete();
-                    } else if (data.type === 'error') {
+                    }
+                } else if (data.type === 'error') {
+                    if (!isCompleted) {
+                        isCompleted = true;
                         onError(data.message);
                     }
-                } catch (error) {
-                    console.error('Error parsing stream data:', error);
                 }
+            } catch (error) {
+                console.error('Error parsing stream data:', error);
             }
         }
-    } catch (error) {
-        if (timeoutId) clearTimeout(timeoutId);
-        throw error;
-    } finally {
-        if (timeoutId) clearTimeout(timeoutId);
     }
 }