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,161 @@
+const { Worker } = require('bullmq');
+const { PassThrough } = require('stream');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { GetObjectCommand } = require('@aws-sdk/client-s3');
+const { getSignedUrl } = require('@aws-sdk/s3-request-presigner');
+
+const { 
+    redis, 
+    exportQueue, 
+    emailQueue,
+    Project, 
+    getConnection, 
+    getCompiledModel,
+    getS3CompatibleStorage,
+    getStorage,
+    decrypt,
+    getBucket
+} = require('@urbackend/common');
+
+const initExportWorker = () => {
+    const worker = new Worker(exportQueue.name, async (job) => {
+        const { projectId, collectionName, userId, email } = job.data;
+        console.log(`[ExportWorker] Starting export for collection ${collectionName} in project ${projectId} requested by ${email}`);
+
+        const project = await Project.findById(projectId);
+        if (!project) throw new Error('Project not found');
+
+        const col = project.collections.find(c => c.name === collectionName);
+        if (!col) throw new Error(`Collection ${collectionName} not found`);
+
+        const connection = await getConnection(projectId);
+        const bucket = getBucket(project);
+        const storagePath = `${projectId}/exports/${collectionName}_export_${Date.now()}.json`;
+
+        let provider = 'supabase';
+        if (project.resources?.storage?.isExternal) {
+            try {
+                const decrypted = decrypt(project.resources.storage.config);
+                const config = JSON.parse(decrypted);
+                provider = config.storageProvider || 'supabase';
+            } catch (err) {
+                console.error("[ExportWorker] Error decrypting storage config:", err);
+            }
+        }
+
+        const client = await getStorage(project);
+
+        console.log(`[ExportWorker] Preparing upload to storage (Provider: ${provider})...`);
+
+        if (provider === 'supabase') {
+            const tempFilePath = path.join(os.tmpdir(), `export_${projectId}_${collectionName}_${Date.now()}.json`);
+            const writeStream = fs.createWriteStream(tempFilePath);
+
+            try {
+                writeStream.write('{\n');
+                const Model = getCompiledModel(connection, col, projectId, project.resources.db.isExternal);
+
+                writeStream.write(`  "${col.name}": [\n`);
+
+                const cursor = Model.find().lean().cursor();
+                let first = true;
+
+                for await (const doc of cursor) {
+                    if (!first) writeStream.write(',\n');
+                    writeStream.write(`    ${JSON.stringify(doc)}`);
+                    first = false;
+                }
+
+                writeStream.write('\n  ]\n');
+                writeStream.write('}\n');
+                writeStream.end();
+
+                await new Promise((resolve, reject) => {
+                    writeStream.on('finish', resolve);
+                    writeStream.on('error', reject);
+                });
+
+                console.log(`[ExportWorker] Temp file created, uploading...`);
+                const fileBuffer = fs.readFileSync(tempFilePath);
+
+                const { error } = await client.storage.from(bucket).upload(storagePath, fileBuffer, {
+                    contentType: 'application/json'
+                });
+
+                if (error) throw error;
+            } finally {
+                if (fs.existsSync(tempFilePath)) {
+                    fs.unlinkSync(tempFilePath);
+                }
+            }
+
+        } else if (provider === 's3' || provider === 'cloudflare_r2') {
+            const passThrough = new PassThrough();
+
+            // Start the upload promise in parallel using the getStorage client
+            const uploadPromise = client.storage.from(bucket).upload(storagePath, passThrough, {
+                contentType: 'application/json'
+            });
+
+            try {
+                passThrough.write('{\n');
+
+                const Model = getCompiledModel(connection, col, projectId, project.resources.db.isExternal);
+
+                passThrough.write(`  "${col.name}": [\n`);
+
+                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  ]\n');
+
+                passThrough.write('}\n');
+                passThrough.end();
+
+                console.log(`[ExportWorker] Database stream ended. Awaiting final storage upload...`);
+                const { error } = await uploadPromise;
+                if (error) throw error;
+            } catch (error) {
+                passThrough.destroy(error);
+                throw error;
+            }
+        } else {
+            throw new Error(`Unknown storage provider: ${provider}`);
+        }
+
+        let downloadUrl;
+        if (provider === 'supabase') {
+            const { data, error } = await client.storage.from(bucket).createSignedUrl(storagePath, 86400);
+            if (error) throw error;
+            downloadUrl = data?.signedUrl;
+        } else {
+            const { s3Client } = await getS3CompatibleStorage(project);
+            const command = new GetObjectCommand({ Bucket: bucket, Key: storagePath });
+            downloadUrl = await getSignedUrl(s3Client, command, { expiresIn: 86400 });
+        }
+
+        // queue the email to be sent to the user
+        await emailQueue.add('send-export-email', { email, downloadUrl, projectName: project.name });
+        console.log(`[ExportWorker] Export completed! Email queued for ${email}`);
+    }, { 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,66 @@
+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, collectionName } = 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."));
+        }
+
+        if (!project.collections.some(c => c.name === collectionName)) {
+            return next(new AppError(404, "Collection not found in 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 collection ${collectionName} in 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, collectionName, userId, email });
+
+        return res.status(202).json({
+            success: true,
+            data: {},
+            message: `Collection 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/collections/:collectionName/export', authMiddleware, exportController.dbExportHandler);
+
 module.exports = router;