Add the per-chat backfill queue with progress and ETA

media-scrape/src/backfill.ts

@@ -0,0 +1,418 @@
+/*
+ * The backfill queue: preview caching for scrapes that already walked.
+ *
+ * One job per chat. Every paused, done, or failed scrape of that chat becomes a member, so a
+ * scrape that was paused and resumed stays in the same job. Jobs run one at a time, batch by
+ * batch, and report progress, throughput, and an ETA to the Bot tab.
+ * See docs/media-scraper.md.
+ */
+
+import {
+    ScrapeEngine,
+    type PreviewBatchResult
+} from "./engine.js";
+import { listScrapes, ScrapeStore, scrapeDir } from "./store.js";
+import type { BackendName, MediaKind, ScrapeMode, ScrapeState } from "./types.js";
+
+export interface BackfillSettings {
+    /** Preview rows per batch. */
+    batchSize: number;
+    /** Cooldown between batches, ms. */
+    batchPauseMs: number;
+    /** Start jobs with pending work on daemon boot. */
+    auto: boolean;
+}
+
+export interface BackfillMember {
+    scrapeId: string;
+    state: ScrapeState;
+    backend: BackendName;
+    mode: ScrapeMode;
+    pending: number;
+    cached: number;
+    failed: number;
+    total: number;
+    downloaded: number;
+    bytes: number;
+    pendingByKind: Record<MediaKind, number>;
+}
+
+export interface BackfillJobStatus {
+    chatId: string;
+    guildId: string | null;
+    state: "idle" | "running" | "paused" | "done";
+    members: BackfillMember[];
+    pending: number;
+    cached: number;
+    failed: number;
+    total: number;
+    downloaded: number;
+    bytes: number;
+    /** Rows handled since this run started. */
+    processed: number;
+    startedAt: string | null;
+    finishedAt: string | null;
+    /** Measured rows per minute while running; a settings-based estimate when idle. */
+    perMinute: number;
+    /** Estimated ms to clear the pending rows. Null when there is nothing to do. */
+    etaMs: number | null;
+}
+
+export interface BackfillTotals {
+    pending: number;
+    cached: number;
+    failed: number;
+    total: number;
+    downloaded: number;
+    bytes: number;
+    perMinute: number;
+    etaMs: number | null;
+    byBackend: Record<BackendName, { pending: number; cached: number; failed: number; total: number }>;
+}
+
+export interface BackfillStatus {
+    jobs: BackfillJobStatus[];
+    totals: BackfillTotals;
+    settings: BackfillSettings;
+    /** ISO time while a Discord rate limit holds the queue. */
+    limitedUntil: string | null;
+    /** Rows per minute the current settings allow, ignoring cooldowns. */
+    estimatedPerMinute: number;
+}
+
+interface Job {
+    chatId: string;
+    guildId: string | null;
+    state: "idle" | "running" | "paused" | "done";
+    stop: boolean;
+    claimed: boolean;
+    startedAt: number | null;
+    finishedAt: number | null;
+    processed: number;
+    perMinute: number | null;
+}
+
+export interface BackfillQueueOptions {
+    rootDir: string;
+    getSettings: () => BackfillSettings;
+    /** Gap range used for the idle estimate, read from the live engine options. */
+    getGaps: () => { minGapMs: number; maxGapMs: number };
+    log?: (line: string) => void;
+    now?: () => number;
+    sleep?: (ms: number) => Promise<void>;
+}
+
+const defaultSleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms));
+
+export class BackfillQueue {
+    private readonly jobs = new Map<string, Job>();
+    private active: Promise<void> = Promise.resolve();
+    private limitedUntil = 0;
+
+    constructor(
+        private readonly engine: ScrapeEngine,
+        private readonly options: BackfillQueueOptions
+    ) {}
+
+    private now(): number {
+        return this.options.now?.() ?? Date.now();
+    }
+
+    private sleep(ms: number): Promise<void> {
+        return (this.options.sleep ?? defaultSleep)(ms);
+    }
+
+    /**
+     * One entry per chat that has any scrape left on disk, with per-kind pending counts. Active
+     * scrapes stay out: their walk caches previews inline.
+     */
+    private plan(includeBytes = true): Map<string, { guildId: string | null; members: BackfillMember[] }> {
+        const groups = new Map<string, { guildId: string | null; members: BackfillMember[] }>();
+
+        for (const meta of listScrapes(this.options.rootDir)) {
+            if (meta.state === "deleted" || meta.state === "deleting" || meta.state === "pending-delete") {
+                continue;
+            }
+
+            const member: BackfillMember = {
+                scrapeId: meta.id,
+                state: meta.state,
+                backend: meta.backend,
+                mode: meta.mode,
+                pending: 0,
+                cached: 0,
+                failed: 0,
+                total: 0,
+                downloaded: meta.downloaded ?? 0,
+                bytes: 0,
+                pendingByKind: { image: 0, gif: 0, video: 0, audio: 0 }
+            };
+
+            try {
+                const store = new ScrapeStore(scrapeDir(this.options.rootDir, meta.id));
+                try {
+                    const counts = store.previewCounts();
+                    member.pending = counts.pending;
+                    member.cached = counts.cached;
+                    member.failed = counts.failed;
+                    member.total = counts.total;
+                    for (const kind of Object.keys(member.pendingByKind) as MediaKind[]) {
+                        member.pendingByKind[kind] = counts.byKind[kind].pending;
+                    }
+                    if (includeBytes) {
+                        member.bytes = Object.values(store.storedBytesByKind()).reduce(
+                            (sum, value) => sum + value,
+                            0
+                        );
+                    }
+                } finally {
+                    store.close();
+                }
+            } catch {
+                // An unreadable index counts as nothing to backfill.
+            }
+
+            const group = groups.get(meta.channelId) ?? { guildId: meta.guildId, members: [] };
+            group.guildId = group.guildId ?? meta.guildId;
+            group.members.push(member);
+            groups.set(meta.channelId, group);
+        }
+
+        return groups;
+    }
+
+    /** The live status of every job, plus totals and the current settings. */
+    list(): BackfillStatus {
+        const settings = this.options.getSettings();
+        const gaps = this.options.getGaps();
+        const groups = this.plan();
+        const jobs: BackfillJobStatus[] = [];
+
+        for (const [chatId, group] of groups) {
+            // Hide chats that are fully backfilled unless their job did work this session.
+            const tracked = this.jobs.get(chatId);
+            const pending = group.members.reduce((sum, member) => sum + member.pending, 0);
+            if (!tracked && pending === 0) continue;
+
+            const perMinute = tracked?.perMinute ?? this.estimatedPerMinute(settings, gaps);
+            const etaMs = pending > 0 && perMinute > 0 ? Math.round((pending / perMinute) * 60_000) : null;
+
+            jobs.push({
+                chatId,
+                guildId: group.guildId,
+                state: tracked?.state ?? "idle",
+                members: group.members,
+                pending,
+                cached: group.members.reduce((sum, member) => sum + member.cached, 0),
+                failed: group.members.reduce((sum, member) => sum + member.failed, 0),
+                total: group.members.reduce((sum, member) => sum + member.total, 0),
+                downloaded: group.members.reduce((sum, member) => sum + member.downloaded, 0),
+                bytes: group.members.reduce((sum, member) => sum + member.bytes, 0),
+                processed: tracked?.processed ?? 0,
+                startedAt: tracked?.startedAt ? new Date(tracked.startedAt).toISOString() : null,
+                finishedAt: tracked?.finishedAt ? new Date(tracked.finishedAt).toISOString() : null,
+                perMinute: Math.round(perMinute * 10) / 10,
+                etaMs
+            });
+        }
+
+        jobs.sort((a, b) => b.pending - a.pending);
+
+        // Totals span every scrape on disk, not just the jobs the list shows.
+        const byBackend: BackfillTotals["byBackend"] = {
+            puppet: { pending: 0, cached: 0, failed: 0, total: 0 },
+            slopbot: { pending: 0, cached: 0, failed: 0, total: 0 }
+        };
+
+        let pending = 0;
+        let cached = 0;
+        let failed = 0;
+        let total = 0;
+        let downloaded = 0;
+        let bytes = 0;
+
+        for (const group of groups.values()) {
+            for (const member of group.members) {
+                pending += member.pending;
+                cached += member.cached;
+                failed += member.failed;
+                total += member.total;
+                downloaded += member.downloaded;
+                bytes += member.bytes;
+
+                const bucket = byBackend[member.backend] ?? byBackend.puppet;
+                bucket.pending += member.pending;
+                bucket.cached += member.cached;
+                bucket.failed += member.failed;
+                bucket.total += member.total;
+            }
+        }
+
+        const estimatedPerMinute = this.estimatedPerMinute(settings, gaps);
+        const measured = [...this.jobs.values()].find(job => job.state === "running" && job.perMinute);
+        const rate = measured?.perMinute ?? estimatedPerMinute;
+
+        return {
+            jobs,
+            totals: {
+                pending,
+                cached,
+                failed,
+                total,
+                downloaded,
+                bytes,
+                perMinute: Math.round(rate * 10) / 10,
+                etaMs: pending > 0 && rate > 0 ? Math.round((pending / rate) * 60_000) : null,
+                byBackend
+            },
+            settings,
+            limitedUntil: this.limitedUntil > 0 ? new Date(this.limitedUntil).toISOString() : null,
+            estimatedPerMinute: Math.round(estimatedPerMinute * 10) / 10
+        };
+    }
+
+    /** Rows per minute implied by the current settings, pauses and gaps included. */
+    private estimatedPerMinute(
+        settings: BackfillSettings,
+        gaps: { minGapMs: number; maxGapMs: number }
+    ): number {
+        const perRow = (gaps.minGapMs + gaps.maxGapMs) / 2 + settings.batchPauseMs / Math.max(1, settings.batchSize);
+        return perRow > 0 ? 60_000 / perRow : 6000;
+    }
+
+    /** Start every job with pending rows, or one chat's job. */
+    start(chatId?: string): void {
+        const status = this.list();
+
+        for (const job of status.jobs) {
+            if (chatId && job.chatId !== chatId) continue;
+            if (job.pending <= 0) continue;
+
+            const tracked = this.jobs.get(job.chatId) ?? {
+                chatId: job.chatId,
+                guildId: job.guildId,
+                state: "idle" as const,
+                stop: false,
+                claimed: false,
+                startedAt: null,
+                finishedAt: null,
+                processed: 0,
+                perMinute: null
+            };
+
+            if (tracked.state === "running" && !tracked.stop) continue;
+
+            tracked.stop = false;
+            tracked.claimed = false;
+            tracked.state = "running";
+            tracked.processed = 0;
+            tracked.perMinute = null;
+            tracked.startedAt = this.now();
+            tracked.finishedAt = null;
+            this.jobs.set(job.chatId, tracked);
+            this.options.log?.(`backfill: started ${job.chatId} (${job.pending} pending)`);
+        }
+
+        this.schedule();
+    }
+
+    /** Ask jobs to stop after the current batch. */
+    pause(chatId?: string): void {
+        for (const job of this.jobs.values()) {
+            if (chatId && job.chatId !== chatId) continue;
+            if (job.state !== "running") continue;
+
+            job.stop = true;
+            job.state = "paused";
+            this.options.log?.(`backfill: pausing ${job.chatId}`);
+        }
+    }
+
+    /** Resolves when every started job has stopped for now. */
+    idle(): Promise<void> {
+        return this.active;
+    }
+
+    /** Start on daemon boot when the setting allows it. */
+    autoStart(): void {
+        if (!this.options.getSettings().auto) return;
+        this.start();
+    }
+
+    private schedule(): void {
+        this.active = this.active.then(async () => {
+            for (;;) {
+                const job = [...this.jobs.values()].find(
+                    candidate => candidate.state === "running" && !candidate.claimed
+                );
+                if (!job) return;
+
+                job.claimed = true;
+                try {
+                    await this.runJob(job);
+                } finally {
+                    job.claimed = false;
+                }
+            }
+        });
+    }
+
+    private async runJob(job: Job): Promise<void> {
+        let stalled = 0;
+
+        while (!job.stop) {
+            const group = this.plan(false).get(job.chatId);
+            const member = group?.members.find(candidate => candidate.pending > 0);
+            if (!member) break;
+
+            const limit = this.engine.rateLimitStatus();
+            if (limit.active) {
+                this.limitedUntil = limit.until ?? this.now() + limit.remainingMs;
+                this.options.log?.(`backfill: waiting out a rate limit (${limit.remainingMs} ms)`);
+                await this.sleep(limit.remainingMs);
+                if (job.stop) break;
+            } else {
+                this.limitedUntil = 0;
+            }
+
+            const settings = this.options.getSettings();
+            const result: PreviewBatchResult = await this.engine.previewBatch(
+                member.scrapeId,
+                Math.max(1, settings.batchSize)
+            );
+            const progressed = result.cached + result.skipped + result.failed;
+            job.processed += progressed + result.ratelimited;
+
+            const elapsed = Math.max(1, this.now() - (job.startedAt ?? this.now()));
+            job.perMinute = job.processed / (elapsed / 60_000);
+
+            const after = this.engine.rateLimitStatus();
+            if (after.active) this.limitedUntil = after.until ?? this.now() + after.remainingMs;
+
+            if (progressed === 0) {
+                stalled++;
+                if (stalled >= 3) {
+                    this.options.log?.(`backfill: ${job.chatId} made no progress, pausing`);
+                    job.state = "paused";
+                    break;
+                }
+            } else {
+                stalled = 0;
+            }
+
+            if (settings.batchPauseMs > 0 && progressed > 0) {
+                await this.sleep(settings.batchPauseMs);
+            }
+        }
+
+        if (job.state === "running") job.state = job.stop ? "paused" : "done";
+        job.finishedAt = job.state === "done" ? this.now() : null;
+        job.perMinute = job.perMinute ?? null;
+        this.limitedUntil = 0;
+        this.options.log?.(
+            job.state === "done"
+                ? `backfill: ${job.chatId} finished`
+                : `backfill: ${job.chatId} ${job.state}`
+        );
+    }
+}

