宜博 / h3livestream

// Static file server + same-origin proxy to the fal.ai queue API.
// ponytail: keeping the proxy (rather than calling queue.fal.run directly
// from the browser) sidesteps having to verify fal's CORS behavior at all —
// same trick used for the MiniMax-native integration this replaced.
const http = require("http");
const https = require("https");
const fs = require("fs");
const path = require("path");

try {
  process.loadEnvFile(path.join(__dirname, ".env"));
} catch {
  // no .env file — fine, env vars may be set another way
}

const PORT = process.env.PORT || 8934;
const FAL_HOST = "queue.fal.run";
const H3_CONSOLE_BASE = process.env.H3_CONSOLE_BASE || "";
const H3_CONSOLE_TOKEN = process.env.H3_CONSOLE_TOKEN || "";
const H3_CONSOLE_URL = H3_CONSOLE_BASE ? new URL(H3_CONSOLE_BASE) : null;

const MIME = { ".html": "text/html", ".js": "text/javascript", ".css": "text/css", ".json": "application/json" };

const { syncLibrary } = require("./ingest.js");

const LIBRARY_PATH = path.join(__dirname, "library.json");
function getLibrary() {
  try {
    return JSON.parse(fs.readFileSync(LIBRARY_PATH, "utf8"));
  } catch {
    return [];
  }
}
function saveLibrary(lib) {
  fs.writeFileSync(LIBRARY_PATH, JSON.stringify(lib, null, 2));
}

const LIVE_STATE_PATH = path.join(__dirname, "live-state.json");
function loadLiveState() {
  try {
    const s = JSON.parse(fs.readFileSync(LIVE_STATE_PATH, "utf8"));
    return { schedule: s.schedule || [], promptQueue: s.promptQueue || [], chatMessages: s.chatMessages || [] };
  } catch {
    return { schedule: [], promptQueue: [], chatMessages: [] };
  }
}
function saveLiveState() {
  fs.writeFileSync(LIVE_STATE_PATH, JSON.stringify({ schedule, promptQueue, chatMessages }, null, 2));
}

function readJsonBody(req) {
  return new Promise((resolve, reject) => {
    let data = "";
    req.on("data", (chunk) => (data += chunk));
    req.on("end", () => {
      try {
        resolve(data ? JSON.parse(data) : {});
      } catch (err) {
        reject(err);
      }
    });
    req.on("error", reject);
  });
}

function h3ConsoleFetch(pathAndQuery, options) {
  return new Promise((resolve, reject) => {
    const proxyReq = http.request(
      {
        hostname: H3_CONSOLE_URL.hostname,
        port: H3_CONSOLE_URL.port || 80,
        path: `${H3_CONSOLE_URL.pathname}${pathAndQuery}`,
        method: (options && options.method) || "GET",
        headers: { Authorization: `Bearer ${H3_CONSOLE_TOKEN}`, "Content-Type": "application/json" },
      },
      (proxyRes) => {
        let data = "";
        proxyRes.on("data", (c) => (data += c));
        proxyRes.on("end", () => {
          try {
            resolve(JSON.parse(data));
          } catch (err) {
            reject(err);
          }
        });
      }
    );
    proxyReq.on("error", reject);
    if (options && options.body) proxyReq.write(JSON.stringify(options.body));
    proxyReq.end();
  });
}

// ---- Live channel: shared virtual playhead + prompt queue ----
// Persisted to live-state.json (see saveLiveState/loadLiveState) so a
// restart resumes the schedule instead of starting cold; ensureSchedule's
// own past-entry trimming naturally handles a state file that's gone stale
// (server was down long enough that every persisted entry is in the past).
const FILLER_DURATION_MS = 30000; // keep the schedule filled at least this far ahead
const H3_GEN_PARAMS = { width: 832, height: 480, frames: 120, steps: 20, turbo: false, seed: 0 };
const FPS = 24;

const initialLiveState = loadLiveState();
let schedule = initialLiveState.schedule; // [{clip, startAt, endAt, requestedBy?}]
let promptQueue = initialLiveState.promptQueue; // [{id, prompt, nickname?, status, clip?, error?, submittedAt, h3JobId?}]
let chatMessages = initialLiveState.chatMessages; // [{id, uid, nickname, text, createdAt, status}]

