apps/backend/src/modules/external-integrations/workers/external-outbound.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { ExternalProviderRegistry } from '../providers/external-provider.registry';
import { ExternalIntegrationService } from '../application/external-integration.service';
import { ExternalConflictService } from '../application/external-conflict.service';
@Injectable()
export class ExternalOutboundWorker {
constructor(
private readonly database:DatabaseService,
private readonly providers:ExternalProviderRegistry,
private readonly integrations:ExternalIntegrationService,
private readonly conflicts:ExternalConflictService,
) {}
async runBatch(limit=50) {
const jobs=await this.database.query<any>(
`with claim as (
select o.id,c.credential_reference,c.config_json,c.location_id
from external_integration_outbox o
join integration_connections c on c.id=o.integration_connection_id
where o.status in ('queued','failed')
and coalesce(o.next_attempt_at,now())<=now()
and o.attempt_count<10
order by o.created_at
limit $1
for update of o skip locked
)
update external_integration_outbox o
set status='processing',attempt_count=o.attempt_count+1
from claim
where o.id=claim.id
returning o.*,claim.credential_reference,claim.config_json,claim.location_id`,
[limit],
);
let completed=0;
for(const job of jobs) {
try {
const provider=this.providers.get(job.provider);
if(job.command_type==='push_purchase_order') {
if(!provider.pushPurchaseOrder || !provider.capabilities.supportsPurchaseOrders) {
await this.database.query(
`update external_integration_outbox
set status='unsupported',completed_at=now()
where id=$1 and organization_id=$2 and integration_connection_id=$3`,
[job.id,job.organization_id,job.integration_connection_id],
);
continue;
}
const po=await this.database.query<any>(
`select po.*,jsonb_agg(jsonb_build_object(
'id',poi.id,'stockItemId',poi.stock_item_id,'quantity',poi.ordered_quantity,
'unitCode',poi.purchase_unit_code,'normalizedUnitCode',poi.normalized_unit_code,
'unitsPerPurchaseUnit',poi.units_per_purchase_unit_snapshot,
'unitPriceMinor',poi.unit_price_minor_snapshot
) order by poi.created_at) as items
from purchase_orders po
left join purchase_order_lines poi on poi.purchase_order_id=po.id
where po.id=$1 and po.organization_id=$2
and ($3::uuid is null or po.location_id=$3)
group by po.id`,
[job.local_entity_id,job.organization_id,job.location_id??null],
);
if(!po[0]) throw new Error('purchase_order.not_found');
const connection=await this.integrations.load(job.organization_id,job.integration_connection_id,job.location_id);
const result=await provider.pushPurchaseOrder(
this.integrations.context(connection),
{
externalSupplierId:job.payload_json.externalSupplierId,
purchaseOrder:po[0],
idempotencyKey:job.idempotency_key,
},
);
await this.database.query(
`update external_integration_outbox
set status=$1,provider_reference=$2,provider_status=$3,
completed_at=case when $1='completed' then now() else null end
where id=$4 and organization_id=$5 and integration_connection_id=$6`,
[
result.status==='completed'?'completed':result.status==='uncertain'?'uncertain':'processing',
result.externalId,result.status,job.id,job.organization_id,job.integration_connection_id,
],
);
if(result.status==='completed') completed++;
continue;
}
await this.database.query(
`update external_integration_outbox
set status='unsupported',completed_at=now()
where id=$1 and organization_id=$2 and integration_connection_id=$3`,
[job.id,job.organization_id,job.integration_connection_id],
);
} catch(error:any) {
const attempts=Number(job.attempt_count);
const dead=attempts>=10;
const retry=Math.min(3600,30*Math.pow(2,Math.max(0,attempts-1)));
await this.database.query(
`update external_integration_outbox
set status=$1,last_error=$2,
next_attempt_at=case when $1='failed' then now()+make_interval(secs=>$3) else null end
where id=$4 and organization_id=$5 and integration_connection_id=$6`,
[dead?'dead_letter':'failed',String(error?.message??error).slice(0,1000),retry,job.id,job.organization_id,job.integration_connection_id],
);
if(dead) {
await this.conflicts.create({
organizationId:job.organization_id,locationId:job.location_id,
connectionId:job.integration_connection_id,conflictType:'provider_rejected',
severity:'high',localEntityType:job.local_entity_type,localEntityId:job.local_entity_id,
summary:'External provider outbound command reached dead letter.',
localSnapshot:{commandType:job.command_type,idempotencyKey:job.idempotency_key},
});
}
}
}
return {processed:jobs.length,completed};
}
}