msemes/PriceHunter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switch streaming search to sequential queue instead of parallel

Browse Source 
Parallel scraping with Puppeteer blocks the Node.js event loop,
preventing SSE events from flushing. Sequential processing means
each store completes, sends its event, and the client sees it
before the next store starts.

Also sorts stores so cheerio-based (fast) stores run first,
giving the user results sooner while Puppeteer stores load.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
mariosemes
2026-03-26 22:37:27 +01:00
parent 0e6ec21e81
commit 2c8ae5f628
1 changed files with 44 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/server/scraper/engine.ts
View File

@@ -155,21 +155,23 @@ export async function searchStreaming(
return;
}
// Sort: cheerio stores first (fast), Puppeteer stores last (slow)
stores.sort((a, b) => (a.render_js || 0) - (b.render_js || 0));
// Send start event with store list
onProgress({
type: 'start',
stores: stores.map((s) => ({ id: s.id, name: s.name, renderJs: !!s.render_js })),
});
// Yield to event loop so the SSE start event flushes to client
// Yield so the SSE start event flushes to client
await new Promise((r) => setTimeout(r, 0));
const limit = pLimit(MAX_CONCURRENCY);
const errors: SearchResult['meta']['errors'] = [];
let totalResults = 0;
const scrapePromises = stores.map((store) =>
limit(async () => {
// Process stores sequentially so SSE events flush between each
for (const store of stores) {
const searchUrl = store.search_url.replace('{query}', encodeURIComponent(query));
const storeStart = Date.now();
const rateLimiter = getLimiter(store.id, 1, Math.floor(store.rate_window / store.rate_limit));
@@ -196,9 +198,6 @@ export async function searchStreaming(
resultCount: products.length,
duration,
});
// Yield so SSE event flushes to client
await new Promise((r) => setTimeout(r, 0));
} catch (err) {
const duration = Date.now() - storeStart;
const errorMessage = err instanceof Error ? err.message : String(err);
@@ -213,10 +212,10 @@ export async function searchStreaming(
duration,
});
}
})
);
await Promise.all(scrapePromises);
// Yield between stores so SSE events flush
await new Promise((r) => setTimeout(r, 0));
}
onProgress({
type: 'done',