src/server.js

// src/server.js — Crawlr v3
require("dotenv").config();

const express = require("express");
const { WebSocketServer } = require("ws");
const { v4: uuidv4 } = require("uuid");
const cors = require("cors");
const http = require("http");
const path = require("path");
const pLimit = require("p-limit");
const pdfParse = require("pdf-parse");
const fetch = require("node-fetch");

const { launchBrowser, newStealthPage, pageToMarkdown, getPageMeta, smartWait } = require("./scraper");
const { extractWithAI, extractWithSelectors } = require("./ai");
const { crawlSite, mapSite } = require("./crawler");
const { domainThrottle, withRetry, sendWebhook } = require("./queue");
const proxyManager = require("./proxy");
const captcha = require("./captcha");
const { setupMCP } = require("./mcp");

const app = express();
const server = http.createServer(app);
const wss = new WebSocketServer({ server });

app.use(cors());
app.use(express.json({ limit: "10mb" }));
app.use(express.static(path.join(__dirname, "../public")));

// ── API Key auth (optional) ───────────────────────────────────────────────
const API_KEY = process.env.CRAWLR_API_KEY;
function authMiddleware(req, res, next) {
  if (!API_KEY) return next();
  const key = req.headers["x-api-key"] || req.query.apiKey;
  if (key !== API_KEY) return res.status(401).json({ ok: false, error: "Invalid API key" });
  next();
}
app.use(["/scrape", "/crawl", "/map", "/extract", "/batch", "/session", "/mcp/call"], authMiddleware);

// ── Session + Job stores ──────────────────────────────────────────────────
const sessions = new Map();
const crawlJobs = new Map();
const agentEvents = new Map(); // sessionId → events[]

// ── MCP setup ─────────────────────────────────────────────────────────────
setupMCP(app, sessions);

// ══════════════════════════════════════════════════════════════════════════
//  /scrape
// ══════════════════════════════════════════════════════════════════════════
app.post("/scrape", async (req, res) => {
  const { url, formats = ["markdown"], useProxy = false, waitFor = 0, waitForSelector, networkIdle = false, actions = [], cookies, headers: reqHeaders } = req.body;
  if (!url) return res.status(400).json({ ok: false, error: "url দরকার" });

  let browser, proxy = null;
  try {
    if (useProxy) { await proxyManager.fetchProxies(); proxy = proxyManager.getNextProxy(); }
    browser = await launchBrowser(proxy);
    const page = await newStealthPage(browser);
    if (cookies?.length) await page.setCookie(...cookies);
    if (reqHeaders) await page.setExtraHTTPHeaders(reqHeaders);

    await domainThrottle(url);
    await withRetry(() => page.goto(url, { waitUntil: "domcontentloaded", timeout: 30000 }), { retries: 2 });
    await smartWait(page, { waitFor, waitForSelector, networkIdle });

    // CAPTCHA check
    const hasCaptcha = await page.evaluate(() => !!document.querySelector(".g-recaptcha,[data-sitekey],.h-captcha"));
    if (hasCaptcha) await captcha.solveRecaptchaV2(page).catch(() => {});

    // Actions
    for (const a of actions) {
      try {
        if (a.type === "click") { await page.click(a.selector); await page.waitForNavigation({ timeout: 5000 }).catch(() => {}); }
        if (a.type === "type") { await page.type(a.selector, a.text); }
        if (a.type === "wait") await new Promise((r) => setTimeout(r, Math.min(a.ms || 1000, 10000)));
        if (a.type === "scroll") await page.evaluate(() => window.scrollTo(0, document.body.scrollHeight));
        if (a.type === "navigate") await page.goto(a.url, { waitUntil: "domcontentloaded" });
      } catch (_) {}
    }

    const meta = await getPageMeta(page);
    const result = { ok: true, url: meta.url, title: meta.title, description: meta.description };
    if (formats.includes("markdown")) result.markdown = await pageToMarkdown(page);
    if (formats.includes("html")) result.html = await page.content();
    if (formats.includes("links")) result.links = meta.links;
    if (formats.includes("meta")) result.meta = meta;
    if (formats.includes("screenshot")) {
      const buf = await page.screenshot({ type: "jpeg", quality: 70 });
      result.screenshot = buf.toString("base64");
    }
    res.json(result);
  } catch (err) {
    if (proxy) proxyManager.removeCurrentProxy();
    res.status(500).json({ ok: false, error: err.message });
  } finally {
    if (browser) await browser.close().catch(() => {});
  }
});

