feat: Implement queue-based async DB export pipeline

apps/consumer/Dockerfile

@@ -0,0 +1,20 @@
+FROM node:22-alpine
+
+WORKDIR /app
+
+# Copy root package files
+COPY package.json package-lock.json ./
+
+# Copy workspace package.json files
+COPY packages/common/package.json ./packages/common/
+COPY apps/consumer/package.json ./apps/consumer/
+
+RUN npm ci
+
+# Copy actual source code
+COPY packages/common ./packages/common
+COPY apps/consumer ./apps/consumer
+
+WORKDIR /app/apps/consumer
+
+CMD ["npm", "run", "start"]

apps/consumer/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "consumer",
+  "version": "1.0.0",
+  "description": "",
+  "main": "src/app.js",
+  "scripts": {
+    "dev": "node src/index.js",
+    "start": "node src/index.js",
+    "test": "jest --testPathPatterns=src/"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "type": "commonjs"
+}

apps/consumer/src/index.js

@@ -0,0 +1,35 @@
+const dotenv = require('dotenv');
+dotenv.config({ path: require('path').join(__dirname, '../../../.env') });
+
+const { validateEnv } = require('@urbackend/common');
+
+if (process.env.NODE_ENV !== 'test') {
+    validateEnv();
+}
+
+const { initExportWorker } = require('./workers/export.worker');
+
+const { connectDB } = require('@urbackend/common');
+
+    (async () => {
+        try {
+            await connectDB();
+
+            const worker = initExportWorker();
+
+            console.log('[CONSUMER] Export worker started and listening for jobs...');
+
+            const shutdown = async () => {
+                console.log('Shutting down worker...');
+                await worker.close();
+                process.exit(0);
+            };
+
+            process.on('SIGINT', shutdown);
+            process.on('SIGTERM', shutdown);
+
+        } catch (err) {
+            console.error('Failed to start worker:', err);
+            process.exit(1);
+        }
+    })();

apps/consumer/src/workers/export.worker.js

@@ -0,0 +1,103 @@
+const { Worker } = require('bullmq');
+const { PassThrough } = require('stream');
+const { Upload } = require('@aws-sdk/lib-storage');
+const { GetObjectCommand } = require('@aws-sdk/client-s3');
+const { getSignedUrl } = require('@aws-sdk/s3-request-presigner');
+
+const { 
+    redis, 
+    exportQueue, 
+    emailQueue,
+    Project, 
+    getConnection, 
+    getCompiledModel,
+    getS3CompatibleStorage,
+    getBucket
+} = require('@urbackend/common');
+
+const initExportWorker = () => {
+    const worker = new Worker(exportQueue.name, async (job) => {
+        const { projectId, userId, email } = job.data;
+        console.log(`[ExportWorker] Starting export for project ${projectId} requested by ${email}`);
+
+        const project = await Project.findById(projectId);
+        if (!project) throw new Error('Project not found');
+
+        const connection = await getConnection(projectId);
+
+        console.log(`[ExportWorker] Preparing streaming upload to storage...`);
+
+        const { s3Client } = await getS3CompatibleStorage(project);
+        const bucket = await getBucket(project);
+        const storagePath = `${projectId}/exports/db_export_${Date.now()}.json`;
+
+        const passThrough = new PassThrough();
+
+        const upload = new Upload({
+            client: s3Client,
+            params: {
+                Bucket: bucket,
+                Key: storagePath,
+                Body: passThrough,
+                ContentType: 'application/json'
+            }
+        });
+
+        // Start the upload promise in parallel
+        const uploadPromise = upload.done();
+
+        try {
+            passThrough.write('{\n');
+
+            for (let i = 0; i < project.collections.length; i++) {
+                const col = project.collections[i];
+                const Model = getCompiledModel(connection, col, projectId, project.resources.db.isExternal);
+
+                passThrough.write(`  "${col.name}": [\n`);
+
+                // use a mongoose cursor to stream documents one by one
+                const cursor = Model.find().lean().cursor();
+                let first = true;
+
+                for await (const doc of cursor) {
+                    if (!first) passThrough.write(',\n');
+                    passThrough.write(`    ${JSON.stringify(doc)}`);
+                    first = false;
+                }
+
+                passThrough.write('\n  ]');
+                if (i < project.collections.length - 1) passThrough.write(',\n');
+            }
+
+            passThrough.write('\n}\n');
+            passThrough.end();
+
+            console.log(`[ExportWorker] Database stream ended. Awaiting final storage upload chunks...`);
+            await uploadPromise;
+
+            // create a signed URL valid for 24 hrs (86400 seconds)
+            const command = new GetObjectCommand({ Bucket: bucket, Key: storagePath });
+            const signedUrl = await getSignedUrl(s3Client, command, { expiresIn: 86400 });
+
+            // queue the email to be sent to the user
+            await emailQueue.add('send-export-email', { email, downloadUrl: signedUrl, projectName: project.name });
+            console.log(`[ExportWorker] Export completed! Email queued for ${email}`);
+
+        } catch (error) {
+            passThrough.destroy(error);
+            throw error;
+        }
+    }, { connection: redis, concurrency: 2 });
+
+    worker.on('completed', (job) => {
+        console.log(`[ExportWorker] Job ${job.id} for project ${job.data.projectId} completed.`);
+    });
+
+    worker.on('failed', (job, err) => {
+        console.error(`[ExportWorker] Job ${job?.id} for project ${job?.data?.projectId} failed:`, err.message);
+    });
+
+    return worker;
+};
+
+module.exports = { initExportWorker };

