Expose preview batches, rate-limit status, and live option updates

media-scrape/src/engine.ts

@@ -46,6 +46,18 @@ export interface ScrapeBackend {
     fetchPage(channelId: string, before: string | null, limit: number): Promise<FetchPage>;
 }
 
+/** One batch of preview work, as returned by `previewBatch`. */
+export interface PreviewBatchResult {
+    cached: number;
+    skipped: number;
+    failed: number;
+    ratelimited: number;
+    /** Rows still pending after the batch. */
+    remaining: number;
+    /** True when a stop request interrupted the batch. */
+    stopped: boolean;
+}
+
 export interface ScrapeEngineOptions {
     rootDir: string;
     pageSize?: number;
@@ -440,41 +452,101 @@ export class ScrapeEngine {
         }
     }
 
-    /** Cache small previews. Embed thumbnails are cached in both modes. */
+    /** Download options shared by the walk-cycle previews and the backfill queue. */
-    private async runPreviews(store: ScrapeStore, dir: string, meta: ScrapeMeta, limit = 0): Promise<void> {
+    private previewOptions(meta: ScrapeMeta) {
-        const options = {
+        return {
             ...(this.options.download ?? {}),
             embedPreviews: true,
             destinationDir: meta.downloadDir ?? null
         };
-        const pending = store.pendingPreviews(limit > 0 ? limit : 5000, true);
+    }
-        let cached = 0;
-        let skipped = 0;
-        let failed = 0;
 
+    /** Cache small previews. Embed thumbnails are cached in both modes. */
+    private async runPreviews(store: ScrapeStore, dir: string, meta: ScrapeMeta, limit = 0): Promise<void> {
+        const options = this.previewOptions(meta);
+        const total = limit > 0 ? limit : 5000;
+        const pending = store.pendingPreviews(total, true);
         if (pending.length) this.log(`${meta.id}: preview pass: ${pending.length} pending`);
+
         const batchSize = Math.max(1, options.previewBatchSize ?? 25);
         const batchPauseMs = Math.max(0, options.previewBatchPauseMs ?? 60_000);
-        let batch = 0;
+        let processed = 0;
+        let limitedBatches = 0;
 
         // Previews are fetched in batches: Discord and the provider APIs throttle bursts.
-        const afterRow = async () => {
+        while (processed < total) {
-            batch++;
+            await this.respectRateLimit();
-            if (batchPauseMs > 0 && batch >= batchSize) {
+
-                batch = 0;
+            const result = await this.previewRows(
+                store,
+                dir,
+                meta,
+                Math.min(batchSize, total - processed),
+                options
+            );
+            processed += result.cached + result.skipped + result.failed + result.ratelimited;
+
+            if (result.stopped || result.remaining === 0 || processed >= total) break;
+
+            // A batch that only hit 429s means the cooldown has not cleared. Wait it out and
+            // retry, but stop after a few rounds so a host that never recovers cannot spin.
+            if (result.ratelimited > 0 && result.cached + result.skipped + result.failed === 0) {
+                limitedBatches++;
+                if (limitedBatches >= 3) break;
+                continue;
+            }
+            limitedBatches = 0;
+
+            if (batchPauseMs > 0) {
                 this.log(`${meta.id}: preview batch of ${batchSize} done, pausing ${batchPauseMs} ms`);
                 await this.sleep(batchPauseMs);
             }
+        }
+    }
+
+    /**
+     * Process up to `limit` pending preview rows for one scrape. Paces nothing and never waits
+     * out a cooldown, so the backfill queue can drive it and report progress; a 429 inside the
+     * batch registers the cooldown for the caller to respect before the next batch.
+     */
+    async previewBatch(scrapeId: string, limit: number): Promise<PreviewBatchResult> {
+        const dir = join(this.options.rootDir, scrapeId);
+        const meta = this.active.get(scrapeId) ?? readScrapeMeta(dir);
+        if (!meta) return { cached: 0, skipped: 0, failed: 0, ratelimited: 0, remaining: 0, stopped: false };
+
+        const store = new ScrapeStore(dir);
+        try {
+            return await this.previewRows(store, dir, meta, Math.max(1, limit), this.previewOptions(meta));
+        } finally {
+            store.close();
+        }
+    }
+
+    /** One batch of preview rows. Mutates the store and the meta counters. */
+    private async previewRows(
+        store: ScrapeStore,
+        dir: string,
+        meta: ScrapeMeta,
+        limit: number,
+        options: DownloadOptions & { embedPreviews: boolean; destinationDir: string | null }
+    ): Promise<PreviewBatchResult> {
+        const result: PreviewBatchResult = {
+            cached: 0,
+            skipped: 0,
+            failed: 0,
+            ratelimited: 0,
+            remaining: 0,
+            stopped: false
         };
+        const pending = store.pendingPreviews(limit, true);
 
         for (const row of pending) {
             if (this.stopRequested.has(meta.id)) {
                 meta.state = "paused";
-                return;
+                result.stopped = true;
+                break;
             }
 
-            await this.respectRateLimit();
-
             // Audio has no image proxy: render a waveform strip with ffmpeg.
             if (row.kind === "audio" && row.source !== "embed") {
                 const waveform = await remoteFrame(row.url, join(dir, "thumbs"), { waveform: true });
@@ -485,14 +557,14 @@ export class ScrapeEngine {
                         ext: "webp"
                     });
                     meta.previewed = (meta.previewed ?? 0) + 1;
+                    result.cached++;
                 } else {
                     store.markPreviewError(row.id, "waveform failed");
-                    failed++;
+                    result.failed++;
                     this.log(`${meta.id}: preview #${row.id} audio waveform failed (${row.url.slice(0, 60)})`);
                 }
                 this.persist(dir, meta);
                 await this.sleep(previewGapMs(options));
-                await afterRow();
                 continue;
             }
 
@@ -503,11 +575,10 @@ export class ScrapeEngine {
                 const resolved = await resolveEmbedPreview(row, options);
                 if (!resolved) {
                     store.markPreviewError(row.id, "no preview source");
-                    skipped++;
+                    result.skipped++;
                     this.log(`${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} has no preview source`);
                     this.persist(dir, meta);
                     await this.sleep(previewGapMs(options));
-                    await afterRow();
                     continue;
                 }
                 previewRow = { ...row, preview_url: resolved };
@@ -535,7 +606,7 @@ export class ScrapeEngine {
 
                 store.markPreview(row.id, { hash, bytes, ext, animHash });
                 meta.previewed = (meta.previewed ?? 0) + 1;
-                cached++;
+                result.cached++;
 
                 // Embed thumbnails also land in the custom destination, so the download has a
                 // viewable poster for embeds while the media tab keeps its local copy.
@@ -554,41 +625,44 @@ export class ScrapeEngine {
                 if (frame) {
                     store.markPreview(row.id, { hash: frame.hash, bytes: frame.bytes, ext: "webp" });
                     meta.previewed = (meta.previewed ?? 0) + 1;
+                    result.cached++;
                 } else {
                     store.markPreviewError(row.id, "too large");
-                    skipped++;
+                    result.skipped++;
                     this.log(`${meta.id}: preview #${row.id} gif too large and no frame (${row.url.slice(0, 60)})`);
                 }
             } else if (outcome.state === "ratelimited") {
                 // Leave the row pending and hold off: the server asked for a break.
                 this.registerRateLimit(outcome.retryAfterMs);
+                result.ratelimited++;
                 this.log(`${meta.id}: preview #${row.id} rate limited (${outcome.error ?? "429"})`);
             } else if (outcome.state === "failed") {
                 // Transient network failure: retry on the next pass, but give up on hosts that
                 // stay unreachable (dead services like Gfycat and Streamable never recover).
                 const attempts = store.bumpPreviewAttempt(row.id);
-                const limit = this.options.previewMaxAttempts ?? 3;
+                const maxAttempts = this.options.previewMaxAttempts ?? 3;
 
-                if (attempts >= limit) {
+                if (attempts >= maxAttempts) {
                     store.markPreviewError(
                         row.id,
                         `unreachable after ${attempts} attempt(s): ${outcome.error ?? "unknown"}`
                     );
-                    skipped++;
+                    result.skipped++;
                     this.log(
                         `${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} gave up after ` +
                             `${attempts} transient failures (${row.url.slice(0, 60)})`
                     );
                 } else {
-                    failed++;
+                    result.failed++;
                     this.log(
                         `${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} failed ` +
-                            `(transient ${attempts}/${limit}): ${outcome.error ?? "unknown"} (${row.url.slice(0, 60)})`
+                            `(transient ${attempts}/${maxAttempts}): ${outcome.error ?? "unknown"} ` +
+                            `(${row.url.slice(0, 60)})`
                     );
                 }
             } else {
                 store.markPreviewError(row.id, outcome.error ?? outcome.state);
-                skipped++;
+                result.skipped++;
                 this.log(
                     `${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} skipped: ` +
                         `${outcome.error ?? outcome.state} (${row.url.slice(0, 60)})`
@@ -597,16 +671,36 @@ export class ScrapeEngine {
 
             this.persist(dir, meta);
             await this.sleep(previewGapMs(options));
-            await afterRow();
         }
 
-        const remaining = store.pendingPreviews(5000, true).length;
+        result.remaining = store.pendingPreviews(5000, true).length;
         if (pending.length) {
             this.log(
-                `${meta.id}: preview pass done: ${cached} cached, ${skipped} skipped, ${failed} failed, ` +
+                `${meta.id}: preview pass done: ${result.cached} cached, ${result.skipped} skipped, ` +
-                    `${remaining} still pending`
+                    `${result.failed} failed, ${result.remaining} still pending`
             );
         }
+
+        return result;
+    }
+
+    /** Whether media traffic is currently holding off because of a rate limit. */
+    rateLimitStatus(): { active: boolean; until: number | null; remainingMs: number } {
+        const remaining = Math.max(0, this.rateLimitUntil - this.now());
+        return {
+            active: remaining > 0,
+            until: this.rateLimitUntil > 0 ? this.rateLimitUntil : null,
+            remainingMs: remaining
+        };
+    }
+
+    /** Retune pacing and limits while jobs run. `download` merges into the download options. */
+    updateOptions(
+        patch: Partial<ScrapeEngineOptions> & { download?: Partial<DownloadOptions> }
+    ): void {
+        const { download, ...rest } = patch;
+        Object.assign(this.options, rest);
+        if (download) this.options.download = { ...(this.options.download ?? {}), ...download };
     }
 
     /** A rate limit was seen: hold all walk and preview traffic for a while. */

media-scrape/src/index.ts

@@ -30,6 +30,7 @@ export {
 export {
     ScrapeEngine,
     type FetchPage,
+    type PreviewBatchResult,
     type ScrapeBackend,
     type ScrapeEngineOptions
 } from "./engine.js";

media-scrape/src/store.ts

@@ -384,6 +384,59 @@ export class ScrapeStore {
             .run(preview.hash, preview.bytes, preview.ext ?? "webp", id);
     }
 
+    /** Preview state for the whole scrape and per kind, for the backfill UI. */
+    previewCounts(): {
+        total: number;
+        pending: number;
+        cached: number;
+        failed: number;
+        byKind: Record<MediaKind, { total: number; pending: number; cached: number; failed: number }>;
+    } {
+        const empty = () => ({ total: 0, pending: 0, cached: 0, failed: 0 });
+        const byKind: Record<MediaKind, { total: number; pending: number; cached: number; failed: number }> = {
+            image: empty(),
+            gif: empty(),
+            video: empty(),
+            audio: empty()
+        };
+
+        const rows = this.db
+            .prepare(
+                `SELECT kind,
+                        COUNT(*) AS total,
+                        SUM(preview_hash IS NOT NULL) AS cached,
+                        SUM(preview_error IS NOT NULL) AS failed,
+                        SUM(preview_hash IS NULL AND preview_error IS NULL) AS pending
+                 FROM media
+                 GROUP BY kind`
+            )
+            .all() as unknown as Array<{
+            kind: MediaKind;
+            total: number;
+            cached: number;
+            failed: number;
+            pending: number;
+        }>;
+
+        const totals = empty();
+        for (const row of rows) {
+            const bucket = byKind[row.kind] ?? empty();
+            bucket.total += row.total;
+            bucket.pending += row.pending ?? 0;
+            bucket.cached += row.cached ?? 0;
+            bucket.failed += row.failed ?? 0;
+            byKind[row.kind] = bucket;
+
+            totals.total += row.total;
+            totals.pending += row.pending ?? 0;
+            totals.cached += row.cached ?? 0;
+            totals.failed += row.failed ?? 0;
+        }
+
+        return { ...totals, byKind };
+    }
+
+    /** Count one transient preview failure and return the new total. */
     /** Count one transient preview failure and return the new total. */
     bumpPreviewAttempt(id: number): number {
         this.db.prepare("UPDATE media SET preview_attempts = preview_attempts + 1 WHERE id = ?").run(id);

media-scrape/test/run.ts

@@ -1118,6 +1118,136 @@ void (async () => {
         store.close();
     });
 
+    await test("previewBatch respects the batch limit and reports the remainder", async () => {
+        const backend = new FakeBackend([
+            {
+                messages: [
+                    message("655", {
+                        attachments: [
+                            attachment({ filename: "one.png", contentType: "image/png" }),
+                            attachment({ filename: "two.png", contentType: "image/png" })
+                        ]
+                    })
+                ]
+            }
+        ]);
+
+        const engine = new ScrapeEngine({
+            rootDir: root,
+            pageSize: 2,
+            pageMinMs: 0,
+            pageMaxMs: 0,
+            sleep: async () => {},
+            random: () => 0,
+            log: () => {},
+            previews: false,
+            download: {
+                fetchImpl: async () =>
+                    new Response("png", { status: 200, headers: { "content-type": "image/png" } }),
+                minGapMs: 0,
+                maxGapMs: 0,
+                sleep: async () => {}
+            }
+        });
+
+        const started = await engine.start({ channelId: "129", guildId: null, mode: "links", backend });
+        await engine.processQueue();
+
+        const first = await engine.previewBatch(started.id, 1);
+        assert.equal(first.cached, 1);
+        assert.equal(first.remaining, 1);
+
+        const second = await engine.previewBatch(started.id, 5);
+        assert.equal(second.cached, 1);
+        assert.equal(second.remaining, 0);
+    });
+
+    await test("previewBatch registers a rate limit for the caller", async () => {
+        const backend = new FakeBackend([
+            {
+                messages: [
+                    message("656", {
+                        attachments: [attachment({ filename: "slow.png", contentType: "image/png" })]
+                    })
+                ]
+            }
+        ]);
+
+        const engine = new ScrapeEngine({
+            rootDir: root,
+            pageSize: 2,
+            pageMinMs: 0,
+            pageMaxMs: 0,
+            sleep: async () => {},
+            random: () => 0,
+            log: () => {},
+            previews: false,
+            rateLimitPauseMs: 30_000,
+            download: {
+                fetchImpl: async () =>
+                    new Response("slow down", { status: 429, headers: { "retry-after": "5" } }),
+                minGapMs: 0,
+                maxGapMs: 0,
+                sleep: async () => {}
+            }
+        });
+
+        const started = await engine.start({ channelId: "130", guildId: null, mode: "links", backend });
+        await engine.processQueue();
+
+        const result = await engine.previewBatch(started.id, 5);
+        assert.equal(result.ratelimited, 1);
+        assert.equal(result.remaining, 1);
+        assert.equal(engine.rateLimitStatus().active, true);
+        assert.ok(engine.rateLimitStatus().remainingMs > 0);
+    });
+
+    await test("updateOptions retunes the attempt cap live", async () => {
+        const backend = new FakeBackend([
+            {
+                messages: [
+                    message("657", {
+                        attachments: [attachment({ filename: "dead.png", contentType: "image/png" })]
+                    })
+                ]
+            }
+        ]);
+
+        const engine = new ScrapeEngine({
+            rootDir: root,
+            pageSize: 2,
+            pageMinMs: 0,
+            pageMaxMs: 0,
+            sleep: async () => {},
+            random: () => 0,
+            log: () => {},
+            previews: false,
+            previewMaxAttempts: 5,
+            download: {
+                fetchImpl: async () => {
+                    throw new Error("fetch failed");
+                },
+                minGapMs: 0,
+                maxGapMs: 0,
+                sleep: async () => {}
+            }
+        });
+
+        const started = await engine.start({ channelId: "131", guildId: null, mode: "links", backend });
+        await engine.processQueue();
+
+        engine.updateOptions({ previewMaxAttempts: 1 });
+
+        const result = await engine.previewBatch(started.id, 5);
+        assert.equal(result.skipped, 1);
+        assert.equal(result.remaining, 0);
+
+        const store = new ScrapeStore(join(root, started.id));
+        const row = store.queryMedia({ kind: "image" })[0];
+        assert.match(String(row.preview_error), /^unreachable after 1 attempt/);
+        store.close();
+    });
+
     await test("backfillPreviews fills rows left by a preview-less walk", async () => {
         const backend = new FakeBackend([
             {