1+ const { Worker } = require ( 'bullmq' ) ;
2+ const { PassThrough } = require ( 'stream' ) ;
3+ const fs = require ( 'fs' ) ;
4+ const os = require ( 'os' ) ;
5+ const path = require ( 'path' ) ;
6+ const { GetObjectCommand } = require ( '@aws-sdk/client-s3' ) ;
7+ const { getSignedUrl } = require ( '@aws-sdk/s3-request-presigner' ) ;
8+
9+ const {
10+ redis,
11+ exportQueue,
12+ emailQueue,
13+ Project,
14+ getConnection,
15+ getCompiledModel,
16+ getS3CompatibleStorage,
17+ getStorage,
18+ decrypt,
19+ getBucket
20+ } = require ( '@urbackend/common' ) ;
21+
22+ const initExportWorker = ( ) => {
23+ const worker = new Worker ( exportQueue . name , async ( job ) => {
24+ const { projectId, collectionName, userId, email } = job . data ;
25+ console . log ( `[ExportWorker] Starting export for collection ${ collectionName } ${ projectId } ${ email } ) ;
26+
27+ const project = await Project . findById ( projectId ) ;
28+ if ( ! project ) throw new Error ( 'Project not found' ) ;
29+
30+ const col = project . collections . find ( c => c . name === collectionName ) ;
31+ if ( ! col ) throw new Error ( `Collection ${ collectionName } ) ;
32+
33+ const connection = await getConnection ( projectId ) ;
34+ const bucket = getBucket ( project ) ;
35+ const storagePath = `${ projectId } ${ collectionName } ${ Date . now ( ) } ;
36+
37+ let provider = 'supabase' ;
38+ if ( project . resources ?. storage ?. isExternal ) {
39+ try {
40+ const decrypted = decrypt ( project . resources . storage . config ) ;
41+ const config = JSON . parse ( decrypted ) ;
42+ provider = config . storageProvider || 'supabase' ;
43+ } catch ( err ) {
44+ console . error ( "[ExportWorker] Error decrypting storage config:" , err ) ;
45+ }
46+ }
47+
48+ const client = await getStorage ( project ) ;
49+
50+ console . log ( `[ExportWorker] Preparing upload to storage (Provider: ${ provider } ) ;
51+
52+ if ( provider === 'supabase' ) {
53+ const tempFilePath = path . join ( os . tmpdir ( ) , `export_${ projectId } ${ collectionName } ${ Date . now ( ) } ) ;
54+ const writeStream = fs . createWriteStream ( tempFilePath ) ;
55+
56+ try {
57+ writeStream . write ( '{\n' ) ;
58+ const Model = getCompiledModel ( connection , col , projectId , project . resources . db . isExternal ) ;
59+
60+ writeStream . write ( ` "${ col . name } ) ;
61+
62+ const cursor = Model . find ( ) . lean ( ) . cursor ( ) ;
63+ let first = true ;
64+
65+ for await ( const doc of cursor ) {
66+ if ( ! first ) writeStream . write ( ',\n' ) ;
67+ writeStream . write ( ` ${ JSON . stringify ( doc ) } ) ;
68+ first = false ;
69+ }
70+
71+ writeStream . write ( '\n ]\n' ) ;
72+ writeStream . write ( '}\n' ) ;
73+ writeStream . end ( ) ;
74+
75+ await new Promise ( ( resolve , reject ) => {
76+ writeStream . on ( 'finish' , resolve ) ;
77+ writeStream . on ( 'error' , reject ) ;
78+ } ) ;
79+
80+ console . log ( `[ExportWorker] Temp file created, uploading...` ) ;
81+ const fileBuffer = fs . readFileSync ( tempFilePath ) ;
82+
83+ const { error } = await client . storage . from ( bucket ) . upload ( storagePath , fileBuffer , {
84+ contentType : 'application/json'
85+ } ) ;
86+
87+ if ( error ) throw error ;
88+ } finally {
89+ if ( fs . existsSync ( tempFilePath ) ) {
90+ fs . unlinkSync ( tempFilePath ) ;
91+ }
92+ }
93+
94+ } else if ( provider === 's3' || provider === 'cloudflare_r2' ) {
95+ const passThrough = new PassThrough ( ) ;
96+
97+ // Start the upload promise in parallel using the getStorage client
98+ const uploadPromise = client . storage . from ( bucket ) . upload ( storagePath , passThrough , {
99+ contentType : 'application/json'
100+ } ) ;
101+
102+ try {
103+ passThrough . write ( '{\n' ) ;
104+
105+ const Model = getCompiledModel ( connection , col , projectId , project . resources . db . isExternal ) ;
106+
107+ passThrough . write ( ` "${ col . name } ) ;
108+
109+ const cursor = Model . find ( ) . lean ( ) . cursor ( ) ;
110+ let first = true ;
111+
112+ for await ( const doc of cursor ) {
113+ if ( ! first ) passThrough . write ( ',\n' ) ;
114+ passThrough . write ( ` ${ JSON . stringify ( doc ) } ) ;
115+ first = false ;
116+ }
117+
118+ passThrough . write ( '\n ]\n' ) ;
119+
120+ passThrough . write ( '}\n' ) ;
121+ passThrough . end ( ) ;
122+
123+ console . log ( `[ExportWorker] Database stream ended. Awaiting final storage upload...` ) ;
124+ const { error } = await uploadPromise ;
125+ if ( error ) throw error ;
126+ } catch ( error ) {
127+ passThrough . destroy ( error ) ;
128+ throw error ;
129+ }
130+ } else {
131+ throw new Error ( `Unknown storage provider: ${ provider } ) ;
132+ }
133+
134+ let downloadUrl ;
135+ if ( provider === 'supabase' ) {
136+ const { data, error } = await client . storage . from ( bucket ) . createSignedUrl ( storagePath , 86400 ) ;
137+ if ( error ) throw error ;
138+ downloadUrl = data ?. signedUrl ;
139+ } else {
140+ const { s3Client } = await getS3CompatibleStorage ( project ) ;
141+ const command = new GetObjectCommand ( { Bucket : bucket , Key : storagePath } ) ;
142+ downloadUrl = await getSignedUrl ( s3Client , command , { expiresIn : 86400 } ) ;
143+ }
144+
145+ // queue the email to be sent to the user
146+ await emailQueue . add ( 'send-export-email' , { email, downloadUrl, projectName : project . name } ) ;
147+ console . log ( `[ExportWorker] Export completed! Email queued for ${ email } ) ;
148+ } , { connection : redis , concurrency : 2 } ) ;
149+
150+ worker . on ( 'completed' , ( job ) => {
151+ console . log ( `[ExportWorker] Job ${ job . id } ${ job . data . projectId } ) ;
152+ } ) ;
153+
154+ worker . on ( 'failed' , ( job , err ) => {
155+ console . error ( `[ExportWorker] Job ${ job ?. id } ${ job ?. data ?. projectId } , err . message ) ;
156+ } ) ;
157+
158+ return worker ;
159+ } ;
160+
161+ module . exports = { initExportWorker } ;
0 commit comments