// ══════════════════════════════════════════════════════════════════════════
//  /crawl
// ══════════════════════════════════════════════════════════════════════════
app.post("/crawl", async (req, res) => {
  const { url, maxPages = 20, maxDepth = 3, includeMarkdown = true, allowedPaths = [], excludedPaths = [], webhookUrl, useProxy = false } = req.body;
  if (!url) return res.status(400).json({ ok: false, error: "url দরকার" });

  const jobId = uuidv4();
  const job = { id: jobId, status: "running", startUrl: url, progress: 0, total: maxPages, results: [], errors: [] };
  crawlJobs.set(jobId, job);

  res.json({ ok: true, jobId, statusUrl: `/crawl/${jobId}` });

  // WebSocket এ live events পাঠাবো
  const broadcast = (data) => {
    const msg = JSON.stringify({ jobId, ...data });
    wss.clients.forEach((c) => { if (c.readyState === 1) c.send(msg); });
  };

  (async () => {
    let browser;
    try {
      let proxy = null;
      if (useProxy) { await proxyManager.fetchProxies(); proxy = proxyManager.getNextProxy(); }
      browser = await launchBrowser(proxy);

      await crawlSite(browser, url, {
        maxPages, maxDepth, includeMarkdown, allowedPaths, excludedPaths, webhookUrl,
        onEvent: (event) => {
          broadcast(event);
          if (event.type === "done") { job.progress = event.done; job.results.push({ url: event.url, title: event.title }); }
          if (event.type === "error") job.errors.push({ url: event.url, error: event.error });
        },
      });

      job.status = "completed";
      broadcast({ type: "job_complete", jobId, pagesFound: job.progress });
      if (webhookUrl) await sendWebhook(webhookUrl, { event: "crawl.complete", jobId, pagesFound: job.progress });
    } catch (err) {
      job.status = "failed"; job.error = err.message;
      broadcast({ type: "job_failed", jobId, error: err.message });
    } finally {
      if (browser) await browser.close().catch(() => {});
    }
  })();
});

app.get("/crawl/:jobId", (req, res) => {
  const job = crawlJobs.get(req.params.jobId);
  if (!job) return res.status(404).json({ ok: false, error: "Job নেই" });
  res.json({ ok: true, ...job });
});

// ══════════════════════════════════════════════════════════════════════════
//  /map
// ══════════════════════════════════════════════════════════════════════════
app.post("/map", async (req, res) => {
  const { url, maxLinks = 200 } = req.body;
  if (!url) return res.status(400).json({ ok: false, error: "url দরকার" });
  let browser;
  try {
    browser = await launchBrowser();
    const page = await newStealthPage(browser);
    const result = await mapSite(page, url, { maxLinks });
    res.json({ ok: true, ...result });
  } catch (err) {
    res.status(500).json({ ok: false, error: err.message });
  } finally {
    if (browser) await browser.close().catch(() => {});
  }
});

// ══════════════════════════════════════════════════════════════════════════
//  /extract — AI structured extraction
// ══════════════════════════════════════════════════════════════════════════
app.post("/extract", async (req, res) => {
  const { url, schema, selectors, prompt, useProxy = false } = req.body;
  if (!url) return res.status(400).json({ ok: false, error: "url দরকার" });
  if (!schema && !selectors) return res.status(400).json({ ok: false, error: "schema বা selectors দিন" });

  let browser;
  try {
    let proxy = null;
    if (useProxy) { await proxyManager.fetchProxies(); proxy = proxyManager.getNextProxy(); }
    browser = await launchBrowser(proxy);
    const page = await newStealthPage(browser);
    await domainThrottle(url);
    await page.goto(url, { waitUntil: "domcontentloaded", timeout: 30000 });

    const meta = await getPageMeta(page);
    let extracted;
    if (schema) {
      const md = await pageToMarkdown(page);
      extracted = await extractWithAI(md, schema, {
        apiKey: req.headers["x-api-key"],
        apiUrl: process.env.AI_API_URL,
        model: process.env.AI_MODEL,
        prompt,
      });
    } else {
      extracted = await extractWithSelectors(page, selectors);
    }
    res.json({ ok: true, url, title: meta.title, data: extracted });
  } catch (err) {
    res.status(500).json({ ok: false, error: err.message });
  } finally {
    if (browser) await browser.close().catch(() => {});
  }
});

// ══════════════════════════════════════════════════════════════════════════
//  /batch/scrape
// ══════════════════════════════════════════════════════════════════════════
app.post("/batch/scrape", async (req, res) => {
  const { urls, formats = ["markdown"], concurrency = 3 } = req.body;
  if (!urls?.length) return res.status(400).json({ ok: false, error: "urls array দিন" });
  if (urls.length > 50) return res.status(400).json({ ok: false, error: "max 50 URLs" });

  const limit = pLimit(Math.min(concurrency, 5));
  const results = await Promise.all(urls.map((url) =>
    limit(async () => {
      let browser;
      try {
        browser = await launchBrowser();
        const page = await newStealthPage(browser);
        await domainThrottle(url);
        await page.goto(url, { waitUntil: "domcontentloaded", timeout: 25000 });
        const meta = await getPageMeta(page);
        const r = { url, ok: true, title: meta.title };
        if (formats.includes("markdown")) r.markdown = await pageToMarkdown(page);
        if (formats.includes("links")) r.links = meta.links;
        return r;
      } catch (err) {
        return { url, ok: false, error: err.message };
      } finally {
        if (browser) await browser.close().catch(() => {});
      }
    })
  ));
  res.json({ ok: true, total: urls.length, results });
});

