Add the media downloader with hash storage and local file serving

media-scrape/src/download.ts

@@ -0,0 +1,129 @@
+/*
+ * Attachment downloader for the media archive. Serial, hash-addressed, conservative.
+ * Rules: docs/media-scraper.md. Third-party embeds are never targets here.
+ */
+
+import { createHash } from "node:crypto";
+import { writeFileSync } from "node:fs";
+import { join } from "node:path";
+import type { MediaRow } from "./store.js";
+
+export interface DownloadOptions {
+    fetchImpl?: typeof fetch;
+    /** Skip files larger than this. Default 50 MiB. */
+    maxFileBytes?: number;
+    /** Gap between files. Default 1000-3000 ms. */
+    minGapMs?: number;
+    maxGapMs?: number;
+    /** Retries after the first attempt, for network and 5xx errors only. */
+    maxRetries?: number;
+    sleep?: (ms: number) => Promise<void>;
+    random?: () => number;
+    log?: (line: string) => void;
+}
+
+export interface DownloadOutcome {
+    id: number;
+    state: "downloaded" | "skipped" | "failed";
+    path?: string;
+    hash?: string;
+    bytes?: number;
+    error?: string;
+}
+
+/** 4xx that will not improve on retry: the signed URL expired or the file is gone. */
+const SKIP_STATUS = new Set([400, 401, 403, 404, 410]);
+
+const TYPE_EXTENSIONS: Record<string, string> = {
+    "image/png": "png",
+    "image/jpeg": "jpg",
+    "image/gif": "gif",
+    "image/webp": "webp",
+    "image/avif": "avif",
+    "video/mp4": "mp4",
+    "video/webm": "webm",
+    "video/quicktime": "mov",
+    "video/x-matroska": "mkv",
+    "audio/mpeg": "mp3",
+    "audio/ogg": "ogg",
+    "audio/wav": "wav",
+    "audio/x-wav": "wav",
+    "audio/mp4": "m4a",
+    "audio/flac": "flac"
+};
+
+function extensionFor(contentType: string | null, url: string): string {
+    const type = (contentType ?? "").split(";")[0].trim().toLowerCase();
+    if (TYPE_EXTENSIONS[type]) return TYPE_EXTENSIONS[type];
+
+    try {
+        const match = /\.([a-z0-9]{2,5})$/i.exec(new URL(url).pathname);
+        if (match) return match[1].toLowerCase();
+    } catch {
+        // fall through
+    }
+    return "bin";
+}
+
+const defaultSleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms));
+
+/** Download one media row into `dir`. The file name is the SHA-256 of the bytes. */
+export async function downloadOne(
+    media: Pick<MediaRow, "id" | "url" | "kind">,
+    dir: string,
+    options: DownloadOptions = {}
+): Promise<DownloadOutcome> {
+    const fetchImpl = options.fetchImpl ?? fetch;
+    const maxBytes = options.maxFileBytes ?? 50 * 1024 * 1024;
+    const maxRetries = options.maxRetries ?? 2;
+    const sleep = options.sleep ?? defaultSleep;
+
+    let lastError = "download failed";
+
+    for (let attempt = 0; attempt <= maxRetries; attempt++) {
+        try {
+            const response = await fetchImpl(media.url);
+
+            if (!response.ok) {
+                if (SKIP_STATUS.has(response.status)) {
+                    return { id: media.id, state: "skipped", error: `http ${response.status}` };
+                }
+                lastError = `http ${response.status}`;
+            } else {
+                const declared = Number(response.headers.get("content-length") ?? 0);
+                if (declared > maxBytes) {
+                    return { id: media.id, state: "skipped", error: `too large (${declared} bytes)` };
+                }
+
+                const buffer = Buffer.from(await response.arrayBuffer());
+                if (buffer.byteLength > maxBytes) {
+                    return { id: media.id, state: "skipped", error: "too large" };
+                }
+
+                const hash = createHash("sha256").update(buffer).digest("hex");
+                const extension = extensionFor(response.headers.get("content-type"), media.url);
+                const file = `${hash}.${extension}`;
+                writeFileSync(join(dir, file), buffer);
+
+                return { id: media.id, state: "downloaded", path: join("media", file), hash, bytes: buffer.byteLength };
+            }
+        } catch (error) {
+            lastError = error instanceof Error ? error.message : String(error);
+        }
+
+        if (attempt < maxRetries) {
+            options.log?.(`download ${media.id} failed (${lastError}), retrying`);
+            await sleep(2 ** attempt * 1000 + 1000);
+        }
+    }
+
+    return { id: media.id, state: "failed", error: lastError };
+}
+
+/** The gap before the next file, in milliseconds. */
+export function downloadGapMs(options: DownloadOptions = {}): number {
+    const min = options.minGapMs ?? 1000;
+    const max = Math.max(min, options.maxGapMs ?? 3000);
+    const random = options.random ?? Math.random;
+    return min + Math.floor(random() * (max - min + 1));
+}

