apps/backend/src/modules/imports/application/import-execution.service.tsTypeScript
import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import {
DatabaseService,
type DatabaseQueryClient,
} from '../../../infrastructure/database/database.service';
import { StockMovementService } from '../../inventory/application/stock-movement.service';
import { DomainConflictError } from '../../../common/errors/domain-conflict.error';
import { OutboxService } from '../../../infrastructure/outbox/outbox.service';
import { assertImportJobInput } from './import-input-validation';
@Injectable()
export class ImportExecutionService {
constructor(
private readonly database:DatabaseService,
private readonly stockMovements:StockMovementService,
private readonly outbox:OutboxService,
) {}
async execute(input:unknown) {
assertImportJobInput(input);
const jobs=await this.database.query<any>(
`select *
from import_jobs
where id=$1 and organization_id=$2 and location_id=$3 and status='importing'`,
[input.importJobId,input.organizationId,input.locationId],
);
const job=jobs[0];
if(!job) return {processed:false};
const rows=await this.database.query<any>(
`select ipr.*
from import_preview_rows ipr
join import_jobs ij on ij.id=ipr.import_job_id
where ipr.import_job_id=$1
and ij.organization_id=$2 and ij.location_id=$3
and ipr.valid=true
order by ipr.row_number`,
[input.importJobId,input.organizationId,input.locationId],
);
let applied=0;
let skipped=0;
for(const row of rows) {
const outcome=await this.database.transaction(async(tx)=>{
// Serialize each deterministic row key. The unique constraint is a second guard.
await tx.query(
`select pg_advisory_xact_lock(hashtext($1))`,
[`import-row:${input.importJobId}:${row.deterministic_row_key}`],
);
const currentJob=await tx.query<any>(
`select id,status,organization_id,location_id,import_type
from import_jobs
where id=$1 and organization_id=$2 and location_id=$3
for share`,
[input.importJobId,input.organizationId,input.locationId],
);
if(!currentJob[0]||currentJob[0].status!=='importing') {
throw new DomainConflictError('imports.execution_invalid_state');
}
const replay=await tx.query<any>(
`select id
from import_applied_rows
where import_job_id=$1 and deterministic_row_key=$2`,
[input.importJobId,row.deterministic_row_key],
);
if(replay[0]) return {skipped:true};
const target=await this.applyRow(job,row.normalized_json,tx);
await tx.query(
`insert into import_applied_rows(
id,import_job_id,deterministic_row_key,target_entity_type,target_entity_id
) values($1,$2,$3,$4,$5)`,
[
randomUUID(),input.importJobId,row.deterministic_row_key,
target.type,target.id??null,
],
);
return {skipped:false};
},{lockTimeoutMs:3000,statementTimeoutMs:15000});
if(outcome.skipped) skipped++;
else applied++;
}
await this.database.transaction(async(tx)=>{
const state=await tx.query<{status:string}>(
`select status from import_jobs
where id=$1 and organization_id=$2 and location_id=$3
for update`,
[input.importJobId,input.organizationId,input.locationId],
);
if(!state[0]||state[0].status!=='importing') {
throw new DomainConflictError('imports.execution_invalid_state');
}
await tx.query(
`update import_jobs
set status='completed',completed_at=now(),
summary_json=coalesce(summary_json,'{}'::jsonb)||$1::jsonb,
version=version+1
where id=$2 and organization_id=$3 and location_id=$4`,
[
JSON.stringify({applied,skipped}),
input.importJobId,input.organizationId,input.locationId,
],
);
await this.outbox.append({
organizationId:input.organizationId,
aggregateType:'ImportJob',
aggregateId:input.importJobId,
eventType:'ImportCompleted',
payload:{
importJobId:input.importJobId,locationId:input.locationId,applied,skipped,
},
},tx);
});
return {processed:true,applied,skipped};
}
private async applyRow(
job:any,row:any,tx:DatabaseQueryClient,
):Promise<{type:string;id?:string}> {
switch(job.import_type) {
case 'products': {
const id=randomUUID();
await tx.query(
`insert into products(id,organization_id,name,status)
values($1,$2,$3,'active')`,
[id,job.organization_id,row.name],
);
return {type:'Product',id};
}
case 'ingredients': {
const id=randomUUID();
await tx.query(
`insert into ingredients(id,organization_id,name,base_unit_code,status)
values($1,$2,$3,$4,'active')`,
[id,job.organization_id,row.name,row.base_unit_code],
);
return {type:'Ingredient',id};
}
case 'employees': {
const id=randomUUID();
await tx.query(
`insert into employees(id,organization_id,first_name,last_name,status)
values($1,$2,$3,$4,'active')`,
[id,job.organization_id,row.first_name,row.last_name],
);
return {type:'Employee',id};
}
case 'guests': {
const id=randomUUID();
await tx.query(
`insert into guests(id,organization_id,display_name,email,phone)
values($1,$2,$3,$4,$5)`,
[id,job.organization_id,row.display_name,row.email??null,row.phone??null],
);
return {type:'Guest',id};
}
case 'suppliers': {
const id=randomUUID();
await tx.query(
`insert into suppliers(id,organization_id,name,status,email,phone)
values($1,$2,$3,'active',$4,$5)`,
[id,job.organization_id,row.name,row.email??null,row.phone??null],
);
return {type:'Supplier',id};
}
case 'inventory_opening_balance': {
if(!job.location_id) throw new DomainConflictError('imports.location_required');
const result=await this.stockMovements.append({
organizationId:job.organization_id,
locationId:job.location_id,
stockItemId:row.stock_item_id,
storageLocationId:row.storage_location_id,
movementType:'manual_adjustment',
quantityDelta:row.quantity,
unitCode:row.unit_code,
normalizedQuantityDelta:row.normalized_quantity??row.quantity,
normalizedUnitCode:row.normalized_unit_code??row.unit_code,
sourceType:'import_applied_row',
sourceId:row.id??undefined,
businessDate:row.business_date,
reason:'opening_balance_import',
} as any,tx);
return {type:'StockMovement',id:result.id};
}
default:
throw new Error(`imports.execution_not_implemented:${job.import_type}`);
}
}
}