apps/backend/src/modules/external-integrations/application/supplier-catalog.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';
@Injectable()
export class SupplierCatalogService {
constructor(private readonly database:DatabaseService) {}
async preview(input:{
organizationId:string;locationId?:string;connectionId:string;supplierId:string;
externalCatalogId?:string;items:Array<any>;sourceFileAssetId?:string;
}) {
const connections=await this.database.query<any>(
`select id from integration_connections
where id=$1 and organization_id=$2 and category='supplier'
and (location_id is null or location_id=$3)`,
[input.connectionId,input.organizationId,input.locationId??null],
);
if(!connections[0]) throw new DomainConflictError('supplier_catalog.connection_not_found');
const id=randomUUID();
let rejected=0;
const normalized:any[]=[];
for(const item of input.items) {
if(!item.externalId || !item.name || !item.purchaseUnitCode || !item.normalizedUnitCode ||
item.unitsPerPurchaseUnit==null || Number(item.unitsPerPurchaseUnit)<=0 ||
item.priceMinor==null || !item.currencyCode) {
rejected++; continue;
}
normalized.push({
externalId:String(item.externalId),
name:String(item.name),
purchaseUnitCode:String(item.purchaseUnitCode),
unitsPerPurchaseUnit:String(item.unitsPerPurchaseUnit),
normalizedUnitCode:String(item.normalizedUnitCode),
priceMinor:Number(item.priceMinor),
currencyCode:String(item.currencyCode),
sku:item.sku?String(item.sku):null,
});
}
await this.database.query(
`insert into supplier_catalog_imports(
id,organization_id,integration_connection_id,supplier_id,external_catalog_id,status,
rejected_items,source_file_asset_id,payload_json
) values($1,$2,$3,$4,$5,'preview',$6,$7,$8::jsonb)`,
[
id,input.organizationId,input.connectionId,input.supplierId,input.externalCatalogId??null,
rejected,input.sourceFileAssetId??null,JSON.stringify({items:normalized}),
],
);
return {id,status:'preview',validItems:normalized.length,rejectedItems:rejected};
}
async commit(input:{organizationId:string;locationId:string;importId:string}) {
return this.database.transaction(async(tx)=>{
const rows=await tx.query<any>(
`select i.* from supplier_catalog_imports i
join integration_connections c on c.id=i.integration_connection_id
where i.id=$1 and i.organization_id=$2 and i.status='preview'
and c.category='supplier'
and (c.location_id is null or c.location_id=$3)
for update`,
[input.importId,input.organizationId,input.locationId],
);
const job=rows[0];
if(!job) throw new DomainConflictError('supplier_catalog.import_invalid');
const items=job.payload_json?.items??[];
let changed=0;
for(const item of items) {
const mapping=await tx.query<any>(
`select local_entity_id from external_system_mappings
where integration_connection_id=$1 and external_entity_type='supplier_product'
and external_id=$2 and status='active'
and (location_id is null or location_id=$3)`,
[job.integration_connection_id,item.externalId,input.locationId],
);
if(!mapping[0]?.local_entity_id) continue;
await tx.query(
`insert into supplier_catalog_items(
id,organization_id,supplier_id,stock_item_id,supplier_sku,supplier_product_name,
purchase_unit_code,units_per_purchase_unit,normalized_unit_code,unit_price_minor,
currency_code,is_active,last_price_at
) values(gen_random_uuid(),$1,$2,$3,$4,$5,$6,$7::numeric,$8,$9,$10,true,now())
on conflict(supplier_id,stock_item_id,purchase_unit_code)
do update set supplier_sku=excluded.supplier_sku,
supplier_product_name=excluded.supplier_product_name,
units_per_purchase_unit=excluded.units_per_purchase_unit,
normalized_unit_code=excluded.normalized_unit_code,
unit_price_minor=excluded.unit_price_minor,
currency_code=excluded.currency_code,
is_active=true,last_price_at=now(),updated_at=now(),
version=supplier_catalog_items.version+1`,
[
job.organization_id,job.supplier_id,mapping[0].local_entity_id,item.sku,item.name,
item.purchaseUnitCode,item.unitsPerPurchaseUnit,item.normalizedUnitCode,
item.priceMinor,item.currencyCode,
],
);
changed++;
}
await tx.query(
`update supplier_catalog_imports
set status='completed',imported_items=$1,changed_items=$1,completed_at=now()
where id=$2`,
[changed,job.id],
);
return {id:job.id,status:'completed',changedItems:changed};
});
}
}