// ══════════════════════════════════════════════════════════════════════════
//  /pdf
// ══════════════════════════════════════════════════════════════════════════
app.post("/pdf", async (req, res) => {
  const { url } = req.body;
  if (!url) return res.status(400).json({ ok: false, error: "url দিন" });
  try {
    const r = await fetch(url);
    if (!r.ok) throw new Error(`PDF fetch failed: ${r.status}`);
    const buf = await r.buffer();
    const data = await pdfParse(buf);
    res.json({ ok: true, text: data.text, pages: data.numpages, info: data.info });
  } catch (err) {
    res.status(500).json({ ok: false, error: err.message });
  }
});

// ══════════════════════════════════════════════════════════════════════════
//  Session routes
// ══════════════════════════════════════════════════════════════════════════
app.post("/session/create", async (req, res) => {
  try {
    const browser = await launchBrowser();
    const page = await newStealthPage(browser);
    if (req.body.cookies?.length) await page.setCookie(...req.body.cookies);
    if (req.body.headers) await page.setExtraHTTPHeaders(req.body.headers);
    const sessionId = uuidv4();
    sessions.set(sessionId, { browser, page, events: [], createdAt: Date.now() });
    res.json({ ok: true, sessionId, wsUrl: `ws://localhost:${PORT}?session=${sessionId}` });
  } catch (err) {
    res.status(500).json({ ok: false, error: err.message });
  }
});

app.post("/session/:id/navigate", async (req, res) => {
  const s = sessions.get(req.params.id);
  if (!s) return res.status(404).json({ ok: false, error: "Session নেই" });
  try {
    await domainThrottle(req.body.url);
    await withRetry(() => s.page.goto(req.body.url, { waitUntil: "domcontentloaded", timeout: 30000 }));
    await smartWait(s.page, req.body);
    const [md, meta] = await Promise.all([pageToMarkdown(s.page), getPageMeta(s.page)]);
    s.events.push({ type: "navigate", url: req.body.url, ts: Date.now() });
    broadcastSession(req.params.id, { type: "navigate", url: meta.url, title: meta.title });
    res.json({ ok: true, ...meta, markdown: md });
  } catch (err) { res.status(500).json({ ok: false, error: err.message }); }
});

app.post("/session/:id/click", async (req, res) => {
  const s = sessions.get(req.params.id);
  if (!s) return res.status(404).json({ ok: false, error: "Session নেই" });
  try {
    await s.page.click(req.body.selector);
    await s.page.waitForNavigation({ timeout: 5000 }).catch(() => {});
    const [md, meta] = await Promise.all([pageToMarkdown(s.page), getPageMeta(s.page)]);
    s.events.push({ type: "click", selector: req.body.selector, ts: Date.now() });
    broadcastSession(req.params.id, { type: "click", selector: req.body.selector });
    res.json({ ok: true, ...meta, markdown: md });
  } catch (err) { res.status(500).json({ ok: false, error: err.message }); }
});

app.post("/session/:id/type", async (req, res) => {
  const s = sessions.get(req.params.id);
  if (!s) return res.status(404).json({ ok: false, error: "Session নেই" });
  try {
    const { selector, text, pressEnter = false } = req.body;
    await s.page.click(selector);
    await s.page.evaluate((sel) => { const el = document.querySelector(sel); if (el) el.value = ""; }, selector);
    await s.page.type(selector, text, { delay: 40 });
    if (pressEnter) { await s.page.keyboard.press("Enter"); await s.page.waitForNavigation({ timeout: 5000 }).catch(() => {}); }
    s.events.push({ type: "type", selector, text, ts: Date.now() });
    broadcastSession(req.params.id, { type: "type", selector, text });
    const [md, meta] = await Promise.all([pageToMarkdown(s.page), getPageMeta(s.page)]);
    res.json({ ok: true, ...meta, markdown: md });
  } catch (err) { res.status(500).json({ ok: false, error: err.message }); }
});

app.post("/session/:id/screenshot", async (req, res) => {
  const s = sessions.get(req.params.id);
  if (!s) return res.status(404).json({ ok: false, error: "Session নেই" });
  try {
    const buf = await s.page.screenshot({ type: "jpeg", quality: 70, fullPage: req.body.fullPage || false });
    res.json({ ok: true, image: buf.toString("base64"), mimeType: "image/jpeg" });
  } catch (err) { res.status(500).json({ ok: false, error: err.message }); }
});

