feat(cleanup): add stale chunked upload cleanup worker

server/controllers/fileController.js

@@ -27,7 +27,7 @@ export const getUserFiles = async (req, res, next) => {
     const userId = req.user.id;
     const { folderId, page = 1, limit = 50 } = req.query; // 'null' string or objectId
 
-    const query = { userId, isDeleted: false };
+    const query = { userId, isDeleted: false, status: { $ne: "PENDING" } };
 
     if (folderId !== undefined) {
       query.folderId = folderId === "null" || folderId === "" ? null : folderId;
@@ -110,6 +110,7 @@ if (checksum) {
   const duplicateFile = await File.findOne({
     checksum,
     userId,
+    status: { $ne: "PENDING" },
   });
 
   if (duplicateFile) {
@@ -125,7 +126,54 @@ if (checksum) {
   }
 }
 
-    const existingFile = await File.findOne({ fileName, userId });
+    // Try to find a pending file to complete
+    const pendingFile = await File.findOne({
+      fileName,
+      userId,
+      status: "PENDING",
+    }).sort({ updatedAt: -1 });
+
+    if (pendingFile) {
+      let hashedPassword = null;
+      if (password) {
+        const salt = await bcrypt.genSalt(10);
+        hashedPassword = await bcrypt.hash(password, salt);
+      }
+
+      // Update the pending file with final completed info
+      pendingFile.fileUrl = fileUrl;
+      pendingFile.fileType = fileType || pendingFile.fileType;
+      pendingFile.fileSize = fileSize || pendingFile.fileSize;
+      pendingFile.fileSizeBytes = fileSizeBytes || pendingFile.fileSizeBytes;
+      pendingFile.checksum = checksum || pendingFile.checksum;
+      pendingFile.status = "COMPLETED";
+      pendingFile.folderId = folderId || pendingFile.folderId;
+      pendingFile.tags = Array.isArray(tags) ? tags : [];
+      pendingFile.password = hashedPassword;
+
+      await pendingFile.save();
+
+      // Enqueue file for malware scanning and document indexing
+      try {
+        await enqueueScan(pendingFile._id);
+        await enqueueIndex(pendingFile._id);
+      } catch (queueErr) {
+        console.error("Failed to enqueue background jobs:", queueErr);
+      }
+
+      const io = req.app.get("io");
+      if (io) io.to(`user_${userId}`).emit("FILE_UPLOADED", pendingFile);
+
+      await invalidateDashboardStats(userId);
+
+      return res.status(201).json({
+        success: true,
+        message: "File info saved successfully",
+        file: pendingFile,
+      });
+    }
+
+    const existingFile = await File.findOne({ fileName, userId, status: { $ne: "PENDING" } });
 
     if (existingFile) {
       // Archive the current state into versions array
@@ -583,9 +631,9 @@ export const getFileStats = async (req, res, next) => {
       }
     }
 
-    const totalFiles = await File.countDocuments({ userId });
+    const totalFiles = await File.countDocuments({ userId, status: { $ne: "PENDING" } });
     const aggregations = await File.aggregate([
-      { $match: { userId } },
+      { $match: { userId, status: { $ne: "PENDING" } } },
       { 
         $group: { 
           _id: null, 
@@ -750,7 +798,7 @@ export const getFavoriteFiles = async (req, res, next) => {
     const { page = 1, limit = 50 } = req.query;
     const skip = (parseInt(page) - 1) * parseInt(limit);
 
-    const query = { userId, isFavorite: true };
+    const query = { userId, isFavorite: true, status: { $ne: "PENDING" } };
     const files = await File.find(query)
       .sort({ updatedAt: -1 })
       .skip(skip)
@@ -816,6 +864,7 @@ export const getFilesByTag = async (req, res, next) => {
 
     const files = await File.find({
       userId,
+      status: { $ne: "PENDING" },
       tags: { $in: [tag] },
     }).sort({ createdAt: -1 });
 
@@ -1007,6 +1056,7 @@ export const searchFiles = async (req, res, next) => {
     const query = {
       userId,
       isDeleted: false,
+      status: { $ne: "PENDING" },
       fileName: { $regex: q, $options: "i" }
     };
 

server/controllers/folderController.js

@@ -83,7 +83,7 @@ export const getFolderContents = async (req, res, next) => {
     const subfolders = await Folder.find({ parentId: folderId, userId }).sort({ name: 1 });
 
     const skip = (parseInt(page) - 1) * parseInt(limit);
-    const query = { folderId: folderId, userId, isDeleted: false };
+    const query = { folderId: folderId, userId, isDeleted: false, status: { $ne: "PENDING" } };
 
     const files = await File.find(query)
       .sort({ createdAt: -1 })

server/controllers/uploadController.js

@@ -105,7 +105,7 @@ async function loadUploadSessionForChunk(req, res, next) {
   }
 }
 
-export const initUpload = async (req, res) => {
+export const initUpload = async (req, res, next) => {
   try {
     const { fileName, fileSizeBytes, mimeType, expectedChecksum, sessionId, forceUpload } = req.body;
     const userId = req.user.id;
@@ -165,6 +165,19 @@ export const initUpload = async (req, res) => {
     const totalChunks = calculateTotalChunks(Number(fileSizeBytes));
     const expiresAt = new Date(Date.now() + UPLOAD_SESSION_TTL_HOURS * 60 * 60 * 1000);
 
+    // Create the File record in PENDING state
+    const pendingFile = new File({
+      fileName,
+      fileUrl: `${req.protocol || "http"}://${(typeof req.get === "function" ? req.get("host") : null) || "localhost"}/pending/${uploadId}`,
+      fileType: mimeType ? (mimeType.split("/")[0] || "application") : "application",
+      fileSize: "0 KB",
+      fileSizeBytes: Number(fileSizeBytes),
+      userId,
+      status: "PENDING",
+      folderId: req.body.folderId || null,
+    });
+    await pendingFile.save();
+
     const session = await UploadSession.create({
       userId,
       uploadId,
@@ -177,6 +190,7 @@ export const initUpload = async (req, res) => {
       tempDir,
       expiresAt,
       status: "uploading",
+      fileId: pendingFile._id,
     });
 
     res.status(201).json({
@@ -188,7 +202,7 @@ export const initUpload = async (req, res) => {
   }
 };
 
-export const getUploadStatus = async (req, res) => {
+export const getUploadStatus = async (req, res, next) => {
   try {
     const session = await UploadSession.findOne({
       _id: req.params.sessionId,
@@ -216,7 +230,7 @@ export const getUploadStatus = async (req, res) => {
   }
 };
 
-export const getResumableUploads = async (req, res) => {
+export const getResumableUploads = async (req, res, next) => {
   try {
     const sessions = await UploadSession.find({
       userId: req.user.id,

server/index.js

@@ -21,6 +21,7 @@ import { startWebhookWorker } from "./jobs/webhookWorker.js";
 import webhookRoutes from "./routes/webhooks.js";
 import { startUploadSessionCleanupJob } from "./jobs/uploadSessionCleanup.js";
 import { initQuotaResetJob } from "./jobs/quotaResetJob.js";
+import { initUploadCleanupWorker } from "./jobs/uploadCleanupWorker.js";
 import { connectRedis } from "./config/redis.js";
 import { ensureUploadTempRoot } from "./utils/chunkStorage.js";
 import {
@@ -53,6 +54,7 @@ startIndexWorker();
 startWebhookWorker();
 startUploadSessionCleanupJob();
 initQuotaResetJob();
+initUploadCleanupWorker();
 ensureUploadTempRoot().catch(console.error);
 
 app.use(express.json());

server/jobs/indexQueue.js

@@ -11,6 +11,10 @@ const connection = {
 export const indexQueue = new Queue("document-indexing", { connection });
 
 export const enqueueIndex = async (fileId) => {
+  if (process.env.ENABLE_DOCUMENT_INDEXING === "false") {
+    console.log("Document indexing is disabled. Skipping for file:", fileId);
+    return;
+  }
   console.log(`[Queue] Enqueuing document indexing job for file: ${fileId}`);
   await indexQueue.add("index-file", { fileId }, {
     attempts: 3,

server/jobs/uploadCleanupWorker.js

@@ -0,0 +1,47 @@
+import cron from "node-cron";
+import File from "../models/File.js";
+import UploadSession from "../models/UploadSession.js";
+import { removeSessionTempDir } from "../utils/chunkStorage.js";
+
+/**
+ * Initializes the background cron worker to clean up stale PENDING uploads.
+ * Runs daily at midnight (0 0 * * *).
+ */
+export const initUploadCleanupWorker = () => {
+  cron.schedule("0 0 * * *", async () => {
+    const thresholdTime = new Date(Date.now() - 24 * 60 * 60 * 1000); // 24 Hours ago
+    console.log(`[Cleanup Worker] Starting sweep of stale PENDING uploads older than ${thresholdTime.toISOString()}`);
+
+    try {
+      // Find all incomplete files older than 24 hours
+      const staleUploads = await File.find({
+        status: "PENDING",
+        updatedAt: { $lt: thresholdTime }
+      });
+
+      let clearedCount = 0;
+      for (const file of staleUploads) {
+        try {
+          // Find matching UploadSession to cleanup temp directory
+          const session = await UploadSession.findOne({ fileId: file._id });
+          if (session) {
+            if (session.tempDir) {
+              await removeSessionTempDir(session.tempDir);
+            }
+            await UploadSession.findByIdAndDelete(session._id);
+          }
+
+          // Delete the stale File document from MongoDB
+          await File.findByIdAndDelete(file._id);
+          clearedCount++;
+        } catch (fileErr) {
+          console.error(`[Cleanup Worker] Failed to clean up file ${file._id}:`, fileErr);
+        }
+      }
+
+      console.log(`[Cleanup Worker] Successfully cleared ${clearedCount} stale uploads.`);
+    } catch (error) {
+      console.error("[Cleanup Worker Error]:", error);
+    }
+  });
+};

server/models/File.js

@@ -170,6 +170,12 @@ const fileSchema = new mongoose.Schema(
       enum: ['pending', 'processing', 'completed', 'failed', 'skipped'],
       default: 'pending',
     },
+    status: {
+      type: String,
+      enum: ['PENDING', 'COMPLETED'],
+      default: 'PENDING',
+      index: true,
+    },
   },
   {
     timestamps: true,

server/models/UploadSession.js

@@ -54,6 +54,11 @@ const uploadSessionSchema = new mongoose.Schema(
       type: String,
       required: true,
     },
+    fileId: {
+      type: mongoose.Schema.Types.ObjectId,
+      ref: "File",
+      default: null,
+    },
     fileUrl: {
       type: String,
       default: null,