apps/backend/src/modules/external-integrations/workers/external-domain-event-router.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { DatabaseService } from '../../../infrastructure/database/database.service';
@Injectable()
export class ExternalDomainEventRouterWorker {
constructor(private readonly database:DatabaseService) {}
async runBatch(limit=100) {
const events=await this.database.query<any>(
`select * from outbox_events
where event_type in ('PurchaseOrderSubmitted','PurchaseOrderApproved')
order by occurred_at limit $1`,
[limit],
);
let routed=0;
for(const event of events) {
const purchaseOrderId=event.payload?.purchaseOrderId??event.aggregate_id;
if(!purchaseOrderId) continue;
const po=await this.database.query<any>(
`select id,organization_id,location_id,supplier_id,status
from purchase_orders where id=$1 and organization_id=$2`,
[purchaseOrderId,event.organization_id],
);
if(!po[0]) continue;
const conns=await this.database.query<any>(
`select id,provider
from integration_connections
where organization_id=$1 and category='supplier'
and status in ('active','connected','degraded')
and sync_direction in ('outbound_only','bidirectional')
and (location_id is null or location_id=$2)`,
[event.organization_id,po[0].location_id],
);
for(const c of conns) {
const claimed=await this.database.transaction(async(tx)=>{
const rows=await tx.query<any>(
`insert into external_outbound_event_consumption(
outbox_event_id,integration_connection_id,status,attempt_count,updated_at
) values($1,$2,'processing',1,now())
on conflict(outbox_event_id,integration_connection_id)
do update set
status='processing',
attempt_count=external_outbound_event_consumption.attempt_count+1,
last_error=null,
updated_at=now()
where external_outbound_event_consumption.status='failed'
or (external_outbound_event_consumption.status='processing'
and external_outbound_event_consumption.updated_at < now()-interval '15 minutes')
returning outbox_event_id`,
[event.id,c.id],
);
return Boolean(rows[0]);
});
if(!claimed) continue;
const supplierMap=await this.database.query<any>(
`select external_id
from external_system_mappings
where integration_connection_id=$1 and local_entity_type='supplier'
and local_entity_id=$2 and status='active'`,
[c.id,po[0].supplier_id],
);
if(!supplierMap[0]) {
await this.mark(event.id,c.id,'skipped');
continue;
}
await this.database.query(
`insert into external_integration_outbox(
id,organization_id,integration_connection_id,provider,local_entity_type,
local_entity_id,command_type,idempotency_key,payload_json
) values(gen_random_uuid(),$1,$2,$3,'purchase_order',$4,'push_purchase_order',$5,$6::jsonb)
on conflict(integration_connection_id,idempotency_key) do nothing`,
[
event.organization_id,c.id,c.provider,po[0].id,
`outbox:${event.id}:${c.id}:push_purchase_order`,
JSON.stringify({purchaseOrderId:po[0].id,externalSupplierId:supplierMap[0].external_id}),
],
);
await this.mark(event.id,c.id,'processed');
routed++;
}
}
return {routed};
}
private async mark(eventId:string,connectionId:string,status:'processed'|'skipped'|'failed',error?:unknown) {
await this.database.query(
`update external_outbound_event_consumption
set status=$3,last_error=$4,updated_at=now()
where outbox_event_id=$1 and integration_connection_id=$2`,
[eventId,connectionId,status,error?String(error).slice(0,1000):null],
);
}
}