app.post("/session/:id/scroll", async (req, res) => {
  const s = sessions.get(req.params.id);
  if (!s) return res.status(404).json({ ok: false, error: "Session নেই" });
  const { direction = "down", amount = 500 } = req.body;
  await s.page.evaluate((d, a) => window.scrollBy(0, d === "down" ? a : -a), direction, amount);
  s.events.push({ type: "scroll", direction, ts: Date.now() });
  res.json({ ok: true });
});

app.get("/session/:id/events", (req, res) => {
  const s = sessions.get(req.params.id);
  if (!s) return res.status(404).json({ ok: false, error: "Session নেই" });
  res.json({ ok: true, events: s.events });
});

app.post("/session/:id/close", async (req, res) => {
  const s = sessions.get(req.params.id);
  if (s) { await s.browser.close().catch(() => {}); sessions.delete(req.params.id); }
  res.json({ ok: true });
});

// ══════════════════════════════════════════════════════════════════════════
//  Proxy + CAPTCHA status
// ══════════════════════════════════════════════════════════════════════════
app.get("/proxy/status", (req, res) => res.json({ ok: true, ...proxyManager.getStats() }));
app.post("/proxy/refresh", async (req, res) => { await proxyManager.fetchProxies(); res.json({ ok: true, ...proxyManager.getStats() }); });
app.get("/captcha/status", (req, res) => res.json({ ok: true, available: captcha.isAvailable(), service: process.env.CAPTCHA_SERVICE || "2captcha" }));

// ══════════════════════════════════════════════════════════════════════════
//  Status
// ══════════════════════════════════════════════════════════════════════════
app.get("/status", (req, res) => {
  res.json({
    ok: true, name: "Crawlr", version: "3.0.0",
    sessions: sessions.size, crawlJobs: crawlJobs.size,
    captchaAvailable: captcha.isAvailable(),
    aiProvider: process.env.GEMINI_API_KEY ? "gemini" : process.env.AI_API_KEY ? "custom" : "none",
    proxy: proxyManager.getStats(),
    mcp: { tools: 22, url: "/mcp" },
  });
});

// ══════════════════════════════════════════════════════════════════════════
//  WebSocket — Live Agent View + session streaming
// ══════════════════════════════════════════════════════════════════════════
function broadcastSession(sessionId, data) {
  const msg = JSON.stringify({ sessionId, ...data });
  wss.clients.forEach((c) => { if (c.readyState === 1) c.send(msg); });
}

wss.on("connection", (ws, req) => {
  const params = new URL(req.url, "http://localhost").searchParams;
  const sessionId = params.get("session");
  const s = sessionId ? sessions.get(sessionId) : null;

  ws.send(JSON.stringify({ type: "connected", sessionId: sessionId || null }));

  if (s) {
    // Live screenshot stream
    let streaming = true;
    const sendFrame = async () => {
      if (!streaming || ws.readyState !== 1) return;
      try {
        const buf = await s.page.screenshot({ type: "jpeg", quality: 45 });
        ws.send(JSON.stringify({ type: "frame", data: buf.toString("base64") }));
      } catch (_) {}
      if (streaming) setTimeout(sendFrame, 600);
    };
    sendFrame();
    ws.on("close", () => { streaming = false; });
  }

  ws.on("message", async (raw) => {
    try {
      const msg = JSON.parse(raw);
      if (s) {
        if (msg.type === "navigate") await s.page.goto(msg.url, { waitUntil: "domcontentloaded" }).catch(() => {});
        if (msg.type === "click") await s.page.click(msg.selector).catch(() => {});
        if (msg.type === "type") await s.page.type(msg.selector, msg.text).catch(() => {});
      }
    } catch (_) {}
  });
});

// ── Auto-cleanup ──────────────────────────────────────────────────────────
setInterval(() => {
  const now = Date.now();
  for (const [id, s] of sessions) {
    if (now - s.createdAt > 30 * 60 * 1000) {
      s.browser.close().catch(() => {});
      sessions.delete(id);
    }
  }
}, 5 * 60 * 1000);

const PORT = process.env.PORT || 7860;
server.listen(PORT, () => {
  console.log(`\n🕷️  Crawlr v3 → http://localhost:${PORT}`);
  console.log(`🤖 MCP → http://localhost:${PORT}/mcp`);
  console.log(`🔑 Auth: ${API_KEY ? "enabled" : "disabled"}`);
  console.log(`🧠 AI: ${process.env.GEMINI_API_KEY ? "Gemini ✅" : process.env.AI_API_KEY ? "Custom ✅" : "none ❌"}\n`);
});