Wire the media archive into the daemon: command, service, and API

poc/src/control-bridge.ts

@@ -25,6 +25,7 @@ import { DEFAULT_VAD } from "./vad.js";
 import { CoreClient, coreSocketPath } from "./core-client.js";
 import { CoreProcess, staleBundleWarning } from "./core-process.js";
 import { dedupe, getQueryCount, paced, pacedSend, pacedVoice, sleep } from "./pacing.js";
+import { MediaService } from "./media.js";
 import { PreferencesStore } from "./preferences.js";
 import { Puppet } from "./puppet.js";
 import { startWebUi } from "./webui.js";
@@ -39,6 +40,12 @@ let puppet: Puppet | undefined;
 let coreConnected = false;
 let connecting = false;
 
+const media = new MediaService({
+    getCore: () => core,
+    isCoreConnected: () => coreConnected,
+    log: line => console.log(`[media] ${line}`)
+});
+
 async function connectLoop(): Promise<void> {
     if (connecting) return;
     connecting = true;
@@ -49,7 +56,7 @@ async function connectLoop(): Promise<void> {
                 const client = await CoreClient.connect();
 
                 core = client;
-                puppet = new Puppet(client, audio);
+                puppet = new Puppet(client, audio, media);
 
                 client.onEvent(event => {
                     // Speaking hints add to the reconciled set so the poller can
@@ -277,6 +284,7 @@ const webui = startWebUi({
     audio,
     transcripts,
     preferences,
+    media,
     connectAudioDevices: () => autoConnectAudio(requireCore())
 });
 

poc/src/media.ts

@@ -0,0 +1,188 @@
+/*
+ * Media archive service: the daemon side of media-scrape/. See docs/media.md.
+ *
+ * Owns the scrape engine, the slopbot client, and the read path the Web UI and
+ * (later) MCP tools call.
+ */
+
+import {
+    ScrapeEngine,
+    ScrapeStore,
+    scrapeDir,
+    resolveDbRoot,
+    type MediaKind,
+    type ScrapeBackend,
+    type ScrapeMeta
+} from "../../media-scrape/src/index.js";
+import { PuppetBackend } from "../../media-scrape/src/puppet-backend.js";
+import {
+    loadSlopbotConfig,
+    slopbotFetcher,
+    slopbotStatus,
+    SlopbotBackend,
+    startSlopbot,
+    type SlopbotFetcher,
+    type SlopbotStatus
+} from "../../media-scrape/slopbot/index.js";
+import type { CoreClient } from "./core-client.js";
+
+type BotClient = Awaited<ReturnType<typeof startSlopbot>>;
+
+export interface MediaServiceDeps {
+    getCore: () => CoreClient | undefined;
+    isCoreConnected: () => boolean;
+    log?: (line: string) => void;
+}
+
+export interface StartScrapeInput {
+    channelId: string;
+    guildId?: string | null;
+    mode?: "links" | "download";
+    backend?: "auto" | "slopbot" | "puppet";
+}
+
+export interface MediaItemResult {
+    scrapeId: string;
+    channelId: string;
+    kind: MediaKind;
+    url: string;
+    provider: string | null;
+    downloaded: boolean;
+    createdAt: string;
+}
+
+export interface MediaStatus {
+    slopbot: SlopbotStatus;
+    scrapes: ScrapeMeta[];
+}
+
+export class MediaService {
+    private readonly rootDir = resolveDbRoot();
+    private readonly engine: ScrapeEngine;
+
+    private bot: BotClient | null = null;
+    private botFetcher: SlopbotFetcher | null = null;
+    private botStarting: Promise<void> | null = null;
+
+    constructor(private readonly deps: MediaServiceDeps) {
+        this.engine = new ScrapeEngine({
+            rootDir: this.rootDir,
+            log: line => this.deps.log?.(line)
+        });
+    }
+
+    /** Log slopbot in on first use. False when it is disabled or the login failed. */
+    private async ensureSlopbot(): Promise<boolean> {
+        if (this.bot) return true;
+
+        const config = loadSlopbotConfig();
+        if (!config.enabled) return false;
+
+        if (!this.botStarting) {
+            this.botStarting = startSlopbot(config)
+                .then(client => {
+                    this.bot = client;
+                    this.botFetcher = slopbotFetcher(client);
+                    this.deps.log?.("slopbot connected");
+                })
+                .catch(error => {
+                    this.deps.log?.(`slopbot login failed: ${String(error)}`);
+                });
+        }
+
+        await this.botStarting;
+        return this.bot !== null;
+    }
+
+    private puppetBackend(): ScrapeBackend {
+        return new PuppetBackend(async (channelId, limit, options) => {
+            const core = this.deps.getCore();
+            if (!core || !this.deps.isCoreConnected()) {
+                throw new Error("the core bridge is not connected");
+            }
+            return (await core.fetchMessages(channelId, limit, options)) as { messages?: unknown[] };
+        });
+    }
+
+    async startScrape(input: StartScrapeInput): Promise<ScrapeMeta> {
+        const mode = input.mode ?? "links";
+        let backend: ScrapeBackend;
+
+        if (input.backend === "puppet") {
+            backend = this.puppetBackend();
+        } else {
+            const available = await this.ensureSlopbot();
+            const inGuild = !!input.guildId && (this.bot?.guilds.cache.has(input.guildId) ?? false);
+
+            if (input.backend === "slopbot" && (!available || !inGuild)) {
+                throw new Error("slopbot is not connected to that guild");
+            }
+
+            const fetcher = this.botFetcher;
+            backend =
+                available && inGuild && fetcher ? new SlopbotBackend(fetcher) : this.puppetBackend();
+        }
+
+        return this.engine.start({
+            channelId: input.channelId,
+            guildId: input.guildId ?? null,
+            mode,
+            backend
+        });
+    }
+
+    stopScrape(id: string): void {
+        this.engine.stop(id);
+    }
+
+    listScrapes(): ScrapeMeta[] {
+        return this.engine.list();
+    }
+
+    getScrape(id: string): ScrapeMeta | null {
+        return this.engine.get(id);
+    }
+
+    status(): MediaStatus {
+        return {
+            slopbot: slopbotStatus(this.bot),
+            scrapes: this.engine.list()
+        };
+    }
+
+    /** Read media rows across every scrape, newest scrape first. */
+    queryMedia(options: { kind?: MediaKind; limit?: number; offset?: number } = {}): MediaItemResult[] {
+        const limit = Math.max(1, Math.min(options.limit ?? 60, 500));
+        const offset = Math.max(0, options.offset ?? 0);
+        const results: MediaItemResult[] = [];
+        let skipped = 0;
+
+        for (const meta of this.engine.list()) {
+            if (meta.state === "deleted") continue;
+
+            const store = new ScrapeStore(scrapeDir(this.rootDir, meta.id));
+            try {
+                const rows = store.queryMedia({ kind: options.kind, limit: limit + offset });
+                for (const row of rows) {
+                    if (skipped++ < offset) continue;
+                    if (results.length >= limit) break;
+                    results.push({
+                        scrapeId: meta.id,
+                        channelId: meta.channelId,
+                        kind: row.kind,
+                        url: row.url,
+                        provider: row.provider,
+                        downloaded: row.downloaded === 1,
+                        createdAt: row.created_at
+                    });
+                }
+            } finally {
+                store.close();
+            }
+
+            if (results.length >= limit) break;
+        }
+
+        return results;
+    }
+}

poc/src/puppet.ts

@@ -110,6 +110,24 @@ export interface PuppetAudio {
     stop(): void;
 }
 
+/** The scrape state a chat command reads. */
+export interface PuppetScrape {
+    id: string;
+    channelId: string;
+    state: string;
+}
+
+/** The media actions a chat command can trigger. */
+export interface PuppetMedia {
+    startScrape(input: {
+        channelId: string;
+        guildId: string | null;
+        mode: "links" | "download";
+    }): Promise<{ id: string }>;
+    stopScrape(id: string): void;
+    listScrapes(): PuppetScrape[];
+}
+
 interface PollState {
     channelId: string;
     authorId: string;
@@ -130,6 +148,15 @@ const WHITELIST_HELP = [
     "  whitelist remove @user      Remove a user"
 ].join("\n");
 
+const SCRAPE_HELP = [
+    "scrape media — archive this channel's media.",
+    "Usage:",
+    "  scrape media            Index links (default)",
+    "  scrape media download   Also download attachment bytes (admin)",
+    "  scrape media status     Show this channel's latest scrape",
+    "  scrape media stop       Stop it (admin)"
+].join("\n");
+
 function clamp(value: number, min: number, max: number): number {
     return Math.min(Math.max(value, min), max);
 }
@@ -164,7 +191,8 @@ export class Puppet {
 
     constructor(
         private readonly core: PuppetTransport,
-        private readonly audio?: PuppetAudio
+        private readonly audio?: PuppetAudio,
+        private readonly media?: PuppetMedia
     ) {
         try {
             const state = JSON.parse(readFileSync(STATE_FILE, "utf8")) as PersistedState;
@@ -531,6 +559,8 @@ export class Puppet {
             await this.handleWhitelistCommand(command, author.id, tier, channelId);
         } else if (command?.name === "speak") {
             await this.handleSpeakCommand(command, author.id, channelId);
+        } else if (command?.name === "scrape") {
+            await this.handleScrapeCommand(command, author.id, channelId, guildId, tier);
         } else if (POLL_ENABLED && command?.name === "poll") {
             await this.handlePollCommand(author.id, channelId, text);
         }
@@ -671,6 +701,73 @@ export class Puppet {
         }
     }
 
+    private async handleScrapeCommand(
+        command: ParsedCommand,
+        authorId: string,
+        channelId: string,
+        guildId: string | undefined,
+        tier: TrustTier
+    ) {
+        const [kind, sub] = command.args;
+        const knownSub = sub === undefined || sub === "download" || sub === "status" || sub === "stop";
+
+        if (kind !== "media" || sub === "help" || !knownSub) {
+            return this.respond(authorId, channelId, "scrape", SCRAPE_HELP);
+        }
+
+        if (!this.media) {
+            return this.respond(authorId, channelId, "scrape", "The media archive is not available.");
+        }
+
+        if (sub === "status") {
+            const scrape = this.media.listScrapes().find(entry => entry.channelId === channelId);
+            return this.respond(
+                authorId,
+                channelId,
+                "scrape",
+                scrape
+                    ? `Media scrape ${scrape.id}: ${scrape.state}.`
+                    : "No media scrape for this channel yet."
+            );
+        }
+
+        if (sub === "stop") {
+            if (!hasTier(tier, "admin")) {
+                return this.respond(authorId, channelId, "scrape", "You do not have permission for this command.");
+            }
+
+            const scrape = this.media
+                .listScrapes()
+                .find(entry => entry.channelId === channelId && ["queued", "running", "paused"].includes(entry.state));
+            if (!scrape) {
+                return this.respond(authorId, channelId, "scrape", "No active media scrape for this channel.");
+            }
+
+            this.media.stopScrape(scrape.id);
+            this.actionLog.append({ type: "media", action: "stop", channelId, scrapeId: scrape.id });
+            return this.respond(authorId, channelId, "scrape", `Stopping media scrape ${scrape.id}.`);
+        }
+
+        const mode = sub === "download" ? "download" : "links";
+        if (mode === "download" && !hasTier(tier, "admin")) {
+            return this.respond(authorId, channelId, "scrape", "Download mode needs the admin tier.");
+        }
+
+        try {
+            const scrape = await this.media.startScrape({ channelId, guildId: guildId ?? null, mode });
+            this.actionLog.append({ type: "media", action: "scrape", channelId, mode, scrapeId: scrape.id });
+            return this.respond(authorId, channelId, "scrape", `Media scrape started (${mode}): ${scrape.id}.`);
+        } catch (error) {
+            this.actionLog.append({ type: "error", scope: "media", channelId, message: String(error) });
+            return this.respond(
+                authorId,
+                channelId,
+                "scrape",
+                `Media scrape failed: ${error instanceof Error ? error.message : String(error)}`
+            );
+        }
+    }
+
     /* Polls */
 
     private async handlePollCommand(authorId: string, channelId: string, text: string) {

poc/src/webui.ts

@@ -20,6 +20,7 @@ import type { AuthService, AuthStatus } from "./auth-service.js";
 import type { AudioService } from "./audio-service.js";
 import { CoreError, type CoreClient } from "./core-client.js";
 import type { CoreProcess } from "./core-process.js";
+import type { MediaService } from "./media.js";
 import { dedupe, getQueryCount, paced, pacedSend, pacedVoice } from "./pacing.js";
 import type { PreferencesStore, WebPreferences } from "./preferences.js";
 import type { Puppet } from "./puppet.js";
@@ -72,6 +73,7 @@ export interface WebUiDeps {
     audio: AudioService;
     transcripts: TranscriptStore;
     preferences: PreferencesStore;
+    media: MediaService;
     connectAudioDevices: () => Promise<{ connected: boolean; message: string }>;
 }
 
@@ -713,6 +715,66 @@ export function startWebUi(deps: WebUiDeps): WebUiHandle {
         })
     );
 
+    app.get(
+        "/api/media/scrapes",
+        wrap(async (_req, res) => {
+            res.json({ ok: true, result: deps.media.listScrapes() });
+        })
+    );
+
+    app.get(
+        "/api/media/status",
+        wrap(async (_req, res) => {
+            res.json({ ok: true, result: deps.media.status() });
+        })
+    );
+
+    app.get(
+        "/api/media",
+        wrap(async (req, res) => {
+            const kind = ["image", "gif", "video", "audio"].includes(String(req.query.kind ?? ""))
+                ? (String(req.query.kind) as "image" | "gif" | "video" | "audio")
+                : undefined;
+            const limit = clampLimit(req.query.limit, 60, 500);
+            const offset = Math.max(0, Number(req.query.offset ?? 0) || 0);
+
+            res.json({ ok: true, result: deps.media.queryMedia({ kind, limit, offset }) });
+        })
+    );
+
+    app.post(
+        "/api/media/scrape",
+        wrap(async (req, res) => {
+            const channelId = String(req.body?.channelId ?? "");
+            if (!channelId) throw new HttpError(400, "channelId is required");
+
+            const mode = req.body?.mode === "download" ? "download" : "links";
+            const guildId = req.body?.guildId ? String(req.body.guildId) : null;
+
+            const scrape = await deps.media.startScrape({ channelId, guildId, mode });
+            deps.getPuppet()?.appendAction({
+                type: "media",
+                action: "scrape",
+                channelId,
+                mode,
+                scrapeId: scrape.id
+            });
+            res.json({ ok: true, result: scrape });
+        })
+    );
+
+    app.post(
+        "/api/media/scrape/stop",
+        wrap(async (req, res) => {
+            const id = String(req.body?.id ?? "");
+            if (!id) throw new HttpError(400, "id is required");
+
+            deps.media.stopScrape(id);
+            deps.getPuppet()?.appendAction({ type: "media", action: "stop", scrapeId: id });
+            res.json({ ok: true, result: { stopped: id } });
+        })
+    );
+
     app.get("/api/logs", (req, res) => {
         const puppet = requirePuppet();
         const limit = clampLimit(req.query.limit, 100, 500);