media-scrape/src/engine.ts

@@ -4,6 +4,7 @@
  */
 
 import { join } from "node:path";
+import { downloadGapMs, downloadOne, type DownloadOptions } from "./download.js";
 import {
     createScrapeDirs,
     listScrapes,
@@ -45,6 +46,8 @@ export interface ScrapeEngineOptions {
     random?: () => number;
     now?: () => number;
     log?: (line: string) => void;
+    /** Download settings for `mode: "download"`. */
+    download?: DownloadOptions;
 }
 
 interface QueuedJob {
@@ -214,6 +217,10 @@ export class ScrapeEngine {
 
                 await this.sleep(this.gapMs());
             }
+
+            if (meta.state === "done" && meta.mode === "download") {
+                await this.runDownloads(store, dir, meta);
+            }
         } catch (error) {
             meta.state = "failed";
             meta.lastError = error instanceof Error ? error.message : String(error);
@@ -225,6 +232,39 @@ export class ScrapeEngine {
         }
     }
 
+    /** Fetch attachment bytes for a finished walk. Serial, hash-addressed, resumable. */
+    private async runDownloads(store: ScrapeStore, dir: string, meta: ScrapeMeta): Promise<void> {
+        const options = this.options.download ?? {};
+        const pending = store.pendingDownloads();
+
+        for (const row of pending) {
+            if (this.stopRequested.has(meta.id)) {
+                meta.state = "paused";
+                return;
+            }
+
+            const outcome = await downloadOne(row, join(dir, "media"), {
+                ...options,
+                log: line => this.log(`${meta.id}: ${line}`)
+            });
+
+            if (outcome.state === "downloaded" && outcome.path && outcome.hash) {
+                store.markDownloaded(row.id, {
+                    path: outcome.path,
+                    hash: outcome.hash,
+                    bytes: outcome.bytes ?? 0
+                });
+                meta.downloaded = (meta.downloaded ?? 0) + 1;
+            } else {
+                store.markDownloadError(row.id, outcome.error ?? outcome.state);
+                if (outcome.state === "failed") meta.downloadErrors = (meta.downloadErrors ?? 0) + 1;
+            }
+
+            this.persist(dir, meta);
+            await this.sleep(downloadGapMs(options));
+        }
+    }
+
     private async fetchWithBackoff(meta: ScrapeMeta, backend: ScrapeBackend): Promise<FetchPage> {
         const pageSize = this.options.pageSize ?? 100;
         const maxRetries = this.options.maxRetries ?? 8;

media-scrape/src/extract.ts

@@ -13,10 +13,14 @@ const AUDIO_EXT = new Set(["mp3", "m4a", "ogg", "wav", "flac"]);
 /** Sticker formatType to file extension. 3 is Lottie JSON; not archived. */
 const STICKER_EXT: Record<number, string | undefined> = { 1: "png", 2: "apng", 4: "gif" };
 
+/** Where the item came from. Embeds are never downloaded. */
+export type MediaSource = "attachment" | "sticker" | "embed";
+
 /** A media item extracted from a message. `url` is the original source link. */
 export interface ExtractedMedia {
     kind: MediaKind;
     url: string;
+    source: MediaSource;
     provider: string | null;
     contentType: string | null;
     size: number | null;
@@ -68,6 +72,7 @@ function hostname(url: string | null): string | null {
 function extractEmbed(embed: CoreEmbed): ExtractedMedia | null {
     const provider = embed.provider ?? hostname(embed.url);
     const common = {
+        source: "embed" as const,
         provider,
         downloaded: false,
         size: null,
@@ -117,6 +122,7 @@ function extractSticker(sticker: CoreSticker): ExtractedMedia | null {
 
     return {
         kind: ext === "gif" ? "gif" : "image",
+        source: "sticker",
         url: `https://cdn.discordapp.com/stickers/${sticker.id}.${ext}`,
         provider: "discord",
         contentType: null,
@@ -138,6 +144,7 @@ function extractFrom(message: MediaMessage): ExtractedMedia[] {
         if (!kind) continue;
         items.push({
             kind,
+            source: "attachment",
             url: attachment.url,
             provider: "discord",
             contentType: attachment.contentType,

media-scrape/src/index.ts

@@ -5,6 +5,7 @@
 
 export { extractMedia, type ExtractedMedia } from "./extract.js";
 export { isNearLimit, planRetention, type RetentionPlan, type RetentionScrape } from "./retention.js";
+export { downloadGapMs, downloadOne, type DownloadOptions, type DownloadOutcome } from "./download.js";
 export {
     ScrapeEngine,
     type FetchPage,

media-scrape/src/store.ts

@@ -98,6 +98,7 @@ export function uniqueScrapeId(root: string, channelId: string, now = new Date()
 export interface MediaRow {
     id: number;
     kind: MediaKind;
+    source: string | null;
     url: string;
     provider: string | null;
     content_type: string | null;
@@ -109,6 +110,10 @@ export interface MediaRow {
     preview_url: string | null;
     file_hash: string | null;
     preview_hash: string | null;
+    file_bytes: number | null;
+    preview_bytes: number | null;
+    file_path: string | null;
+    download_error: string | null;
     created_at: string;
 }
 
@@ -124,6 +129,7 @@ CREATE TABLE IF NOT EXISTS messages (
 CREATE TABLE IF NOT EXISTS media (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     kind TEXT NOT NULL,
+    source TEXT,
     url TEXT NOT NULL UNIQUE,
     provider TEXT,
     content_type TEXT,
@@ -137,6 +143,8 @@ CREATE TABLE IF NOT EXISTS media (
     preview_hash TEXT,
     file_bytes INTEGER,
     preview_bytes INTEGER,
+    file_path TEXT,
+    download_error TEXT,
     created_at TEXT NOT NULL
 );
 
@@ -167,6 +175,9 @@ export class ScrapeStore {
 
         if (!names.has("file_bytes")) this.db.exec("ALTER TABLE media ADD COLUMN file_bytes INTEGER");
         if (!names.has("preview_bytes")) this.db.exec("ALTER TABLE media ADD COLUMN preview_bytes INTEGER");
+        if (!names.has("source")) this.db.exec("ALTER TABLE media ADD COLUMN source TEXT");
+        if (!names.has("file_path")) this.db.exec("ALTER TABLE media ADD COLUMN file_path TEXT");
+        if (!names.has("download_error")) this.db.exec("ALTER TABLE media ADD COLUMN download_error TEXT");
     }
 
     close(): void {
@@ -180,7 +191,7 @@ export class ScrapeStore {
             "INSERT OR REPLACE INTO messages (id, author_id, author_name, timestamp, content) VALUES (?, ?, ?, ?, ?)"
         );
         const insertMedia = this.db.prepare(
-            "INSERT OR IGNORE INTO media (kind, url, provider, content_type, size, width, height, duration_secs, downloaded, preview_url, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR IGNORE INTO media (kind, source, url, provider, content_type, size, width, height, duration_secs, downloaded, preview_url, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
         );
         const findMedia = this.db.prepare("SELECT id FROM media WHERE url = ?");
         const linkSource = this.db.prepare(
@@ -202,6 +213,7 @@ export class ScrapeStore {
                 for (const item of extractMedia(message)) {
                     const result = insertMedia.run(
                         item.kind,
+                        item.source,
                         item.url,
                         item.provider,
                         item.contentType,
@@ -255,6 +267,30 @@ export class ScrapeStore {
         return result;
     }
 
+    /** Media rows that still need bytes: never downloaded and not skipped. */
+    pendingDownloads(limit = 5000): MediaRow[] {
+        return this.db
+            .prepare(
+                "SELECT * FROM media WHERE downloaded = 0 AND file_path IS NULL AND download_error IS NULL AND (source IS NULL OR source != 'embed') ORDER BY id LIMIT ?"
+            )
+            .all(limit) as unknown as MediaRow[];
+    }
+
+    markDownloaded(
+        id: number,
+        file: { path: string; hash: string; bytes: number }
+    ): void {
+        this.db
+            .prepare(
+                "UPDATE media SET downloaded = 1, file_path = ?, file_hash = ?, file_bytes = ?, download_error = NULL WHERE id = ?"
+            )
+            .run(file.path, file.hash, file.bytes, id);
+    }
+
+    markDownloadError(id: number, error: string): void {
+        this.db.prepare("UPDATE media SET download_error = ? WHERE id = ?").run(error.slice(0, 200), id);
+    }
+
     getMedia(id: number): MediaRow | null {
         const row = this.db.prepare("SELECT * FROM media WHERE id = ?").get(id) as unknown as
             | MediaRow

media-scrape/src/types.ts

@@ -83,6 +83,10 @@ export interface ScrapeMeta {
     media: Record<MediaKind, number>;
     errors: number;
     lastError: string | null;
+    /** Files fetched in download mode. */
+    downloaded?: number;
+    /** Files that failed after retries; skipped URLs are not counted here. */
+    downloadErrors?: number;
 }
 
 export function emptyMediaCounts(): Record<MediaKind, number> {