apps/backend/src/modules/fiscal/workers/real-fiscal-recovery.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { RealFiscalService } from '../application/real-fiscal.service';
@Injectable()
export class RealFiscalRecoveryWorker {
constructor(
private readonly database:DatabaseService,
private readonly fiscal:RealFiscalService,
) {}
async runBatch(limit=25) {
const cases=await this.database.query<any>(
`with claim as (
select id
from fiscal_recovery_cases
where status in ('open','retrying') and next_attempt_at<=now()
order by case severity when 'critical' then 0 else 1 end,created_at
limit $1
for update skip locked
)
update fiscal_recovery_cases c
set status='retrying',attempt_count=c.attempt_count+1,updated_at=now()
from claim
where c.id=claim.id
returning c.*`,
[limit],
);
let recovered=0;
for(const c of cases) {
try {
const result=await this.fiscal.status({
organizationId:c.organization_id,
locationId:c.location_id,
fiscalTransactionId:c.fiscal_transaction_id,
});
if(result.state==='completed') {
// The finish path is idempotent and applies the normalized provider result.
await this.database.query(
`update fiscal_transactions
set status='started',provider_status=$1,provider_revision=coalesce($2,provider_revision)
where id=$3 and organization_id=$4 and location_id=$5 and status='recovery_required'`,
[result.providerStatus,result.revision??null,c.fiscal_transaction_id,c.organization_id,c.location_id],
);
const finish=await this.fiscal.finish({
organizationId:c.organization_id,
locationId:c.location_id,
fiscalTransactionId:c.fiscal_transaction_id,
});
if(finish.status==='completed') {
await this.database.query(
`update fiscal_recovery_cases
set status='resolved',resolved_at=now(),updated_at=now()
where id=$1 and organization_id=$2 and location_id=$3`,
[c.id,c.organization_id,c.location_id],
);
recovered++;
continue;
}
}
await this.database.query(
`update fiscal_recovery_cases
set status=case when attempt_count>=10 then 'provider_support' else 'open' end,
next_attempt_at=now()+make_interval(secs=>least(3600,30*power(2,attempt_count)::int)),
last_error=$1,updated_at=now()
where id=$2 and organization_id=$3 and location_id=$4`,
[`provider_state:${result.providerStatus}`,c.id,c.organization_id,c.location_id],
);
} catch(error:any) {
await this.database.query(
`update fiscal_recovery_cases
set status=case when attempt_count>=10 then 'provider_support' else 'open' end,
next_attempt_at=now()+make_interval(secs=>least(3600,30*power(2,attempt_count)::int)),
last_error=$1,updated_at=now()
where id=$2 and organization_id=$3 and location_id=$4`,
[String(error?.message??error).slice(0,1000),c.id,c.organization_id,c.location_id],
);
}
}
return {processed:cases.length,recovered};
}
}