apps/dashboard-api/package.json

@@ -14,6 +14,8 @@
     "transform": {}
   },
   "dependencies": {
+    "@bull-board/api": "^7.1.5",
+    "@bull-board/express": "^7.1.5",
     "@kiroo/sdk": "^0.1.2",
     "@supabase/supabase-js": "^2.84.0",
     "@urbackend/common": "*",

apps/dashboard-api/src/app.js

@@ -119,7 +119,6 @@ app.use('/api/admin/metrics', dashboardLimiter, adminMetricsRoute);
 
 
 
-
 app.get('/api/server-ip', async (req, res) => {
     const ip = await getPublicIp();
     res.json({ ip });

apps/dashboard-api/src/controllers/dbExport.controller.js

@@ -0,0 +1,60 @@
+const { AppError } = require('@urbackend/common');
+const { Developer } = require('@urbackend/common');
+const { Project } = require('@urbackend/common');
+const { exportQueue } = require('@urbackend/common');
+const { redis } = require('@urbackend/common');
+const { getProjectById, setProjectById } = require('@urbackend/common');
+
+module.exports.dbExportHandler = async (req, res, next) => {
+    try {
+        const { projectId } = req.params;
+        const { _id: userId } = req.user;
+
+        let project = await getProjectById(projectId);
+        if (!project) {
+            project = await Project.findById(projectId).lean();
+            if (!project) {
+                return next(new AppError(404, "Project not found."));
+            }
+            await setProjectById(projectId, project);
+        }
+
+        if (project.owner.toString() !== userId.toString()) {
+            return next(new AppError(403, "Access denied. You are not the owner of this project."));
+        }
+
+
+        const developer = await Developer.findById(userId).select('email plan').lean();
+        if (!developer) {
+            return next(new AppError(404, "Authenticated developer not found."));
+        }
+        const { email, plan = 'free' } = developer;
+
+        console.log(`[Dashboard API] Received export request for project ${projectId} from user ${userId} (${email})`);
+
+
+        const maxExports = plan === 'pro' ? 5 : 1;
+        const today = new Date().toISOString().split('T')[0];
+        const key = `project:${projectId}:export_limit:${today}`;
+
+        const currentCount = await redis.get(key);
+        if (currentCount && Number(currentCount) >= maxExports) {
+            return next(new AppError(429, `Daily export limit reached (${maxExports}/${maxExports}). Please try again tomorrow.`));
+        }
+
+        const newCount = await redis.incr(key);
+        if (newCount === 1) {
+            await redis.expire(key, 86400); // Set expiry to 24 hours
+        }
+
+        await exportQueue.add('export-database', { projectId, userId, email });
+
+        return res.status(202).json({
+            message: `Database export request received. You will receive an email with a download link shortly. Usage today: ${newCount}/${maxExports}.`,
+        });
+
+    } catch (err) {
+        console.error("[Dashboard API] Error handling export request for project - ", req.params.projectId, ": ", err);
+        return next(new AppError(500, err.message || "Failed to initiate database export."));
+    }
+};

apps/dashboard-api/src/routes/projects.js

@@ -50,6 +50,7 @@ const {
 
 const { createAdminUser, resetPassword, getUserDetails, updateAdminUser, listUserSessions, revokeUserSession } = require('../controllers/userAuth.controller');
 
+const exportController = require('../controllers/dbExport.controller');
 
 // POST REQ FOR CREATE PROJECT
 router.post('/', authMiddleware, verifyEmail, planEnforcement.checkProjectLimit, createProject);
@@ -152,4 +153,7 @@ router.put('/:projectId/admin/users/:userId', authMiddleware, loadProjectForAdmi
 router.get('/:projectId/admin/users/:userId/sessions', authMiddleware, loadProjectForAdmin, checkAuthEnabled, listUserSessions);
 router.delete('/:projectId/admin/users/:userId/sessions/:tokenId', authMiddleware, loadProjectForAdmin, checkAuthEnabled, revokeUserSession);
 
+// POST req for DB EXPORT
+router.post('/:projectId/export', authMiddleware, exportController.dbExportHandler);
+
 module.exports = router;