diff --git a/CURRENT.md b/CURRENT.md index 4a0cc6e..7fd9a5d 100644 --- a/CURRENT.md +++ b/CURRENT.md @@ -26,7 +26,7 @@ mobile draft at 390px. Checkpoints, push after each: - [x] C1 design docs, AGENTS/INDEX pointers, `.gitignore` - [x] C2 core message shape: embeds, forwards, attachment metadata - - [ ] C3 media-scrape store and puppet walker, tests + - [x] C3 media-scrape store and puppet walker, tests - [ ] C4 slopbot backend, tests - [ ] C5 Web UI Media and Bot tabs, API routes - [ ] C6 storage limit, warning flow, clean modal, settings diff --git a/docs/media.md b/docs/media.md index 78e6989..109c9e8 100644 --- a/docs/media.md +++ b/docs/media.md @@ -59,6 +59,8 @@ media-scrape/db/ - The database is an index. It holds the scrape state, the messages seen, the media items, and the link between a media item and its source messages. Source channel, author, and timestamp are indexed so filters can be added later without a re-scrape. +- Storage uses `node:sqlite` (experimental in Node 22, prints a warning). If its API changes, + `better-sqlite3` is the fallback; it needs a native build, so that switch is an owner install. - The file name in `media/` is the SHA-256 of the bytes, never the original name. The original name is metadata. - Deleting a scrape is deleting its directory. Nothing is shared between scrapes. diff --git a/docs/testing.md b/docs/testing.md index 0ab0e74..f963865 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -14,10 +14,11 @@ cd poc pnpm test:comm pnpm test:cache pnpm test:audio +pnpm test:media ``` -The files are `poc/src/test-communication.ts` and `poc/src/test-cache.ts`. The tests do not need -Discord or Vesktop. +The files are `poc/src/test-communication.ts`, `poc/src/test-cache.ts`, and +`media-scrape/test/run.ts` (media archive). The tests do not need Discord or Vesktop. ## Type checks diff --git a/media-scrape/src/engine.ts b/media-scrape/src/engine.ts new file mode 100644 index 0000000..e730084 --- /dev/null +++ b/media-scrape/src/engine.ts @@ -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; +} + +export interface ScrapeEngineOptions { + rootDir: string; + pageSize?: number; + pageMinMs?: number; + pageMaxMs?: number; + pagesPerHour?: number; + pagesPerDay?: number; + maxRetries?: number; + sleep?: (ms: number) => Promise; + random?: () => number; + now?: () => number; + log?: (line: string) => void; +} + +interface QueuedJob { + meta: ScrapeMeta; + dir: string; + backend: ScrapeBackend; +} + +const defaultSleep = (ms: number) => new Promise(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(); + private readonly stopRequested = new Set(); + private readonly pageTimes: number[] = []; + private processingPromise: Promise | 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 { + 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 { + if (!this.processingPromise) { + this.processingPromise = this.drainQueue().finally(() => { + this.processingPromise = null; + }); + } + return this.processingPromise; + } + + private async drainQueue(): Promise { + 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(); + 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 { + 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 { + 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 { + return (this.options.sleep ?? defaultSleep)(ms); + } + + private log(line: string): void { + this.options.log?.(line); + } +} diff --git a/media-scrape/src/extract.ts b/media-scrape/src/extract.ts new file mode 100644 index 0000000..c324a59 --- /dev/null +++ b/media-scrape/src/extract.ts @@ -0,0 +1,180 @@ +/* + * Media extraction: classify one core message into media items. + * Rules: docs/media-scraper.md. Third-party embeds are never downloaded. + */ + +import type { CoreAttachment, CoreEmbed, CoreSticker, CoreMessage, MediaKind, MediaMessage } from "./types.js"; + +const IMAGE_EXT = new Set(["png", "jpg", "jpeg", "webp", "bmp", "avif"]); +const GIF_EXT = new Set(["gif"]); +const VIDEO_EXT = new Set(["mp4", "webm", "mov", "mkv"]); +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 = { 1: "png", 2: "apng", 4: "gif" }; + +/** A media item extracted from a message. `url` is the original source link. */ +export interface ExtractedMedia { + kind: MediaKind; + url: string; + provider: string | null; + contentType: string | null; + size: number | null; + width: number | null; + height: number | null; + durationSecs: number | null; + /** False for every embed. Downloads are a later step, never for third-party embeds. */ + downloaded: boolean; + previewUrl: string | null; +} + +function extension(filename: string | null | undefined): string { + const match = /\.([a-z0-9]+)$/i.exec(filename ?? ""); + return match ? match[1].toLowerCase() : ""; +} + +function classifyAttachment(attachment: CoreAttachment, voiceMessage: boolean): MediaKind | null { + const ext = extension(attachment.filename); + const type = (attachment.contentType ?? "").toLowerCase(); + + // A voice-message flag wins unless the file is clearly an image or a video. + if ( + voiceMessage && + !IMAGE_EXT.has(ext) && + !GIF_EXT.has(ext) && + !VIDEO_EXT.has(ext) && + !type.startsWith("image/") && + !type.startsWith("video/") + ) { + return "audio"; + } + + if (AUDIO_EXT.has(ext) || type.startsWith("audio/")) return "audio"; + if (VIDEO_EXT.has(ext) || type.startsWith("video/")) return "video"; + if (GIF_EXT.has(ext) || type === "image/gif") return "gif"; + if (IMAGE_EXT.has(ext) || type.startsWith("image/")) return "image"; + return null; +} + +function hostname(url: string | null): string | null { + if (!url) return null; + try { + return new URL(url).hostname; + } catch { + return null; + } +} + +function extractEmbed(embed: CoreEmbed): ExtractedMedia | null { + const provider = embed.provider ?? hostname(embed.url); + const common = { + provider, + downloaded: false, + size: null, + width: null, + height: null, + durationSecs: null + }; + + if (embed.videoUrl) { + return { + ...common, + kind: "video", + url: embed.url ?? embed.videoUrl, + contentType: null, + previewUrl: embed.thumbnailUrl ?? embed.imageUrl + }; + } + + const providerKey = (provider ?? "").toLowerCase(); + if (embed.type === "gifv" || providerKey.includes("tenor") || providerKey.includes("giphy")) { + return { + ...common, + kind: "gif", + url: embed.url ?? embed.imageUrl ?? "", + contentType: null, + previewUrl: embed.thumbnailUrl ?? embed.imageUrl + }; + } + + if (embed.imageUrl) { + return { + ...common, + kind: "image", + url: embed.url ?? embed.imageUrl, + contentType: null, + previewUrl: embed.thumbnailUrl ?? embed.imageUrl + }; + } + + return null; +} + +function extractSticker(sticker: CoreSticker): ExtractedMedia | null { + if (!sticker.id) return null; + const ext = STICKER_EXT[sticker.formatType ?? 1]; + if (!ext) return null; + + return { + kind: ext === "gif" ? "gif" : "image", + url: `https://cdn.discordapp.com/stickers/${sticker.id}.${ext}`, + provider: "discord", + contentType: null, + size: null, + width: null, + height: null, + durationSecs: null, + downloaded: false, + previewUrl: null + }; +} + +function extractFrom(message: MediaMessage): ExtractedMedia[] { + const items: ExtractedMedia[] = []; + + for (const attachment of message.attachments ?? []) { + if (!attachment.url) continue; + const kind = classifyAttachment(attachment, message.voiceMessage); + if (!kind) continue; + items.push({ + kind, + url: attachment.url, + provider: "discord", + contentType: attachment.contentType, + size: attachment.size, + width: attachment.width, + height: attachment.height, + durationSecs: attachment.durationSecs, + downloaded: false, + previewUrl: null + }); + } + + for (const embed of message.embeds ?? []) { + const item = extractEmbed(embed); + if (item?.url) items.push(item); + } + + for (const sticker of message.stickerItems ?? []) { + const item = extractSticker(sticker); + if (item) items.push(item); + } + + return items; +} + +/** Extract every media item from a message, including forwarded snapshots. Deduped by URL. */ +export function extractMedia(message: CoreMessage): ExtractedMedia[] { + const items = extractFrom(message); + + for (const snapshot of message.forwarded ?? []) { + items.push(...extractFrom(snapshot)); + } + + const seen = new Set(); + return items.filter(item => { + if (seen.has(item.url)) return false; + seen.add(item.url); + return true; + }); +} diff --git a/media-scrape/src/index.ts b/media-scrape/src/index.ts new file mode 100644 index 0000000..04b2215 --- /dev/null +++ b/media-scrape/src/index.ts @@ -0,0 +1,25 @@ +/* + * Media archive entry point. The daemon mounts this module at checkpoint C5; + * nothing imports it before that. + */ + +export { extractMedia, type ExtractedMedia } from "./extract.js"; +export { + ScrapeEngine, + type FetchPage, + type ScrapeBackend, + type ScrapeEngineOptions +} from "./engine.js"; +export { + DEFAULT_DB_ROOT, + dirSizeBytes, + listScrapes, + readScrapeMeta, + resolveDbRoot, + ScrapeStore, + scrapeDir, + uniqueScrapeId, + writeScrapeMeta, + type MediaRow +} from "./store.js"; +export * from "./types.js"; diff --git a/media-scrape/src/store.ts b/media-scrape/src/store.ts new file mode 100644 index 0000000..53f7ae7 --- /dev/null +++ b/media-scrape/src/store.ts @@ -0,0 +1,245 @@ +/* + * Per-scrape storage: one directory, one SQLite index, one media folder. + * Layout and rules: docs/media.md. + */ + +import { mkdirSync, readdirSync, readFileSync, statSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { fileURLToPath } from "node:url"; +import { extractMedia } from "./extract.js"; +import { emptyMediaCounts, type CoreMessage, type MediaKind, type ScrapeMeta } from "./types.js"; + +/** Default DB root: `media-scrape/db`, beside the module. Override with VESKSLOP_MEDIA_DB_DIR. */ +export const DEFAULT_DB_ROOT = fileURLToPath(new URL("../db", import.meta.url)); + +export function resolveDbRoot(env: NodeJS.ProcessEnv = process.env): string { + return env.VESKSLOP_MEDIA_DB_DIR ?? DEFAULT_DB_ROOT; +} + +export function scrapeDir(root: string, id: string): string { + return join(root, id); +} + +export function writeScrapeMeta(dir: string, meta: ScrapeMeta): void { + writeFileSync(join(dir, "scrape.json"), JSON.stringify(meta, null, 2) + "\n"); +} + +export function readScrapeMeta(dir: string): ScrapeMeta | null { + try { + return JSON.parse(readFileSync(join(dir, "scrape.json"), "utf8")) as ScrapeMeta; + } catch { + return null; + } +} + +/** Every scrape on disk, newest first. */ +export function listScrapes(root: string): ScrapeMeta[] { + let entries: string[]; + try { + entries = readdirSync(root); + } catch { + return []; + } + + const metas: ScrapeMeta[] = []; + for (const entry of entries) { + const meta = readScrapeMeta(join(root, entry)); + if (meta) metas.push(meta); + } + + return metas.sort((a, b) => (a.createdAt < b.createdAt ? 1 : -1)); +} + +/** Combined size of a directory tree, used by the storage limit. */ +export function dirSizeBytes(dir: string): number { + let entries; + try { + entries = readdirSync(dir, { withFileTypes: true }); + } catch { + return 0; + } + + let total = 0; + for (const entry of entries) { + const path = join(dir, entry.name); + if (entry.isDirectory()) { + total += dirSizeBytes(path); + } else { + try { + total += statSync(path).size; + } catch { + // A file that vanished mid-walk counts as zero. + } + } + } + + return total; +} + +export function createScrapeDirs(root: string, id: string): string { + const dir = scrapeDir(root, id); + mkdirSync(join(dir, "media"), { recursive: true }); + mkdirSync(join(dir, "thumbs"), { recursive: true }); + return dir; +} + +/** `-`, with a numeric suffix on a same-second collision. */ +export function uniqueScrapeId(root: string, channelId: string, now = new Date()): string { + const stamp = now.toISOString().replace(/\.\d+Z$/, "").replace(/[-:T]/g, ""); + const base = `${channelId}-${stamp}`; + + let id = base; + let suffix = 2; + while (readScrapeMeta(join(root, id))) id = `${base}-${suffix++}`; + return id; +} + +export interface MediaRow { + id: number; + kind: MediaKind; + url: string; + provider: string | null; + content_type: string | null; + size: number | null; + width: number | null; + height: number | null; + duration_secs: number | null; + downloaded: number; + preview_url: string | null; + file_hash: string | null; + preview_hash: string | null; + created_at: string; +} + +const SCHEMA = ` +CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY, + author_id TEXT NOT NULL, + author_name TEXT NOT NULL, + timestamp TEXT NOT NULL, + content TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS media ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + url TEXT NOT NULL UNIQUE, + provider TEXT, + content_type TEXT, + size INTEGER, + width INTEGER, + height INTEGER, + duration_secs REAL, + downloaded INTEGER NOT NULL DEFAULT 0, + preview_url TEXT, + file_hash TEXT, + preview_hash TEXT, + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS media_sources ( + media_id INTEGER NOT NULL REFERENCES media(id), + message_id TEXT NOT NULL, + PRIMARY KEY (media_id, message_id) +); + +CREATE INDEX IF NOT EXISTS idx_media_kind ON media(kind); +CREATE INDEX IF NOT EXISTS idx_media_sources_message ON media_sources(message_id); +`; + +/** One open SQLite index. The caller closes it. */ +export class ScrapeStore { + private readonly db: DatabaseSync; + + constructor(readonly dir: string) { + this.db = new DatabaseSync(join(dir, "index.db")); + this.db.exec(SCHEMA); + } + + close(): void { + this.db.close(); + } + + /** Write one page of messages in a transaction. Returns the newly inserted media per kind. */ + insertPage(messages: CoreMessage[]): Record { + const added = emptyMediaCounts(); + const insertMessage = this.db.prepare( + "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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + ); + const findMedia = this.db.prepare("SELECT id FROM media WHERE url = ?"); + const linkSource = this.db.prepare( + "INSERT OR IGNORE INTO media_sources (media_id, message_id) VALUES (?, ?)" + ); + const createdAt = new Date().toISOString(); + + this.db.exec("BEGIN"); + try { + for (const message of messages) { + insertMessage.run( + message.id, + message.author.id, + message.author.username, + message.timestamp, + message.content + ); + + for (const item of extractMedia(message)) { + const result = insertMedia.run( + item.kind, + item.url, + item.provider, + item.contentType, + item.size, + item.width, + item.height, + item.durationSecs, + item.downloaded ? 1 : 0, + item.previewUrl, + createdAt + ); + if (result.changes > 0) added[item.kind]++; + + const row = findMedia.get(item.url) as { id: number } | undefined; + if (row) linkSource.run(row.id, message.id); + } + } + this.db.exec("COMMIT"); + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + + return added; + } + + counts(): Record { + const rows = this.db + .prepare("SELECT kind, COUNT(*) AS n FROM media GROUP BY kind") + .all() as unknown as Array<{ kind: string; n: number }>; + + const result = emptyMediaCounts(); + for (const row of rows) { + if (row.kind in result) result[row.kind as MediaKind] = row.n; + } + return result; + } + + queryMedia(options: { kind?: MediaKind; limit?: number; offset?: number } = {}): MediaRow[] { + const limit = Math.max(1, Math.min(options.limit ?? 50, 500)); + const offset = Math.max(0, options.offset ?? 0); + + const statement = options.kind + ? this.db.prepare("SELECT * FROM media WHERE kind = ? ORDER BY id DESC LIMIT ? OFFSET ?") + : this.db.prepare("SELECT * FROM media ORDER BY id DESC LIMIT ? OFFSET ?"); + + const rows = options.kind + ? statement.all(options.kind, limit, offset) + : statement.all(limit, offset); + + return rows as unknown as MediaRow[]; + } +} diff --git a/media-scrape/src/types.ts b/media-scrape/src/types.ts new file mode 100644 index 0000000..4dee596 --- /dev/null +++ b/media-scrape/src/types.ts @@ -0,0 +1,90 @@ +/* + * Shared types for the media archive. See docs/media.md and docs/media-scraper.md. + */ + +export type MediaKind = "image" | "gif" | "video" | "audio"; + +export const MEDIA_KINDS: MediaKind[] = ["image", "gif", "video", "audio"]; + +export type ScrapeState = + | "queued" + | "running" + | "paused" + | "done" + | "failed" + | "pending-delete" + | "deleting" + | "deleted"; + +export type ScrapeMode = "links" | "download"; + +export type BackendName = "puppet" | "slopbot"; + +export interface CoreAttachment { + id: string | null; + filename: string; + url: string | null; + contentType: string | null; + size: number | null; + width: number | null; + height: number | null; + durationSecs: number | null; +} + +export interface CoreEmbed { + type: string | null; + url: string | null; + title: string | null; + provider: string | null; + videoUrl: string | null; + imageUrl: string | null; + thumbnailUrl: string | null; +} + +export interface CoreSticker { + id: string | null; + name: string | null; + formatType: number | null; +} + +/** The part of a core message the extractor reads. Forwarded snapshots use the same shape. */ +export interface MediaMessage { + attachments: CoreAttachment[]; + embeds: CoreEmbed[]; + stickerItems: CoreSticker[]; + voiceMessage: boolean; +} + +/** A message as shaped by the core bridge (`message.fetch`, `message.read`). */ +export interface CoreMessage extends MediaMessage { + id: string; + author: { id: string; username: string; avatar: string | null }; + content: string; + timestamp: string; + editedAt: string | null; + replyTo: { id: string; authorId: string | null; authorName: string | null } | null; + forwarded: Array; +} + +/** `scrape.json`: the state of one scrape. The directory is the scrape's identity. */ +export interface ScrapeMeta { + id: string; + channelId: string; + guildId: string | null; + backend: BackendName; + mode: ScrapeMode; + state: ScrapeState; + /** The oldest message ID seen. The next page is fetched before it. */ + cursor: string | null; + createdAt: string; + updatedAt: string; + pages: number; + messages: number; + media: Record; + errors: number; + lastError: string | null; +} + +export function emptyMediaCounts(): Record { + return { image: 0, gif: 0, video: 0, audio: 0 }; +} diff --git a/media-scrape/test/run.ts b/media-scrape/test/run.ts new file mode 100644 index 0000000..c590772 --- /dev/null +++ b/media-scrape/test/run.ts @@ -0,0 +1,269 @@ +/* + * Media archive unit tests. No Discord, no network: a fake backend and a temp DB root. + * Run with: pnpm test:media (from poc/). + */ + +import assert from "node:assert/strict"; +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { ScrapeEngine, type FetchPage, type ScrapeBackend } from "../src/engine.js"; +import { extractMedia } from "../src/extract.js"; +import { dirSizeBytes, listScrapes, ScrapeStore } from "../src/store.js"; +import type { CoreAttachment, CoreMessage } from "../src/types.js"; + +let failures = 0; + +async function test(name: string, fn: () => void | Promise): Promise { + try { + await fn(); + console.log(`ok - ${name}`); + } catch (error) { + failures++; + console.error(`not ok - ${name}`); + console.error(error); + } +} + +function message(id: string, extra: Partial = {}): CoreMessage { + return { + id, + author: { id: "1", username: "tester", avatar: null }, + content: "", + timestamp: "2026-09-16T00:00:00.000Z", + editedAt: null, + replyTo: null, + attachments: [], + embeds: [], + stickerItems: [], + forwarded: [], + voiceMessage: false, + ...extra + }; +} + +function attachment(extra: Partial = {}): CoreAttachment { + const filename = extra.filename ?? "file.bin"; + return { + id: null, + filename, + url: `https://cdn.discordapp.com/attachments/1/2/${filename}`, + contentType: null, + size: null, + width: null, + height: null, + durationSecs: null, + ...extra + }; +} + +class FakeBackend implements ScrapeBackend { + readonly name = "puppet" as const; + readonly calls: Array<{ before: string | null; limit: number }> = []; + + constructor(private readonly pages: Array) {} + + async fetchPage(_channelId: string, before: string | null, limit: number): Promise { + this.calls.push({ before, limit }); + const page = this.pages.shift(); + if (!page) return { messages: [] }; + if (page instanceof Error) throw page; + return page; + } +} + +function engineFor(root: string, backend: ScrapeBackend, extra: Record = {}) { + return new ScrapeEngine({ + rootDir: root, + pageSize: 2, + pageMinMs: 0, + pageMaxMs: 0, + sleep: async () => {}, + random: () => 0, + log: () => {}, + ...extra + }); +} + +void (async () => { + const root = mkdtempSync(join(tmpdir(), "veskslop-media-")); + + await test("extract classifies attachments, embeds, and stickers", () => { + const items = extractMedia( + message("100", { + attachments: [ + attachment({ filename: "cat.png", contentType: "image/png" }), + attachment({ filename: "loop.gif" }), + attachment({ filename: "clip.mp4", contentType: "video/mp4", size: 1234 }), + attachment({ filename: "note.ogg", durationSecs: 3.2 }) + ], + embeds: [ + { + type: "video", + url: "https://www.youtube.com/watch?v=x", + title: "x", + provider: "YouTube", + videoUrl: "https://www.youtube.com/embed/x", + imageUrl: null, + thumbnailUrl: "https://i.ytimg.com/vi/x/hqdefault.jpg" + }, + { + type: "gifv", + url: "https://tenor.com/view/x", + title: null, + provider: "Tenor", + videoUrl: null, + imageUrl: null, + thumbnailUrl: "https://media.tenor.com/x.gif" + } + ], + stickerItems: [ + { id: "55", name: "wave", formatType: 4 }, + { id: "56", name: "lottie", formatType: 3 } + ] + }) + ); + + const kinds = items.map(item => item.kind).sort(); + assert.deepEqual(kinds, ["audio", "gif", "gif", "gif", "image", "video", "video"]); + const youtube = items.find(item => item.url.includes("youtube.com/watch")); + assert.equal(youtube?.kind, "video"); + assert.equal(youtube?.provider, "YouTube"); + assert.equal(youtube?.downloaded, false); + assert.equal(items.find(item => item.provider === "Tenor")?.kind, "gif"); + assert.equal(items.find(item => item.url.endsWith("/stickers/55.gif"))?.kind, "gif"); + assert.equal(items.some(item => item.url.includes("lottie")), false); + assert.equal(items.find(item => item.kind === "audio")?.durationSecs, 3.2); + }); + + await test("extract finds media in forwarded snapshots", () => { + const items = extractMedia( + message("101", { + forwarded: [ + { + content: "", + timestamp: null, + attachments: [attachment({ filename: "fwd.webm", contentType: "video/webm" })], + embeds: [], + stickerItems: [], + voiceMessage: false + } + ] + }) + ); + + assert.equal(items.length, 1); + assert.equal(items[0].kind, "video"); + }); + + await test("extract treats a voice message as audio", () => { + const items = extractMedia( + message("102", { + voiceMessage: true, + attachments: [attachment({ filename: "voice-message.ogg" })] + }) + ); + + assert.equal(items.length, 1); + assert.equal(items[0].kind, "audio"); + }); + + await test("engine walks pages, persists meta, and stores media", async () => { + const backend = new FakeBackend([ + { + messages: [ + message("100", { attachments: [attachment({ filename: "one.png", contentType: "image/png" })] }), + message("101") + ] + }, + { messages: [message("99")] } + ]); + + const engine = engineFor(root, backend); + const started = await engine.start({ channelId: "123", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const meta = engine.get(started.id); + assert.equal(meta?.state, "done"); + assert.equal(meta?.pages, 2); + assert.equal(meta?.messages, 3); + assert.equal(meta?.cursor, "99"); + assert.equal(meta?.media.image, 1); + assert.equal(backend.calls[0].limit, 2); + assert.equal(backend.calls[1].before, "100"); + + const dir = join(root, started.id); + assert.equal(existsSync(join(dir, "scrape.json")), true); + assert.equal(existsSync(join(dir, "index.db")), true); + assert.ok(dirSizeBytes(dir) > 0); + + const stored = new ScrapeStore(dir); + assert.equal(stored.counts().image, 1); + assert.equal(stored.queryMedia({ kind: "image" }).length, 1); + assert.equal(stored.queryMedia({ kind: "video" }).length, 0); + stored.close(); + + assert.equal(listScrapes(root).length, 1); + }); + + await test("engine backs off and retries with retryAfterMs", async () => { + const rate = Object.assign(new Error("rate limited"), { retryAfterMs: 250 }); + const backend = new FakeBackend([rate, new Error("boom"), { messages: [] }]); + const sleeps: number[] = []; + + const engine = engineFor(root, backend, { + maxRetries: 3, + sleep: async (ms: number) => { + sleeps.push(ms); + } + }); + const started = await engine.start({ channelId: "124", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const meta = engine.get(started.id); + assert.equal(meta?.state, "done"); + assert.equal(meta?.errors, 2); + assert.deepEqual(sleeps, [250, 3000]); + }); + + await test("engine pauses when the page budget is reached", async () => { + const backend = new FakeBackend([ + { messages: [message("200"), message("201")] }, + { messages: [message("199"), message("198")] } + ]); + + const engine = engineFor(root, backend, { pagesPerDay: 1 }); + const started = await engine.start({ channelId: "125", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const meta = engine.get(started.id); + assert.equal(meta?.state, "paused"); + assert.equal(meta?.pages, 1); + assert.equal(backend.calls.length, 1); + }); + + await test("engine stop pauses after the current page", async () => { + const backend = new FakeBackend([ + { messages: [message("300"), message("301")] }, + { messages: [message("299")] } + ]); + + const engine = engineFor(root, backend); + const started = await engine.start({ channelId: "126", guildId: null, mode: "links", backend }); + engine.stop(started.id); + await engine.processQueue(); + + const meta = engine.get(started.id); + assert.equal(meta?.state, "paused"); + assert.equal(meta?.pages, 1); + }); + + rmSync(root, { recursive: true, force: true }); + + if (failures > 0) { + console.error(`${failures} test(s) failed`); + process.exit(1); + } + + console.log("all media tests passed"); +})(); diff --git a/poc/README.md b/poc/README.md index b558316..d9d7cca 100644 --- a/poc/README.md +++ b/poc/README.md @@ -21,6 +21,7 @@ pnpm install | `pnpm poc:dm` | Send one test DM to zzib. | | `pnpm test:comm` | Run the unit tests for the pure logic. | | `pnpm test:cache` | Run the unit tests for the message cache. | +| `pnpm test:media` | Run the media archive tests (fake backend, temp DB). | | `pnpm cdp:probe` | Check CDP access and event delivery. | | `pnpm cdp:dispatch` | Run one CDP expression in the client. | | `pnpm check` | Type check the daemon code. | diff --git a/poc/package.json b/poc/package.json index 5af36f6..38fd96d 100644 --- a/poc/package.json +++ b/poc/package.json @@ -16,6 +16,7 @@ "cdp:dispatch": "tsx src/cdp-dispatch.ts", "test:cache": "tsx src/test-cache.ts", "test:audio": "tsx src/test-audio.ts", + "test:media": "tsx ../media-scrape/test/run.ts", "audio:doctor": "bash scripts/audio-doctor.sh" }, "dependencies": { diff --git a/poc/tsconfig.json b/poc/tsconfig.json index 4a41f93..42e72bc 100644 --- a/poc/tsconfig.json +++ b/poc/tsconfig.json @@ -12,5 +12,5 @@ "forceConsistentCasingInFileNames": true, "resolveJsonModule": true }, - "include": ["src", "../media-scrape/slopbot"] + "include": ["src", "../media-scrape/slopbot", "../media-scrape/src", "../media-scrape/test"] }