apps/backend/src/modules/imports/workers/import-preview.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { createHash,randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { OutboxService } from '../../../infrastructure/outbox/outbox.service';
import { ImportFileReaderService } from './import-file-reader.service';
import { assertImportJobInput } from '../application/import-input-validation';
@Injectable()
export class ImportPreviewWorker {
constructor(
private readonly database:DatabaseService,
private readonly outbox:OutboxService,
private readonly reader:ImportFileReaderService,
) {}
async validate(input:unknown) {
assertImportJobInput(input);
const importJobId=input.importJobId;
const jobs=await this.database.query<any>(
`select ij.*,fa.storage_key,fa.mime_type
from import_jobs ij
join file_assets fa on fa.id=ij.file_asset_id
where ij.id=$1 and ij.organization_id=$2 and ij.location_id=$3
and ij.status='uploaded'`,
[importJobId,input.organizationId,input.locationId],
);
const job=jobs[0];
if(!job) return {processed:false};
const claimed=await this.database.query<any>(
`update import_jobs
set status='validating',started_at=now(),version=version+1
where id=$1 and organization_id=$2 and location_id=$3 and status='uploaded'
returning id`,
[importJobId,input.organizationId,input.locationId],
);
if(!claimed[0]) return {processed:false};
try {
const records=await this.reader.read({
storageKey:job.storage_key,
mimeType:job.mime_type,
});
if(records.length>50_000) throw new Error('imports.row_limit_exceeded');
await this.database.query(`delete from import_row_errors where import_job_id=$1`,[importJobId]);
await this.database.query(`delete from import_preview_rows where import_job_id=$1`,[importJobId]);
let valid=0,invalid=0;
for(let i=0;i<records.length;i++) {
const row=records[i];
const normalized=this.normalize(job.import_type,row,job.mapping_json??{});
const errors=this.validateRow(job.import_type,normalized);
const key=createHash('sha256').update(JSON.stringify(normalized)).digest('hex');
await this.database.query(
`insert into import_preview_rows(
id,import_job_id,row_number,normalized_json,valid,deterministic_row_key
) values($1,$2,$3,$4::jsonb,$5,$6)`,
[randomUUID(),importJobId,i+2,JSON.stringify(normalized),errors.length===0,key],
);
if(errors.length===0) valid++;
else {
invalid++;
for(const error of errors) {
await this.database.query(
`insert into import_row_errors(
id,import_job_id,row_number,field_name,error_code,message,raw_value_preview,severity
) values($1,$2,$3,$4,$5,$6,$7,'error')`,
[
randomUUID(),importJobId,i+2,error.field,error.code,error.message,
error.rawValue==null?null:String(error.rawValue).slice(0,120),
],
);
}
}
}
const summary={rowsTotal:records.length,valid,errors:invalid};
await this.database.query(
`update import_jobs set status='preview_ready',summary_json=$1::jsonb,version=version+1
where id=$2 and organization_id=$3 and location_id=$4`,
[JSON.stringify(summary),importJobId,input.organizationId,input.locationId],
);
await this.outbox.append({
organizationId:job.organization_id,
aggregateType:'ImportJob',
aggregateId:importJobId,
eventType:'ImportPreviewReady',
payload:{importJobId,...summary},
});
return {processed:true,...summary};
} catch(error:any) {
await this.database.query(
`update import_jobs set status='failed',completed_at=now(),
summary_json=$1::jsonb,version=version+1 where id=$2 and organization_id=$3 and location_id=$4`,
[JSON.stringify({error:String(error?.message??error)}),importJobId,input.organizationId,input.locationId],
);
await this.outbox.append({
organizationId:job.organization_id,
aggregateType:'ImportJob',
aggregateId:importJobId,
eventType:'ImportFailed',
payload:{importJobId,error:String(error?.message??error)},
});
return {processed:true,failed:true};
}
}
private normalize(type:string,row:Record<string,string>,mapping:Record<string,string>) {
const out:Record<string,string>={};
for(const [source,value] of Object.entries(row)) {
const target=mapping[source] ?? source;
out[target]=value;
}
return out;
}
private validateRow(type:string,row:Record<string,string>) {
const errors:Array<{field?:string;code:string;message:string;rawValue?:unknown}>=[];
const required:Record<string,string[]>={
products:['name'],
ingredients:['name','base_unit_code'],
employees:['first_name','last_name'],
guests:['display_name'],
suppliers:['name'],
supplier_products:['supplier_id','stock_item_id','supplier_product_name'],
inventory_opening_balance:['stock_item_id','storage_location_id','quantity','unit_code'],
recipes_basic:['name'],
};
for(const field of required[type]??[]) {
if(!row[field]?.trim()) {
errors.push({field,code:'required',message:`Missing required field: ${field}`});
}
}
// Allergen truth is never inferred from free text.
if(Object.keys(row).some((x)=>x.toLowerCase().includes('allergen')) &&
!Object.keys(row).some((x)=>x==='allergen_codes')) {
errors.push({
field:'allergen_codes',
code:'allergen_structured_required',
message:'Allergens require explicit structured codes; free text is not authoritative.',
});
}
return errors;
}
}