// ---- chat + moderation ----
// ponytail: a starter blocklist, not a real content-policy list — swap in
// whatever the actual moderation requirements are. Substring match on the
// lowercased text; cheap and deterministic. Anything this doesn't catch
// (content that's fine as *text* but unsafe to *generate*) falls through to
// the H3 console's own generation-time safety checker — a rejected/failed
// job there just surfaces as a normal "failed" queue entry.
const BANNED_WORDS = ["色情", "porn", "自杀", "血腥暴力", "毒品"];
function moderateText(text) {
  const lower = text.toLowerCase();
  const hit = BANNED_WORDS.find((w) => lower.includes(w));
  if (hit) return { ok: false, reason: `包含违禁词` };
  return { ok: true };
}

const CHAT_HISTORY_LIMIT = 300;
function pushChatMessage(msg) {
  chatMessages.push(msg);
  if (chatMessages.length > CHAT_HISTORY_LIMIT) chatMessages = chatMessages.slice(-CHAT_HISTORY_LIMIT);
}
function updateChatStatus(id, status) {
  const msg = chatMessages.find((m) => m.id === id);
  if (msg) msg.status = status;
}

// Shared by /api/chat (after moderation) and /api/prompt (direct API use,
// e.g. testing) — looks up a library match or kicks off generation, exactly
// like before chat existed. `nickname` is optional (empty for /api/prompt).
function submitPromptToQueue(id, prompt, nickname) {
  const entry = { id, prompt, nickname: nickname || null, submittedAt: Date.now() };
  const match = findLibraryMatch(normalizePrompt(prompt));
  if (match) {
    entry.status = "matched";
    entry.clip = match;
    promptQueue.push(entry);
    return { status: "matched", clip: match };
  }
  entry.status = "generating";
  promptQueue.push(entry);
  generateForPrompt(entry); // fire and forget — polled via /api/live/state
  return { status: "generating", id: entry.id };
}

function normalizePrompt(s) {
  return s.trim().toLowerCase().replace(/\s+/g, " ");
}
function findLibraryMatch(promptNorm) {
  return getLibrary().find((c) => {
    const zh = normalizePrompt(c.prompt);
    const en = c.promptEn ? normalizePrompt(c.promptEn) : null;
    return (
      zh === promptNorm ||
      zh.includes(promptNorm) ||
      promptNorm.includes(zh) ||
      (en && (en === promptNorm || en.includes(promptNorm) || promptNorm.includes(en)))
    );
  });
}

// Idle fallback pick avoids anything played within this window, so a clip
// that just aired doesn't come back around too soon. Falls back to ignoring
// the cooldown (keeping only the immediate-repeat guard) if literally
// everything is on cooldown, rather than ever stalling the schedule.
const REPEAT_COOLDOWN_MS = 20 * 60 * 1000;

function markPlayed(id) {
  const lib = getLibrary();
  const clip = lib.find((c) => c.id === id);
  if (!clip) return;
  clip.lastPlayedAt = Date.now();
  saveLibrary(lib);
}

// Consumed entries are spliced out immediately rather than left as
// status:"scheduled" — once a clip is in `schedule` it shows up via
// current/upNext, so keeping a second, never-cleared copy in promptQueue
// just grows forever and duplicates the same info.
// Returns {clip, requestedBy, chatId} — chatId lets the caller flip the
// originating chat message to "scheduled" once it's actually about to air.
function pickNextClip() {
  const readyIdx = promptQueue.findIndex((q) => q.status === "ready");
  if (readyIdx !== -1) {
    const q = promptQueue.splice(readyIdx, 1)[0];
    return { clip: q.clip, requestedBy: q.nickname, chatId: q.id };
  }
  const matchedIdx = promptQueue.findIndex((q) => q.status === "matched");
  if (matchedIdx !== -1) {
    const q = promptQueue.splice(matchedIdx, 1)[0];
    return { clip: q.clip, requestedBy: q.nickname, chatId: q.id };
  }

  const lib = getLibrary();
  if (!lib.length) return null;
  const prev = schedule.length ? schedule[schedule.length - 1].clip.id : null;
  const notPrev = lib.length > 1 ? lib.filter((c) => c.id !== prev) : lib;
  const now = Date.now();
  // Never-played clips (no lastPlayedAt) are always eligible — genuinely
  // random among everything that isn't on cooldown, no bias toward "fresh".
  const offCooldown = notPrev.filter((c) => !c.lastPlayedAt || now - c.lastPlayedAt > REPEAT_COOLDOWN_MS);
  const pool = offCooldown.length ? offCooldown : notPrev;
  return { clip: pool[Math.floor(Math.random() * pool.length)], requestedBy: null, chatId: null };
}

