apps/backend/src/modules/imports/application/import.service.tsTypeScript
import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { DomainConflictError } from '../../../common/errors/domain-conflict.error';
import { AuditService } from '../../audit/audit.service';
import {
assertImportConfirmationInput,
assertImportCreateInput,
assertImportJobInput,
} from './import-input-validation';
@Injectable()
export class ImportService {
constructor(
private readonly database:DatabaseService,
private readonly audit:AuditService,
) {}
async create(input:unknown) {
assertImportCreateInput(input);
const asset=await this.database.query<any>(
`select id,mime_type,status,purpose,location_id
from file_assets
where id=$1 and organization_id=$2
and (location_id is null or location_id=$3)`,
[input.fileAssetId,input.organizationId,input.locationId],
);
if(!asset[0]||asset[0].status!=='available'||asset[0].purpose!=='import_source') {
throw new DomainConflictError('imports.file_invalid');
}
const id=randomUUID();
await this.database.query(
`insert into import_jobs(
id,organization_id,location_id,import_type,file_asset_id,
mapping_json,options_json,created_by_user_id
) values($1,$2,$3,$4,$5,$6::jsonb,$7::jsonb,$8)`,
[
id,input.organizationId,input.locationId,input.importType,input.fileAssetId,
JSON.stringify(input.mapping??{}),JSON.stringify(input.options??{}),input.userId??null,
],
);
return {id,status:'uploaded'};
}
async get(input:unknown) {
assertImportJobInput(input);
const rows=await this.database.query<any>(
`select *
from import_jobs
where id=$1 and organization_id=$2 and location_id=$3`,
[input.id,input.organizationId,input.locationId],
);
if(!rows[0]) throw new DomainConflictError('imports.not_found');
return rows[0];
}
async preview(input:unknown) {
assertImportJobInput(input);
const job=await this.get(input);
const errors=await this.database.query<any>(
`select ire.row_number,ire.field_name,ire.error_code,ire.message,
ire.raw_value_preview,ire.severity
from import_row_errors ire
join import_jobs ij on ij.id=ire.import_job_id
where ire.import_job_id=$1
and ij.organization_id=$2 and ij.location_id=$3
order by ire.row_number,ire.id`,
[input.id,input.organizationId,input.locationId],
);
const rows=await this.database.query<any>(
`select ipr.row_number,ipr.normalized_json,ipr.valid
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
order by ipr.row_number
limit 200`,
[input.id,input.organizationId,input.locationId],
);
return {job,rows,errors};
}
async cancel(input:unknown) {
assertImportJobInput(input);
const rows=await this.database.query<any>(
`update import_jobs
set status='cancelled',completed_at=now(),version=version+1
where id=$1 and organization_id=$2 and location_id=$3
and status in ('uploaded','validating','preview_ready')
returning id,status`,
[input.id,input.organizationId,input.locationId],
);
if(!rows[0]) throw new DomainConflictError('imports.cancel_invalid');
return rows[0];
}
async markConfirmed(input:unknown) {
assertImportConfirmationInput(input);
const rows=await this.database.query<any>(
`update import_jobs
set status='importing',started_at=coalesce(started_at,now()),version=version+1
where id=$1 and organization_id=$2 and location_id=$3 and status='preview_ready'
returning id,status`,
[input.id,input.organizationId,input.locationId],
);
if(!rows[0]) throw new DomainConflictError('imports.confirm_invalid');
// Audit is intentionally best effort after business state is committed.
await this.audit.record({
organizationId:input.organizationId,
actorUserId:input.userId,
eventType:'ImportConfirmed',
entityType:'ImportJob',
entityId:input.id,
metadata:{locationId:input.locationId},
}).catch(()=>undefined);
return rows[0];
}
}