apps/backend/src/modules/fiscal/workers/fiscal-domain-event-router.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { RealFiscalService } from '../application/real-fiscal.service';
@Injectable()
export class FiscalDomainEventRouterWorker {
constructor(
private readonly database:DatabaseService,
private readonly fiscal:RealFiscalService,
) {}
async runBatch(limit=100) {
const events=await this.database.query<any>(
`select *
from outbox_events
where event_type in ('PaymentCaptured','RefundCompleted','ReservationCancelled','OrderCancelled')
order by occurred_at
limit $1`,
[limit],
);
let routed=0;
let skipped=0;
for(const event of events) {
const locationId=event.payload?.locationId ?? await this.locationForEvent(event);
if(!locationId) { skipped++; continue; }
const connections=await this.database.query<any>(
`select ic.id,ic.status,ic.fiscal_outage_mode,
fdm.id as mapping_id
from integration_connections ic
join fiscal_device_mappings fdm on fdm.integration_connection_id=ic.id
where ic.organization_id=$1
and ic.category='fiscal'
and ic.status in ('connected','degraded','active')
and fdm.location_id=$2
and fdm.effective_until is null
and fdm.status<>'disabled'
order by case when fdm.cash_register_id is null then 1 else 0 end,fdm.created_at
limit 1`,
[event.organization_id,locationId],
);
if(!connections[0]) { skipped++; continue; }
const connection=connections[0];
const claimed=await this.database.transaction(async(tx)=>{
const rows=await tx.query<any>(
`insert into fiscal_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=fiscal_event_consumption.attempt_count+1,
fiscal_transaction_id=null,
last_error=null,
updated_at=now()
where fiscal_event_consumption.status='failed'
or (fiscal_event_consumption.status='processing'
and fiscal_event_consumption.updated_at < now()-interval '15 minutes')
returning outbox_event_id`,
[event.id,connection.id],
);
return Boolean(rows[0]);
});
if(!claimed) continue;
const mapping=this.mapEvent(event);
if(!mapping) {
await this.mark(event.id,connection.id,null,'skipped');
skipped++;
continue;
}
try {
const result=await this.fiscal.start({
organizationId:event.organization_id,
locationId,
checkId:mapping.checkId,
paymentId:mapping.paymentId,
transactionType:mapping.transactionType,
mappingId:connection.mapping_id,
clientTransactionRef:`event:${event.id}`,
});
await this.mark(event.id,connection.id,result.id,'processed');
routed++;
} catch {
await this.mark(event.id,connection.id,null,'failed');
}
}
return {routed,skipped};
}
private mapEvent(event:any) {
if(event.event_type==='PaymentCaptured') {
return {
transactionType:'sale' as const,
checkId:event.payload?.checkId,
paymentId:event.payload?.paymentId,
};
}
if(event.event_type==='RefundCompleted') {
return {
transactionType:'refund' as const,
checkId:event.payload?.checkId ?? null,
paymentId:event.payload?.paymentId ?? null,
};
}
if(['ReservationCancelled','OrderCancelled'].includes(event.event_type)) {
// Only route cancellation when a fiscal check reference is explicitly present.
if(!event.payload?.checkId) return null;
return {
transactionType:'cancellation' as const,
checkId:event.payload.checkId,
paymentId:null,
};
}
return null;
}
private async locationForEvent(event:any) {
const checkId=event.payload?.checkId;
if(checkId) {
const rows=await this.database.query<any>(
`select location_id from checks
where id=$1 and organization_id=$2`,
[checkId,event.organization_id],
);
return rows[0]?.location_id??null;
}
const paymentId=event.payload?.paymentId;
if(paymentId) {
const rows=await this.database.query<any>(
`select location_id from payments
where id=$1 and organization_id=$2`,
[paymentId,event.organization_id],
);
return rows[0]?.location_id??null;
}
return null;
}
private async mark(eventId:string,connectionId:string,fiscalTransactionId:string|null,status:string) {
await this.database.query(
`update fiscal_event_consumption
set fiscal_transaction_id=$3,status=$4,
last_error=case when $4='failed' then 'fiscal routing failed' else null end,
updated_at=now()
where outbox_event_id=$1 and integration_connection_id=$2`,
[eventId,connectionId,fiscalTransactionId,status],
);
}
}