diff --git a/src/index.js b/src/index.js new file mode 100644 index 0000000..41de4b7 --- /dev/null +++ b/src/index.js @@ -0,0 +1,93 @@ +const express = require("express"); +const http = require("http"); +const path = require("path"); +const fs = require("fs"); + +const config = require("./config"); +const apiRoutes = require("./routes/api"); +const wsManager = require("./utils/ws"); +const { startWorker } = require("./workers/image.worker"); +const imageQueue = require("./config/queue"); + +// Bull Board +const { createBullBoard } = require("@bull-board/api"); +const { BullAdapter } = require("@bull-board/api/bullAdapter"); +const { ExpressAdapter } = require("@bull-board/express"); + +// Ensure dirs exist +fs.mkdirSync(config.storage.uploadDir, { recursive: true }); +fs.mkdirSync(config.storage.outputDir, { recursive: true }); + +const app = express(); +const server = http.createServer(app); + +// --- Middleware --- +app.use(express.json()); + +// --- Bull Board dashboard (basic auth) --- +const bullBoardAdapter = new ExpressAdapter(); +bullBoardAdapter.setBasePath("/admin/queues"); + +createBullBoard({ + queues: [new BullAdapter(imageQueue)], + serverAdapter: bullBoardAdapter, +}); + +// Simple basic auth for dashboard +app.use("/admin", (req, res, next) => { + const authHeader = req.headers.authorization; + if (!authHeader) { + res.setHeader("WWW-Authenticate", 'Basic realm="ImgForge Admin"'); + return res.status(401).send("Authentication required"); + } + + const [user, pass] = Buffer.from(authHeader.split(" ")[1], "base64") + .toString() + .split(":"); + + if (user === config.bullBoard.user && pass === config.bullBoard.pass) { + return next(); + } + + res.setHeader("WWW-Authenticate", 'Basic realm="ImgForge Admin"'); + return res.status(401).send("Invalid credentials"); +}); +app.use("/admin/queues", bullBoardAdapter.getRouter()); + +// --- Static files (landing page) --- +app.use(express.static(path.join(__dirname, "../public"))); + +// --- API routes --- +app.use("/api", apiRoutes); + +// --- Error handler --- +app.use((err, _req, res, _next) => { + if (err.code === "LIMIT_FILE_SIZE") { + return res.status(413).json({ error: "File too large (max 20MB)" }); + } + if (err.message?.includes("Unsupported format")) { + return res.status(415).json({ error: err.message }); + } + console.error("[Server] Unhandled error:", err); + res.status(500).json({ error: "Internal server error" }); +}); + +// --- WebSocket --- +wsManager.attach(server); + +// --- Start worker in same process (for single-server setup) --- +// For production scale: run `npm run worker` as a separate PM2 process +startWorker(); + +// --- Boot --- +server.listen(config.port, config.host, () => { + console.log(` +╔══════════════════════════════════════════╗ +║ 🔥 ImgForge v1.0.0 ║ +╠══════════════════════════════════════════╣ +║ API: http://${config.host}:${config.port}/api ║ +║ Dashboard: http://${config.host}:${config.port}/admin/queues ║ +║ WebSocket: ws://${config.host}:${config.port}/ws ║ +╚══════════════════════════════════════════╝ + `); +}); diff --git a/src/routes/api.js b/src/routes/api.js new file mode 100644 index 0000000..d84ce66 --- /dev/null +++ b/src/routes/api.js @@ -0,0 +1,208 @@ +const { Router } = require("express"); +const path = require("path"); +const fs = require("fs"); +const upload = require("../middleware/upload"); +const imageQueue = require("../config/queue"); +const { validateOperations, OPERATIONS } = require("../utils/validators"); +const config = require("../config"); + +const router = Router(); + +/** + * POST /api/jobs + * Multipart: file + operations (JSON string) + * + * Example operations: + * [ + * { "op": "resize", "params": { "width": 800 } }, + * { "op": "convert", "params": { "format": "webp" } }, + * { "op": "compress", "params": { "quality": 80 } } + * ] + */ +router.post("/jobs", upload.single("file"), async (req, res) => { + try { + if (!req.file) { + return res.status(400).json({ error: "No file uploaded" }); + } + + // Parse operations from body + let operations; + try { + operations = JSON.parse(req.body.operations || "[]"); + } catch { + return res.status(400).json({ error: "operations must be valid JSON" }); + } + + const validationErr = validateOperations(operations); + if (validationErr) { + // Clean up uploaded file + fs.unlinkSync(req.file.path); + return res.status(400).json({ error: validationErr }); + } + + // Enqueue the job + const job = await imageQueue.add("process", { + inputPath: req.file.path, + originalName: req.file.originalname, + mimetype: req.file.mimetype, + size: req.file.size, + operations, + }, { + priority: operations.length > 5 ? 10 : 1, // heavy jobs = lower priority + }); + + res.status(202).json({ + jobId: job.id, + status: "queued", + operations: operations.map((o) => o.op), + ws: `ws://${req.headers.host}/ws?jobId=${job.id}`, + poll: `/api/jobs/${job.id}`, + }); + } catch (err) { + console.error("[API] POST /jobs error:", err); + res.status(500).json({ error: "Internal server error" }); + } +}); + +/** + * GET /api/jobs/:id — poll job status + */ +router.get("/jobs/:id", async (req, res) => { + try { + const job = await imageQueue.getJob(req.params.id); + + if (!job) { + return res.status(404).json({ error: "Job not found" }); + } + + const state = await job.getState(); + const progress = job.progress(); + + const response = { + jobId: job.id, + status: state, + progress, + operations: job.data.operations.map((o) => o.op), + createdAt: new Date(job.timestamp).toISOString(), + }; + + if (state === "completed") { + response.result = job.returnvalue; + response.download = `/api/jobs/${job.id}/download`; + } + + if (state === "failed") { + response.error = job.failedReason; + response.attempts = job.attemptsMade; + } + + res.json(response); + } catch (err) { + console.error("[API] GET /jobs/:id error:", err); + res.status(500).json({ error: "Internal server error" }); + } +}); + +/** + * GET /api/jobs/:id/download — download processed file + */ +router.get("/jobs/:id/download", async (req, res) => { + try { + const job = await imageQueue.getJob(req.params.id); + + if (!job) return res.status(404).json({ error: "Job not found" }); + + const state = await job.getState(); + if (state !== "completed") { + return res.status(409).json({ error: `Job is ${state}, not ready for download` }); + } + + const { outputPath, outputFilename } = job.returnvalue; + + if (!fs.existsSync(outputPath)) { + return res.status(410).json({ error: "Processed file has been cleaned up" }); + } + + res.download(outputPath, outputFilename); + } catch (err) { + console.error("[API] Download error:", err); + res.status(500).json({ error: "Internal server error" }); + } +}); + +/** + * GET /api/operations — list available operations + */ +router.get("/operations", (_req, res) => { + const ops = Object.entries(OPERATIONS).map(([name, def]) => ({ + name, + description: def.description, + })); + res.json({ operations: ops }); +}); + +/** + * GET /api/jobs — list recent jobs across all states + * Query: ?limit=50 (default 50, max 200) + */ +router.get("/jobs", async (req, res) => { + try { + const limit = Math.min(parseInt(req.query.limit || "50"), 200); + + // Fetch from all states + const [waiting, active, completed, failed, delayed] = await Promise.all([ + imageQueue.getWaiting(0, limit), + imageQueue.getActive(0, limit), + imageQueue.getCompleted(0, limit), + imageQueue.getFailed(0, limit), + imageQueue.getDelayed(0, limit), + ]); + + const format = async (job, state) => ({ + jobId: job.id, + status: state === "waiting" ? "queued" : state, + operations: (job.data.operations || []).map((o) => o.op), + operationsDetail: job.data.operations, + progress: job.progress(), + createdAt: new Date(job.timestamp).toISOString(), + result: job.returnvalue || null, + error: job.failedReason || null, + attempts: job.attemptsMade || 0, + }); + + const all = await Promise.all([ + ...waiting.map((j) => format(j, "waiting")), + ...active.map((j) => format(j, "active")), + ...completed.map((j) => format(j, "completed")), + ...failed.map((j) => format(j, "failed")), + ...delayed.map((j) => format(j, "delayed")), + ]); + + // Sort by creation time descending + all.sort((a, b) => new Date(b.createdAt) - new Date(a.createdAt)); + + res.json({ jobs: all.slice(0, limit) }); + } catch (err) { + console.error("[API] GET /jobs error:", err); + res.status(500).json({ error: "Internal server error" }); + } +}); + +/** + * GET /api/health + */ +router.get("/health", async (_req, res) => { + try { + // Check Redis connectivity through Bull + const counts = await imageQueue.getJobCounts(); + res.json({ + status: "ok", + queue: counts, + uptime: process.uptime(), + }); + } catch (err) { + res.status(503).json({ status: "degraded", error: err.message }); + } +}); + +module.exports = router;