apps/backend/src/modules/exports/workers/export.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { PutObjectCommand,S3Client } from '@aws-sdk/client-s3';
import ExcelJS from 'exceljs';
import { randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { OutboxService } from '../../../infrastructure/outbox/outbox.service';
import { ExportAuthorizationService } from '../application/export-authorization.service';
@Injectable()
export class ExportWorker {
private readonly bucket=process.env.OBJECT_STORAGE_BUCKET ?? 'gastro-os';
private readonly s3=new S3Client({
region:process.env.OBJECT_STORAGE_REGION ?? 'local',
endpoint:process.env.OBJECT_STORAGE_ENDPOINT,
forcePathStyle:true,
credentials:{
accessKeyId:process.env.OBJECT_STORAGE_ACCESS_KEY ?? 'minio',
secretAccessKey:process.env.OBJECT_STORAGE_SECRET_KEY ?? 'change-me',
},
});
constructor(
private readonly database:DatabaseService,
private readonly outbox:OutboxService,
private readonly authorization:ExportAuthorizationService,
) {}
async runOne(exportJobId:string) {
const jobs=await this.database.query<any>(
`select * from export_jobs where id=$1 and status='queued'`,
[exportJobId],
);
const job=jobs[0];
if(!job) return {processed:false};
await this.authorization.assertGenerationAllowed(job);
const claimed=await this.database.query<any>(
`update export_jobs
set status='processing',started_at=now()
where id=$1 and status='queued'
returning id`,
[exportJobId],
);
if(!claimed[0]) return {processed:false};
try {
const rows=await this.dataset(job);
const generated=job.format==='xlsx'
? await this.xlsx(rows)
: this.csv(rows);
if(job.format==='pdf') throw new Error('exports.pdf_curated_layout_required');
const mime=job.format==='xlsx'
? 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
: 'text/csv';
const ext=job.format==='xlsx'?'xlsx':'csv';
const assetId=randomUUID();
const key=`org/${job.organization_id}/exports/${randomUUID()}.${ext}`;
await this.s3.send(new PutObjectCommand({
Bucket:this.bucket,Key:key,Body:generated,ContentType:mime,
}));
await this.database.query(
`insert into file_assets(
id,organization_id,location_id,purpose,storage_key,original_file_name,
mime_type,size_bytes,status,created_by_user_id
) values($1,$2,$3,'export_result',$4,$5,$6,$7,'available',$8)`,
[
assetId,job.organization_id,job.location_id,key,
`${job.export_type}-${exportJobId}.${ext}`,mime,
Buffer.byteLength(generated),job.requested_by_user_id,
],
);
await this.database.query(
`update export_jobs
set status='completed',file_asset_id=$1,completed_at=now(),
expires_at=now()+interval '7 days'
where id=$2`,
[assetId,exportJobId],
);
await this.outbox.append({
organizationId:job.organization_id,
aggregateType:'ExportJob',
aggregateId:exportJobId,
eventType:'ExportCompleted',
payload:{exportJobId,fileAssetId:assetId,requestedByUserId:job.requested_by_user_id},
});
return {processed:true,fileAssetId:assetId};
} catch(error:any) {
await this.database.query(
`update export_jobs set status='failed',completed_at=now(),error=$1 where id=$2`,
[String(error?.message??error),exportJobId],
);
await this.outbox.append({
organizationId:job.organization_id,
aggregateType:'ExportJob',
aggregateId:exportJobId,
eventType:'ExportFailed',
payload:{exportJobId,error:String(error?.message??error)},
});
return {processed:true,failed:true};
}
}
private async dataset(job:any):Promise<Record<string,unknown>[]> {
const loc=job.location_id;
switch(job.export_type) {
case 'inventory':
return this.database.query<any>(
`select si.name,sl.name as storage_location,
sbp.quantity_normalized::text as quantity, sbp.normalized_unit_code as unit
from stock_balance_projection sbp
join stock_items si on si.id=sbp.stock_item_id
join storage_locations sl on sl.id=sbp.storage_location_id
where sbp.organization_id=$1 and ($2::uuid is null or sbp.location_id=$2)
order by si.name,sl.name`,
[job.organization_id,loc],
);
case 'waste':
return this.database.query<any>(
`select wr.recorded_at,si.name,wr.normalized_quantity::text as quantity,
wr.normalized_unit_code as unit,wr.reason
from waste_records wr
join stock_items si on si.id=wr.stock_item_id
where wr.organization_id=$1 and ($2::uuid is null or wr.location_id=$2)
order by wr.recorded_at desc limit 100000`,
[job.organization_id,loc],
);
case 'suppliers':
default:
if(job.export_type==='purchasing') {
return this.database.query<any>(
`select gr.received_at,s.name as supplier,gr.receipt_number,
grl.received_quantity::text,grl.purchase_unit_code,
grl.unit_price_minor_snapshot,grl.currency_code
from goods_receipts gr
join suppliers s on s.id=gr.supplier_id
join goods_receipt_lines grl on grl.goods_receipt_id=gr.id
where gr.organization_id=$1 and ($2::uuid is null or gr.location_id=$2)
order by gr.received_at desc limit 100000`,
[job.organization_id,loc],
);
}
if(job.export_type==='incidents') {
return this.database.query<any>(
`select incident_type,severity,status,title,occurred_at,reported_at,resolution
from incidents
where organization_id=$1 and ($2::uuid is null or location_id=$2)
order by reported_at desc limit 100000`,
[job.organization_id,loc],
);
}
if(job.export_type==='safety') {
return this.database.query<any>(
`select sc.business_date,sc.status,sc.result,sc.performed_at,sct.name
from safety_checks sc
join safety_check_templates sct on sct.id=sc.safety_check_template_id
where sc.organization_id=$1 and ($2::uuid is null or sc.location_id=$2)
order by sc.business_date desc limit 100000`,
[job.organization_id,loc],
);
}
if(job.export_type==='employees') {
return this.database.query<any>(
`select id,first_name,last_name,status,created_at
from employees where organization_id=$1 order by last_name,first_name`,
[job.organization_id],
);
}
if(job.export_type==='guests') {
return this.database.query<any>(
`select id,display_name,email,phone,created_at
from guests where organization_id=$1 order by created_at desc limit 100000`,
[job.organization_id],
);
}
if(job.export_type==='audit') {
return this.database.query<any>(
`select occurred_at,event_type,entity_type,entity_id,actor_user_id,metadata
from audit_events where organization_id=$1
order by occurred_at desc limit 100000`,
[job.organization_id],
);
}
return this.database.query<any>(
`select business_date,gross_sales_minor,refunds_minor,order_count,guest_count,
theoretical_food_cost_percent,gross_margin_percent
from reporting_daily_location_kpis
where organization_id=$1 and ($2::uuid is null or location_id=$2)
order by business_date desc limit 10000`,
[job.organization_id,loc],
);
}
}
private csv(rows:Record<string,unknown>[]) {
if(rows.length===0) return Buffer.from('', 'utf8');
const headers=Object.keys(rows[0]);
const cell=(value:unknown)=>{
let s=value==null?'':typeof value==='object'?JSON.stringify(value):String(value);
// Protect spreadsheet consumers from formula injection.
if(/^[=+\-@]/.test(s)) s=`'${s}`;
return `"${s.replace(/"/g,'""')}"`;
};
const lines=[headers.map(cell).join(',')];
for(const row of rows) lines.push(headers.map((h)=>cell(row[h])).join(','));
return Buffer.from(lines.join('\r\n'),'utf8');
}
private async xlsx(rows:Record<string,unknown>[]) {
const workbook=new ExcelJS.Workbook();
const sheet=workbook.addWorksheet('Export');
if(rows.length) {
const headers=Object.keys(rows[0]);
sheet.addRow(headers);
for(const row of rows) {
sheet.addRow(headers.map((h)=>{
const v=row[h];
if(v==null) return '';
if(typeof v==='object') return JSON.stringify(v);
let s=String(v);
if(/^[=+\-@]/.test(s)) s=`'${s}`;
return s;
}));
}
}
const result=await workbook.xlsx.writeBuffer();
return Buffer.from(result);
}
}