const FAILED_ENTRY_TTL_MS = 20000; // how long a "generation failed" entry stays visible before cleanup

function ensureSchedule() {
  const now = Date.now();
  while (schedule.length > 1 && schedule[1].endAt <= now) schedule.shift();
  while (schedule.length === 0 || schedule[schedule.length - 1].endAt < now + FILLER_DURATION_MS) {
    const picked = pickNextClip();
    if (!picked || !picked.clip) break;
    const { clip, requestedBy, chatId } = picked;
    markPlayed(clip.id);
    if (chatId) updateChatStatus(chatId, "scheduled");
    const startAt = schedule.length ? schedule[schedule.length - 1].endAt : now;
    schedule.push({ clip, startAt, endAt: startAt + clip.duration * 1000, requestedBy });
  }
  promptQueue = promptQueue.filter((q) => q.status !== "failed" || now - q.failedAt < FAILED_ENTRY_TTL_MS);

  for (const [cid, lastSeen] of viewers) if (now - lastSeen > VIEWER_TTL_MS) viewers.delete(cid);
  saveLiveState();
}
setInterval(ensureSchedule, 2000);

// ---- viewer count: anyone whose /api/live/state poll landed in the last VIEWER_TTL_MS ----
const VIEWER_TTL_MS = 12000;
const viewers = new Map(); // clientId -> lastSeenAt

// ---- rate limiting: cheap per-IP throttle on prompt submission ----
const RATE_LIMIT_MAX = 3;
const RATE_LIMIT_WINDOW_MS = 5 * 60 * 1000;
const rateLimitHits = new Map(); // ip -> [timestamps]
function isRateLimited(ip) {
  const now = Date.now();
  const hits = (rateLimitHits.get(ip) || []).filter((t) => now - t < RATE_LIMIT_WINDOW_MS);
  hits.push(now);
  rateLimitHits.set(ip, hits);
  return hits.length > RATE_LIMIT_MAX;
}

// Runs in the background; never blocks the live schedule or the POST /api/prompt response.
// `entry.h3JobId` set (and persisted via saveLiveState) as soon as we have it,
// so a server restart mid-generation can resume polling instead of losing
// track of the job (see resumeGenerations() at startup).
async function generateForPrompt(entry) {
  try {
    if (!entry.h3JobId) {
      const submitted = await h3ConsoleFetch("/v1/videos", {
        method: "POST",
        body: { prompt: entry.prompt, ...H3_GEN_PARAMS },
      });
      if (!submitted.id) throw new Error(submitted.detail || "submit failed");
      entry.h3JobId = submitted.id;
      saveLiveState();
    }

    for (let i = 0; i < 120; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const job = await h3ConsoleFetch(`/v1/videos/${entry.h3JobId}`);
      if (job.deleted_at) throw new Error("job was deleted on the H3 console before it finished");
      if (job.status === "done") {
        const clip = {
          id: job.id,
          source: "h3console",
          prompt: entry.prompt,
          fullPrompt: entry.prompt,
          width: job.req.width,
          height: job.req.height,
          duration: Math.round((job.req.frames / FPS) * 10) / 10,
          partition: job.partition,
          likes: 0,
          createdAt: Date.now(),
        };
        const lib = getLibrary();
        lib.push(clip);
        saveLibrary(lib);
        entry.status = "ready";
        entry.clip = clip;
        updateChatStatus(entry.id, "ready");
        saveLiveState();
        return;
      }
      if (job.status === "error" || job.status === "failed") throw new Error(job.error || "generation failed");
    }
    throw new Error("generation timed out");
  } catch (err) {
    entry.status = "failed";
    entry.error = err.message;
    entry.failedAt = Date.now();
    updateChatStatus(entry.id, "failed");
    saveLiveState();
  }
}

