Add the media scrape engine, store, and puppet walker
This commit is contained in:
Fisto
@@ -0,0 +1,268 @@
|
||||
/*
|
||||
* The scrape engine: queue, walk, pacing, budgets, backoff, resume.
|
||||
* Rules: docs/media-scraper.md.
|
||||
*/
|
||||
|
||||
import { join } from "node:path";
|
||||
import {
|
||||
createScrapeDirs,
|
||||
listScrapes,
|
||||
readScrapeMeta,
|
||||
ScrapeStore,
|
||||
uniqueScrapeId,
|
||||
writeScrapeMeta
|
||||
} from "./store.js";
|
||||
import {
|
||||
emptyMediaCounts,
|
||||
MEDIA_KINDS,
|
||||
type BackendName,
|
||||
type CoreMessage,
|
||||
type ScrapeMeta,
|
||||
type ScrapeMode
|
||||
} from "./types.js";
|
||||
|
||||
export interface FetchPage {
|
||||
messages: CoreMessage[];
|
||||
/** Set by a backend that saw a rate-limit hint. The engine sleeps before the next page. */
|
||||
rateLimitedMs?: number;
|
||||
}
|
||||
|
||||
export interface ScrapeBackend {
|
||||
readonly name: BackendName;
|
||||
fetchPage(channelId: string, before: string | null, limit: number): Promise<FetchPage>;
|
||||
}
|
||||
|
||||
export interface ScrapeEngineOptions {
|
||||
rootDir: string;
|
||||
pageSize?: number;
|
||||
pageMinMs?: number;
|
||||
pageMaxMs?: number;
|
||||
pagesPerHour?: number;
|
||||
pagesPerDay?: number;
|
||||
maxRetries?: number;
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
random?: () => number;
|
||||
now?: () => number;
|
||||
log?: (line: string) => void;
|
||||
}
|
||||
|
||||
interface QueuedJob {
|
||||
meta: ScrapeMeta;
|
||||
dir: string;
|
||||
backend: ScrapeBackend;
|
||||
}
|
||||
|
||||
const defaultSleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms));
|
||||
|
||||
function oldestId(messages: CoreMessage[]): string | null {
|
||||
let oldest: string | null = null;
|
||||
for (const message of messages) {
|
||||
if (oldest === null || BigInt(message.id) < BigInt(oldest)) oldest = message.id;
|
||||
}
|
||||
return oldest;
|
||||
}
|
||||
|
||||
export class ScrapeEngine {
|
||||
private readonly queue: QueuedJob[] = [];
|
||||
private readonly active = new Map<string, ScrapeMeta>();
|
||||
private readonly stopRequested = new Set<string>();
|
||||
private readonly pageTimes: number[] = [];
|
||||
private processingPromise: Promise<void> | null = null;
|
||||
|
||||
constructor(private readonly options: ScrapeEngineOptions) {}
|
||||
|
||||
/** Create a scrape and queue the walk. Returns immediately. */
|
||||
async start(input: {
|
||||
channelId: string;
|
||||
guildId: string | null;
|
||||
mode: ScrapeMode;
|
||||
backend: ScrapeBackend;
|
||||
}): Promise<ScrapeMeta> {
|
||||
const now = this.now();
|
||||
const id = uniqueScrapeId(this.options.rootDir, input.channelId, new Date(now));
|
||||
const dir = createScrapeDirs(this.options.rootDir, id);
|
||||
const timestamp = new Date(now).toISOString();
|
||||
|
||||
const meta: ScrapeMeta = {
|
||||
id,
|
||||
channelId: input.channelId,
|
||||
guildId: input.guildId,
|
||||
backend: input.backend.name,
|
||||
mode: input.mode,
|
||||
state: "queued",
|
||||
cursor: null,
|
||||
createdAt: timestamp,
|
||||
updatedAt: timestamp,
|
||||
pages: 0,
|
||||
messages: 0,
|
||||
media: emptyMediaCounts(),
|
||||
errors: 0,
|
||||
lastError: null
|
||||
};
|
||||
|
||||
writeScrapeMeta(dir, meta);
|
||||
this.active.set(id, meta);
|
||||
this.queue.push({ meta, dir, backend: input.backend });
|
||||
void this.processQueue();
|
||||
|
||||
return { ...meta };
|
||||
}
|
||||
|
||||
/** Run queued jobs until the queue is empty. Tests await this; the daemon lets it run. */
|
||||
processQueue(): Promise<void> {
|
||||
if (!this.processingPromise) {
|
||||
this.processingPromise = this.drainQueue().finally(() => {
|
||||
this.processingPromise = null;
|
||||
});
|
||||
}
|
||||
return this.processingPromise;
|
||||
}
|
||||
|
||||
private async drainQueue(): Promise<void> {
|
||||
while (this.queue.length > 0) {
|
||||
const job = this.queue.shift();
|
||||
if (!job) break;
|
||||
await this.runJob(job);
|
||||
}
|
||||
}
|
||||
|
||||
get(id: string): ScrapeMeta | null {
|
||||
const active = this.active.get(id);
|
||||
if (active) return { ...active };
|
||||
return readScrapeMeta(join(this.options.rootDir, id));
|
||||
}
|
||||
|
||||
/** Every scrape on disk, merged with the jobs this engine still holds in memory. */
|
||||
list(): ScrapeMeta[] {
|
||||
const merged = new Map<string, ScrapeMeta>();
|
||||
for (const meta of listScrapes(this.options.rootDir)) merged.set(meta.id, meta);
|
||||
for (const [id, meta] of this.active) merged.set(id, meta);
|
||||
|
||||
return [...merged.values()].sort((a, b) => (a.createdAt < b.createdAt ? 1 : -1));
|
||||
}
|
||||
|
||||
/** Ask a running or queued scrape to stop. It pauses after the current page. */
|
||||
stop(id: string): void {
|
||||
this.stopRequested.add(id);
|
||||
}
|
||||
|
||||
private async runJob(job: QueuedJob): Promise<void> {
|
||||
const { meta, dir, backend } = job;
|
||||
const store = new ScrapeStore(dir);
|
||||
const pageSize = this.options.pageSize ?? 100;
|
||||
|
||||
try {
|
||||
if (this.stopRequested.has(meta.id)) {
|
||||
meta.state = "paused";
|
||||
return;
|
||||
}
|
||||
|
||||
meta.state = "running";
|
||||
this.persist(dir, meta);
|
||||
|
||||
for (;;) {
|
||||
if (this.stopRequested.has(meta.id)) {
|
||||
meta.state = "paused";
|
||||
break;
|
||||
}
|
||||
|
||||
if (!this.allowPage()) {
|
||||
meta.state = "paused";
|
||||
this.log(`${meta.id}: page budget reached, paused`);
|
||||
break;
|
||||
}
|
||||
|
||||
const page = await this.fetchWithBackoff(meta, backend);
|
||||
if (page.rateLimitedMs) await this.sleep(page.rateLimitedMs);
|
||||
|
||||
if (page.messages.length === 0) {
|
||||
meta.state = "done";
|
||||
break;
|
||||
}
|
||||
|
||||
const added = store.insertPage(page.messages);
|
||||
for (const kind of MEDIA_KINDS) meta.media[kind] += added[kind];
|
||||
meta.pages++;
|
||||
meta.messages += page.messages.length;
|
||||
meta.cursor = oldestId(page.messages) ?? meta.cursor;
|
||||
this.pageTimes.push(this.now());
|
||||
this.persist(dir, meta);
|
||||
|
||||
if (page.messages.length < pageSize) {
|
||||
meta.state = "done";
|
||||
break;
|
||||
}
|
||||
|
||||
await this.sleep(this.gapMs());
|
||||
}
|
||||
} catch (error) {
|
||||
meta.state = "failed";
|
||||
meta.lastError = error instanceof Error ? error.message : String(error);
|
||||
this.log(`${meta.id}: failed: ${meta.lastError}`);
|
||||
} finally {
|
||||
this.persist(dir, meta);
|
||||
store.close();
|
||||
this.active.set(meta.id, meta);
|
||||
}
|
||||
}
|
||||
|
||||
private async fetchWithBackoff(meta: ScrapeMeta, backend: ScrapeBackend): Promise<FetchPage> {
|
||||
const pageSize = this.options.pageSize ?? 100;
|
||||
const maxRetries = this.options.maxRetries ?? 8;
|
||||
|
||||
let attempt = 0;
|
||||
for (;;) {
|
||||
try {
|
||||
return await backend.fetchPage(meta.channelId, meta.cursor, pageSize);
|
||||
} catch (error) {
|
||||
meta.errors++;
|
||||
if (attempt >= maxRetries) throw error;
|
||||
|
||||
const retryAfter = (error as { retryAfterMs?: number }).retryAfterMs;
|
||||
const delay = retryAfter ?? Math.pow(2, attempt) * 1000 + 1000;
|
||||
this.log(`${meta.id}: page failed, retrying in ${delay} ms`);
|
||||
await this.sleep(delay);
|
||||
attempt++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private allowPage(): boolean {
|
||||
const now = this.now();
|
||||
while (this.pageTimes.length > 0 && this.pageTimes[0] < now - 86_400_000) {
|
||||
this.pageTimes.shift();
|
||||
}
|
||||
|
||||
const lastHour = this.pageTimes.filter(time => time > now - 3_600_000).length;
|
||||
if (lastHour >= (this.options.pagesPerHour ?? 300)) return false;
|
||||
if (this.pageTimes.length >= (this.options.pagesPerDay ?? 3000)) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
private gapMs(): number {
|
||||
const min = this.options.pageMinMs ?? 1200;
|
||||
const max = Math.max(min, this.options.pageMaxMs ?? 3500);
|
||||
return min + Math.floor(this.random() * (max - min + 1));
|
||||
}
|
||||
|
||||
private persist(dir: string, meta: ScrapeMeta): void {
|
||||
meta.updatedAt = new Date(this.now()).toISOString();
|
||||
writeScrapeMeta(dir, meta);
|
||||
}
|
||||
|
||||
private now(): number {
|
||||
return (this.options.now ?? Date.now)();
|
||||
}
|
||||
|
||||
private random(): number {
|
||||
return (this.options.random ?? Math.random)();
|
||||
}
|
||||
|
||||
private sleep(ms: number): Promise<void> {
|
||||
return (this.options.sleep ?? defaultSleep)(ms);
|
||||
}
|
||||
|
||||
private log(line: string): void {
|
||||
this.options.log?.(line);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user