Run the title catch-up in its own YouTube lane
This commit is contained in:
@@ -0,0 +1,202 @@
|
||||
/*
|
||||
* The YouTube title catch-up lane.
|
||||
*
|
||||
* Titles normally arrive with the scrape (Discord embeds carry them), so this lane only exists
|
||||
* for rows indexed before titles were stored. It talks to YouTube's oEmbed directly, never to
|
||||
* Discord, so it takes no media-traffic lock and can run beside a walk. Pacing is hard-coded
|
||||
* (100 rows per batch, 10 s between batches): the lane is a one-time catch-up, not a tuned knob.
|
||||
*/
|
||||
|
||||
import { ScrapeEngine } from "./engine.js";
|
||||
import { listScrapes, ScrapeStore, scrapeDir } from "./store.js";
|
||||
|
||||
export interface TitleLaneStatus {
|
||||
state: "idle" | "running" | "paused" | "done";
|
||||
pending: number;
|
||||
resolved: number;
|
||||
failed: number;
|
||||
/** Rows handled since the lane started. */
|
||||
processed: number;
|
||||
startedAt: string | null;
|
||||
finishedAt: string | null;
|
||||
perMinute: number | null;
|
||||
etaMs: number | null;
|
||||
/** The scrape being worked on right now. */
|
||||
currentScrapeId: string | null;
|
||||
}
|
||||
|
||||
export interface TitleLaneOptions {
|
||||
rootDir: string;
|
||||
log?: (line: string) => void;
|
||||
now?: () => number;
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
/** Hard-coded in production; tests override. */
|
||||
batchSize?: number;
|
||||
batchPauseMs?: number;
|
||||
}
|
||||
|
||||
const BATCH_SIZE = 100;
|
||||
const BATCH_PAUSE_MS = 10_000;
|
||||
|
||||
interface LaneState {
|
||||
state: TitleLaneStatus["state"];
|
||||
stop: boolean;
|
||||
startedAt: number | null;
|
||||
finishedAt: number | null;
|
||||
processed: number;
|
||||
perMinute: number | null;
|
||||
currentScrapeId: string | null;
|
||||
}
|
||||
|
||||
const defaultSleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms));
|
||||
|
||||
export class TitleLane {
|
||||
private readonly state: LaneState = {
|
||||
state: "idle",
|
||||
stop: false,
|
||||
startedAt: null,
|
||||
finishedAt: null,
|
||||
processed: 0,
|
||||
perMinute: null,
|
||||
currentScrapeId: null
|
||||
};
|
||||
|
||||
private active: Promise<void> = Promise.resolve();
|
||||
|
||||
constructor(
|
||||
private readonly engine: ScrapeEngine,
|
||||
private readonly options: TitleLaneOptions
|
||||
) {}
|
||||
|
||||
private now(): number {
|
||||
return this.options.now?.() ?? Date.now();
|
||||
}
|
||||
|
||||
private sleep(ms: number): Promise<void> {
|
||||
return (this.options.sleep ?? defaultSleep)(ms);
|
||||
}
|
||||
|
||||
/** Totals across every scrape, with the lane's own run state. */
|
||||
status(): TitleLaneStatus {
|
||||
let pending = 0;
|
||||
let resolved = 0;
|
||||
let failed = 0;
|
||||
|
||||
for (const meta of listScrapes(this.options.rootDir)) {
|
||||
if (meta.state === "deleted" || meta.state === "deleting" || meta.state === "pending-delete") {
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const store = new ScrapeStore(scrapeDir(this.options.rootDir, meta.id));
|
||||
try {
|
||||
const counts = store.previewCounts();
|
||||
pending += counts.titlePending;
|
||||
resolved += counts.titled ?? 0;
|
||||
failed += counts.titleFailed ?? 0;
|
||||
} finally {
|
||||
store.close();
|
||||
}
|
||||
} catch {
|
||||
// an unreadable index counts as nothing to do
|
||||
}
|
||||
}
|
||||
|
||||
const runMs = this.state.startedAt ? Math.max(1, this.now() - this.state.startedAt) : 0;
|
||||
const perMinute =
|
||||
this.state.perMinute ??
|
||||
(runMs && this.state.processed ? this.state.processed / (runMs / 60_000) : null);
|
||||
|
||||
return {
|
||||
state: this.state.state,
|
||||
pending,
|
||||
resolved,
|
||||
failed,
|
||||
processed: this.state.processed,
|
||||
startedAt: this.state.startedAt ? new Date(this.state.startedAt).toISOString() : null,
|
||||
finishedAt: this.state.finishedAt ? new Date(this.state.finishedAt).toISOString() : null,
|
||||
perMinute: perMinute === null ? null : Math.round(perMinute * 10) / 10,
|
||||
etaMs:
|
||||
pending > 0 && perMinute && perMinute > 0
|
||||
? Math.round((pending / perMinute) * 60_000)
|
||||
: null,
|
||||
currentScrapeId: this.state.currentScrapeId
|
||||
};
|
||||
}
|
||||
|
||||
start(): void {
|
||||
this.state.stop = false;
|
||||
this.state.state = "running";
|
||||
this.state.startedAt = this.state.startedAt ?? this.now();
|
||||
this.state.finishedAt = null;
|
||||
this.options.log?.("titles: lane started");
|
||||
this.schedule();
|
||||
}
|
||||
|
||||
pause(): void {
|
||||
if (this.state.state !== "running") return;
|
||||
this.state.stop = true;
|
||||
this.state.state = "paused";
|
||||
this.options.log?.("titles: pausing");
|
||||
}
|
||||
|
||||
/** Resolves when the lane has stopped for now. */
|
||||
idle(): Promise<void> {
|
||||
return this.active;
|
||||
}
|
||||
|
||||
private schedule(): void {
|
||||
this.active = this.active.then(async () => {
|
||||
if (this.state.state !== "running") return;
|
||||
await this.run();
|
||||
});
|
||||
}
|
||||
|
||||
private async run(): Promise<void> {
|
||||
const batchSize = Math.max(1, this.options.batchSize ?? BATCH_SIZE);
|
||||
const batchPauseMs = Math.max(0, this.options.batchPauseMs ?? BATCH_PAUSE_MS);
|
||||
|
||||
while (!this.state.stop) {
|
||||
const meta = listScrapes(this.options.rootDir).find(candidate => {
|
||||
if (candidate.state === "deleted" || candidate.state === "deleting") return false;
|
||||
try {
|
||||
const store = new ScrapeStore(scrapeDir(this.options.rootDir, candidate.id));
|
||||
try {
|
||||
return store.previewCounts().titlePending > 0;
|
||||
} finally {
|
||||
store.close();
|
||||
}
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
});
|
||||
|
||||
if (!meta) break;
|
||||
|
||||
this.state.currentScrapeId = meta.id;
|
||||
const result = await this.engine.titleBatch(meta.id, batchSize);
|
||||
this.state.processed += result.titled + result.failed;
|
||||
|
||||
const elapsed = Math.max(1, this.now() - (this.state.startedAt ?? this.now()));
|
||||
this.state.perMinute = this.state.processed / (elapsed / 60_000);
|
||||
|
||||
this.options.log?.(
|
||||
`titles: ${meta.id} batch done: +${result.titled} titled, ${result.failed} closed, ` +
|
||||
`${result.remaining} left in scrape`
|
||||
);
|
||||
|
||||
if (this.state.stop) break;
|
||||
if (result.stopped) {
|
||||
this.state.state = "paused";
|
||||
break;
|
||||
}
|
||||
|
||||
if (batchPauseMs > 0) await this.sleep(batchPauseMs);
|
||||
}
|
||||
|
||||
this.state.currentScrapeId = null;
|
||||
if (this.state.state === "running") this.state.state = this.state.stop ? "paused" : "done";
|
||||
this.state.finishedAt = this.state.state === "done" ? this.now() : null;
|
||||
this.options.log?.(`titles: lane ${this.state.state} (${this.state.processed} rows this run)`);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user