// Startup recovery: resume polling any generation that was mid-flight when
// the process last exited. Anything genuinely orphaned (no h3JobId — should
// never happen given the flow above) just gets dropped as failed.
function resumeGenerations() {
  for (const entry of promptQueue) {
    if (entry.status !== "generating") continue;
    if (entry.h3JobId) generateForPrompt(entry);
    else {
      entry.status = "failed";
      entry.error = "lost track of this generation across a restart";
      entry.failedAt = Date.now();
    }
  }
}

http
  .createServer((req, res) => {
    if (req.url.startsWith("/fal/")) {
      const proxyReq = https.request(
        {
          hostname: FAL_HOST,
          path: req.url.slice(4),
          method: req.method,
          headers: { ...req.headers, host: FAL_HOST },
        },
        (proxyRes) => {
          res.writeHead(proxyRes.statusCode, proxyRes.headers);
          proxyRes.pipe(res);
        }
      );
      proxyReq.on("error", (err) => {
        res.writeHead(502, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: { message: err.message } }));
      });
      req.pipe(proxyReq);
      return;
    }

    const pathname = req.url.split("?")[0];

    if (pathname === "/api/library") {
      const sort = new URL(req.url, "http://x").searchParams.get("sort") || "newest";
      const lib = getLibrary().slice();
      const sorters = {
        newest: (a, b) => b.createdAt - a.createdAt,
        oldest: (a, b) => a.createdAt - b.createdAt,
        top: (a, b) => (b.likes || 0) - (a.likes || 0) || b.createdAt - a.createdAt,
      };
      lib.sort(sorters[sort] || sorters.newest);
      res.writeHead(200, { "Content-Type": "application/json" });
      res.end(JSON.stringify(lib));
      return;
    }

    if (pathname === "/api/live/state") {
      ensureSchedule();
      const now = Date.now();
      const cid = new URL(req.url, "http://x").searchParams.get("cid");
      if (cid) viewers.set(cid, now);
      let idx = schedule.findIndex((s) => s.startAt <= now && now < s.endAt);
      if (idx === -1) idx = schedule.length - 1;
      const cur = schedule[idx];
      res.writeHead(200, { "Content-Type": "application/json" });
      res.end(
        JSON.stringify({
          now,
          watching: viewers.size,
          current: cur ? { clip: cur.clip, startAt: cur.startAt, endAt: cur.endAt, requestedBy: cur.requestedBy || null } : null,
          upNext: schedule.slice(idx + 1, idx + 4).map((s) => ({ ...s.clip, requestedBy: s.requestedBy || null })),
          queue: promptQueue
            .filter((q) => ["matched", "generating", "failed"].includes(q.status))
            .map((q) => ({ prompt: q.prompt, nickname: q.nickname, status: q.status, error: q.error })),
        })
      );
      return;
    }

    if (pathname === "/api/prompt" && req.method === "POST") {
      if (isRateLimited(req.socket.remoteAddress)) {
        res.writeHead(429, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: `最多每 ${RATE_LIMIT_WINDOW_MS / 60000} 分钟提交 ${RATE_LIMIT_MAX} 次，请稍后再试` }));
        return;
      }
      readJsonBody(req)
        .then((body) => {
          const prompt = (body.prompt || "").trim();
          if (!prompt || prompt.length > 300) {
            res.writeHead(400, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ error: "prompt required, max 300 chars" }));
            return;
          }
          const result = submitPromptToQueue(Date.now().toString(36), prompt, null);
          saveLiveState();
          res.writeHead(200, { "Content-Type": "application/json" });
          res.end(JSON.stringify(result));
        })
        .catch(() => {
          res.writeHead(400, { "Content-Type": "application/json" });
          res.end(JSON.stringify({ error: "invalid JSON body" }));
        });
      return;
    }

    if (pathname === "/api/chat" && req.method === "GET") {
      const cid = new URL(req.url, "http://x").searchParams.get("cid");
      if (cid) viewers.set(cid, Date.now());
      res.writeHead(200, { "Content-Type": "application/json" });
      res.end(JSON.stringify(chatMessages.slice(-100)));
      return;
    }

    if (pathname === "/api/chat" && req.method === "POST") {
      if (isRateLimited(req.socket.remoteAddress)) {
        res.writeHead(429, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: `最多每 ${RATE_LIMIT_WINDOW_MS / 60000} 分钟发送 ${RATE_LIMIT_MAX} 条，请稍后再试` }));
        return;
      }
      readJsonBody(req)
        .then((body) => {
          const uid = (body.uid || "").trim();
          const nickname = (body.nickname || "").trim().slice(0, 20);
          const text = (body.text || "").trim();
          if (!uid || !nickname || !text || text.length > 300) {
            res.writeHead(400, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ error: "uid, nickname, text required, text max 300 chars" }));
            return;
          }
          const id = `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 6)}`;
          const msg = { id, uid, nickname, text, createdAt: Date.now() };

          const mod = moderateText(text);
          if (!mod.ok) {
            msg.status = "rejected";
            msg.rejectReason = mod.reason;
            pushChatMessage(msg);
            saveLiveState();
            res.writeHead(200, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ status: "rejected", reason: mod.reason }));
            return;
          }

          msg.status = "approved";
          pushChatMessage(msg);
          const result = submitPromptToQueue(id, text, nickname);
          msg.status = result.status; // "matched" | "generating"
          saveLiveState();
          res.writeHead(200, { "Content-Type": "application/json" });
          res.end(JSON.stringify(result));
        })
        .catch(() => {
          res.writeHead(400, { "Content-Type": "application/json" });
          res.end(JSON.stringify({ error: "invalid JSON body" }));
        });
      return;
    }

    if (pathname === "/api/admin/library" && req.method === "POST") {
      readJsonBody(req)
        .then((body) => {
          const prompt = (body.prompt || "").trim();
          const videoUrl = (body.videoUrl || "").trim();
          if (!prompt || !videoUrl) {
            res.writeHead(400, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ error: "prompt and videoUrl required" }));
            return;
          }
          const clip = {
            id: `manual-${Date.now().toString(36)}${Math.random().toString(36).slice(2, 6)}`,
            source: "manual",
            edited: true,
            prompt,
            promptEn: (body.promptEn || "").trim(),
            fullPrompt: (body.fullPrompt || prompt).trim(),
            videoUrl,
            width: body.width || null,
            height: body.height || null,
            duration: body.duration || null,
            partition: "manual",
            likes: 0,
            createdAt: Date.now(),
          };
          const lib = getLibrary();
          lib.push(clip);
          saveLibrary(lib);
          res.writeHead(200, { "Content-Type": "application/json" });
          res.end(JSON.stringify(clip));
        })
        .catch(() => {
          res.writeHead(400, { "Content-Type": "application/json" });
          res.end(JSON.stringify({ error: "invalid JSON body" }));
        });
      return;
    }

    if (pathname.startsWith("/api/admin/library/") && req.method === "PATCH") {
      const id = pathname.slice("/api/admin/library/".length);
      readJsonBody(req)
        .then((body) => {
          const lib = getLibrary();
          const clip = lib.find((c) => c.id === id);
          if (!clip) {
            res.writeHead(404, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ error: "not found" }));
            return;
          }
          for (const field of ["prompt", "promptEn", "fullPrompt", "videoUrl"]) {
            if (typeof body[field] === "string" && body[field].trim()) clip[field] = body[field].trim();
          }
          // Storing a videoUrl (whether this clip started as h3console or
          // manual) means /lib/video/:id should serve that directly instead
          // of proxying the H3 console — see the handler below.
          if (body.videoUrl) clip.source = "manual";
          // Marks this entry as admin-owned so the periodic H3 library sync
          // (ingest.js's syncLibrary) carries it forward as-is instead of
          // overwriting it with a freshly re-extracted prompt.
          clip.edited = true;
          saveLibrary(lib);
          res.writeHead(200, { "Content-Type": "application/json" });
          res.end(JSON.stringify(clip));
        })
        .catch(() => {
          res.writeHead(400, { "Content-Type": "application/json" });
          res.end(JSON.stringify({ error: "invalid JSON body" }));
        });
      return;
    }

    if (pathname.startsWith("/api/admin/library/") && req.method === "DELETE") {
      const id = pathname.slice("/api/admin/library/".length);
      const lib = getLibrary();
      const clip = lib.find((c) => c.id === id);
      if (!clip) {
        res.writeHead(404, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: "not found" }));
        return;
      }
      saveLibrary(lib.filter((c) => c.id !== id));
      // Without this, the next periodic syncLibrary() would just re-add it —
      // it's still a "done" job on the H3 console until explicitly deleted there too.
      const cleanup = clip.source === "h3console" ? h3ConsoleFetch(`/v1/videos/${id}`, { method: "DELETE" }).catch(() => {}) : Promise.resolve();
      cleanup.then(() => {
        res.writeHead(200, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ deleted: true }));
      });
      return;
    }

    if (pathname.startsWith("/api/like/") && req.method === "POST") {
      const id = pathname.slice("/api/like/".length);
      const lib = getLibrary();
      const clip = lib.find((c) => c.id === id);
      if (!clip) {
        res.writeHead(404, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: "not found" }));
        return;
      }
      clip.likes = (clip.likes || 0) + 1;
      saveLibrary(lib);
      res.writeHead(200, { "Content-Type": "application/json" });
      res.end(JSON.stringify({ likes: clip.likes }));
      return;
    }

    if (pathname.startsWith("/lib/video/")) {
      const id = pathname.slice("/lib/video/".length);
      const clip = getLibrary().find((c) => c.id === id);
      if (clip && clip.videoUrl) {
        res.writeHead(302, { Location: clip.videoUrl });
        res.end();
        return;
      }
      if (!H3_CONSOLE_URL) {
        res.writeHead(500, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: "H3_CONSOLE_BASE not configured" }));
        return;
      }
      const upstreamHeaders = { Authorization: `Bearer ${H3_CONSOLE_TOKEN}` };
      if (req.headers.range) upstreamHeaders.range = req.headers.range;
      const proxyReq = http.request(
        {
          hostname: H3_CONSOLE_URL.hostname,
          port: H3_CONSOLE_URL.port || 80,
          path: `${H3_CONSOLE_URL.pathname}/v1/videos/${id}/file`,
          method: "GET",
          headers: upstreamHeaders,
        },
        (proxyRes) => {
          res.writeHead(proxyRes.statusCode, proxyRes.headers);
          proxyRes.pipe(res);
        }
      );
      proxyReq.on("error", (err) => {
        res.writeHead(502, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: { message: err.message } }));
      });
      proxyReq.end();
      return;
    }
    const filePath = path.join(__dirname, pathname === "/" ? "/index.html" : pathname);
    fs.readFile(filePath, (err, data) => {
      if (err) {
        res.writeHead(404);
        res.end("Not found");
        return;
      }
      res.writeHead(200, { "Content-Type": MIME[path.extname(filePath)] || "application/octet-stream" });
      res.end(data);
    });
  })
  .listen(PORT, () => console.log(`http://localhost:${PORT}`));

resumeGenerations();

// Auto-sync the library from the H3 Console so content added there directly
// (not through our own /api/prompt) shows up without a manual `node ingest.js`.
const LIBRARY_SYNC_INTERVAL_MS = 5 * 60 * 1000;
syncLibrary()
  .then((n) => console.log(`library synced — ${n} clips`))
  .catch((err) => console.error(`library sync failed: ${err.message}`));
setInterval(() => {
  syncLibrary().catch((err) => console.error(`library sync failed: ${err.message}`));
}, LIBRARY_SYNC_INTERVAL_MS);