media-scrape/src/index.ts

@@ -27,6 +27,15 @@ export {
     type DownloadOptions,
     type DownloadOutcome
 } from "./download.js";
+export {
+    BackfillQueue,
+    type BackfillJobStatus,
+    type BackfillMember,
+    type BackfillQueueOptions,
+    type BackfillSettings,
+    type BackfillStatus,
+    type BackfillTotals
+} from "./backfill.js";
 export {
     ScrapeEngine,
     type FetchPage,

media-scrape/test/run.ts

@@ -19,6 +19,7 @@ import {
 } from "../src/download.js";
 import { ogImageFromHtml, resolveEmbedPreview } from "../src/embed-preview.js";
 import { ScrapeEngine, type FetchPage, type ScrapeBackend } from "../src/engine.js";
+import { BackfillQueue } from "../src/backfill.js";
 import { extractMedia } from "../src/extract.js";
 import { PuppetBackend } from "../src/puppet-backend.js";
 import { betterCandidate } from "../src/dedupe.js";
@@ -1248,6 +1249,149 @@ void (async () => {
         store.close();
     });
 
+    await test("backfill queue groups scrapes per chat and estimates an ETA", async () => {
+        const backend = new FakeBackend([
+            {
+                messages: [
+                    message("660", {
+                        attachments: [
+                            attachment({ filename: "a.png", contentType: "image/png" }),
+                            attachment({ filename: "b.png", contentType: "image/png" })
+                        ]
+                    })
+                ]
+            },
+            {
+                messages: [
+                    message("661", {
+                        attachments: [attachment({ filename: "c.png", contentType: "image/png" })]
+                    })
+                ]
+            }
+        ]);
+
+        const engine = engineFor(root, backend);
+        const first = await engine.start({ channelId: "200", guildId: null, mode: "links", backend });
+        await engine.processQueue();
+        const second = await engine.start({ channelId: "200", guildId: null, mode: "links", backend });
+        await engine.processQueue();
+
+        const queue = new BackfillQueue(engine, {
+            rootDir: root,
+            getSettings: () => ({ batchSize: 2, batchPauseMs: 0, auto: false }),
+            getGaps: () => ({ minGapMs: 0, maxGapMs: 0 }),
+            sleep: async () => {}
+        });
+
+        const status = queue.list();
+        const job = status.jobs.find(candidate => candidate.chatId === "200");
+        assert.ok(job);
+        assert.equal(job.members.length, 2);
+        assert.equal(job.pending, 3);
+        assert.ok((job.etaMs ?? 0) > 0);
+        assert.ok(status.totals.byBackend.puppet.pending >= 3);
+        assert.equal(status.totals.byBackend.slopbot.pending, 0);
+        assert.ok(status.totals.total >= 3);
+        assert.equal(job.members.some(member => member.scrapeId === first.id), true);
+        assert.equal(job.members.some(member => member.scrapeId === second.id), true);
+    });
+
+    await test("backfill queue processes batches and finishes", async () => {
+        const backend = new FakeBackend([
+            {
+                messages: [
+                    message("662", {
+                        attachments: [
+                            attachment({ filename: "x.png", contentType: "image/png" }),
+                            attachment({ filename: "y.png", contentType: "image/png" }),
+                            attachment({ filename: "z.png", contentType: "image/png" })
+                        ]
+                    })
+                ]
+            }
+        ]);
+
+        const engine = engineFor(root, backend, {
+            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: "201", guildId: null, mode: "links", backend });
+        await engine.processQueue();
+
+        const queue = new BackfillQueue(engine, {
+            rootDir: root,
+            getSettings: () => ({ batchSize: 1, batchPauseMs: 0, auto: false }),
+            getGaps: () => ({ minGapMs: 0, maxGapMs: 0 }),
+            sleep: async () => {}
+        });
+
+        queue.start();
+        await queue.idle();
+
+        const status = queue.list();
+        const job = status.jobs.find(candidate => candidate.chatId === "201");
+        assert.ok(job);
+        assert.equal(job.state, "done");
+        assert.equal(job.pending, 0);
+        assert.equal(job.cached, 3);
+        assert.ok(job.processed >= 3);
+
+        const store = new ScrapeStore(join(root, started.id));
+        assert.equal(store.previewCounts().cached, 3);
+        store.close();
+    });
+
+    await test("backfill queue pause leaves rows pending", async () => {
+        const backend = new FakeBackend([
+            {
+                messages: [
+                    message("663", {
+                        attachments: [
+                            attachment({ filename: "p.png", contentType: "image/png" }),
+                            attachment({ filename: "q.png", contentType: "image/png" })
+                        ]
+                    })
+                ]
+            }
+        ]);
+
+        const engine = engineFor(root, backend, {
+            download: {
+                fetchImpl: async () =>
+                    new Response("png", { status: 200, headers: { "content-type": "image/png" } }),
+                minGapMs: 0,
+                maxGapMs: 0,
+                sleep: async () => {}
+            }
+        });
+
+        await engine.start({ channelId: "202", guildId: null, mode: "links", backend });
+        await engine.processQueue();
+
+        const queue = new BackfillQueue(engine, {
+            rootDir: root,
+            getSettings: () => ({ batchSize: 1, batchPauseMs: 0, auto: false }),
+            getGaps: () => ({ minGapMs: 0, maxGapMs: 0 }),
+            sleep: async () => {}
+        });
+
+        queue.start();
+        queue.pause();
+        await queue.idle();
+
+        const status = queue.list();
+        const job = status.jobs.find(candidate => candidate.chatId === "202");
+        assert.ok(job);
+        assert.equal(job.state, "paused");
+        assert.equal(job.pending, 2);
+    });
+
     await test("backfillPreviews fills rows left by a preview-less walk", async () => {
         const backend = new FakeBackend([
             {